33 lines
1.0 KiB
Python
33 lines
1.0 KiB
Python
|
"""
|
||
|
|
API 依赖项
|
||
|
|
"""
|
||
|
|
from typing import AsyncGenerator
|
||
|
|
from tortoise.contrib.fastapi import HTTPNotFoundError
|
||
|
|
from fastapi import Depends, HTTPException, status
|
||
|
|
from tortoise.exceptions import DBConnectionError, DoesNotExist
|
||
|
|
from tortoise import Tortoise
|
||
|
|
|
||
|
|
async def get_db() -> AsyncGenerator[None, None]:
|
||
|
|
"""获取数据库连接依赖"""
|
||
|
|
try:
|
||
|
|
# 检查数据库连接
|
||
|
|
conn = Tortoise.get_connection("default")
|
||
|
|
# 执行简单查询验证连接
|
||
|
|
await conn.execute_query("SELECT 1")
|
||
|
|
yield
|
||
|
|
except DBConnectionError as e:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
||
|
|
detail="数据库连接失败"
|
||
|
|
)
|
||
|
|
except Exception as e:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||
|
|
detail=f"数据库错误: {str(e)}"
|
||
|
|
)
|
||
|
|
|
||
|
|
def get_current_user():
|
||
|
|
"""获取当前用户（示例）"""
|
||
|
|
# 这里可以实现 JWT 验证等逻辑
|
||
|
|
pass
|