61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
"""
|
||
数据库连接与初始化模块。
|
||
|
||
职责：
|
||
- 建立 SQLite + SQLAlchemy 连接（引擎）
|
||
- 提供会话工厂 SessionLocal 与 FastAPI 依赖 get_db
|
||
- 提供数据库初始化函数 init_db()
|
||
"""
|
||
|
||
import os
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import declarative_base, sessionmaker
|
||
|
||
# 定位 backend 目录与项目根目录（与 backend 同级）
|
||
BACKEND_DIR = Path(__file__).resolve().parent
|
||
PROJECT_ROOT = BACKEND_DIR.parent
|
||
|
||
# 读取项目根目录下的 .env 配置文件（若存在）
|
||
load_dotenv(PROJECT_ROOT / ".env")
|
||
|
||
# 数据库地址：默认使用 backend 目录下的 SQLite 文件，可通过 .env 的 DATABASE_URL 覆盖（留空视为使用默认值）
|
||
DATABASE_URL = os.getenv("DATABASE_URL") or f"sqlite:///{(BACKEND_DIR / 'blog.db').as_posix()}"
|
||
|
||
# 创建 SQLAlchemy 引擎
|
||
# SQLite 需关闭 check_same_thread：FastAPI 多线程处理请求时会跨线程复用会话
|
||
engine = create_engine(
|
||
DATABASE_URL,
|
||
connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {},
|
||
)
|
||
|
||
# 会话工厂：每个请求创建独立会话，避免线程安全问题
|
||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||
|
||
# 所有 ORM 模型的公共声明基类
|
||
Base = declarative_base()
|
||
|
||
|
||
def get_db():
|
||
"""FastAPI 依赖注入：提供数据库会话，请求结束后自动关闭。"""
|
||
db = SessionLocal()
|
||
try:
|
||
yield db
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def init_db() -> None:
|
||
"""初始化数据库：创建所有已注册 ORM 模型对应的数据表。"""
|
||
# 延迟导入模型，确保全部模型注册到 Base.metadata 后再建表
|
||
from . import models # noqa: F401
|
||
|
||
Base.metadata.create_all(bind=engine)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 支持直接运行：python -m backend.database
|
||
init_db()
|
||
print(f"数据库初始化完成：{DATABASE_URL}") |