47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
"""
|
|
SQLAlchemy 引擎：使用环境变量 ``database_url`` / ``DATABASE_URL``（与项目根目录 ``.env`` 一致）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.engine import Engine
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_engine: Engine | None = None
|
|
|
|
|
|
def _database_url_from_env() -> str:
|
|
for key in ("database_url", "DATABASE_URL"):
|
|
v = os.getenv(key)
|
|
if v is not None and str(v).strip():
|
|
return str(v).strip()
|
|
return ""
|
|
|
|
|
|
def get_engine(*, url: Optional[str] = None, reset: bool = False) -> Engine:
|
|
"""
|
|
返回默认业务库引擎（单例）。未配置 ``database_url`` 时抛出 ``ValueError``。
|
|
|
|
:param url: 若传入，则忽略单例并为此 URL 新建引擎（便于测试）。
|
|
:param reset: 为 True 时丢弃已缓存的单例，下次再按环境变量创建。
|
|
"""
|
|
global _engine
|
|
if reset:
|
|
_engine = None
|
|
if url is not None:
|
|
return create_engine(url, pool_pre_ping=True)
|
|
if _engine is not None:
|
|
return _engine
|
|
u = _database_url_from_env()
|
|
if not u:
|
|
raise ValueError("database_url 未配置")
|
|
_engine = create_engine(u, pool_pre_ping=True)
|
|
logger.info("SQLAlchemy engine initialized from database_url")
|
|
return _engine
|