your-workflow/backend/app/core/db.py

25 lines
786 B
Python

from tortoise.contrib.fastapi import Tortoise
async def init_tortoise(db_type: str, db_config: dict):
if db_type == "mysql":
db_url = (
f"mysql://{db_config['db_user']}:{db_config['db_password']}"
f"@{db_config['db_host']}:{db_config['db_port']}/{db_config['db_name']}"
)
elif db_type == "sqlite":
db_url = f"sqlite://{db_config['db_name']}"
else:
raise ValueError("Unsupported db_type")
try:
await Tortoise.init(
db_url=db_url,
modules={"models": ["models.db"]}, # 这里写你模型所在路径
)
await Tortoise.generate_schemas()
return True
except Exception as e:
raise e
async def close_tortoise():
await Tortoise.close_connections()