66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
"""数据库访问层。
|
||
|
||
提供 run_query：读取 .sql 文件、剥离注释、参数化绑定执行、返回 list[dict]。
|
||
所有报表查询必须走参数绑定，禁止字符串拼接 SQL（防注入）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from sqlalchemy import create_engine, text
|
||
from sqlalchemy.engine import Engine
|
||
|
||
from core.settings import settings
|
||
|
||
_engine: Engine | None = None
|
||
|
||
|
||
def get_engine() -> Engine:
|
||
"""惰性创建并缓存 SQLAlchemy engine。"""
|
||
global _engine
|
||
if _engine is None:
|
||
_engine = create_engine(
|
||
settings.database_url,
|
||
pool_pre_ping=True,
|
||
pool_recycle=1800,
|
||
)
|
||
return _engine
|
||
|
||
|
||
def load_sql(sql_path: str | Path) -> str:
|
||
"""读取 .sql 文件并剥离行级 `--` 注释，返回纯 SQL 文本。
|
||
|
||
支持 SQL 文件里自由书写中文注释（便于阅读与 AI 理解）。
|
||
仅剥离以 `--` 开头的整行注释（行首可有空白），不处理行内注释，
|
||
避免误删含 `--` 的字符串字面量。
|
||
"""
|
||
path = Path(sql_path)
|
||
if not path.is_absolute():
|
||
from core.settings import PROJECT_ROOT
|
||
|
||
path = PROJECT_ROOT / path
|
||
raw = path.read_text(encoding="utf-8")
|
||
lines = []
|
||
for line in raw.splitlines():
|
||
# 去掉行首空白后判断是否为注释行
|
||
if re.match(r"\s*--", line):
|
||
continue
|
||
lines.append(line)
|
||
return "\n".join(lines).strip()
|
||
|
||
|
||
def run_query(sql_path: str | Path, params: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""执行参数化查询，返回 list[dict]（每行一个 dict，键为列别名）。
|
||
|
||
:param sql_path: .sql 文件路径（相对项目根或绝对路径）。
|
||
:param params: 绑定参数，键名对应 SQL 中的 :name 占位符。
|
||
"""
|
||
sql_text = load_sql(sql_path)
|
||
engine = get_engine()
|
||
with engine.connect() as conn:
|
||
result = conn.execute(text(sql_text), params)
|
||
cols = list(result.keys())
|
||
return [dict(zip(cols, row)) for row in result.fetchall()]
|