tools-api/app/call_center/service/intent_records_service.py

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

from app.call_center.crud.crud_intent_records import intent_records_dao
from model.intent_records import IntentRecords
from app.call_center.schema.intent_records import CreateIntentRecordsParam, UpdateIntentRecordsParam, \
    GetIntentRecordsByIdParam
from common.exception import errors
from database.db_mysql import async_db_session
from sqlalchemy import Select


class IntentRecordsService:
    @staticmethod
    async def get(*, pk: str) -> IntentRecords:
        async with async_db_session() as db:
            intent_records = await intent_records_dao.get(db, pk)
            if not intent_records:
                raise errors.NotFoundError(msg='record不存在')
            return intent_records

    @staticmethod
    async def get_select_by_id(*, obj: GetIntentRecordsByIdParam) -> IntentRecords | None:
        async with async_db_session.begin() as db:
            return await intent_records_dao.get(db, obj)

    @staticmethod
    async def get_select() -> Select:
        return await intent_records_dao.get_list()

    @staticmethod
    async def get_all() -> Sequence[IntentRecords]:
        async with async_db_session() as db:
            intent_records = await intent_records_dao.get_all(db)
            return intent_records

    @staticmethod
    async def create(*, obj: CreateIntentRecordsParam) -> None:
        async with async_db_session.begin() as db:
            await intent_records_dao.create(db, obj)

    @staticmethod
    async def update(*, pk: int, obj: UpdateIntentRecordsParam) -> int:
        async with async_db_session.begin() as db:
            count = await intent_records_dao.update(db, pk, obj)
            return count

    @staticmethod
    async def update_manual_intent(*, obj: UpdateIntentRecordsParam) -> int:
        async with async_db_session.begin() as db:
            count = await intent_records_dao.update_manual_intent(db, obj)
            return count

    @staticmethod
    async def delete(*, pk: list[int]) -> int:
        async with async_db_session.begin() as db:
            count = await intent_records_dao.delete(db, pk)
            return count


intent_records_service: IntentRecordsService = IntentRecordsService()