payment-platform/backend/app/plugin/module_payment/apikey/crud.py

from collections.abc import Sequence
from datetime import datetime
from typing import Any

from app.api.v1.module_system.auth.schema import AuthSchema
from app.plugin.module_payment.apikey.model import TenantApiKeyModel, TenantApiLogModel
from app.plugin.module_payment.apikey.schema import (
    TenantApiKeyCreate,
    TenantApiKeyUpdate,
    TenantApiLogCreate,
)
from app.core.base_crud import CRUDBase


class TenantApiKeyCRUD(CRUDBase[TenantApiKeyModel, TenantApiKeyCreate, TenantApiKeyUpdate]):
    """租户API Key数据层"""

    def __init__(self, auth: AuthSchema) -> None:
        self.auth = auth
        super().__init__(model=TenantApiKeyModel, auth=auth)

    async def get_by_id_crud(
        self, id: int, preload: list[str | Any] | None = None
    ) -> TenantApiKeyModel | None:
        return await self.get(id=id, preload=preload)

    async def get_by_api_key(self, api_key: str) -> TenantApiKeyModel | None:
        return await self.get(
            preload=[],
            api_key=api_key,
            status="0",
            # expired_at=("None", None),
        )

    async def create_crud(
        self,
        api_key: str,
        api_secret: str,
        tenant_id: int,
        expired_at: datetime,
        description: str | None = None,
    ) -> TenantApiKeyModel:
        data = {
            "tenant_id": tenant_id,
            "description": description,
            "api_key": api_key,
            "api_secret": api_secret,
            "expired_at": expired_at,
        }
        obj = await self.create(data=data, skip_tenant_id=True)
        return obj

    async def update_status_crud(self, api_key_id: int, status: str) -> TenantApiKeyModel | None:
        return await self.update(id=api_key_id, data={"status": status})

    async def update_last_used_crud(self, api_key_id: int) -> None:
        api_key_obj = await self.get_by_id_crud(api_key_id)
        if api_key_obj:
            api_key_obj.last_used_at = datetime.now()
            await self.auth.db.flush()

    async def delete_crud(self, api_key_id: int) -> None:
        await self.delete(ids=[api_key_id])

class TenantApiLogCRUD(CRUDBase[TenantApiLogModel, TenantApiLogCreate, TenantApiLogCreate]):
    """租户API调用日志数据层"""

    def __init__(self, auth: AuthSchema) -> None:
        self.auth = auth
        super().__init__(model=TenantApiLogModel, auth=auth)

    async def create_crud(
        self,
        api_key_id: int | None,
        tenant_id: int,
        endpoint: str,
        method: str,
        request_ip: str,
        request_data: str | None,
        response_code: int,
        response_time: float,
    ) -> TenantApiLogModel:
        data = TenantApiLogCreate(
            api_key_id=api_key_id,
            tenant_id=tenant_id,
            endpoint=endpoint,
            method=method,
            request_ip=request_ip,
            request_data=request_data,
            response_code=response_code,
            response_time=response_time,
        )
        return await self.create(data=data, skip_tenant_id=True)