modify: config sqlite 配置规范
This commit is contained in:
+139
-76
@@ -1,135 +1,198 @@
|
||||
"""
|
||||
配置管理模块
|
||||
配置管理模块 - 简化版本
|
||||
基于 Pydantic Settings 和 TOML 文件
|
||||
"""
|
||||
from typing import Optional, Dict, Any
|
||||
|
||||
from typing import Optional, List, Dict, Any
|
||||
from pydantic_settings import BaseSettings
|
||||
from pydantic import Field
|
||||
from pydantic import model_validator
|
||||
import toml
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import os
|
||||
|
||||
try:
|
||||
from ...utils.launch import args
|
||||
except:
|
||||
from utils.launch import args # type: ignore
|
||||
from utils.launch import args # type: ignore
|
||||
|
||||
|
||||
class SQLiteConfig(BaseSettings):
|
||||
"""SQLite 配置"""
|
||||
|
||||
engine: str = "tortoise.backends.sqlite"
|
||||
file_path: str = "db.sqlite3"
|
||||
journal_mode: str = "WAL"
|
||||
busy_timeout: int = 10000
|
||||
maxsize: int = 10
|
||||
minsize: int = 1
|
||||
cache_size: int = -2000
|
||||
synchronous: str = "NORMAL"
|
||||
temp_store: str = "MEMORY"
|
||||
mmap_size: int = 1073741824
|
||||
foreign_keys: bool = True
|
||||
auto_vacuum: str = "INCREMENTAL"
|
||||
max_cached_statements: int = 100
|
||||
readonly: bool = False
|
||||
|
||||
@property
|
||||
def tortoise_config(self) -> Dict[str, Any]:
|
||||
return {
|
||||
"default": {
|
||||
"engine": self.engine,
|
||||
"credentials": {
|
||||
"file_path": self.file_path,
|
||||
"journal_mode": self.journal_mode,
|
||||
"busy_timeout": self.busy_timeout,
|
||||
"maxsize": self.maxsize,
|
||||
"minsize": self.minsize,
|
||||
"cache_size": self.cache_size,
|
||||
"synchronous": self.synchronous,
|
||||
"temp_store": self.temp_store,
|
||||
"mmap_size": self.mmap_size,
|
||||
"foreign_keys": self.foreign_keys,
|
||||
"auto_vacuum": self.auto_vacuum,
|
||||
"max_cached_statements": self.max_cached_statements,
|
||||
"readonly": self.readonly,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
"""应用配置"""
|
||||
|
||||
|
||||
# 应用信息
|
||||
app_name: str = "FastAPI Demo"
|
||||
app_version: str = "1.0.0"
|
||||
description: str = "FastAPI 应用"
|
||||
|
||||
|
||||
# 服务器配置
|
||||
host: str = "0.0.0.0"
|
||||
port: int = 8000
|
||||
reload: bool = False
|
||||
workers: int = 4
|
||||
|
||||
# 数据库配置
|
||||
database_url: str = "sqlite://db.sqlite3"
|
||||
generate_schemas: bool = False
|
||||
|
||||
workers: int = 1 # 开发环境默认1，避免与 reload 冲突
|
||||
|
||||
database_type: str = "sqlite"
|
||||
|
||||
# SQLite 配置
|
||||
sqlite_config: SQLiteConfig = SQLiteConfig()
|
||||
|
||||
# 环境配置
|
||||
env: str = "development"
|
||||
debug: bool = False
|
||||
|
||||
|
||||
# 数据库配置
|
||||
generate_schemas: bool = False
|
||||
|
||||
# 日志配置
|
||||
logging_level: str = "INFO"
|
||||
logging_file: Optional[str] = None
|
||||
|
||||
|
||||
# CORS 配置
|
||||
cors_allow_origins: list = ["*"]
|
||||
cors_allow_origins: List[str] = ["*"]
|
||||
cors_allow_credentials: bool = True
|
||||
cors_allow_methods: list = ["*"]
|
||||
cors_allow_headers: list = ["*"]
|
||||
|
||||
cors_allow_methods: List[str] = ["*"]
|
||||
cors_allow_headers: List[str] = ["*"]
|
||||
|
||||
# 安全配置
|
||||
secret_key: str = "your-secret-key-change-in-production"
|
||||
algorithm: str = "HS256"
|
||||
access_token_expire_minutes: int = 30
|
||||
|
||||
|
||||
# 模块系统
|
||||
module_system_type: str = "local"
|
||||
module_system_path: list = ["plugins", "app"]
|
||||
|
||||
module_system_path: List[str] = ["plugins", "app"]
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
extra = "ignore"
|
||||
|
||||
|
||||
@model_validator(mode="after")
|
||||
def validate_settings(self):
|
||||
"""验证和调整设置"""
|
||||
# 如果启用 reload，workers 必须为 1
|
||||
if self.reload and self.workers != 1:
|
||||
print(f"⚠️ reload 启用，将 workers 从 {self.workers} 调整为 1")
|
||||
self.workers = 1
|
||||
return self
|
||||
|
||||
@classmethod
|
||||
def from_toml(cls, config_path: Optional[str] = None) -> "Settings":
|
||||
"""从 TOML 文件加载配置"""
|
||||
|
||||
"""从 TOML 文件加载配置 - 简化版本"""
|
||||
|
||||
# 1. 确定配置文件路径
|
||||
if config_path is None:
|
||||
# 尝试多个可能的配置路径
|
||||
possible_paths = [
|
||||
"config.toml",
|
||||
"./config.toml",
|
||||
"../config.toml",
|
||||
Path(__file__).parent.parent.parent / "config.toml",
|
||||
]
|
||||
|
||||
for path in possible_paths:
|
||||
if Path(path).exists():
|
||||
config_path = str(path)
|
||||
print(f"📄 加载配置文件: {config_path}")
|
||||
break
|
||||
|
||||
if config_path is None:
|
||||
print("⚠️ 未找到配置文件，使用默认配置")
|
||||
return cls()
|
||||
|
||||
config_path = "config.toml"
|
||||
|
||||
config_file = Path(config_path)
|
||||
if not config_file.exists():
|
||||
print(f"⚠️ 配置文件不存在: {config_path}，使用默认配置")
|
||||
return cls()
|
||||
|
||||
try:
|
||||
config_file = Path(config_path)
|
||||
if not config_file.exists():
|
||||
print(f"⚠️ 配置文件不存在: {config_path}")
|
||||
return cls()
|
||||
|
||||
# 2. 加载 TOML 文件
|
||||
config_data = toml.load(config_file)
|
||||
|
||||
# 获取默认配置
|
||||
print(f"📄 加载配置文件: {config_path}")
|
||||
|
||||
# 3. 确定环境
|
||||
default_config = config_data.get("default", {})
|
||||
|
||||
# 确定环境
|
||||
env = default_config.get("env", "development")
|
||||
|
||||
# 获取环境特定配置
|
||||
env_config = config_data.get(env, {})
|
||||
|
||||
# 合并配置
|
||||
merged_config = {**default_config, **env_config}
|
||||
|
||||
# 处理嵌套配置
|
||||
for section in ["server", "database", "security", "logging", "cors", "module_system"]:
|
||||
if section in config_data:
|
||||
if isinstance(config_data[section], dict):
|
||||
merged_config.update(config_data[section])
|
||||
|
||||
# 从环境特定配置中覆盖
|
||||
for section in ["server", "database", "security", "logging", "cors", "module_system"]:
|
||||
env_section_key = f"{env}.{section}"
|
||||
if env_section_key in config_data:
|
||||
if isinstance(config_data[env_section_key], dict):
|
||||
merged_config.update(config_data[env_section_key])
|
||||
|
||||
print(f"🌍 环境: {env}")
|
||||
return cls(**merged_config)
|
||||
|
||||
|
||||
# 4. 获取环境特定配置
|
||||
env_config = config_data.get(env, {})
|
||||
|
||||
# 5. 构建配置字典
|
||||
config_dict = {}
|
||||
|
||||
# 5.1 处理默认配置
|
||||
for key, value in default_config.items():
|
||||
if key.startswith("app."):
|
||||
config_dict[key.replace("app.", "app_")] = value
|
||||
elif key.startswith("server."):
|
||||
config_dict[key.replace("server.", "")] = value
|
||||
elif key == "env":
|
||||
config_dict["env"] = value
|
||||
|
||||
# 5.2 处理环境特定配置
|
||||
for key, value in env_config.items():
|
||||
if key.startswith("server."):
|
||||
config_dict[key.replace("server.", "")] = value
|
||||
elif key == "debug":
|
||||
config_dict["debug"] = value
|
||||
|
||||
# 5.3 处理 database 配置
|
||||
database_config = {}
|
||||
if "database" in env_config:
|
||||
database_type = env_config["database"].get("type", "sqlite")
|
||||
database_config = env_config["database"].get(database_type, {})
|
||||
config_dict["database_type"] = database_type
|
||||
# 5.4 处理 generate_schemas
|
||||
if "database" in env_config:
|
||||
db_config = env_config["database"]
|
||||
if isinstance(db_config, dict) and "generate_schemas" in db_config:
|
||||
config_dict["generate_schemas"] = db_config["generate_schemas"]
|
||||
|
||||
if database_type == "sqlite":
|
||||
if database_config:
|
||||
config_dict["sqlite_config"] = SQLiteConfig(**database_config)
|
||||
|
||||
return cls(**config_dict)
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ 加载配置失败: {e}")
|
||||
import traceback
|
||||
|
||||
traceback.print_exc()
|
||||
return cls()
|
||||
|
||||
|
||||
# 全局配置实例
|
||||
_settings = None
|
||||
|
||||
|
||||
def get_settings() -> Settings:
|
||||
"""获取配置实例（单例模式）"""
|
||||
global _settings
|
||||
if _settings is None:
|
||||
_settings = Settings.from_toml(args.config)
|
||||
return _settings
|
||||
return _settings
|
||||
|
||||
+26
-23
@@ -1,19 +1,21 @@
|
||||
"""
|
||||
Tortoise-ORM 数据库配置
|
||||
"""
|
||||
from typing import Optional
|
||||
|
||||
from tortoise import Tortoise
|
||||
from tortoise.contrib.fastapi import register_tortoise
|
||||
from .config import get_settings
|
||||
|
||||
|
||||
settings = get_settings()
|
||||
|
||||
|
||||
async def init_db(run_mode: str):
|
||||
"""初始化数据库连接"""
|
||||
settings = get_settings()
|
||||
db_config = getattr(settings, f"{settings.database_type}_config").tortoise_config
|
||||
if run_mode == "production":
|
||||
db_config = {
|
||||
"connections": {
|
||||
"default": settings.database_url
|
||||
},
|
||||
"connections": db_config,
|
||||
"apps": {
|
||||
"models": {
|
||||
"models": ["app.models", "aerich.models"],
|
||||
@@ -23,11 +25,10 @@ async def init_db(run_mode: str):
|
||||
"use_tz": False,
|
||||
"timezone": "UTC",
|
||||
}
|
||||
|
||||
elif run_mode == "development":
|
||||
db_config = {
|
||||
"connections": {
|
||||
"default": settings.database_url
|
||||
},
|
||||
"connections": db_config,
|
||||
"apps": {
|
||||
"models": {
|
||||
"models": ["{{ cookiecutter.project_name }}.app.models", "aerich.models"],
|
||||
@@ -39,9 +40,9 @@ async def init_db(run_mode: str):
|
||||
}
|
||||
else:
|
||||
raise ValueError("Invalid run mode")
|
||||
|
||||
|
||||
await Tortoise.init(db_config)
|
||||
|
||||
|
||||
# 根据配置决定是否生成表结构
|
||||
if settings.generate_schemas:
|
||||
print("🗃️ 生成数据库表结构...")
|
||||
@@ -49,41 +50,44 @@ async def init_db(run_mode: str):
|
||||
print("✅ 数据库表结构生成完成")
|
||||
else:
|
||||
print("⏭️ 跳过数据库表结构生成（建议使用迁移工具）")
|
||||
|
||||
print(f"✅ 数据库连接成功: {settings.database_url}")
|
||||
|
||||
print(f"✅ 数据库连接成功: {db_config}")
|
||||
|
||||
|
||||
async def close_db():
|
||||
"""关闭数据库连接"""
|
||||
await Tortoise.close_connections()
|
||||
print("✅ 数据库连接已关闭")
|
||||
|
||||
|
||||
def register_db(app, run_mode: str):
|
||||
"""将数据库注册到 FastAPI 应用"""
|
||||
settings = get_settings()
|
||||
|
||||
modules_path = "app.models" if run_mode == "production" else "{{ cookiecutter.project_name }}.app.models"
|
||||
|
||||
modules_path = "app.models" if run_mode == "production" else "{{ cookiecutter.project_name }}.app.models"
|
||||
file_path = getattr(settings, f"{settings.database_type}_config").file_path
|
||||
register_tortoise(
|
||||
app,
|
||||
db_url=settings.database_url,
|
||||
db_url=f"sqlite://{file_path}",
|
||||
modules={"models": [modules_path]},
|
||||
generate_schemas=settings.generate_schemas,
|
||||
add_exception_handlers=True,
|
||||
)
|
||||
|
||||
|
||||
return app
|
||||
|
||||
|
||||
# 在 app/core/database.py 中添加
|
||||
from aerich import Command
|
||||
|
||||
|
||||
async def run_migrations():
|
||||
"""运行数据库迁移"""
|
||||
try:
|
||||
command = Command(
|
||||
tortoise_config={
|
||||
"connections": {
|
||||
"default": get_settings().database_url
|
||||
},
|
||||
"connections": getattr(
|
||||
settings, f"{settings.database_type}_config"
|
||||
).tortoise_config,
|
||||
"apps": {
|
||||
"models": {
|
||||
"models": ["app.models", "aerich.models"],
|
||||
@@ -91,13 +95,12 @@ async def run_migrations():
|
||||
}
|
||||
},
|
||||
},
|
||||
app="models"
|
||||
app="models",
|
||||
)
|
||||
|
||||
|
||||
# 检查并应用迁移
|
||||
await command.init()
|
||||
await command.upgrade()
|
||||
print("✅ 数据库迁移完成")
|
||||
except Exception as e:
|
||||
print(f"⚠️ 数据库迁移失败: {e}")
|
||||
|
||||
Reference in New Issue
Block a user