DataMiddleware/SourceCode/DataMiddleware/tools/stores/sqlserver_store.py

from typing import List, Any, Sequence

from tools.utils.db_helper import sqlserver_helper
import tools.utils as utils
from sqlalchemy.orm import Session
from tools.models.dinge_model import DingEKu
from sqlalchemy import select, Row
from collections import defaultdict
class SqlServerStore:
    def __init__(self):
        self._db_helper = sqlserver_helper.SQLServerHelper()
        self._logger = utils.get_logger()

    def group_by_book_no(self) -> List[str]:
        try:
            with Session(self._db_helper.get_engine()) as session:
                stmt = select(DingEKu.shuhao).group_by(DingEKu.shuhao).order_by(DingEKu.shuhao.asc())
                result = session.execute(stmt)
                return [row[0] for row in result]
        except Exception as e:
            self._logger.error(f"查询定额库失败: {str(e)}")
            raise




    def query_ding_e_ku(self, shuhao: str = None) -> List[dict]:
        """
        查询定额库表
        参数：
        - shuhao: 书号
        """
        try:
            with Session(self._db_helper.get_engine()) as session:
                # 构建查询条件
                conditions = []
                if shuhao:
                   conditions.append(DingEKu.shuhao == shuhao)

                # 分页查询
                stmt = select(DingEKu.shuhao,
                    DingEKu.dingebianhao,
                    DingEKu.dingemingcheng,
                    DingEKu.danwei,
                    DingEKu.gongzuoneirong,
                    DingEKu.jibendinge,
                    DingEKu.jijia,
                    DingEKu.gongfei,
                    DingEKu.liaofei,
                    DingEKu.jifei,
                    DingEKu.danzhong,
                    DingEKu.jiehao,
                    DingEKu.liushuihao,
                    DingEKu.lock,
                    DingEKu.shiyong,
                    DingEKu.tiaoxuhao,
                    DingEKu.daorushijian).where(*conditions).order_by(DingEKu.shuhao.asc())

                result = session.execute(stmt)
                # 动态生成字段映射（排除消耗字段）
                return [{
                    'shuhao': row.shuhao,
                    'dingebianhao': row.dingebianhao,
                    'dingemingcheng': row.dingemingcheng,
                    'danwei': row.danwei,
                    'gongzuoneirong': row.gongzuoneirong,
                    'jibendinge': row.jibendinge,
                    'jijia': row.jijia,
                    'gongfei': row.gongfei,
                    'liaofei': row.liaofei,
                    'jifei': row.jifei,
                    'danzhong': row.danzhong,
                    'jiehao': row.jiehao,
                    'liushuihao': row.liushuihao,
                    'lock': row.lock,
                    'shiyong': row.shiyong,
                    'tiaoxuhao': row.tiaoxuhao,
                    'daorushijian': row.daorushijian
                } for row in result]

        except Exception as e:
            self._logger.error(f"查询定额库失败: {str(e)}")
            raise