mirror of
https://github.com/Cccc-owo/CheckInApp.git
synced 2026-06-17 14:06:28 +00:00
57 lines
1.7 KiB
Plaintext
57 lines
1.7 KiB
Plaintext
from sqlalchemy import create_engine
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
from datetime import datetime, timezone
|
|
from backend.config import settings
|
|
import sqlite3
|
|
|
|
# SQLite 类型转换器：将从数据库读取的字符串转换为 timezone-aware datetime
|
|
def convert_timestamp(val):
|
|
"""将从数据库读取的字符串转换为 timezone-aware datetime (UTC)"""
|
|
if val is None:
|
|
return None
|
|
# 解析 ISO 8601 格式的字符串
|
|
try:
|
|
dt = datetime.fromisoformat(val.decode() if isinstance(val, bytes) else val)
|
|
# 如果是 naive datetime，添加 UTC timezone
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt
|
|
except (ValueError, AttributeError):
|
|
return None
|
|
|
|
# 注册 SQLite 类型转换器（全局）
|
|
sqlite3.register_converter("DATETIME", convert_timestamp)
|
|
sqlite3.register_converter("TIMESTAMP", convert_timestamp)
|
|
|
|
# 创建数据库引擎
|
|
# 为 SQLite 连接添加 detect_types 参数以启用类型转换
|
|
engine = create_engine(
|
|
settings.DATABASE_URL,
|
|
connect_args={
|
|
"check_same_thread": False, # SQLite 特定配置
|
|
"detect_types": sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES # 启用类型转换
|
|
},
|
|
echo=False, # 生产环境设为 False
|
|
)
|
|
|
|
# 创建会话工厂
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
# 创建基类
|
|
Base = declarative_base()
|
|
|
|
|
|
def get_db():
|
|
"""依赖注入：获取数据库会话"""
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def init_db():
|
|
"""初始化数据库：创建所有表"""
|
|
Base.metadata.create_all(bind=engine)
|