#!/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()