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) -> bool:
        try:
            await self.delete(ids=[api_key_id])
            return True
        except Exception:
            return False

    async def get_list_crud(
        self,
        tenant_id: int | None = None,
        status: str | None = None,
        skip: int = 0,
        limit: int = 100,
    ) -> Sequence[TenantApiKeyModel]:
        search: dict[str, Any] = {}
        if tenant_id:
            search["tenant_id"] = tenant_id
        if status:
            search["status"] = status
        return await self.list(
            search=search if search else None,
            order_by=[{"created_time": "desc"}],
        )


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)

    async def get_list_crud(
        self,
        tenant_id: int | None = None,
        api_key_id: int | None = None,
        endpoint: str | None = None,
        skip: int = 0,
        limit: int = 100,
    ) -> Sequence[TenantApiLogModel]:
        search: dict[str, Any] = {}
        if tenant_id:
            search["tenant_id"] = tenant_id
        if api_key_id:
            search["api_key_id"] = api_key_id
        if endpoint:
            search["endpoint"] = ("like", endpoint)
        return await self.list(
            search=search if search else None,
            order_by=[{"created_time": "desc"}],
        )