LQAgentPlatform/foundation/database/base/sql/bus_data_query.py

from typing import List, Tuple, Any, Optional, Dict
from foundation.observability.logger.loggering import server_logger
from foundation.utils.common import handler_err
from foundation.database.base.sql.async_mysql_base_dao import AsyncBaseDAO


class BasisOfPreparationDAO(AsyncBaseDAO):
    """异步编制依据 对象"""
    
    
    async def get_info_by_id(self, id: int) -> Optional[Dict]:
        """根据ID获取编制依据"""
        query = "SELECT * FROM t_basis_of_preparation WHERE id = %s"
        return await self.fetch_one(query, (id,))
    
    async def get_list(self) -> List[Dict]:
        """获取所有编制依据"""
        query = "SELECT * FROM t_basis_of_preparation WHERE status = 'current' ORDER BY created_at DESC"
        return await self.fetch_all(query)
    

    async def get_info_by_condition(self, conditions: Dict) -> List[Dict]:
        """根据条件查询编制依据"""
        if not conditions:
            return await self.get_list()
        
        try:
            where_clause = " AND ".join([f"{field} = %s" for field in conditions.keys()])
            where_values = list(conditions.values())
            
            query = f"SELECT * FROM t_basis_of_preparation WHERE {where_clause} AND status = 'current' ORDER BY created_at DESC"
            return await self.fetch_all(query, tuple(where_values))
            
        except Exception as err:
            handler_err(logger=server_logger, err=err, err_name="条件查询失败")
            raise