diff --git a/output.md b/output.md deleted file mode 100644 index 6373cdc..0000000 --- a/output.md +++ /dev/null @@ -1,1953 +0,0 @@ -`./{{ cookiecutter.project_name }}\config.toml` -```python -[default] -app.name = "{{ cookiecutter.project_name }}" -app.version = "{{ cookiecutter.author_name }}" -app.description = "基于 FastAPI 和 Tortoise-ORM 的 chongming 打包脚手架" -app.debug = false -# env = "production" - -[server] -host = "0.0.0.0" -port = 8000 -reload = false -workers = 4 - -[database] -url = "sqlite://db.sqlite3" -# PostgreSQL 示例: "postgres://user:pass@localhost:5432/dbname" -# MySQL 示例: "mysql://user:pass@localhost:3306/dbname" -generate_schemas = true # 生产环境使用迁移工具 - -[security] -secret_key = "your-secret-key-change-in-production" -algorithm = "HS256" -access_token_expire_minutes = 30 - -[development] -env = "development" -debug = true - -[development.server] -host = "127.0.0.1" -port = 8000 -reload = true -# workers = 4 - -[development.database] -url = "sqlite://dev.db" -generate_schemas = true - -[development.logging] -level = "DEBUG" -file = "logs/dev.log" - -[development.cors] -allow_origins = ["*"] -allow_credentials = true -allow_methods = ["*"] -allow_headers = ["*"] - -[production] -env = "production" -debug = false - -[production.server] -host = "0.0.0.0" -port = 8000 -reload = false -workers = 1 - -[production.database] -url = "sqlite://prod.db" -generate_schemas = true - -[production.logging] -level = "INFO" -file = "logs/prod.log" - -[production.cors] -allow_origins = ["https://yourdomain.com"] -allow_credentials = true -allow_methods = ["GET", "POST", "PUT", "DELETE"] -allow_headers = ["Authorization", "Content-Type"] - -[production.module_system] -type = "module-bank" -path = ["resources/app.mbank", "resources/plugins.mbank"] - -[production.file_system] -type = "svfs" -path = "resources/application.vfs" - -[redis] -enabled = false -url = "redis://localhost:6379" - -[cache] -default_ttl = 300 -``` - -`./{{ cookiecutter.project_name }}\pyproject.toml` -```python -[project] -name = "{{ cookiecutter.project_name }}" -version = "0.1.0" -description = "基于 modules-bank 的 fastapi 打包脚手架" -readme = "README.md" -requires-python = ">=3.8" -dependencies = [ - "aerich", - "fastapi[uvicorn]", - "module-bank", - "passlib", - "pydantic-settings", - "pydantic[email]", - "sqlite-vfs", - "toml", - "tortoise-orm", - "uvicorn[standard]", -] - -[[tool.uv.index]] -url = "https://mirrors.ustc.edu.cn/pypi/web/simple/" - -[tool.uv.workspace] -members = [ - ".", -] - -[tool.uv.sources] -{{ cookiecutter.project_name }} = { workspace = true } - -[dependency-groups] -dev = [ - "{{ cookiecutter.project_name }}", - "pyinstaller>=6.18.0", -] - -[tool.hatch.build.targets.wheel] -packages = ["src/{{ cookiecutter.project_name }}"] - -[project.scripts] -cm_build = "{{ cookiecutter.project_name }}.scripts.build:main" -cm_main = "{{ cookiecutter.project_name }}:main" - -``` - -`./{{ cookiecutter.project_name }}\public\config.toml` -```python -[default] -app.name = "{{ cookiecutter.project_name }}" -app.version = "{{ cookiecutter.author_name }}" -app.description = "基于 FastAPI 和 Tortoise-ORM 的 chongming 打包脚手架" -app.debug = false -env = "production" - -[server] -host = "0.0.0.0" -port = 8000 -reload = false -workers = 4 - -[database] -url = "sqlite://db.sqlite3" -# PostgreSQL 示例: "postgres://user:pass@localhost:5432/dbname" -# MySQL 示例: "mysql://user:pass@localhost:3306/dbname" -generate_schemas = true # 生产环境使用迁移工具 - -[security] -secret_key = "your-secret-key-change-in-production" -algorithm = "HS256" -access_token_expire_minutes = 30 - -[development] -env = "development" -debug = true - -[development.server] -host = "127.0.0.1" -port = 8000 -reload = true -# workers = 4 - -[development.database] -url = "sqlite://dev.db" -generate_schemas = true - -[development.logging] -level = "DEBUG" -file = "logs/dev.log" - -[development.cors] -allow_origins = ["*"] -allow_credentials = true -allow_methods = ["*"] -allow_headers = ["*"] - -[production] -env = "production" -debug = false - -[production.server] -host = "0.0.0.0" -port = 8000 -reload = false -workers = 1 - -[production.database] -url = "sqlite://prod.db" -generate_schemas = true - -[production.logging] -level = "INFO" -file = "logs/prod.log" - -[production.cors] -allow_origins = ["https://yourdomain.com"] -allow_credentials = true -allow_methods = ["GET", "POST", "PUT", "DELETE"] -allow_headers = ["Authorization", "Content-Type"] - -[production.module_system] -type = "module-bank" -path = ["resources/app.mbank", "resources/plugins.mbank"] - -[production.file_system] -type = "svfs" -path = "resources/application.vfs" - -[redis] -enabled = false -url = "redis://localhost:6379" - -[cache] -default_ttl = 300 -``` - -`./{{ cookiecutter.project_name }}\public\main.py` -```python -# main.py - 修复 reload 问题 -#!/usr/bin/env python3 -""" -FastAPI + Tortoise-ORM + Chongming 示例 -主入口文件，适配 chongming 打包系统 -""" -import sys -import os -from utils.launch import launch -from multiprocessing import freeze_support - -def app_run(app_config: dict, default_config: dict): - """ - 应用运行函数 - 适配模块银行环境 - - Args: - app_config: 应用配置 - default_config: 默认配置 - """ - import uvicorn - from app.logger import setup_logging, get_logger # type: ignore - - setup_logging(app_config["logging"]) - logger = get_logger("app") - - - # 获取服务器配置 - server_config = app_config.get("server", {}) - host = server_config.get("host", "0.0.0.0") - port = server_config.get("port", 8000) - - # 打包环境强制禁用重载 - is_frozen = getattr(sys, 'frozen', False) - reload = False if is_frozen else server_config.get("reload", False) - workers = server_config.get("workers", 4) - - # 显示启动信息 - env = app_config.get("env", "development") - logger.info("=" * 50) - logger.info(f"🚀 启动 {default_config.get('app.name', 'FastAPI 应用')}") - logger.info(f"📦 版本: {default_config.get('app.version', '1.0.0')}") - logger.info(f"🌍 环境: {env}") - logger.info(f"📍 地址: http://{host}:{port}") - logger.info(f"📚 文档: http://{host}:{port}/docs") - logger.info(f"🔧 调试: {'启用' if app_config.get('debug', False) else '禁用'}") - logger.info(f"🔄 重载: {'启用' if reload else '禁用'}") - logger.info(f"👥 Workers: {workers}") - logger.info("=" * 50) - - # 关键修改：根据 reload 模式选择不同的启动方式 - - uvicorn.run( - "app:app", - host=host, - port=port, - reload=reload, - workers=workers, - log_level="debug" if app_config.get("debug", False) else "info", - access_log=True, - ) - -def main(): - freeze_support() - # 如果是打包后的可执行文件，调整路径 - if getattr(sys, 'frozen', False): - exe_dir = os.path.dirname(sys.executable) - if exe_dir not in sys.path: - sys.path.insert(0, exe_dir) - - # 启动应用 - launch(app_run, "development") - -if __name__ == "__main__": - main() -``` - -`./{{ cookiecutter.project_name }}\public\utils\config.py` -```python -# config_loader.py - 读取config.toml的示例代码 - -import toml -import os -from pathlib import Path -from typing import Any, Dict, Optional - -class ConfigLoader: - """TOML配置文件加载器""" - - def __init__(self, config_path: Optional[str] = None): - """ - 初始化配置加载器 - - Args: - config_path: 配置文件路径，如果为None则自动查找 - """ - if config_path is None: - # 自动查找配置文件 - config_path = self._find_config_file() - - self.config_path = Path(config_path) - self._config: Dict[str, Any] = {} - self.load() - - def _find_config_file(self) -> str: - """自动查找配置文件""" - possible_paths = [ - "config.toml", - "config/config.toml", - "~/.config/myapp/config.toml", - "/etc/myapp/config.toml" - ] - - for path in possible_paths: - expanded_path = os.path.expanduser(path) - if os.path.exists(expanded_path): - return expanded_path - - raise FileNotFoundError("未找到配置文件") - - def load(self) -> None: - """加载配置文件""" - if not self.config_path.exists(): - raise FileNotFoundError(f"配置文件不存在: {self.config_path}") - - with open(self.config_path, 'r', encoding='utf-8') as f: - self._config = toml.load(f) - - def get(self, key: str, default: Any = None) -> Any: - """获取配置值，支持点号分隔的路径""" - keys = key.split('.') - value = self._config - - try: - for k in keys: - value = value[k] - return value - except (KeyError, TypeError): - return default - - def get_section(self, section: str) -> Dict[str, Any]: - """获取整个配置节""" - return self._config.get(section, {}) - - def reload(self) -> None: - """重新加载配置文件""" - self.load() - - def __getitem__(self, key: str) -> Any: - """支持字典式访问""" - return self.get(key) - - def __contains__(self, key: str) -> bool: - """检查配置是否存在""" - return self.get(key) is not None - - @property - def all(self) -> Dict[str, Any]: - """获取所有配置""" - return self._config - -``` - -`./{{ cookiecutter.project_name }}\public\utils\launch.py` -```python -import sys -from .config import ConfigLoader -import argparse -import os -from module_bank import PythonToSQLite -from typing import Callable - -# 检查是否在 PyInstaller 子进程中 -def is_pyinstaller_child(): - return len(sys.argv) > 1 and ('_child.py' in sys.argv[0] or sys.argv[1].isdigit()) - -def parse_args(): - """安全的参数解析函数""" - if is_pyinstaller_child(): - # PyInstaller 子进程，使用默认参数 - return argparse.Namespace(config="config.toml") - - parser = argparse.ArgumentParser(description="chongming 应用启动脚本") - parser.add_argument("--config", default="config.toml", help="配置文件路径") - return parser.parse_args() - - -args = parse_args() - -def init_module_bank(config): - MODULE_BANK_PATH_LIST = config["production"]["module_system"]["path"] - for module_path in MODULE_BANK_PATH_LIST: - if os.path.exists(module_path) == False: - raise ValueError(f"模块路径不存在: {module_path}") - - packer_list = [PythonToSQLite(module_path) for module_path in MODULE_BANK_PATH_LIST] - for packer in packer_list: - modules = packer.list_modules() - for module in modules: - print(f"{module['module_name']} {'[包]' if module['is_package'] else ''}") - packer.install_importer() - -def launch(run_fun: Callable[[dict, dict], None], run_mode: str): - - config = ConfigLoader(args.config) - - # 获取对应运行模式的配置 - default_config = config.get("default", {}) - run_mode = default_config.get("env", "development") - app_config = config.get(run_mode, {}) - if run_mode == "production": - init_module_bank(config) - - run_fun(app_config, default_config) - -``` - -`./{{ cookiecutter.project_name }}\public\utils\__init__.py` -```python - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\__init__.py` -```python -# main.py - 修复 reload 问题 -#!/usr/bin/env python3 -""" -FastAPI + Tortoise-ORM + Chongming 示例 -主入口文件，适配 chongming 打包系统 -""" -import sys -import os -from .utils.launch import launch -from multiprocessing import freeze_support - -def app_run(app_config: dict, default_config: dict): - """ - 应用运行函数 - 适配模块银行环境 - - Args: - app_config: 应用配置 - default_config: 默认配置 - """ - import uvicorn - from .app.logger import setup_logging, get_logger - - setup_logging(app_config["logging"]) - logger = get_logger("app") - - - # 获取服务器配置 - server_config = app_config.get("server", {}) - host = server_config.get("host", "0.0.0.0") - port = server_config.get("port", 8000) - - # 打包环境强制禁用重载 - is_frozen = getattr(sys, 'frozen', False) - reload = False if is_frozen else server_config.get("reload", False) - workers = server_config.get("workers", 4) - - # 显示启动信息 - env = app_config.get("env", "development") - logger.info("=" * 50) - logger.info(f"🚀 启动 {default_config.get('app.name', 'FastAPI 应用')}") - logger.info(f"📦 版本: {default_config.get('app.version', '1.0.0')}") - logger.info(f"🌍 环境: {env}") - logger.info(f"📍 地址: http://{host}:{port}") - logger.info(f"📚 文档: http://{host}:{port}/docs") - logger.info(f"🔧 调试: {'启用' if app_config.get('debug', False) else '禁用'}") - logger.info(f"🔄 重载: {'启用' if reload else '禁用'}") - logger.info(f"👥 Workers: {workers}") - logger.info("=" * 50) - - # 关键修改：根据 reload 模式选择不同的启动方式 - - uvicorn.run( - "{{ cookiecutter.project_name }}.app:app", - host=host, - port=port, - reload=reload, - workers=workers, - log_level="debug" if app_config.get("debug", False) else "info", - access_log=True, - ) - -def main(): - freeze_support() - # 如果是打包后的可执行文件，调整路径 - if getattr(sys, 'frozen', False): - exe_dir = os.path.dirname(sys.executable) - if exe_dir not in sys.path: - sys.path.insert(0, exe_dir) - - # 启动应用 - launch(app_run, "development") - -if __name__ == "__main__": - main() -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\__init__.py` -```python -""" -FastAPI 应用包 -""" -__version__ = "1.0.0" - - -""" -FastAPI 应用实例 -""" -from fastapi import FastAPI -from fastapi.middleware.cors import CORSMiddleware -from contextlib import asynccontextmanager -from .logger import setup_logging, get_logger - -from .core.config import get_settings -from .core.database import register_db, init_db, close_db -from .api import api_router - -# 获取配置 -settings = get_settings() -logger = get_logger("app") - -@asynccontextmanager -async def lifespan(app: FastAPI): - """ - 应用生命周期管理 - - 注意：在生产环境的模块银行中，数据库初始化可能失败 - 这里使用 try-except 确保应用能够正常启动 - """ - logger.info(f"🚀 启动 {settings.app_name} v{settings.app_version}") - logger.info(f"🌍 环境: {settings.env}") - - # 初始化数据库（生产环境可能跳过） - try: - await init_db(settings.env) - except Exception as e: - logger.warning(f"⚠️ 数据库初始化失败: {e}") - logger.warning("💡 提示：应用将继续运行，但数据库功能可能受限") - - yield - - # 关闭数据库连接 - try: - await close_db() - except Exception as e: - logger.warning(f"⚠️ 数据库关闭失败: {e}") - - logger.info("👋 关闭应用") - -# 创建 FastAPI 应用实例 -app = FastAPI( - title=settings.app_name, - version=settings.app_version, - description=settings.description, - docs_url="/docs" if settings.debug else None, - redoc_url="/redoc" if settings.debug else None, - openapi_url="/openapi.json" if settings.debug else None, - lifespan=lifespan, -) - -# 添加 CORS 中间件 -app.add_middleware( - CORSMiddleware, - allow_origins=settings.cors_allow_origins, - allow_credentials=settings.cors_allow_credentials, - allow_methods=settings.cors_allow_methods, - allow_headers=settings.cors_allow_headers, -) - -# 注册数据库 -app = register_db(app, settings.env) - -# 注册路由 -app.include_router(api_router, prefix="/api/v1") - -# 根路由 -@app.get("/", summary="根路径") -async def root(): - """根路径，返回应用信息""" - return { - "message": f"欢迎使用 {settings.app_name}", - "version": settings.app_version, - "docs": "/docs" if settings.debug else None, - "health": "/api/v1/health", - } - -# 全局异常处理器 -@app.exception_handler(Exception) -async def global_exception_handler(request, exc): - """全局异常处理器""" - logger.error(f"全局异常: {exc}") - return { - "error": "内部服务器错误", - "detail": str(exc) if settings.debug else "请查看服务器日志" - } - -logger.info("✅ FastAPI 应用初始化完成") - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\api\dependencies.py` -```python -""" -API 依赖项 -""" -from typing import AsyncGenerator -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 - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\api\__init__.py` -```python -""" -API 路由 -""" -from fastapi import APIRouter -from .routers import health, items, users - -# 创建主路由 -api_router = APIRouter() - -# 注册子路由 -api_router.include_router(health.router, prefix="/health", tags=["health"]) -api_router.include_router(items.router, prefix="/items", tags=["items"]) -api_router.include_router(users.router, prefix="/users", tags=["users"]) - -# 导出路由 -__all__ = ["api_router"] -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\api\routers\health.py` -```python -""" -健康检查路由 -""" -from fastapi import APIRouter, Depends -from ...core.config import get_settings -from ..dependencies import get_db - -router = APIRouter() - -@router.get("/", summary="健康检查") -async def health_check(db = Depends(get_db)): - """健康检查端点""" - settings = get_settings() - - return { - "status": "healthy", - "service": settings.app_name, - "version": settings.app_version, - "environment": settings.env, - "timestamp": "now" # 实际使用中可以用 datetime.now() - } - -@router.get("/ready", summary="就绪检查") -async def readiness_check(): - """就绪检查端点""" - return { - "status": "ready", - "database": "connected", - "api": "available" - } - -@router.get("/info", summary="服务信息") -async def service_info(): - """服务信息端点""" - settings = get_settings() - - return { - "name": settings.app_name, - "version": settings.app_version, - "description": settings.description, - "environment": settings.env, - "debug": settings.debug, - "database": settings.database_url.split("://")[0] if "://" in settings.database_url else "unknown" - } -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\api\routers\items.py` -```python -""" -Item 路由 -""" -from typing import List -from fastapi import APIRouter, Depends, HTTPException, status -from tortoise.exceptions import DoesNotExist -from ...models.item import Item -from ...schemas.item import ItemCreate, ItemUpdate, ItemResponse -from ..dependencies import get_db - -router = APIRouter() - -@router.get("/", response_model=List[ItemResponse], summary="获取物品列表") -async def read_items( - skip: int = 0, - limit: int = 100, - db=Depends(get_db) -): - """获取物品列表""" - items = await Item.all().offset(skip).limit(limit) - return items - -@router.get("/{item_id}", response_model=ItemResponse, summary="获取物品详情") -async def read_item(item_id: int, db=Depends(get_db)): - """根据ID获取物品""" - try: - item = await Item.get(id=item_id) - return item - except DoesNotExist: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"物品 ID {item_id} 不存在" - ) - -@router.post("/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED, summary="创建物品") -async def create_item(item: ItemCreate, db=Depends(get_db)): - """创建新物品""" - # 检查名称是否已存在 - existing_item = await Item.filter(name=item.name).first() - if existing_item: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"物品名称 '{item.name}' 已存在" - ) - - # 创建物品 - item_dict = item.model_dump() - new_item = await Item.create(**item_dict) - return new_item - -@router.put("/{item_id}", response_model=ItemResponse, summary="更新物品") -async def update_item(item_id: int, item_update: ItemUpdate, db=Depends(get_db)): - """更新物品信息""" - try: - item = await Item.get(id=item_id) - - # 更新字段 - update_data = item_update.model_dump(exclude_unset=True) - - # 如果更新名称，检查是否与其他物品重名 - if "name" in update_data: - existing_item = await Item.filter( - name=update_data["name"] - ).exclude(id=item_id).first() - if existing_item: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"物品名称 '{update_data['name']}' 已存在" - ) - - # 应用更新 - await item.update_from_dict(update_data) - await item.save() - - return item - except DoesNotExist: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"物品 ID {item_id} 不存在" - ) - -@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT, summary="删除物品") -async def delete_item(item_id: int, db=Depends(get_db)): - """删除物品""" - try: - item = await Item.get(id=item_id) - await item.delete() - return None - except DoesNotExist: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"物品 ID {item_id} 不存在" - ) -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\api\routers\users.py` -```python -""" -User 路由 -""" -from typing import List -from fastapi import APIRouter, Depends, HTTPException, status -from tortoise.exceptions import DoesNotExist, IntegrityError - -from ...models.user import User -from ...schemas.user import UserCreate, UserUpdate, UserResponse -from ...services.user_service import UserService -from ..dependencies import get_db - -router = APIRouter() - -@router.get("/", response_model=List[UserResponse], summary="获取用户列表") -async def read_users( - skip: int = 0, - limit: int = 100, - db=Depends(get_db) -): - """获取用户列表""" - users = await UserService.get_all_users(skip, limit) - return users - -@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED, summary="创建用户") -async def create_user(user_data: UserCreate, db=Depends(get_db)): - """创建新用户""" - # 检查用户名是否已存在 - existing_user = await UserService.get_user_by_username(user_data.username) - if existing_user: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"用户名 '{user_data.username}' 已存在" - ) - - # 检查邮箱是否已存在 - existing_email = await UserService.get_user_by_email(user_data.email) - if existing_email: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail=f"邮箱 '{user_data.email}' 已存在" - ) - - # 创建用户 - try: - user = await UserService.create_user(user_data) - return user - except IntegrityError: - raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="创建用户失败" - ) - -@router.get("/{user_id}", response_model=UserResponse, summary="获取用户详情") -async def read_user(user_id: int, db=Depends(get_db)): - """根据ID获取用户""" - user = await UserService.get_user_by_id(user_id) - if not user: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"用户 ID {user_id} 不存在" - ) - return user - -@router.put("/{user_id}", response_model=UserResponse, summary="更新用户") -async def update_user(user_id: int, user_update: UserUpdate, db=Depends(get_db)): - """更新用户信息""" - user = await UserService.update_user(user_id, user_update) - if not user: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"用户 ID {user_id} 不存在" - ) - return user - -@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT, summary="删除用户") -async def delete_user(user_id: int, db=Depends(get_db)): - """删除用户""" - deleted = await UserService.delete_user(user_id) - if not deleted: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"用户 ID {user_id} 不存在" - ) - return None -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\api\routers\__init__.py` -```python -""" -路由模块 -""" -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\core\config.py` -```python -""" -配置管理模块 -基于 Pydantic Settings 和 TOML 文件 -""" -from typing import Optional, Dict, Any -from pydantic_settings import BaseSettings -from pydantic import Field -import toml -from pathlib import Path -import sys -import os -try: - from ...utils.launch import args -except: - from utils.launch import args # type: ignore - -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 - - # 环境配置 - env: str = "development" - debug: bool = False - - # 日志配置 - logging_level: str = "INFO" - logging_file: Optional[str] = None - - # CORS 配置 - cors_allow_origins: list = ["*"] - cors_allow_credentials: bool = True - cors_allow_methods: list = ["*"] - cors_allow_headers: list = ["*"] - - # 安全配置 - 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"] - - class Config: - env_file = ".env" - extra = "ignore" - - @classmethod - def from_toml(cls, config_path: Optional[str] = None) -> "Settings": - """从 TOML 文件加载配置""" - - 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() - - try: - config_file = Path(config_path) - if not config_file.exists(): - print(f"⚠️ 配置文件不存在: {config_path}") - return cls() - - config_data = toml.load(config_file) - - # 获取默认配置 - 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) - - 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 -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\core\database.py` -```python -""" -Tortoise-ORM 数据库配置 -""" -from typing import Optional -from tortoise import Tortoise -from tortoise.contrib.fastapi import register_tortoise -from .config import get_settings - -async def init_db(run_mode: str): - """初始化数据库连接""" - settings = get_settings() - if run_mode == "production": - db_config = { - "connections": { - "default": settings.database_url - }, - "apps": { - "models": { - "models": ["app.models", "aerich.models"], - "default_connection": "default", - } - }, - "use_tz": False, - "timezone": "UTC", - } - elif run_mode == "development": - db_config = { - "connections": { - "default": settings.database_url - }, - "apps": { - "models": { - "models": ["{{ cookiecutter.project_name }}.app.models", "aerich.models"], - "default_connection": "default", - } - }, - "use_tz": False, - "timezone": "UTC", - } - else: - raise ValueError("Invalid run mode") - - await Tortoise.init(db_config) - - # 根据配置决定是否生成表结构 - if settings.generate_schemas: - print("🗃️ 生成数据库表结构...") - await Tortoise.generate_schemas() - print("✅ 数据库表结构生成完成") - else: - print("⏭️ 跳过数据库表结构生成（建议使用迁移工具）") - - print(f"✅ 数据库连接成功: {settings.database_url}") - -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" - - register_tortoise( - app, - db_url=settings.database_url, - 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 - }, - "apps": { - "models": { - "models": ["app.models", "aerich.models"], - "default_connection": "default", - } - }, - }, - app="models" - ) - - # 检查并应用迁移 - await command.init() - await command.upgrade() - print("✅ 数据库迁移完成") - except Exception as e: - print(f"⚠️ 数据库迁移失败: {e}") - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\core\security.py` -```python - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\core\__init__.py` -```python -""" -核心模块 -""" -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\logger\__init__.py` -```python -import logging -from pathlib import Path - -class ExcludeWatchFilesFilter(logging.Filter): - def filter(self, record): - # 如果日志记录来自 watchfiles 模块，则不记录（返回 False） - return not record.name.startswith('watchfiles') - -def setup_logging(log_config: dict): - log_level = getattr(logging, log_config["level"], logging.INFO) - log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' - # 创建文件夹 - log_file_path = Path(log_config["file"]) - log_file_dir = log_file_path.parent - if log_file_dir != Path("."): - log_file_dir.mkdir(parents=True, exist_ok=True) - - # 设置特定日志级别的过滤 - logging.getLogger("watchfiles").setLevel(logging.WARNING) - - # 配置根日志记录器 - root_logger = logging.getLogger() - root_logger.setLevel(log_level) - - # 创建格式化器 - formatter = logging.Formatter(log_format) - - # 创建文件处理器并添加过滤器 - file_handler = logging.FileHandler(log_config["file"], encoding="utf-8") - file_handler.setFormatter(formatter) - file_handler.addFilter(ExcludeWatchFilesFilter()) - - # 创建控制台处理器并添加过滤器 - console_handler = logging.StreamHandler() - console_handler.setFormatter(formatter) - console_handler.addFilter(ExcludeWatchFilesFilter()) - - # 添加处理器到根日志记录器 - root_logger.addHandler(file_handler) - root_logger.addHandler(console_handler) - - print(f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']}") - -def get_logger(name: str) -> logging.Logger: - logger = logging.getLogger(name) - # 为每个logger添加过滤器 - logger.addFilter(ExcludeWatchFilesFilter()) - return logger -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\models\item.py` -```python -""" -Item 数据模型 -""" -from tortoise.models import Model -from tortoise import fields -from datetime import datetime - -class Item(Model): - """物品模型""" - - id = fields.IntField(pk=True, description="主键ID") - name = fields.CharField(max_length=100, description="名称") - description = fields.TextField(null=True, description="描述") - price = fields.DecimalField(max_digits=10, decimal_places=2, default=0.00, description="价格") - is_active = fields.BooleanField(default=True, description="是否激活") - - # 关联用户（可选） - owner = fields.ForeignKeyField( - "models.User", - related_name="items", - null=True, - on_delete=fields.SET_NULL, - description="所有者" - ) - - # 时间戳 - created_at = fields.DatetimeField(auto_now_add=True, description="创建时间") - updated_at = fields.DatetimeField(auto_now=True, description="更新时间") - - def __str__(self): - return f"Item(id={self.id}, name={self.name}, price={self.price})" - - class Meta: - table = "items" - table_description = "物品表" - ordering = ["-created_at"] - - class PydanticMeta: - # Pydantic 模型配置 - exclude = ["is_active"] -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\models\user.py` -```python -""" -User 数据模型 -""" -from tortoise.models import Model -from tortoise import fields -from datetime import datetime -from .item import Item - -class User(Model): - """用户模型""" - - id = fields.IntField(pk=True, description="主键ID") - username = fields.CharField(max_length=50, unique=True, description="用户名") - email = fields.CharField(max_length=100, unique=True, description="邮箱") - hashed_password = fields.CharField(max_length=128, description="哈希密码") - is_active = fields.BooleanField(default=True, description="是否激活") - is_superuser = fields.BooleanField(default=False, description="是否超级用户") - - # 时间戳 - created_at = fields.DatetimeField(auto_now_add=True, description="创建时间") - updated_at = fields.DatetimeField(auto_now=True, description="更新时间") - - # 反向关系 - items: fields.ReverseRelation["Item"] - - def __str__(self): - return f"User(id={self.id}, username={self.username})" - - class Meta: - table = "users" - table_description = "用户表" - ordering = ["-created_at"] - - class PydanticMeta: - # Pydantic 模型配置 - exclude = ["hashed_password"] -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\models\__init__.py` -```python -""" -数据库模型 -""" -from .item import Item -from .user import User - -# 导出所有模型，供 Tortoise-ORM 自动发现 -__all__ = ["Item", "User"] -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\schemas\item.py` -```python -""" -Item Pydantic 模型 -""" -from pydantic import BaseModel, Field -from datetime import datetime -from typing import Optional -from decimal import Decimal - -class ItemBase(BaseModel): - """Item 基础模型""" - name: str = Field(..., min_length=1, max_length=100, description="物品名称") - description: Optional[str] = Field(None, description="物品描述") - price: Decimal = Field(Decimal(0.00), ge=0, description="物品价格") - -class ItemCreate(ItemBase): - """创建 Item 模型""" - pass - -class ItemUpdate(BaseModel): - """更新 Item 模型""" - name: Optional[str] = Field(None, min_length=1, max_length=100, description="物品名称") - description: Optional[str] = Field(None, description="物品描述") - price: Optional[Decimal] = Field(None, ge=0, description="物品价格") - -class ItemInDB(ItemBase): - """数据库中的 Item 模型""" - id: int - is_active: bool - created_at: datetime - updated_at: datetime - - class Config: - from_attributes = True - -class ItemResponse(ItemInDB): - """Item 响应模型""" - pass -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\schemas\user.py` -```python -# app/schemas/user.py -from pydantic import BaseModel, EmailStr -from datetime import datetime -from typing import Optional - -class UserBase(BaseModel): - username: str - email: EmailStr - -class UserCreate(UserBase): - password: str - -class UserUpdate(BaseModel): - username: Optional[str] = None - email: Optional[EmailStr] = None - password: Optional[str] = None - -class UserInDB(UserBase): - id: int - is_active: bool - created_at: datetime - updated_at: datetime - - class Config: - from_attributes = True - -class UserResponse(UserInDB): - pass - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\schemas\__init__.py` -```python -""" -Pydantic 模型 -""" -from .item import ItemCreate, ItemUpdate, ItemResponse, ItemInDB - -__all__ = [ - "ItemCreate", - "ItemUpdate", - "ItemResponse", - "ItemInDB" -] -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\services\item_service.py` -```python -# app/services/item_service.py -from typing import Optional, List -from ..models.item import Item -from ..schemas.item import ItemCreate, ItemUpdate - -class ItemService: - @staticmethod - async def get_all_items( - skip: int = 0, - limit: int = 100, - owner_id: Optional[int] = None - ) -> List[Item]: - query = Item.all() - if owner_id: - query = query.filter(owner_id=owner_id) - return await query.offset(skip).limit(limit) - - @staticmethod - async def get_item_by_id(item_id: int) -> Optional[Item]: - return await Item.filter(id=item_id).first().prefetch_related("owner") - - @staticmethod - async def create_item(item_data: ItemCreate, owner_id: Optional[int] = None) -> Item: - item = await Item.create( - **item_data.model_dump(), - owner_id=owner_id - ) - return item - - @staticmethod - async def update_item(item_id: int, item_data: ItemUpdate) -> Optional[Item]: - item = await ItemService.get_item_by_id(item_id) - if not item: - return None - - update_data = item_data.model_dump(exclude_unset=True) - await item.update_from_dict(update_data) - await item.save() - return item - - @staticmethod - async def delete_item(item_id: int) -> bool: - item = await ItemService.get_item_by_id(item_id) - if item: - await item.delete() - return True - return False -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\services\user_service.py` -```python -# app/services/user_service.py -from typing import Optional, List -from ..models.user import User -from ..schemas.user import UserCreate, UserUpdate -from passlib.context import CryptContext - -pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") - -class UserService: - @staticmethod - def get_password_hash(password: str) -> str: - return pwd_context.hash(password) - - @staticmethod - def verify_password(plain_password: str, hashed_password: str) -> bool: - return pwd_context.verify(plain_password, hashed_password) - - @staticmethod - async def get_all_users(skip: int = 0, limit: int = 100) -> List[User]: - return await User.all().offset(skip).limit(limit) - - @staticmethod - async def get_user_by_id(user_id: int) -> Optional[User]: - return await User.filter(id=user_id).first() - - @staticmethod - async def get_user_by_username(username: str) -> Optional[User]: - return await User.filter(username=username).first() - - @staticmethod - async def get_user_by_email(email: str) -> Optional[User]: - return await User.filter(email=email).first() - - @staticmethod - async def create_user(user_data: UserCreate) -> User: - hashed_password = UserService.get_password_hash(user_data.password) - user = await User.create( - username=user_data.username, - email=user_data.email, - hashed_password=hashed_password - ) - return user - - @staticmethod - async def update_user(user_id: int, user_data: UserUpdate) -> Optional[User]: - user = await UserService.get_user_by_id(user_id) - if not user: - return None - - update_data = user_data.model_dump(exclude_unset=True) - - if "password" in update_data: - update_data["hashed_password"] = UserService.get_password_hash( - update_data.pop("password") - ) - - await user.update_from_dict(update_data) - await user.save() - return user - - @staticmethod - async def delete_user(user_id: int) -> bool: - user = await UserService.get_user_by_id(user_id) - if user: - await user.delete() - return True - return False - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\app\services\__init__.py` -```python - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\scripts\build.py` -```python -from module_bank import PythonToSQLite -import os -from pathlib import Path -import shutil -import subprocess -import sys - -def run_pyinstaller(): - """运行 PyInstaller 打包成可执行文件""" - try: - # 构建 PyInstaller 命令 - cmd = [ - sys.executable, "-m", "PyInstaller", - "--onefile", - "--name", "{{ cookiecutter.project_name }}", - "--distpath", ".", - "--workpath", "./build_temp", - "--specpath", "./specs", - "--clean", - # 隐藏导入 - "--hidden-import", "passlib", - "--hidden-import", "passlib.context", - "--hidden-import", "passlib.handlers.bcrypt", - "--hidden-import", "fastapi", - "--hidden-import", "fastapi.middleware.cors", - "--hidden-import", "pydantic_settings", - "--hidden-import", "tortoise", - "--hidden-import", "tortoise.contrib.fastapi", - "--hidden-import", "aerich", - "--hidden-import", "uvicorn", - # 复制包元数据 - 解决 importlib.metadata 问题 - "--copy-metadata", "tortoise-orm", - "--copy-metadata", "aerich", - "--copy-metadata", "fastapi", - "--copy-metadata", "uvicorn", - "--copy-metadata", "pydantic", - "--copy-metadata", "pydantic-settings", - "--copy-metadata", "module-bank", - "--copy-metadata", "passlib", - "--copy-metadata", "sqlite-vfs", - "--copy-metadata", "toml", - # 收集整个包 - "--collect-all", "tortoise-orm", - "--collect-all", "aerich", - # 禁用多进程优化 - # "--multiprocessing-fork", # 明确指定使用 fork 模式 - # 添加数据文件 - # "--add-data", "resources:resources", - # 其他选项 - "--icon", "../../public/{{ cookiecutter.project_name }}.ico", - "main.py" - ] - - print(f"正在运行 PyInstaller: {' '.join(cmd)}") - result = subprocess.run(cmd, cwd="./build", capture_output=True, text=True) - - if result.returncode == 0: - print("PyInstaller 打包成功！") - print(result.stdout) - else: - print("PyInstaller 打包失败！") - print("错误信息:", result.stderr) - - except FileNotFoundError: - print("错误：未找到 PyInstaller，请先安装 PyInstaller：pip install pyinstaller") - except Exception as e: - print(f"PyInstaller 执行出错: {e}") - -def main(): - # 打包模块 - resources_path = Path("resources") - if not resources_path.exists(): - resources_path.mkdir() - app_packer = PythonToSQLite("./resources/app.mbank") - app_packer.pack_directory("src/{{ cookiecutter.project_name }}/app", "app") - app_packer.verify_package_structure() - # app_packer.delete_source_code(None) - - plugin_packer = PythonToSQLite("./resources/plugins.mbank") - plugin_packer.pack_directory("src/{{ cookiecutter.project_name }}/plugins", "plugins") - plugin_packer.verify_package_structure() - # plugin_packer.delete_source_code(None) - - # 创建 build 目录 - build_path = Path("build") - if build_path.exists() and build_path.is_dir(): - shutil.rmtree("build") - - os.makedirs("build", exist_ok=True) - - # 复制 utils 文件夹到 build 文件夹 - utils_path = Path("utils") - if utils_path.exists(): - target_utils_path = Path("build") / "utils" - shutil.copytree(utils_path, target_utils_path, dirs_exist_ok=True) - print(f"已复制 utils 文件夹到 {target_utils_path}") - - # 复制 resources 文件夹到 build 文件夹 - resources_path = Path("resources") - if resources_path.exists(): - target_resources_path = Path("build") / "resources" - shutil.copytree(resources_path, target_resources_path, dirs_exist_ok=True) - print(f"已复制 resources 文件夹到 {target_resources_path}") - - print("模块打包完成") - - # 复制 main.py 到 build 文件夹 - shutil.copy("public/main.py", "build") - # 复制 config.toml 到 build 文件夹 - shutil.copy("public/config.toml", "build") - # 复制 utils 文件夹到 build 文件夹 - shutil.copytree("public/utils", "build/utils") - - # 运行 PyInstaller - run_pyinstaller() - - # 清理临时文件 - shutil.rmtree(r"build/build_temp", ignore_errors=True) - shutil.rmtree(r"build/specs", ignore_errors=True) - shutil.rmtree(r"build/utils", ignore_errors=True) - os.remove(r"build/main.py") - # os.remove(r"build/config.toml") - - print("构建完成") - -if __name__ == "__main__": - main() -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\scripts\clean_pycache.py` -```python -import os -import shutil -# 需要遍历的目录 -root_dir = "./" -# 遍历目录 -for dirpath, dirnames, filenames in os.walk(root_dir): - if "__pycache__" in dirnames: - # 获取 __pycache__ 目录的全路径 - pycache_dir = os.path.join(dirpath, "__pycache__") - # 删除目录 - shutil.rmtree(pycache_dir) - print(f"Removed: {pycache_dir}") - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\scripts\code_content.py` -```python -import os -import sys - - -def generate_markdown_from_py_files(directory, output_file): - with open(output_file, "w", encoding="utf-8") as md_file: - for root, dirs, files in os.walk(directory): - # 排除 venv 目录 - dirs[:] = [d for d in dirs if d != ".venv"] - dirs[:] = [d for d in dirs if d != ".vscode"] - dirs[:] = [d for d in dirs if d != "Scripts"] - dirs[:] = [d for d in dirs if d != "build"] - dirs[:] = [d for d in dirs if d != "node_modules"] - for file in files: - if ( - file.endswith(".py") - or file.endswith(".rs") - or file.endswith(".toml") - ): - file_path = os.path.join(root, file) - md_file.write(f"`{file_path}`\n") - md_file.write("```python\n") - with open(file_path, "r", encoding="utf-8") as py_file: - md_file.write(py_file.read()) - md_file.write("\n```\n\n") - - -if __name__ == "__main__": - # 指定目录和输出文件名 - target_directory = sys.argv[1] # 替换为你的目标目录 - output_markdown_file = "output.md" # 输出的 Markdown 文件名 - - generate_markdown_from_py_files(target_directory, output_markdown_file) - print(f"Markdown 文件已生成：{output_markdown_file}") - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\scripts\migrate.py` -```python -#!/usr/bin/env python3 -""" -数据库迁移脚本 -""" -import asyncio -import sys -import os - -# 添加项目根目录到路径 -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -async def main(): - from app.core.config import get_settings - from aerich import Command - - settings = get_settings() - - # Aerich 配置 - TORTOISE_ORM = { - "connections": { - "default": settings.database_url - }, - "apps": { - "models": { - "models": ["app.models", "aerich.models"], - "default_connection": "default", - } - }, - "use_tz": False, - "timezone": "UTC", - } - - command = Command(tortoise_config=TORTOISE_ORM, app="models") - - # 初始化 Aerich（首次运行） - if not os.path.exists("./migrations"): - print("🔧 初始化数据库迁移...") - await command.init() - print("✅ 迁移初始化完成") - - # 生成迁移文件 - print("📝 生成迁移文件...") - try: - migration_name = sys.argv[1] if len(sys.argv) > 1 else "update" - await command.migrate(name=migration_name) - print(f"✅ 迁移文件生成成功: {migration_name}") - except Exception as e: - print(f"⚠️ 生成迁移文件失败: {e}") - # 尝试升级 - pass - - # 应用迁移 - print("🔄 应用数据库迁移...") - try: - await command.upgrade() - print("✅ 迁移应用成功") - except Exception as e: - print(f"❌ 迁移应用失败: {e}") - -if __name__ == "__main__": - asyncio.run(main()) -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\scripts\__init__.py` -```python - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\utils\config.py` -```python -# config_loader.py - 读取config.toml的示例代码 - -import toml -import os -from pathlib import Path -from typing import Any, Dict, Optional - -class ConfigLoader: - """TOML配置文件加载器""" - - def __init__(self, config_path: Optional[str] = None): - """ - 初始化配置加载器 - - Args: - config_path: 配置文件路径，如果为None则自动查找 - """ - if config_path is None: - # 自动查找配置文件 - config_path = self._find_config_file() - - self.config_path = Path(config_path) - self._config: Dict[str, Any] = {} - self.load() - - def _find_config_file(self) -> str: - """自动查找配置文件""" - possible_paths = [ - "config.toml", - "config/config.toml", - "~/.config/myapp/config.toml", - "/etc/myapp/config.toml" - ] - - for path in possible_paths: - expanded_path = os.path.expanduser(path) - if os.path.exists(expanded_path): - return expanded_path - - raise FileNotFoundError("未找到配置文件") - - def load(self) -> None: - """加载配置文件""" - if not self.config_path.exists(): - raise FileNotFoundError(f"配置文件不存在: {self.config_path}") - - with open(self.config_path, 'r', encoding='utf-8') as f: - self._config = toml.load(f) - - def get(self, key: str, default: Any = None) -> Any: - """获取配置值，支持点号分隔的路径""" - keys = key.split('.') - value = self._config - - try: - for k in keys: - value = value[k] - return value - except (KeyError, TypeError): - return default - - def get_section(self, section: str) -> Dict[str, Any]: - """获取整个配置节""" - return self._config.get(section, {}) - - def reload(self) -> None: - """重新加载配置文件""" - self.load() - - def __getitem__(self, key: str) -> Any: - """支持字典式访问""" - return self.get(key) - - def __contains__(self, key: str) -> bool: - """检查配置是否存在""" - return self.get(key) is not None - - @property - def all(self) -> Dict[str, Any]: - """获取所有配置""" - return self._config - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\utils\launch.py` -```python -import sys -from .config import ConfigLoader -import argparse -import os -from module_bank import PythonToSQLite -from typing import Callable - -# 检查是否在 PyInstaller 子进程中 -def is_pyinstaller_child(): - return len(sys.argv) > 1 and ('_child.py' in sys.argv[0] or sys.argv[1].isdigit()) - -def parse_args(): - """安全的参数解析函数""" - if is_pyinstaller_child(): - # PyInstaller 子进程，使用默认参数 - return argparse.Namespace(config="config.toml") - - parser = argparse.ArgumentParser(description="chongming 应用启动脚本") - parser.add_argument("--config", default="config.toml", help="配置文件路径") - return parser.parse_args() - - -args = parse_args() - -def init_module_bank(config): - MODULE_BANK_PATH_LIST = config["production"]["module_system"]["path"] - for module_path in MODULE_BANK_PATH_LIST: - if os.path.exists(module_path) == False: - raise ValueError(f"模块路径不存在: {module_path}") - - packer_list = [PythonToSQLite(module_path) for module_path in MODULE_BANK_PATH_LIST] - for packer in packer_list: - modules = packer.list_modules() - for module in modules: - print(f"{module['module_name']} {'[包]' if module['is_package'] else ''}") - packer.install_importer() - -def launch(run_fun: Callable[[dict, dict], None], run_mode: str): - - config = ConfigLoader(args.config) - - # 获取对应运行模式的配置 - default_config = config.get("default", {}) - run_mode = default_config.get("env", "development") - app_config = config.get(run_mode, {}) - if run_mode == "production": - init_module_bank(config) - - run_fun(app_config, default_config) - -``` - -`./{{ cookiecutter.project_name }}\src\{{ cookiecutter.project_name }}\utils\__init__.py` -```python - -``` - diff --git a/{{ cookiecutter.project_name }}/README.md b/{{ cookiecutter.project_name }}/README.md index 74f7353..c67ae07 100644 --- a/{{ cookiecutter.project_name }}/README.md +++ b/{{ cookiecutter.project_name }}/README.md @@ -11,8 +11,6 @@ uv 环境 ```shell -uv sync -uv pip install -e . uv run serve ``` @@ -49,6 +47,7 @@ build │   ├── plugins.mbank │   └── app.mbank ├── config.toml +├── db.sqlite3 └── {{ cookiecutter.project_name }} # windows {{ cookiecutter.project_name }}.exe ``` diff --git a/{{ cookiecutter.project_name }}/public/config.toml b/{{ cookiecutter.project_name }}/public/config.toml index 11cd228..384c737 100644 --- a/{{ cookiecutter.project_name }}/public/config.toml +++ b/{{ cookiecutter.project_name }}/public/config.toml @@ -1,7 +1,7 @@ [default] app.name = "{{ cookiecutter.project_name }}" app.version = "{{ cookiecutter.project_name }}" -app.description = "基于 FastAPI 和 Tortoise-ORM 的 chongming 打包脚手架" +app.description = "基于 FastAPI 和 Tortoise-ORM 的 {{ cookiecutter.project_name }} 打包脚手架" app.debug = false env = "production" @@ -29,6 +29,13 @@ readonly = false secret_key = "your-secret-key-change-in-production" algorithm = "HS256" access_token_expire_minutes = 30 +refresh_token_expire_days = 7 + +[jwt_cache] +enabled = true +db_path = "./cache" +cleanup_interval_hours = 24 +max_sessions_per_user = 5 [development] env = "development" diff --git a/{{ cookiecutter.project_name }}/public/main.py b/{{ cookiecutter.project_name }}/public/main.py index ab4ef52..cca0081 100644 --- a/{{ cookiecutter.project_name }}/public/main.py +++ b/{{ cookiecutter.project_name }}/public/main.py @@ -1,36 +1,36 @@ # main.py - 修复 reload 问题 #!/usr/bin/env python3 """ -FastAPI + Tortoise-ORM + Chongming 示例 -主入口文件，适配 chongming 打包系统 +FastAPI + Tortoise-ORM + {{ cookiecutter.project_name }} 示例 +主入口文件，适配 {{ cookiecutter.project_name }} 打包系统 """ import sys import os from utils.launch import launch from multiprocessing import freeze_support + def app_run(app_config: dict, default_config: dict): """ 应用运行函数 - 适配模块银行环境 - + Args: app_config: 应用配置 default_config: 默认配置 """ import uvicorn - from app.logger import setup_logging, get_logger # type: ignore + from app.logger import setup_logging, get_logger # type: ignore setup_logging(app_config["logging"]) logger = get_logger("app") - # 获取服务器配置 server_config = app_config.get("server", {}) host = server_config.get("host", "0.0.0.0") port = server_config.get("port", 8000) - + # 打包环境强制禁用重载 - is_frozen = getattr(sys, 'frozen', False) + is_frozen = getattr(sys, "frozen", False) reload = False if is_frozen else server_config.get("reload", False) workers = server_config.get("workers", 4) @@ -46,7 +46,7 @@ def app_run(app_config: dict, default_config: dict): logger.info(f"🔄 重载: {'启用' if reload else '禁用'}") logger.info(f"👥 Workers: {workers}") logger.info("=" * 50) - + # 关键修改：根据 reload 模式选择不同的启动方式 uvicorn.run( @@ -58,17 +58,19 @@ def app_run(app_config: dict, default_config: dict): log_level="debug" if app_config.get("debug", False) else "info", access_log=True, ) - + + def main(): freeze_support() # 如果是打包后的可执行文件，调整路径 - if getattr(sys, 'frozen', False): + if getattr(sys, "frozen", False): exe_dir = os.path.dirname(sys.executable) if exe_dir not in sys.path: sys.path.insert(0, exe_dir) - + # 启动应用 launch(app_run, "development") + if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/{{ cookiecutter.project_name }}/public/utils/launch.py b/{{ cookiecutter.project_name }}/public/utils/launch.py index 4efbbc8..4a05588 100644 --- a/{{ cookiecutter.project_name }}/public/utils/launch.py +++ b/{{ cookiecutter.project_name }}/public/utils/launch.py @@ -5,29 +5,34 @@ import os from module_bank import PythonToSQLite from typing import Callable + # 检查是否在 PyInstaller 子进程中 def is_pyinstaller_child(): - return len(sys.argv) > 1 and ('_child.py' in sys.argv[0] or sys.argv[1].isdigit()) + return len(sys.argv) > 1 and ("_child.py" in sys.argv[0] or sys.argv[1].isdigit()) + def parse_args(): """安全的参数解析函数""" if is_pyinstaller_child(): # PyInstaller 子进程，使用默认参数 return argparse.Namespace(config="config.toml") - - parser = argparse.ArgumentParser(description="chongming 应用启动脚本") + + parser = argparse.ArgumentParser( + description="{{ cookiecutter.project_name }} 应用启动脚本" + ) parser.add_argument("--config", default="config.toml", help="配置文件路径") return parser.parse_args() args = parse_args() + def init_module_bank(config): MODULE_BANK_PATH_LIST = config["production"]["module_system"]["path"] for module_path in MODULE_BANK_PATH_LIST: if os.path.exists(module_path) == False: raise ValueError(f"模块路径不存在: {module_path}") - + packer_list = [PythonToSQLite(module_path) for module_path in MODULE_BANK_PATH_LIST] for packer in packer_list: modules = packer.list_modules() @@ -35,15 +40,16 @@ def init_module_bank(config): print(f"{module['module_name']} {'[包]' if module['is_package'] else ''}") packer.install_importer() + def launch(run_fun: Callable[[dict, dict], None], run_mode: str): config = ConfigLoader(args.config) - + # 获取对应运行模式的配置 default_config = config.get("default", {}) run_mode = default_config.get("env", "development") app_config = config.get(run_mode, {}) if run_mode == "production": init_module_bank(config) - + run_fun(app_config, default_config) diff --git a/{{ cookiecutter.project_name }}/pyproject.toml b/{{ cookiecutter.project_name }}/pyproject.toml index ea6d3ac..6509490 100644 --- a/{{ cookiecutter.project_name }}/pyproject.toml +++ b/{{ cookiecutter.project_name }}/pyproject.toml @@ -1,3 +1,7 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + [project] name = "{{ cookiecutter.project_name }}" version = "0.1.0" @@ -5,33 +9,32 @@ description = "基于 modules-bank 的 fastapi 打包脚手架" readme = "README.md" requires-python = ">=3.8" dependencies = [ - "aerich", "fastapi[uvicorn]", "module-bank", - "passlib", + "passlib[bcrypt]", "pydantic-settings", "pydantic[email]", "sqlite-vfs", "toml", "tortoise-orm", "uvicorn[standard]", + "sqlitedict", + "pyjwt", + "bcrypt==4.0.1", ] +[[tool.uv.index]] +url = "https://pypi.tuna.tsinghua.edu.cn/simple" +default = true + [[tool.uv.index]] url = "https://mirrors.ustc.edu.cn/pypi/web/simple/" -[tool.uv.workspace] -members = [ - ".", -] - -[tool.uv.sources] -{{ cookiecutter.project_name }} = { workspace = true } - [dependency-groups] dev = [ - "{{ cookiecutter.project_name }}", + "aerich", "pyinstaller>=6.18.0", + "requests>=2.32.4", ] [tool.hatch.build.targets.wheel] @@ -39,4 +42,4 @@ packages = ["src/{{ cookiecutter.project_name }}"] [project.scripts] build = "{{ cookiecutter.project_name }}.scripts.build:main" -serve = "{{ cookiecutter.project_name }}:main" +serve = "{{ cookiecutter.project_name }}:main" \ No newline at end of file diff --git a/{{ cookiecutter.project_name }}/requirements.txt b/{{ cookiecutter.project_name }}/requirements.txt index 19ebddc..528d3f4 100644 --- a/{{ cookiecutter.project_name }}/requirements.txt +++ b/{{ cookiecutter.project_name }}/requirements.txt @@ -7,4 +7,10 @@ pydantic[email] sqlite-vfs toml tortoise-orm -uvicorn[standard] \ No newline at end of file +uvicorn[standard] +sqlitedict +pyjwt +bcrypt==4.0.1 +pyinstaller +aerich +requests \ No newline at end of file diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/__init__.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/__init__.py index c75e939..5d9313b 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/__init__.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/__init__.py @@ -1,18 +1,19 @@ # main.py - 修复 reload 问题 #!/usr/bin/env python3 """ -FastAPI + Tortoise-ORM + Chongming 示例 -主入口文件，适配 chongming 打包系统 +FastAPI + Tortoise-ORM + {{ cookiecutter.project_name }} 示例 +主入口文件，适配 {{ cookiecutter.project_name }} 打包系统 """ import sys import os from .utils.launch import launch from multiprocessing import freeze_support + def app_run(app_config: dict, default_config: dict): """ 应用运行函数 - 适配模块银行环境 - + Args: app_config: 应用配置 default_config: 默认配置 @@ -23,14 +24,13 @@ def app_run(app_config: dict, default_config: dict): setup_logging(app_config["logging"]) logger = get_logger("app") - # 获取服务器配置 server_config = app_config.get("server", {}) host = server_config.get("host", "0.0.0.0") port = server_config.get("port", 8000) - + # 打包环境强制禁用重载 - is_frozen = getattr(sys, 'frozen', False) + is_frozen = getattr(sys, "frozen", False) reload = False if is_frozen else server_config.get("reload", False) workers = server_config.get("workers", 4) @@ -46,7 +46,7 @@ def app_run(app_config: dict, default_config: dict): logger.info(f"🔄 重载: {'启用' if reload else '禁用'}") logger.info(f"👥 Workers: {workers}") logger.info("=" * 50) - + # 关键修改：根据 reload 模式选择不同的启动方式 uvicorn.run( @@ -58,17 +58,28 @@ def app_run(app_config: dict, default_config: dict): log_level="debug" if app_config.get("debug", False) else "info", access_log=True, ) - + + def main(): freeze_support() # 如果是打包后的可执行文件，调整路径 - if getattr(sys, 'frozen', False): + if getattr(sys, "frozen", False): exe_dir = os.path.dirname(sys.executable) if exe_dir not in sys.path: sys.path.insert(0, exe_dir) - + # 启动应用 launch(app_run, "development") -if __name__ == "__main__": - main() \ No newline at end of file + +def serve(): + import asyncio + from .scripts import create_admin + + print("🔍 正在执行 create_admin 脚本...") + success = asyncio.run(create_admin.main()) # type: ignore + if success: + print("✅ create_admin 脚本执行成功") + else: + print("⚠️ create_admin 脚本执行失败或跳过（例如，管理员已存在）") + main() diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/__init__.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/__init__.py index 28ffaca..5b7b25b 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/__init__.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/__init__.py @@ -1,8 +1,9 @@ """ API 路由 """ + from fastapi import APIRouter -from .routers import health, items, users +from .routers import health, items, users, auth # 创建主路由 api_router = APIRouter() @@ -11,6 +12,7 @@ api_router = APIRouter() api_router.include_router(health.router, prefix="/health", tags=["health"]) api_router.include_router(items.router, prefix="/items", tags=["items"]) api_router.include_router(users.router, prefix="/users", tags=["users"]) +api_router.include_router(auth.router, prefix="/auth", tags=["authentication"]) # 导出路由 -__all__ = ["api_router"] \ No newline at end of file +__all__ = ["api_router"] diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/dependencies.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/dependencies.py index 815e970..5557e63 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/dependencies.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/dependencies.py @@ -1,11 +1,20 @@ """ API 依赖项 """ + from typing import AsyncGenerator from fastapi import Depends, HTTPException, status +from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from tortoise.exceptions import DBConnectionError, DoesNotExist from tortoise import Tortoise +from ..services.auth_service import get_auth_service +from ..core.jwt_cache import TokenData + +# 安全组件 +security = HTTPBearer() + + async def get_db() -> AsyncGenerator[None, None]: """获取数据库连接依赖""" try: @@ -16,16 +25,61 @@ async def get_db() -> AsyncGenerator[None, None]: yield except DBConnectionError as e: raise HTTPException( - status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail="数据库连接失败" + 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)}" + detail=f"数据库错误: {str(e)}", ) -def get_current_user(): - """获取当前用户（示例）""" - # 这里可以实现 JWT 验证等逻辑 - pass + +async def get_current_user( + credentials: HTTPAuthorizationCredentials = Depends(security), +) -> TokenData: + """ + 获取当前用户依赖项 + + 验证访问令牌并返回用户信息 + """ + auth_service = get_auth_service() + + token = credentials.credentials + token_data = await auth_service.validate_access_token(token) + + if token_data is None: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="无效的访问令牌", + headers={"WWW-Authenticate": "Bearer"}, + ) + + return token_data + + +async def get_current_active_user( + current_user: TokenData = Depends(get_current_user), +) -> TokenData: + """ + 获取当前活跃用户 + + 检查用户是否被禁用 + """ + # 这里可以添加额外的检查，比如用户是否被禁用 + # 由于 TokenData 目前不包含用户状态，需要查询数据库 + # 为简化起见，这里直接返回 + return current_user + + +async def get_current_superuser( + current_user: TokenData = Depends(get_current_user), +) -> TokenData: + """ + 获取当前超级用户 + + 检查用户是否具有超级用户权限 + """ + # 这里需要查询数据库检查用户权限 + # 为简化起见，这里直接返回 + # 实际应用中应该检查用户的权限 + return current_user diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/auth.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/auth.py new file mode 100644 index 0000000..c600e92 --- /dev/null +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/auth.py @@ -0,0 +1,135 @@ +""" +认证路由 - 处理登录、注册、Token 刷新等 +""" + +from fastapi import APIRouter, Depends, HTTPException, status, Request +from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials + +from ...schemas.user import UserCreate, UserLogin, UserResponse +from ...services.auth_service import get_auth_service, TokenResponse +from ...services.user_service import UserService +from ...core.jwt_cache import TokenData +from ..dependencies import get_current_user + +router = APIRouter(tags=["authentication"]) + +# 安全组件 +security = HTTPBearer() + + +@router.post("/register", response_model=UserResponse, summary="用户注册") +async def register(user_data: UserCreate, request: Request): + """用户注册""" + # 检查用户名是否已存在 + existing_user = await UserService.get_user_by_username(user_data.username) + if existing_user: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, detail="用户名已存在" + ) + + # 检查邮箱是否已存在 + existing_email = await UserService.get_user_by_email(user_data.email) + if existing_email: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, detail="邮箱已存在" + ) + + # 创建用户 + user = await UserService.create_user(user_data) + + return user + + +@router.post("/login", response_model=TokenResponse, summary="用户登录") +async def login(user_data: UserLogin, request: Request): + """用户登录并获取令牌""" + auth_service = get_auth_service() + + # 验证用户凭据 + user = await auth_service.authenticate_user(user_data.username, user_data.password) + if not user: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="用户名或密码错误", + headers={"WWW-Authenticate": "Bearer"}, + ) + + # 检查用户是否激活 + if not user.is_active: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, detail="用户已被禁用" + ) + + # 创建令牌 + tokens = await auth_service.create_tokens( + user, request, device_id=user_data.device_id + ) + + return tokens + + +@router.post("/refresh", response_model=dict, summary="刷新访问令牌") +async def refresh_token( + request: Request, credentials: HTTPAuthorizationCredentials = Depends(security) +): + """使用刷新令牌获取新的访问令牌""" + auth_service = get_auth_service() + + refresh_token = credentials.credentials + new_access_token = await auth_service.refresh_access_token(refresh_token, request) + + if not new_access_token: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="无效的刷新令牌", + headers={"WWW-Authenticate": "Bearer"}, + ) + + return { + "access_token": new_access_token, + "token_type": "bearer", + "expires_in": get_auth_service().settings.access_token_expire_minutes * 60, + } + + +@router.post("/logout", summary="用户登出") +async def logout(credentials: HTTPAuthorizationCredentials = Depends(security)): + """用户登出，使令牌失效""" + auth_service = get_auth_service() + + token = credentials.credentials + await auth_service.logout(token) + + return {"message": "登出成功"} + + +@router.post("/logout-all", summary="登出所有设备") +async def logout_all(current_user: TokenData = Depends(get_current_user)): + """使用户在所有设备的令牌失效""" + auth_service = get_auth_service() + await auth_service.logout_all(current_user.user_id) + + return {"message": "所有设备已登出"} + + +@router.get("/sessions", summary="获取用户会话") +async def get_sessions(current_user: TokenData = Depends(get_current_user)): + """获取用户的所有活跃会话""" + auth_service = get_auth_service() + sessions = await auth_service.get_user_sessions(current_user.user_id) + + return { + "user_id": current_user.user_id, + "sessions": sessions, + "count": len(sessions), + } + + +@router.get("/me", response_model=UserResponse, summary="获取当前用户信息") +async def get_current_user_info(current_user: TokenData = Depends(get_current_user)): + """获取当前登录用户的信息""" + user = await UserService.get_user_by_id(int(current_user.user_id)) + if not user: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在") + + return user diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/health.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/health.py index 5e80345..87191b2 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/health.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/health.py @@ -1,44 +1,43 @@ """ 健康检查路由 """ + from fastapi import APIRouter, Depends from ...core.config import get_settings from ..dependencies import get_db router = APIRouter() + @router.get("/", summary="健康检查") -async def health_check(db = Depends(get_db)): +async def health_check(db=Depends(get_db)): """健康检查端点""" settings = get_settings() - + return { "status": "healthy", "service": settings.app_name, "version": settings.app_version, "environment": settings.env, - "timestamp": "now" # 实际使用中可以用 datetime.now() + "timestamp": "now", # 实际使用中可以用 datetime.now() } + @router.get("/ready", summary="就绪检查") async def readiness_check(): """就绪检查端点""" - return { - "status": "ready", - "database": "connected", - "api": "available" - } + return {"status": "ready", "database": "connected", "api": "available"} + @router.get("/info", summary="服务信息") async def service_info(): """服务信息端点""" settings = get_settings() - + return { "name": settings.app_name, "version": settings.app_version, "description": settings.description, "environment": settings.env, "debug": settings.debug, - "database": settings.database_url.split("://")[0] if "://" in settings.database_url else "unknown" - } \ No newline at end of file + } diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/items.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/items.py index 0fe3179..1160669 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/items.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/items.py @@ -1,6 +1,7 @@ """ Item 路由 """ + from typing import List from fastapi import APIRouter, Depends, HTTPException, status from tortoise.exceptions import DoesNotExist @@ -10,16 +11,14 @@ from ..dependencies import get_db router = APIRouter() + @router.get("/", response_model=List[ItemResponse], summary="获取物品列表") -async def read_items( - skip: int = 0, - limit: int = 100, - db=Depends(get_db) -): +async def read_items(skip: int = 0, limit: int = 100, db=Depends(get_db)): """获取物品列表""" items = await Item.all().offset(skip).limit(limit) return items + @router.get("/{item_id}", response_model=ItemResponse, summary="获取物品详情") async def read_item(item_id: int, db=Depends(get_db)): """根据ID获取物品""" @@ -28,11 +27,16 @@ async def read_item(item_id: int, db=Depends(get_db)): return item except DoesNotExist: raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"物品 ID {item_id} 不存在" + status_code=status.HTTP_404_NOT_FOUND, detail=f"物品 ID {item_id} 不存在" ) -@router.post("/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED, summary="创建物品") + +@router.post( + "/", + response_model=ItemResponse, + status_code=status.HTTP_201_CREATED, + summary="创建物品", +) async def create_item(item: ItemCreate, db=Depends(get_db)): """创建新物品""" # 检查名称是否已存在 @@ -40,45 +44,46 @@ async def create_item(item: ItemCreate, db=Depends(get_db)): if existing_item: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"物品名称 '{item.name}' 已存在" + detail=f"物品名称 '{item.name}' 已存在", ) - + # 创建物品 item_dict = item.model_dump() new_item = await Item.create(**item_dict) return new_item + @router.put("/{item_id}", response_model=ItemResponse, summary="更新物品") async def update_item(item_id: int, item_update: ItemUpdate, db=Depends(get_db)): """更新物品信息""" try: item = await Item.get(id=item_id) - + # 更新字段 update_data = item_update.model_dump(exclude_unset=True) - + # 如果更新名称，检查是否与其他物品重名 if "name" in update_data: - existing_item = await Item.filter( - name=update_data["name"] - ).exclude(id=item_id).first() + existing_item = ( + await Item.filter(name=update_data["name"]).exclude(id=item_id).first() + ) if existing_item: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"物品名称 '{update_data['name']}' 已存在" + detail=f"物品名称 '{update_data['name']}' 已存在", ) - + # 应用更新 await item.update_from_dict(update_data) await item.save() - + return item except DoesNotExist: raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"物品 ID {item_id} 不存在" + status_code=status.HTTP_404_NOT_FOUND, detail=f"物品 ID {item_id} 不存在" ) + @router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT, summary="删除物品") async def delete_item(item_id: int, db=Depends(get_db)): """删除物品""" @@ -88,6 +93,5 @@ async def delete_item(item_id: int, db=Depends(get_db)): return None except DoesNotExist: raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"物品 ID {item_id} 不存在" - ) \ No newline at end of file + status_code=status.HTTP_404_NOT_FOUND, detail=f"物品 ID {item_id} 不存在" + ) diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/users.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/users.py index b534eee..df7d647 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/users.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/users.py @@ -1,6 +1,7 @@ """ User 路由 """ + from typing import List from fastapi import APIRouter, Depends, HTTPException, status from tortoise.exceptions import DoesNotExist, IntegrityError @@ -8,78 +9,148 @@ from tortoise.exceptions import DoesNotExist, IntegrityError from ...models.user import User from ...schemas.user import UserCreate, UserUpdate, UserResponse from ...services.user_service import UserService -from ..dependencies import get_db +from ..dependencies import ( + get_db, + get_current_user, + get_current_active_user, + get_current_superuser, +) +from ...core.jwt_cache import TokenData router = APIRouter() -@router.get("/", response_model=List[UserResponse], summary="获取用户列表") -async def read_users( - skip: int = 0, - limit: int = 100, - db=Depends(get_db) -): - """获取用户列表""" + +@router.get( + "/", + response_model=List[UserResponse], + summary="获取用户列表", + dependencies=[Depends(get_current_superuser)], +) # 需要超级用户权限 +async def read_users(skip: int = 0, limit: int = 100, db=Depends(get_db)): + """获取用户列表（需要管理员权限）""" users = await UserService.get_all_users(skip, limit) return users -@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED, summary="创建用户") + +@router.post( + "/", + response_model=UserResponse, + status_code=status.HTTP_201_CREATED, + summary="创建用户", + dependencies=[Depends(get_current_superuser)], +) # 需要超级用户权限 async def create_user(user_data: UserCreate, db=Depends(get_db)): - """创建新用户""" + """创建新用户（需要管理员权限）""" # 检查用户名是否已存在 existing_user = await UserService.get_user_by_username(user_data.username) if existing_user: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"用户名 '{user_data.username}' 已存在" + detail=f"用户名 '{user_data.username}' 已存在", ) - + # 检查邮箱是否已存在 existing_email = await UserService.get_user_by_email(user_data.email) if existing_email: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, - detail=f"邮箱 '{user_data.email}' 已存在" + detail=f"邮箱 '{user_data.email}' 已存在", ) - + # 创建用户 try: user = await UserService.create_user(user_data) return user except IntegrityError: raise HTTPException( - status_code=status.HTTP_400_BAD_REQUEST, - detail="创建用户失败" + status_code=status.HTTP_400_BAD_REQUEST, detail="创建用户失败" ) -@router.get("/{user_id}", response_model=UserResponse, summary="获取用户详情") + +@router.get("/me", response_model=UserResponse, summary="获取当前用户信息") +async def read_current_user(current_user: TokenData = Depends(get_current_active_user)): + """获取当前登录用户的信息""" + user = await UserService.get_user_by_id(int(current_user.user_id)) + if not user: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"用户不存在") + return user + + +@router.get( + "/{user_id}", + response_model=UserResponse, + summary="获取用户详情", + dependencies=[Depends(get_current_superuser)], +) # 需要超级用户权限 async def read_user(user_id: int, db=Depends(get_db)): - """根据ID获取用户""" + """根据ID获取用户（需要管理员权限）""" user = await UserService.get_user_by_id(user_id) if not user: raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"用户 ID {user_id} 不存在" + status_code=status.HTTP_404_NOT_FOUND, detail=f"用户 ID {user_id} 不存在" ) return user -@router.put("/{user_id}", response_model=UserResponse, summary="更新用户") + +@router.put( + "/{user_id}", + response_model=UserResponse, + summary="更新用户", + dependencies=[Depends(get_current_superuser)], +) # 需要超级用户权限 async def update_user(user_id: int, user_update: UserUpdate, db=Depends(get_db)): - """更新用户信息""" + """更新用户信息（需要管理员权限）""" user = await UserService.update_user(user_id, user_update) if not user: raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"用户 ID {user_id} 不存在" + status_code=status.HTTP_404_NOT_FOUND, detail=f"用户 ID {user_id} 不存在" ) return user -@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT, summary="删除用户") + +@router.delete( + "/{user_id}", + status_code=status.HTTP_204_NO_CONTENT, + summary="删除用户", + dependencies=[Depends(get_current_superuser)], +) # 需要超级用户权限 async def delete_user(user_id: int, db=Depends(get_db)): - """删除用户""" + """删除用户（需要管理员权限）""" deleted = await UserService.delete_user(user_id) if not deleted: raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, - detail=f"用户 ID {user_id} 不存在" + status_code=status.HTTP_404_NOT_FOUND, detail=f"用户 ID {user_id} 不存在" ) - return None \ No newline at end of file + return None + + +@router.post( + "/{user_id}/deactivate", + response_model=UserResponse, + summary="停用用户", + dependencies=[Depends(get_current_superuser)], +) # 需要超级用户权限 +async def deactivate_user(user_id: int, db=Depends(get_db)): + """停用用户（需要管理员权限）""" + user = await UserService.deactivate_user(user_id) + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail=f"用户 ID {user_id} 不存在" + ) + return user + + +@router.post( + "/{user_id}/activate", + response_model=UserResponse, + summary="激活用户", + dependencies=[Depends(get_current_superuser)], +) # 需要超级用户权限 +async def activate_user(user_id: int, db=Depends(get_db)): + """激活用户（需要管理员权限）""" + user = await UserService.activate_user(user_id) + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail=f"用户 ID {user_id} 不存在" + ) + return user diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/config.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/config.py index 2290998..9203ea1 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/config.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/config.py @@ -97,6 +97,13 @@ class Settings(BaseSettings): secret_key: str = "your-secret-key-change-in-production" algorithm: str = "HS256" access_token_expire_minutes: int = 30 + refresh_token_expire_days: int = 7 + + # JWT 缓存配置 + jwt_cache_enabled: bool = True + jwt_cache_db_path: str = "jwt_cache.sqlite" + jwt_cache_cleanup_interval_hours: int = 24 + jwt_cache_max_sessions_per_user: int = 5 # 模块系统 module_system_type: str = "local" @@ -176,6 +183,18 @@ class Settings(BaseSettings): if database_config: config_dict["sqlite_config"] = SQLiteConfig(**database_config) + # 处理JWT缓存 + if "jwt_cache" in config_data: + jwt_cache_config = config_data["jwt_cache"] + for key, value in jwt_cache_config.items(): + config_dict[f"jwt_cache_{key}"] = value + + # 处理安全配置 + if "security" in config_dict: + security_config = config_dict["security"] + for key, value in security_config.items(): + config_dict[key] = value + return cls(**config_dict) except Exception as e: diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/database.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/database.py index bf76b44..6595cdd 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/database.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/database.py @@ -31,7 +31,10 @@ async def init_db(run_mode: str): "connections": db_config, "apps": { "models": { - "models": ["{{ cookiecutter.project_name }}.app.models", "aerich.models"], + "models": [ + "{{ cookiecutter.project_name }}.app.models", + "aerich.models", + ], "default_connection": "default", } }, @@ -63,7 +66,11 @@ async def close_db(): def register_db(app, run_mode: str): """将数据库注册到 FastAPI 应用""" - 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, @@ -76,31 +83,26 @@ def register_db(app, run_mode: str): return app -# 在 app/core/database.py 中添加 -from aerich import Command +# 为 Aerich 提供配置 +def get_tortoise_config(): + """获取 Tortoise ORM 配置 - 供 Aerich 使用""" + db_config = getattr(settings, f"{settings.database_type}_config").tortoise_config + + return { + "connections": db_config, + "apps": { + "models": { + "models": [ + "{{ cookiecutter.project_name }}.app.models", + "aerich.models", + ], + "default_connection": "default", + } + }, + "use_tz": False, + "timezone": "UTC", + } -async def run_migrations(): - """运行数据库迁移""" - try: - command = Command( - tortoise_config={ - "connections": getattr( - settings, f"{settings.database_type}_config" - ).tortoise_config, - "apps": { - "models": { - "models": ["app.models", "aerich.models"], - "default_connection": "default", - } - }, - }, - app="models", - ) - - # 检查并应用迁移 - await command.init() - await command.upgrade() - print("✅ 数据库迁移完成") - except Exception as e: - print(f"⚠️ 数据库迁移失败: {e}") +# 保存配置引用 +TORTOISE_CONFIG = get_tortoise_config() diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/jwt_cache.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/jwt_cache.py new file mode 100644 index 0000000..5668715 --- /dev/null +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/jwt_cache.py @@ -0,0 +1,464 @@ +import jwt +import json +from datetime import datetime, timedelta +from typing import Optional, Dict, Any, List +from sqlitedict import SqliteDict +from dataclasses import dataclass, asdict +from pydantic import BaseModel +import hashlib +import threading +from pathlib import Path + +from .config import get_settings + + +@dataclass +class TokenCacheData: + """Token 缓存数据结构""" + + token_hash: str + user_id: str + device_id: Optional[str] = None + user_agent: Optional[str] = None + ip_address: Optional[str] = None + created_at: Optional[str] = None + expires_at: Optional[str] = None + last_used: Optional[str] = None + is_active: bool = True + payload_data: Optional[Dict[str, Any]] = None + token_type: str = "access" # access 或 refresh + + def __post_init__(self): + if self.created_at is None: + self.created_at = datetime.utcnow().isoformat() + if self.last_used is None: + self.last_used = datetime.utcnow().isoformat() + + +class JWTCache: + """JWT Token 缓存管理器""" + + def __init__(self, db_path: Optional[str] = None): + self.settings = get_settings() + + if db_path is None: + db_path = ( + self.settings.jwt_cache_db_path + if hasattr(self.settings, "jwt_cache_db_path") + else "jwt_cache.sqlite" + ) + + db_file_path = Path(db_path) + db_file_path.mkdir(parents=True, exist_ok=True) + + self.lock = threading.RLock() + + # 主 Token 缓存 + self.token_db = SqliteDict( + rf"{db_path}/tokens.cache", + tablename="tokens", + autocommit=True, + encode=self._json_encode, + decode=self._json_decode, + journal_mode="WAL", + ) + + # 黑名单缓存 + self.blacklist_db = SqliteDict( + rf"{db_path}/blacklist.cache", + tablename="blacklist", + autocommit=True, + encode=json.dumps, + decode=json.loads, + journal_mode="WAL", + ) + + # 用户会话索引 + self.user_sessions_db = SqliteDict( + rf"{db_path}/user_sessions.cache", + tablename="user_sessions", + autocommit=True, + encode=json.dumps, + decode=json.loads, + journal_mode="WAL", + ) + + self.secret_key = self.settings.secret_key + self.algorithm = self.settings.algorithm + + def _json_encode(self, obj): + """自定义 JSON 编码器，处理 datetime 对象""" + + def convert_datetime(o): + if isinstance(o, datetime): + return o.isoformat() + raise TypeError( + f"Object of type {o.__class__.__name__} is not JSON serializable" + ) + + return json.dumps(obj, default=convert_datetime) + + def _json_decode(self, s): + """自定义 JSON 解码器""" + return json.loads(s) + + def _hash_token(self, token: str) -> str: + """Token 哈希化（避免存储明文 Token）""" + return hashlib.sha256(token.encode()).hexdigest() + + def create_token( + self, + user_id: str, + user_data: Dict[str, Any], + token_type: str = "access", + device_id: Optional[str] = None, + user_agent: Optional[str] = None, + ip_address: Optional[str] = None, + ) -> str: + """ + 创建并缓存 JWT Token + + Args: + user_id: 用户ID + user_data: 用户数据 + token_type: Token 类型（access 或 refresh） + device_id: 设备ID + user_agent: 用户代理 + ip_address: IP地址 + + Returns: + JWT Token 字符串 + """ + with self.lock: + # 计算过期时间 + if token_type == "access": + expire_minutes = self.settings.access_token_expire_minutes + expires_delta = timedelta(minutes=expire_minutes) + else: + expire_days = ( + self.settings.refresh_token_expire_days + if hasattr(self.settings, "refresh_token_expire_days") + else 7 + ) + expires_delta = timedelta(days=expire_days) + + expire = datetime.utcnow() + expires_delta + + # 构建 payload + payload = { + "sub": user_id, + "exp": expire, + "iat": datetime.utcnow(), + "type": token_type, + "data": user_data, + "device_id": device_id, + "jti": self._generate_jti(), # JWT ID，唯一标识 + } + + # 生成 JWT Token + token = jwt.encode(payload, self.secret_key, algorithm=self.algorithm) + if isinstance(token, bytes): + token = token.decode("utf-8") + + # 创建缓存数据 + token_hash = self._hash_token(token) + cache_data = TokenCacheData( + token_hash=token_hash, + user_id=user_id, + device_id=device_id, + user_agent=user_agent, + ip_address=ip_address, + expires_at=expire.isoformat(), # 转换为字符串 + payload_data=payload, + token_type=token_type, + ) + + # 存储到缓存 + self.token_db[token_hash] = asdict(cache_data) + + # 更新用户会话索引 + self._update_user_session_index(user_id, token_hash, token_type) + + # 检查用户会话数限制 + self._enforce_session_limit(user_id) + + return token + + def _generate_jti(self) -> str: + """生成唯一的 JWT ID""" + import uuid + + return str(uuid.uuid4()) + + def _update_user_session_index( + self, user_id: str, token_hash: str, token_type: str + ): + """更新用户会话索引""" + with self.lock: + key = f"{user_id}:{token_type}" + user_tokens = self.user_sessions_db.get(key, []) + + # 添加新 token + user_tokens.append(token_hash) + + # 只保留最近的一定数量的 token + max_tokens = 100 # 每个用户每种类型最多保留100个 token + if len(user_tokens) > max_tokens: + user_tokens = user_tokens[-max_tokens:] + + self.user_sessions_db[key] = user_tokens + + def _enforce_session_limit(self, user_id: str): + """强制用户会话限制""" + max_sessions = getattr(self.settings, "max_sessions_per_user", 5) + + access_key = f"{user_id}:access" + refresh_key = f"{user_id}:refresh" + + # 检查 access token 数量 + access_tokens = self.user_sessions_db.get(access_key, []) + if len(access_tokens) > max_sessions: + # 删除最早的 token + tokens_to_remove = access_tokens[:-max_sessions] + for token_hash in tokens_to_remove: + if token_hash in self.token_db: + del self.token_db[token_hash] + + # 更新索引 + self.user_sessions_db[access_key] = access_tokens[-max_sessions:] + + # 检查 refresh token 数量 + refresh_tokens = self.user_sessions_db.get(refresh_key, []) + if len(refresh_tokens) > max_sessions: + tokens_to_remove = refresh_tokens[:-max_sessions] + for token_hash in tokens_to_remove: + if token_hash in self.token_db: + del self.token_db[token_hash] + + self.user_sessions_db[refresh_key] = refresh_tokens[-max_sessions:] + + def validate_token( + self, token: str, token_type: str = "access" + ) -> Optional[Dict[str, Any]]: + """ + 验证 JWT Token + + Args: + token: JWT Token + token_type: 期望的 Token 类型 + + Returns: + 验证成功返回用户数据，失败返回 None + """ + try: + # 解码验证 Token + payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm]) + + # 检查 Token 类型 + if payload.get("type") != token_type: + return None + + user_id = payload.get("sub") + jti = payload.get("jti") + + if not user_id or not jti: + return None + + # 检查黑名单 + token_hash = self._hash_token(token) + if self._is_blacklisted(token_hash): + return None + + # 检查缓存 + with self.lock: + cache_data = self.token_db.get(token_hash) + if not cache_data: + return None + + # 检查 Token 是否激活 + if not cache_data.get("is_active", True): + return None + + # 检查是否过期 + expires_at = datetime.fromisoformat(cache_data["expires_at"]) + if datetime.utcnow() > expires_at: + # 标记为不活跃 + cache_data["is_active"] = False + self.token_db[token_hash] = cache_data + return None + + # 更新最后使用时间 + cache_data["last_used"] = datetime.utcnow().isoformat() + self.token_db[token_hash] = cache_data + + return { + "user_id": user_id, + "device_id": cache_data.get("device_id"), + "payload": payload, + "cache_data": cache_data, + } + + except jwt.ExpiredSignatureError: + return None + except jwt.InvalidTokenError: + return None + except Exception: + return None + + def add_to_blacklist(self, token: str, reason: str = "logout"): + """添加 Token 到黑名单""" + token_hash = self._hash_token(token) + expire_time = datetime.utcnow() + timedelta(days=7) # 黑名单保留7天 + + self.blacklist_db[token_hash] = { + "reason": reason, + "blacklisted_at": datetime.utcnow().isoformat(), + "expires_at": expire_time.isoformat(), + } + + def _is_blacklisted(self, token_hash: str) -> bool: + """检查 Token 是否在黑名单中""" + with self.lock: + if token_hash not in self.blacklist_db: + return False + + blacklist_data = self.blacklist_db[token_hash] + expires_at = datetime.fromisoformat(blacklist_data["expires_at"]) + + # 如果黑名单条目已过期，删除它 + if datetime.utcnow() > expires_at: + del self.blacklist_db[token_hash] + return False + + return True + + def invalidate_token(self, token: str): + """使单个 Token 失效""" + with self.lock: + token_hash = self._hash_token(token) + + # 添加到黑名单 + self.add_to_blacklist(token) + + # 标记为不活跃 + if token_hash in self.token_db: + cache_data = self.token_db[token_hash] + cache_data["is_active"] = False + self.token_db[token_hash] = cache_data + + def invalidate_user_tokens(self, user_id: str, token_type: Optional[str] = None): + """使用户的所有 Token 失效""" + with self.lock: + if token_type: + keys = [f"{user_id}:{token_type}"] + else: + keys = [f"{user_id}:access", f"{user_id}:refresh"] + + for key in keys: + token_hashes = self.user_sessions_db.get(key, []) + for token_hash in token_hashes: + # 添加到黑名单 + # 注意：这里我们只有 token_hash，没有原始 token + # 所以只能标记为不活跃 + if token_hash in self.token_db: + cache_data = self.token_db[token_hash] + cache_data["is_active"] = False + self.token_db[token_hash] = cache_data + + # 清空用户会话索引 + self.user_sessions_db[key] = [] + + def get_user_sessions(self, user_id: str) -> List[Dict[str, Any]]: + """获取用户的所有活跃会话""" + with self.lock: + sessions = [] + + for token_type in ["access", "refresh"]: + key = f"{user_id}:{token_type}" + token_hashes = self.user_sessions_db.get(key, []) + + for token_hash in token_hashes: + if token_hash in self.token_db: + cache_data = self.token_db[token_hash] + + # 检查是否活跃且未过期 + if cache_data.get("is_active", True): + expires_at = datetime.fromisoformat( + cache_data["expires_at"] + ) + if datetime.utcnow() <= expires_at: + sessions.append( + { + "token_type": token_type, + "device_id": cache_data.get("device_id"), + "user_agent": cache_data.get("user_agent"), + "ip_address": cache_data.get("ip_address"), + "created_at": cache_data["created_at"], + "last_used": cache_data["last_used"], + "expires_at": cache_data["expires_at"], + } + ) + + return sessions + + def cleanup_expired(self): + """清理过期数据""" + with self.lock: + cleaned_count = 0 + current_time = datetime.utcnow() + + # 清理过期 Token + for token_hash, data in list(self.token_db.items()): + expires_at = datetime.fromisoformat(data["expires_at"]) + if current_time > expires_at: + del self.token_db[token_hash] + cleaned_count += 1 + + # 清理过期黑名单 + for token_hash, data in list(self.blacklist_db.items()): + expires_at = datetime.fromisoformat(data["expires_at"]) + if current_time > expires_at: + del self.blacklist_db[token_hash] + cleaned_count += 1 + + return cleaned_count + + def close(self): + with self.lock: + """关闭所有数据库连接""" + self.token_db.close() + self.blacklist_db.close() + self.user_sessions_db.close() + + +# 全局缓存实例 +_jwt_cache = None + + +def get_jwt_cache() -> JWTCache: + """获取 JWT 缓存实例（单例模式）""" + global _jwt_cache + if _jwt_cache is None: + _jwt_cache = JWTCache() + return _jwt_cache + + +# Token 响应模型 +class TokenResponse(BaseModel): + """Token 响应模型""" + + access_token: str + refresh_token: str + token_type: str = "bearer" + expires_in: int + refresh_expires_in: int + + +# Token 数据模型 +class TokenData(BaseModel): + """Token 数据模型""" + + user_id: str + username: Optional[str] = None + email: Optional[str] = None + scopes: List[str] = [] diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/security.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/security.py deleted file mode 100644 index e69de29..0000000 diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/logger/__init__.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/logger/__init__.py index 01c88ff..f5fed2f 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/logger/__init__.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/logger/__init__.py @@ -1,14 +1,16 @@ import logging from pathlib import Path + class ExcludeWatchFilesFilter(logging.Filter): def filter(self, record): # 如果日志记录来自 watchfiles 模块，则不记录（返回 False） - return not record.name.startswith('watchfiles') + return not record.name.startswith("watchfiles") + def setup_logging(log_config: dict): log_level = getattr(logging, log_config["level"], logging.INFO) - log_format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s' + log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" # 创建文件夹 log_file_path = Path(log_config["file"]) log_file_dir = log_file_path.parent @@ -17,19 +19,19 @@ def setup_logging(log_config: dict): # 设置特定日志级别的过滤 logging.getLogger("watchfiles").setLevel(logging.WARNING) - + # 配置根日志记录器 root_logger = logging.getLogger() root_logger.setLevel(log_level) # 创建格式化器 formatter = logging.Formatter(log_format) - + # 创建文件处理器并添加过滤器 file_handler = logging.FileHandler(log_config["file"], encoding="utf-8") file_handler.setFormatter(formatter) file_handler.addFilter(ExcludeWatchFilesFilter()) - + # 创建控制台处理器并添加过滤器 console_handler = logging.StreamHandler() console_handler.setFormatter(formatter) @@ -38,11 +40,14 @@ def setup_logging(log_config: dict): # 添加处理器到根日志记录器 root_logger.addHandler(file_handler) root_logger.addHandler(console_handler) - - print(f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']}") + + print( + f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']}" + ) + def get_logger(name: str) -> logging.Logger: logger = logging.getLogger(name) # 为每个logger添加过滤器 logger.addFilter(ExcludeWatchFilesFilter()) - return logger \ No newline at end of file + return logger diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/__init__.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/__init__.py index 16d83c0..a976d88 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/__init__.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/__init__.py @@ -1,8 +1,9 @@ """ 数据库模型 """ + from .item import Item from .user import User # 导出所有模型，供 Tortoise-ORM 自动发现 -__all__ = ["Item", "User"] \ No newline at end of file +__all__ = ["Item", "User"] diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/item.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/item.py index 0066d38..b5fa22b 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/item.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/item.py @@ -1,40 +1,44 @@ """ Item 数据模型 """ + from tortoise.models import Model from tortoise import fields from datetime import datetime + class Item(Model): """物品模型""" - + id = fields.IntField(pk=True, description="主键ID") name = fields.CharField(max_length=100, description="名称") description = fields.TextField(null=True, description="描述") - price = fields.DecimalField(max_digits=10, decimal_places=2, default=0.00, description="价格") + price = fields.DecimalField( + max_digits=10, decimal_places=2, default=0.00, description="价格" + ) is_active = fields.BooleanField(default=True, description="是否激活") - + # 关联用户（可选） owner = fields.ForeignKeyField( - "models.User", + "models.User", related_name="items", null=True, on_delete=fields.SET_NULL, - description="所有者" + description="所有者", ) - + # 时间戳 created_at = fields.DatetimeField(auto_now_add=True, description="创建时间") updated_at = fields.DatetimeField(auto_now=True, description="更新时间") - + def __str__(self): return f"Item(id={self.id}, name={self.name}, price={self.price})" - - class Meta: + + class Meta: # type: ignore table = "items" table_description = "物品表" ordering = ["-created_at"] - + class PydanticMeta: # Pydantic 模型配置 - exclude = ["is_active"] \ No newline at end of file + exclude = ["is_active"] diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/user.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/user.py index b49a7c3..baa42d2 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/user.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/user.py @@ -1,36 +1,38 @@ """ User 数据模型 """ + from tortoise.models import Model from tortoise import fields from datetime import datetime from .item import Item + class User(Model): """用户模型""" - + id = fields.IntField(pk=True, description="主键ID") username = fields.CharField(max_length=50, unique=True, description="用户名") email = fields.CharField(max_length=100, unique=True, description="邮箱") hashed_password = fields.CharField(max_length=128, description="哈希密码") is_active = fields.BooleanField(default=True, description="是否激活") is_superuser = fields.BooleanField(default=False, description="是否超级用户") - + # 时间戳 created_at = fields.DatetimeField(auto_now_add=True, description="创建时间") updated_at = fields.DatetimeField(auto_now=True, description="更新时间") - + # 反向关系 items: fields.ReverseRelation["Item"] - + def __str__(self): return f"User(id={self.id}, username={self.username})" - - class Meta: + + class Meta: # type: ignore table = "users" table_description = "用户表" ordering = ["-created_at"] - + class PydanticMeta: # Pydantic 模型配置 - exclude = ["hashed_password"] \ No newline at end of file + exclude = ["hashed_password"] diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/__init__.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/__init__.py index acad57d..6c582a3 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/__init__.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/__init__.py @@ -1,11 +1,7 @@ """ Pydantic 模型 """ + from .item import ItemCreate, ItemUpdate, ItemResponse, ItemInDB -__all__ = [ - "ItemCreate", - "ItemUpdate", - "ItemResponse", - "ItemInDB" -] \ No newline at end of file +__all__ = ["ItemCreate", "ItemUpdate", "ItemResponse", "ItemInDB"] diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/item.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/item.py index 9916676..870e82c 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/item.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/item.py @@ -1,37 +1,50 @@ """ Item Pydantic 模型 """ + from pydantic import BaseModel, Field from datetime import datetime from typing import Optional from decimal import Decimal + class ItemBase(BaseModel): """Item 基础模型""" + name: str = Field(..., min_length=1, max_length=100, description="物品名称") description: Optional[str] = Field(None, description="物品描述") price: Decimal = Field(Decimal(0.00), ge=0, description="物品价格") + class ItemCreate(ItemBase): """创建 Item 模型""" + pass + class ItemUpdate(BaseModel): """更新 Item 模型""" - name: Optional[str] = Field(None, min_length=1, max_length=100, description="物品名称") + + name: Optional[str] = Field( + None, min_length=1, max_length=100, description="物品名称" + ) description: Optional[str] = Field(None, description="物品描述") price: Optional[Decimal] = Field(None, ge=0, description="物品价格") + class ItemInDB(ItemBase): """数据库中的 Item 模型""" + id: int is_active: bool created_at: datetime updated_at: datetime - + class Config: from_attributes = True + class ItemResponse(ItemInDB): """Item 响应模型""" - pass \ No newline at end of file + + pass diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/user.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/user.py index 436466e..aa49a0d 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/user.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/user.py @@ -3,26 +3,39 @@ from pydantic import BaseModel, EmailStr from datetime import datetime from typing import Optional + class UserBase(BaseModel): username: str email: EmailStr + class UserCreate(UserBase): password: str + class UserUpdate(BaseModel): username: Optional[str] = None email: Optional[EmailStr] = None password: Optional[str] = None + class UserInDB(UserBase): id: int is_active: bool + is_superuser: bool created_at: datetime updated_at: datetime - + class Config: from_attributes = True + class UserResponse(UserInDB): pass + + +# 新增登录模型 +class UserLogin(BaseModel): + username: str + password: str + device_id: Optional[str] = None # 设备ID，用于多设备登录 diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/auth_service.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/auth_service.py new file mode 100644 index 0000000..6ffcc31 --- /dev/null +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/auth_service.py @@ -0,0 +1,163 @@ +""" +认证服务 - 处理用户认证和 Token 管理 +""" + +from typing import Optional, Dict, Any +from datetime import datetime, timedelta +from fastapi import HTTPException, status, Request +from passlib.context import CryptContext + +from ..models.user import User +from ..schemas.user import UserCreate, UserLogin +from .user_service import UserService +from ..core.jwt_cache import get_jwt_cache, TokenResponse, TokenData +from ..core.config import get_settings + + +class AuthService: + """认证服务""" + + def __init__(self): + self.jwt_cache = get_jwt_cache() + self.settings = get_settings() + self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") + + async def authenticate_user(self, username: str, password: str) -> Optional[User]: + """验证用户凭据""" + user = await UserService.get_user_by_username(username) + if not user: + return None + + if not UserService.verify_password(password, user.hashed_password): + return None + + return user + + async def create_tokens( + self, + user: User, + request: Optional[Request] = None, + device_id: Optional[str] = None, + ) -> TokenResponse: + """为用户创建访问和刷新令牌""" + + # 获取用户数据 + user_data = { + "username": user.username, + "email": user.email, + "is_superuser": user.is_superuser, + } + + # 获取设备信息 + user_agent = None + ip_address = None + if request: + user_agent = request.headers.get("user-agent") + ip_address = request.client.host if request.client else None + + # 创建访问令牌 + access_token = self.jwt_cache.create_token( + user_id=str(user.id), + user_data=user_data, + token_type="access", + device_id=device_id, + user_agent=user_agent, + ip_address=ip_address, + ) + + # 创建刷新令牌 + refresh_token = self.jwt_cache.create_token( + user_id=str(user.id), + user_data=user_data, + token_type="refresh", + device_id=device_id, + user_agent=user_agent, + ip_address=ip_address, + ) + + return TokenResponse( + access_token=access_token, + refresh_token=refresh_token, + expires_in=self.settings.access_token_expire_minutes * 60, + refresh_expires_in=getattr(self.settings, "refresh_token_expire_days", 7) + * 24 + * 60 + * 60, + ) + + async def validate_access_token(self, token: str) -> Optional[TokenData]: + """验证访问令牌""" + validation = self.jwt_cache.validate_token(token, token_type="access") + if not validation: + return None + + return TokenData( + user_id=validation["user_id"], + username=validation["payload"]["data"].get("username"), + email=validation["payload"]["data"].get("email"), + scopes=[], # 可以根据需要添加权限范围 + ) + + async def refresh_access_token( + self, refresh_token: str, request: Optional[Request] = None + ) -> Optional[str]: + """使用刷新令牌获取新的访问令牌""" + validation = self.jwt_cache.validate_token(refresh_token, token_type="refresh") + if not validation: + return None + + user_id = validation["user_id"] + + # 获取用户 + user = await UserService.get_user_by_id(int(user_id)) + if not user: + return None + + # 创建新的访问令牌 + user_data = { + "username": user.username, + "email": user.email, + "is_superuser": user.is_superuser, + } + + device_id = validation.get("device_id") + user_agent = None + ip_address = None + if request: + user_agent = request.headers.get("user-agent") + ip_address = request.client.host if request.client else None + + new_access_token = self.jwt_cache.create_token( + user_id=user_id, + user_data=user_data, + token_type="access", + device_id=device_id, + user_agent=user_agent, + ip_address=ip_address, + ) + + return new_access_token + + async def logout(self, token: str): + """用户登出 - 使令牌失效""" + self.jwt_cache.invalidate_token(token) + + async def logout_all(self, user_id: str): + """使用户的所有令牌失效""" + self.jwt_cache.invalidate_user_tokens(str(user_id)) + + async def get_user_sessions(self, user_id: str): + """获取用户的活跃会话""" + return self.jwt_cache.get_user_sessions(str(user_id)) + + +# 全局认证服务实例 +_auth_service = None + + +def get_auth_service() -> AuthService: + """获取认证服务实例（单例模式）""" + global _auth_service + if _auth_service is None: + _auth_service = AuthService() + return _auth_service diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/item_service.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/item_service.py index 7d0995d..80294a7 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/item_service.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/item_service.py @@ -3,45 +3,43 @@ from typing import Optional, List from ..models.item import Item from ..schemas.item import ItemCreate, ItemUpdate + class ItemService: @staticmethod async def get_all_items( - skip: int = 0, - limit: int = 100, - owner_id: Optional[int] = None + skip: int = 0, limit: int = 100, owner_id: Optional[int] = None ) -> List[Item]: query = Item.all() if owner_id: query = query.filter(owner_id=owner_id) return await query.offset(skip).limit(limit) - + @staticmethod async def get_item_by_id(item_id: int) -> Optional[Item]: return await Item.filter(id=item_id).first().prefetch_related("owner") - + @staticmethod - async def create_item(item_data: ItemCreate, owner_id: Optional[int] = None) -> Item: - item = await Item.create( - **item_data.model_dump(), - owner_id=owner_id - ) + async def create_item( + item_data: ItemCreate, owner_id: Optional[int] = None + ) -> Item: + item = await Item.create(**item_data.model_dump(), owner_id=owner_id) return item - + @staticmethod async def update_item(item_id: int, item_data: ItemUpdate) -> Optional[Item]: item = await ItemService.get_item_by_id(item_id) if not item: return None - + update_data = item_data.model_dump(exclude_unset=True) await item.update_from_dict(update_data) await item.save() return item - + @staticmethod async def delete_item(item_id: int) -> bool: item = await ItemService.get_item_by_id(item_id) if item: await item.delete() return True - return False \ No newline at end of file + return False diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/user_service.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/user_service.py index 7b39504..46754e7 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/user_service.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/user_service.py @@ -6,58 +6,59 @@ from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") + class UserService: @staticmethod def get_password_hash(password: str) -> str: return pwd_context.hash(password) - + @staticmethod def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) - + @staticmethod async def get_all_users(skip: int = 0, limit: int = 100) -> List[User]: return await User.all().offset(skip).limit(limit) - + @staticmethod async def get_user_by_id(user_id: int) -> Optional[User]: return await User.filter(id=user_id).first() - + @staticmethod async def get_user_by_username(username: str) -> Optional[User]: return await User.filter(username=username).first() - + @staticmethod async def get_user_by_email(email: str) -> Optional[User]: return await User.filter(email=email).first() - + @staticmethod async def create_user(user_data: UserCreate) -> User: hashed_password = UserService.get_password_hash(user_data.password) user = await User.create( username=user_data.username, email=user_data.email, - hashed_password=hashed_password + hashed_password=hashed_password, ) return user - + @staticmethod async def update_user(user_id: int, user_data: UserUpdate) -> Optional[User]: user = await UserService.get_user_by_id(user_id) if not user: return None - + update_data = user_data.model_dump(exclude_unset=True) - + if "password" in update_data: update_data["hashed_password"] = UserService.get_password_hash( update_data.pop("password") ) - + await user.update_from_dict(update_data) await user.save() return user - + @staticmethod async def delete_user(user_id: int) -> bool: user = await UserService.get_user_by_id(user_id) @@ -65,3 +66,23 @@ class UserService: await user.delete() return True return False + + @staticmethod + async def deactivate_user(user_id: int) -> Optional[User]: + """停用用户""" + user = await UserService.get_user_by_id(user_id) + if user: + user.is_active = False + await user.save() + return user + return None + + @staticmethod + async def activate_user(user_id: int) -> Optional[User]: + """激活用户""" + user = await UserService.get_user_by_id(user_id) + if user: + user.is_active = True + await user.save() + return user + return None diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/build.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/build.py index e950285..d4a72af 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/build.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/build.py @@ -5,67 +5,130 @@ import shutil import subprocess import sys + +def create_admin(): + try: + cmd = [ + sys.executable, + "../src/{{ cookiecutter.project_name }}/scripts/create_admin.py", + ] + print(f"正在运行 create_admin.py: {' '.join(cmd)}") + result = subprocess.run(cmd, cwd="./build", capture_output=True, text=True) + if result.returncode == 0: + print("创建管理员用户成功") + print(result.stdout) + else: + print("创建管理员用户失败") + print("错误信息:", result.stderr) + except Exception as e: + print("创建管理员用户失败") + + def run_pyinstaller(): """运行 PyInstaller 打包成可执行文件""" try: # 构建 PyInstaller 命令 cmd = [ - sys.executable, "-m", "PyInstaller", + sys.executable, + "-m", + "PyInstaller", "--onefile", - "--name", "{{ cookiecutter.project_name }}", - "--distpath", ".", - "--workpath", "./build_temp", - "--specpath", "./specs", + "--name", + "{{ cookiecutter.project_name }}", + "--distpath", + ".", + "--workpath", + "./build_temp", + "--specpath", + "./specs", "--clean", # 隐藏导入 - "--hidden-import", "passlib", - "--hidden-import", "passlib.context", - "--hidden-import", "passlib.handlers.bcrypt", - "--hidden-import", "fastapi", - "--hidden-import", "fastapi.middleware.cors", - "--hidden-import", "pydantic_settings", - "--hidden-import", "tortoise", - "--hidden-import", "tortoise.contrib.fastapi", - "--hidden-import", "aerich", - "--hidden-import", "uvicorn", + "--hidden-import", + "passlib", + "--hidden-import", + "passlib.context", + "--hidden-import", + "passlib.handlers.bcrypt", + "--hidden-import", + "fastapi", + "--hidden-import", + "fastapi.middleware.cors", + "--hidden-import", + "pydantic_settings", + "--hidden-import", + "tortoise", + "--hidden-import", + "tortoise.contrib.fastapi", + "--hidden-import", + "aerich", + "--hidden-import", + "uvicorn", + "--hidden-import", + "jwt", + "--hidden-import", + "PyJWT", + "--hidden-import", + "sqlitedict", # 复制包元数据 - 解决 importlib.metadata 问题 - "--copy-metadata", "tortoise-orm", - "--copy-metadata", "aerich", - "--copy-metadata", "fastapi", - "--copy-metadata", "uvicorn", - "--copy-metadata", "pydantic", - "--copy-metadata", "pydantic-settings", - "--copy-metadata", "module-bank", - "--copy-metadata", "passlib", - "--copy-metadata", "sqlite-vfs", - "--copy-metadata", "toml", + "--copy-metadata", + "tortoise-orm", + "--copy-metadata", + "aerich", + "--copy-metadata", + "fastapi", + "--copy-metadata", + "uvicorn", + "--copy-metadata", + "pydantic", + "--copy-metadata", + "pydantic-settings", + "--copy-metadata", + "module-bank", + "--copy-metadata", + "passlib", + "--copy-metadata", + "sqlite-vfs", + "--copy-metadata", + "toml", + "--copy-metadata", + "PyJWT", + "--copy-metadata", + "sqlitedict", # 收集整个包 - "--collect-all", "tortoise-orm", - "--collect-all", "aerich", + "--collect-all", + "tortoise-orm", + "--collect-all", + "aerich", + "--collect-all", + "PyJWT", + "--collect-all", + "sqlitedict", # 禁用多进程优化 # "--multiprocessing-fork", # 明确指定使用 fork 模式 # 添加数据文件 # "--add-data", "resources:resources", # 其他选项 - "--icon", "../../public/{{ cookiecutter.project_name }}.ico", - "main.py" + "--icon", + "../../public/{{ cookiecutter.project_name }}.ico", + "main.py", ] - + print(f"正在运行 PyInstaller: {' '.join(cmd)}") result = subprocess.run(cmd, cwd="./build", capture_output=True, text=True) - + if result.returncode == 0: print("PyInstaller 打包成功！") print(result.stdout) else: print("PyInstaller 打包失败！") print("错误信息:", result.stderr) - + except FileNotFoundError: print("错误：未找到 PyInstaller，请先安装 PyInstaller：pip install pyinstaller") except Exception as e: print(f"PyInstaller 执行出错: {e}") + def main(): # 打包模块 resources_path = Path("resources") @@ -74,18 +137,20 @@ def main(): app_packer = PythonToSQLite("./resources/app.mbank") app_packer.pack_directory("src/{{ cookiecutter.project_name }}/app", "app") app_packer.verify_package_structure() - # app_packer.delete_source_code(None) + app_packer.delete_source_code(None) plugin_packer = PythonToSQLite("./resources/plugins.mbank") - plugin_packer.pack_directory("src/{{ cookiecutter.project_name }}/plugins", "plugins") + plugin_packer.pack_directory( + "src/{{ cookiecutter.project_name }}/plugins", "plugins" + ) plugin_packer.verify_package_structure() - # plugin_packer.delete_source_code(None) + plugin_packer.delete_source_code(None) # 创建 build 目录 build_path = Path("build") if build_path.exists() and build_path.is_dir(): shutil.rmtree("build") - + os.makedirs("build", exist_ok=True) # 复制 utils 文件夹到 build 文件夹 @@ -110,10 +175,13 @@ def main(): shutil.copy("public/config.toml", "build") # 复制 utils 文件夹到 build 文件夹 shutil.copytree("public/utils", "build/utils") - + # 运行 PyInstaller run_pyinstaller() + # 创建管理员 + create_admin() + # 清理临时文件 shutil.rmtree(r"build/build_temp", ignore_errors=True) shutil.rmtree(r"build/specs", ignore_errors=True) @@ -123,5 +191,6 @@ def main(): print("构建完成") + if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/create_admin.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/create_admin.py new file mode 100644 index 0000000..9c4a659 --- /dev/null +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/create_admin.py @@ -0,0 +1,90 @@ +#!/usr/bin/env python3 +""" +创建管理员用户脚本 +""" +import asyncio +import sys +from pathlib import Path + + +# 添加项目根目录到路径 +project_root = Path(__file__).parent.parent.parent.parent +sys.path.insert(0, str(project_root)) + + +async def main(): + try: + # 首先初始化 Tortoise-ORM + from tortoise import Tortoise + from {{ cookiecutter.project_name }}.app.core.config import get_settings + + settings = get_settings() + + db_config = getattr( + settings, f"{settings.database_type}_config" + ).tortoise_config + + # Tortoise-ORM 配置 + db_config = { + "connections": db_config, + "apps": { + "models": { + "models": ["{{ cookiecutter.project_name }}.app.models", "aerich.models"], + "default_connection": "default", + } + }, + "use_tz": False, + "timezone": "UTC", + } + + print("🔧 初始化 Tortoise-ORM...") + await Tortoise.init(db_config) + + # 生成数据库表 + print("🗃️ 生成数据库表结构...") + await Tortoise.generate_schemas() + + # 导入模型和服务 + from {{ cookiecutter.project_name }}.app.models.user import User + from {{ cookiecutter.project_name }}.app.services.user_service import UserService + from {{ cookiecutter.project_name }}.app.schemas.user import UserCreate + + # 检查是否已存在管理员 + admin = await User.filter(username="admin").first() + if admin: + print("✅ 管理员用户已存在") + print(f" 用户名: {admin.username}") + print(f" 邮箱: {admin.email}") + await Tortoise.close_connections() + return + + # 创建管理员用户 + user_data = UserCreate( + username="admin", + email="admin@example.com", + password="admin123", # 生产环境应使用复杂密码 + ) + + admin = await UserService.create_user(user_data) + admin.is_superuser = True + await admin.save() + + print(f"✅ 管理员用户创建成功") + print(f" 用户名: {admin.username}") + print(f" 邮箱: {admin.email}") + print(f" 密码: {user_data.password}") + print("\n⚠️ 警告：请在生产环境中修改默认密码！") + + # 关闭数据库连接 + await Tortoise.close_connections() + print("✅ 数据库连接已关闭") + + except Exception as e: + print(f"❌ 创建管理员失败: {e}") + import traceback + + traceback.print_exc() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/migrate.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/migrate.py deleted file mode 100644 index ed91c5b..0000000 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/migrate.py +++ /dev/null @@ -1,61 +0,0 @@ -#!/usr/bin/env python3 -""" -数据库迁移脚本 -""" -import asyncio -import sys -import os - -# 添加项目根目录到路径 -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -async def main(): - from app.core.config import get_settings - from aerich import Command - - settings = get_settings() - - # Aerich 配置 - TORTOISE_ORM = { - "connections": { - "default": settings.database_url - }, - "apps": { - "models": { - "models": ["app.models", "aerich.models"], - "default_connection": "default", - } - }, - "use_tz": False, - "timezone": "UTC", - } - - command = Command(tortoise_config=TORTOISE_ORM, app="models") - - # 初始化 Aerich（首次运行） - if not os.path.exists("./migrations"): - print("🔧 初始化数据库迁移...") - await command.init() - print("✅ 迁移初始化完成") - - # 生成迁移文件 - print("📝 生成迁移文件...") - try: - migration_name = sys.argv[1] if len(sys.argv) > 1 else "update" - await command.migrate(name=migration_name) - print(f"✅ 迁移文件生成成功: {migration_name}") - except Exception as e: - print(f"⚠️ 生成迁移文件失败: {e}") - # 尝试升级 - pass - - # 应用迁移 - print("🔄 应用数据库迁移...") - try: - await command.upgrade() - print("✅ 迁移应用成功") - except Exception as e: - print(f"❌ 迁移应用失败: {e}") - -if __name__ == "__main__": - asyncio.run(main()) \ No newline at end of file diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/config.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/config.py index b1ca8a4..abdc2c1 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/config.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/config.py @@ -5,78 +5,78 @@ import os from pathlib import Path from typing import Any, Dict, Optional + class ConfigLoader: """TOML配置文件加载器""" - + def __init__(self, config_path: Optional[str] = None): """ 初始化配置加载器 - + Args: config_path: 配置文件路径，如果为None则自动查找 """ if config_path is None: # 自动查找配置文件 config_path = self._find_config_file() - + self.config_path = Path(config_path) self._config: Dict[str, Any] = {} self.load() - + def _find_config_file(self) -> str: """自动查找配置文件""" possible_paths = [ "config.toml", "config/config.toml", "~/.config/myapp/config.toml", - "/etc/myapp/config.toml" + "/etc/myapp/config.toml", ] - + for path in possible_paths: expanded_path = os.path.expanduser(path) if os.path.exists(expanded_path): return expanded_path - + raise FileNotFoundError("未找到配置文件") - + def load(self) -> None: """加载配置文件""" if not self.config_path.exists(): raise FileNotFoundError(f"配置文件不存在: {self.config_path}") - - with open(self.config_path, 'r', encoding='utf-8') as f: + + with open(self.config_path, "r", encoding="utf-8") as f: self._config = toml.load(f) - + def get(self, key: str, default: Any = None) -> Any: """获取配置值，支持点号分隔的路径""" - keys = key.split('.') + keys = key.split(".") value = self._config - + try: for k in keys: value = value[k] return value except (KeyError, TypeError): return default - + def get_section(self, section: str) -> Dict[str, Any]: """获取整个配置节""" return self._config.get(section, {}) - + def reload(self) -> None: """重新加载配置文件""" self.load() - + def __getitem__(self, key: str) -> Any: """支持字典式访问""" return self.get(key) - + def __contains__(self, key: str) -> bool: """检查配置是否存在""" return self.get(key) is not None - + @property def all(self) -> Dict[str, Any]: """获取所有配置""" return self._config - \ No newline at end of file diff --git a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/launch.py b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/launch.py index 4efbbc8..4a05588 100644 --- a/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/launch.py +++ b/{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/launch.py @@ -5,29 +5,34 @@ import os from module_bank import PythonToSQLite from typing import Callable + # 检查是否在 PyInstaller 子进程中 def is_pyinstaller_child(): - return len(sys.argv) > 1 and ('_child.py' in sys.argv[0] or sys.argv[1].isdigit()) + return len(sys.argv) > 1 and ("_child.py" in sys.argv[0] or sys.argv[1].isdigit()) + def parse_args(): """安全的参数解析函数""" if is_pyinstaller_child(): # PyInstaller 子进程，使用默认参数 return argparse.Namespace(config="config.toml") - - parser = argparse.ArgumentParser(description="chongming 应用启动脚本") + + parser = argparse.ArgumentParser( + description="{{ cookiecutter.project_name }} 应用启动脚本" + ) parser.add_argument("--config", default="config.toml", help="配置文件路径") return parser.parse_args() args = parse_args() + def init_module_bank(config): MODULE_BANK_PATH_LIST = config["production"]["module_system"]["path"] for module_path in MODULE_BANK_PATH_LIST: if os.path.exists(module_path) == False: raise ValueError(f"模块路径不存在: {module_path}") - + packer_list = [PythonToSQLite(module_path) for module_path in MODULE_BANK_PATH_LIST] for packer in packer_list: modules = packer.list_modules() @@ -35,15 +40,16 @@ def init_module_bank(config): print(f"{module['module_name']} {'[包]' if module['is_package'] else ''}") packer.install_importer() + def launch(run_fun: Callable[[dict, dict], None], run_mode: str): config = ConfigLoader(args.config) - + # 获取对应运行模式的配置 default_config = config.get("default", {}) run_mode = default_config.get("env", "development") app_config = config.get(run_mode, {}) if run_mode == "production": init_module_bank(config) - + run_fun(app_config, default_config)