tools-api/app/admin/crud/crud_org.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from typing import Sequence

from model.org import Org
from app.admin.schema.org import CreateOrgParam, UpdateOrgParam
from sqlalchemy import Select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy_crud_plus import CRUDPlus


class CRUDOrg(CRUDPlus[Org]):
    async def get(self, db: AsyncSession, pk: int) -> Org | None:
        """
        获取IntentOrg

        :param db:
        :param pk:
        :return:
        """
        return await self.select_model(db, pk)

    async def get_by_token(self, db: AsyncSession, token: str) -> Org | None:
        """
        获取IntentOrg

        :param db:
        :param token:
        :return:
        """
        return await self.select_model_by_column(db, api_key=token)

    async def get_list(self) -> Select:
        """
        获取IntentOrg列表

        :return:
        """
        return await self.select_order('created_at', 'desc')

    async def get_all(self, db: AsyncSession) -> Sequence[Org]:
        """
        获取所有IntentOrg

        :param db:
        :return:
        """
        return await self.select_models(db)

    async def create(self, db: AsyncSession, obj_in: CreateOrgParam) -> None:
        """
        创建IntentOrg

        :param db:
        :param obj_in:
        :return:
        """
        await self.create_model(db, obj_in)

    async def update(self, db: AsyncSession, pk: int, obj_in: UpdateOrgParam) -> int:
        """
        更新IntentOrg

        :param db:
        :param pk:
        :param obj_in:
        :return:
        """
        return await self.update_model(db, pk, obj_in)

    async def delete(self, db: AsyncSession, pk: list[int]) -> int:
        """
        删除IntentOrg

        :param db:
        :param pk:
        :return:
        """
        return  await self.delete_model_by_column(db, allow_multiple=True, id__in=pk)


org_dao: CRUDOrg = CRUDOrg(Org)