diff --git a/app/db/__init__.py b/app/db/__init__.py index efc09916..43843511 100644 --- a/app/db/__init__.py +++ b/app/db/__init__.py @@ -1,12 +1,60 @@ import asyncio from typing import Any, Generator, List, Optional, Self, Tuple, AsyncGenerator, Union -from sqlalchemy import NullPool, QueuePool, and_, create_engine, inspect, text, select, delete, Column, Integer, \ +from sqlalchemy import NullPool, QueuePool, and_, create_engine, event, inspect, text, select, delete, Column, Integer, \ Sequence, Identity +from sqlalchemy.engine import Engine as SQLAlchemyEngine, ExceptionContext from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker from sqlalchemy.orm import Session, as_declarative, declared_attr, scoped_session, sessionmaker from app.core.config import settings +from app.log import logger + + +def _database_error_metadata(error: BaseException) -> Optional[dict[str, Any]]: + """提取 SQLite 与 PostgreSQL 驱动提供的稳定错误分类字段。""" + metadata = {"error_type": type(error).__name__} + + # DBAPI 驱动字段并不共享统一类型，动态读取可同时兼容 sqlite3、psycopg2 与 asyncpg。 + sqlite_errorcode = getattr(error, "sqlite_errorcode", None) + sqlite_errorname = getattr(error, "sqlite_errorname", None) + if sqlite_errorcode is not None or sqlite_errorname: + if sqlite_errorcode is not None: + metadata["error_code"] = sqlite_errorcode + if sqlite_errorname: + metadata["error_name"] = sqlite_errorname + return metadata + + sqlstate = getattr(error, "sqlstate", None) or getattr(error, "pgcode", None) + if not sqlstate: + sqlstate = getattr(getattr(error, "diag", None), "sqlstate", None) + if sqlstate: + metadata["sqlstate"] = sqlstate + return metadata + + return None + + +def _log_database_error(exception_context: ExceptionContext) -> None: + """记录非敏感驱动错误码，并保持 SQLAlchemy 原有异常传播。""" + metadata = _database_error_metadata(exception_context.original_exception) + if not metadata: + return + + dialect = exception_context.dialect + fields = { + "database": dialect.name, + "driver": dialect.driver, + **metadata, + } + logger.error( + "数据库驱动异常：" + ", ".join(f"{key}={value}" for key, value in fields.items()) + ) + + +def _register_database_error_logging(engine: SQLAlchemyEngine) -> None: + """为主程序 Engine 注册统一的底层驱动错误诊断。""" + event.listen(engine, "handle_error", _log_database_error) def get_id_column(): @@ -71,6 +119,7 @@ def _get_sqlite_engine(is_async: bool = False): # 创建数据库引擎 engine = create_engine(**_db_kwargs) + _register_database_error_logging(engine) # 设置WAL模式 _journal_mode = "WAL" if settings.DB_WAL_ENABLE else "DELETE" @@ -91,6 +140,7 @@ def _get_sqlite_engine(is_async: bool = False): } # 创建异步数据库引擎 async_engine = create_async_engine(**_db_kwargs) + _register_database_error_logging(async_engine.sync_engine) # 设置WAL模式 _journal_mode = "WAL" if settings.DB_WAL_ENABLE else "DELETE" @@ -146,6 +196,7 @@ def _get_postgresql_engine(is_async: bool = False): # 创建数据库引擎 engine = create_engine(**_db_kwargs) + _register_database_error_logging(engine) print(f"PostgreSQL database connected to {settings.DB_POSTGRESQL_TARGET}/{settings.DB_POSTGRESQL_DATABASE}") return engine @@ -163,6 +214,7 @@ def _get_postgresql_engine(is_async: bool = False): } # 创建异步数据库引擎 async_engine = create_async_engine(**_db_kwargs) + _register_database_error_logging(async_engine.sync_engine) print(f"Async PostgreSQL database connected to {settings.DB_POSTGRESQL_TARGET}/{settings.DB_POSTGRESQL_DATABASE}") return async_engine diff --git a/tests/test_db_error_diagnostics.py b/tests/test_db_error_diagnostics.py new file mode 100644 index 00000000..8614237e --- /dev/null +++ b/tests/test_db_error_diagnostics.py @@ -0,0 +1,106 @@ +import asyncio + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.exc import OperationalError + +import app.db as db_module + + +class _SqliteError(Exception): + """模拟 sqlite3 异常暴露的扩展错误字段。""" + + sqlite_errorcode = 266 + sqlite_errorname = "SQLITE_IOERR_READ" + + +class _PsycopgError(Exception): + """模拟 psycopg2 异常暴露的 SQLSTATE 字段。""" + + pgcode = "40001" + + +class _AsyncpgError(Exception): + """模拟 asyncpg 适配异常暴露的 SQLSTATE 字段。""" + + sqlstate = "23505" + + +@pytest.mark.parametrize( + ("error", "expected"), + [ + ( + _SqliteError("disk I/O error"), + { + "error_type": "_SqliteError", + "error_code": 266, + "error_name": "SQLITE_IOERR_READ", + }, + ), + ( + _PsycopgError("serialization failure"), + { + "error_type": "_PsycopgError", + "sqlstate": "40001", + }, + ), + ( + _AsyncpgError("duplicate key"), + { + "error_type": "_AsyncpgError", + "sqlstate": "23505", + }, + ), + ], +) +def test_database_error_metadata_extracts_driver_codes(error, expected) -> None: + """诊断元数据应兼容 SQLite、psycopg2 与 asyncpg 的稳定错误字段。""" + assert db_module._database_error_metadata(error) == expected + + +def test_database_error_listener_omits_statement_and_parameters(monkeypatch) -> None: + """数据库错误日志不得包含 SQL、参数或驱动返回的原始消息。""" + messages = [] + engine = create_engine("sqlite:///:memory:") + monkeypatch.setattr("app.db.logger.error", messages.append) + db_module._register_database_error_logging(engine) + + with pytest.raises(OperationalError): + with engine.connect() as connection: + connection.execute( + text("SELECT * FROM missing_table WHERE token = :token"), + {"token": "private-token"}, + ) + + assert len(messages) == 1 + assert "database=sqlite" in messages[0] + assert "driver=pysqlite" in messages[0] + assert "error_code=1" in messages[0] + assert "error_name=SQLITE_ERROR" in messages[0] + assert "missing_table" not in messages[0] + assert "private-token" not in messages[0] + + +def test_async_database_engine_logs_driver_error_metadata(monkeypatch) -> None: + """异步 Engine 应通过底层 sync engine 记录驱动错误码。""" + messages = [] + monkeypatch.setattr("app.db.logger.error", messages.append) + + async def query_missing_table() -> None: + async with db_module.AsyncEngine.connect() as connection: + await connection.execute(text("SELECT * FROM async_missing_table")) + + with pytest.raises(OperationalError): + asyncio.run(query_missing_table()) + + assert len(messages) == 1 + assert "database=sqlite" in messages[0] + assert "driver=aiosqlite" in messages[0] + assert "error_code=1" in messages[0] + assert "error_name=SQLITE_ERROR" in messages[0] + assert "async_missing_table" not in messages[0] + + +def test_database_error_metadata_ignores_unclassified_errors() -> None: + """没有驱动错误码时不应制造无效诊断日志。""" + assert db_module._database_error_metadata(RuntimeError("plain failure")) is None