| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051 |
- from typing import Any
- from app.api.v1.module_system.auth.schema import AuthSchema
- from app.core.base_crud import CRUDBase
- from .model import IssueBatchModel, QuotaChangeLogModel, QuotaModel
- from .schema import QuotaCreateSchema, QuotaUpdateSchema
- class QuotaCRUD(CRUDBase[QuotaModel, QuotaCreateSchema, QuotaUpdateSchema]):
- """额度 CRUD 操作"""
- def __init__(self, auth: AuthSchema) -> None:
- self.auth = auth
- super().__init__(model=QuotaModel, auth=auth)
- async def get_by_out_biz_no(
- self, out_biz_no: str
- ) -> QuotaModel | None:
- return await self.get(out_biz_no=out_biz_no)
- async def get_by_employee_id(
- self, employee_id: str
- ) -> QuotaModel | None:
- return await self.get(employee_id=employee_id)
- class IssueBatchCRUD(CRUDBase[IssueBatchModel, dict[str, Any], dict[str, Any]]):
- """手工发放批次 CRUD 操作"""
- def __init__(self, auth: AuthSchema) -> None:
- self.auth = auth
- super().__init__(model=IssueBatchModel, auth=auth)
- async def get_by_batch_no(
- self, batch_no: str
- ) -> IssueBatchModel | None:
- return await self.get(batch_no=batch_no)
- async def get_by_issue_batch_id(
- self, issue_batch_id: str
- ) -> IssueBatchModel | None:
- return await self.get(issue_batch_id=issue_batch_id)
- class QuotaChangeLogCRUD(CRUDBase[QuotaChangeLogModel, dict[str, Any], dict[str, Any]]):
- """额度变更记录 CRUD"""
- def __init__(self, auth: AuthSchema) -> None:
- self.auth = auth
- super().__init__(model=QuotaChangeLogModel, auth=auth)
|
