chakcy/chongming_template_fastapi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modify 更换orm框架

Browse Source
This commit is contained in:
chakcy
2026-03-06 14:30:10 +08:00
parent c7266e47d0
commit a4b7c82db0
43 changed files with 769 additions and 1649 deletions
Download Patch File Download Diff File Expand all files Collapse all files
{{ cookiecutter.project_name }}/config.toml
+19 -21
View File
@@ -3,32 +3,30 @@ app.name = "{{ cookiecutter.project_name }}"
app.version = "{{ cookiecutter.project_name }}"
app.description = "基于 FastAPI 和 Tortoise-ORM 的 chongming 打包脚手架"
app.debug = false
# env = "production"
env = "development"
[database]
type = "sqlite"
url = "sqlite+aiosqlite:///./database.db"
generate_schemas = true # 生产环境使用迁移工具
[database.sqlite]
engine = "tortoise.backends.sqlite"
file_path = "db.sqlite3"
journal_mode = "WAL"
busy_timeout = 10000
maxsize = 10
minsize = 1
cache_size = -2000
synchronous = "NORMAL"
temp_store = "MEMORY"
mmap_size = 1073741824
foreign_keys = true
auto_vacuum = "INCREMENTAL"
max_cached_statements = 100
readonly = false
echo = true
pool_size = 20
max_overflow = 10
pool_timeout = 30
pool_recycle = 300
pool_pre_ping = true
connect_args.timeout = 15
connect_args.check_same_thread = false
[security]
secret_key = "your-secret-key-change-in-production"
algorithm = "HS256"
access_token_expire_minutes = 30
refresh_token_expire_days = 7
jwt_cache_db_path = "jwt_cache"
max_sessions_per_user = 5
[development]
env = "development"
@@ -38,11 +36,11 @@ debug = true
host = "127.0.0.1"
port = 8000
reload = true
# workers = 4
[development.logging]
level = "DEBUG"
file = "logs/dev.log"
format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
[development.cors]
allow_origins = ["*"]
@@ -58,14 +56,17 @@ debug = false
host = "0.0.0.0"
port = 8000
reload = false
workers = 1
workers = 4
[production.logging]
level = "INFO"
file = "logs/prod.log"
format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
max_size = 10485760 # 10MB
backup_count = 5 # 最多保存 10 个日志文件
[production.cors]
allow_origins = ["https://yourdomain.com", "http://localhos:8000", "http://127.0.0.1:8000"]
allow_origins = ["https://yourdomain.com", "http://localhost:8000", "http://127.0.0.1:8000"]
allow_credentials = true
allow_methods = ["GET", "POST", "PUT", "DELETE"]
allow_headers = ["Authorization", "Content-Type"]
@@ -77,6 +78,3 @@ path = ["resources/app.mbank", "resources/plugins.mbank"]
[production.file_system]
type = "svfs"
path = "resources/application.vfs"
[cache]
default_ttl = 300
{{ cookiecutter.project_name }}/public/config.toml
+17 -28
View File
@@ -1,40 +1,31 @@
[default]
app.name = "{{ cookiecutter.project_name }}"
app.version = "{{ cookiecutter.project_name }}"
app.description = "基于 FastAPI 和 Tortoise-ORM 的 {{ cookiecutter.project_name }} 打包脚手架"
app.version = "0.1.0"
app.description = "基于 FastAPI 的 chongming 打包脚手架"
app.debug = false
env = "production"
[database]
type = "sqlite"
url = "sqlite+aiosqlite:///./database.db"
generate_schemas = true # 生产环境使用迁移工具
[database.sqlite]
engine = "tortoise.backends.sqlite"
file_path = "db.sqlite3"
journal_mode = "WAL"
busy_timeout = 10000
maxsize = 10
minsize = 1
cache_size = -2000
synchronous = "NORMAL"
temp_store = "MEMORY"
mmap_size = 1073741824
foreign_keys = true
auto_vacuum = "INCREMENTAL"
max_cached_statements = 100
readonly = false
echo = true
pool_size = 20
max_overflow = 10
pool_timeout = 30
pool_recycle = 300
pool_pre_ping = true
connect_args.timeout = 15
connect_args.check_same_thread = false
[security]
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
jwt_cache_db_path = "jwt_cache"
max_sessions_per_user = 5
[development]
@@ -43,13 +34,13 @@ debug = true
[development.server]
host = "127.0.0.1"
port = 8000
port = 8001
reload = true
# workers = 4
[development.logging]
level = "DEBUG"
file = "logs/dev.log"
format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
[development.cors]
allow_origins = ["*"]
@@ -65,14 +56,15 @@ debug = false
host = "0.0.0.0"
port = 8000
reload = false
workers = 1
workers = 4
[production.logging]
level = "INFO"
file = "logs/prod.log"
format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
[production.cors]
allow_origins = ["https://yourdomain.com", "http://localhos:8000", "http://127.0.0.1:8000"]
allow_origins = ["https://yourdomain.com", "http://localhost:8000", "http://127.0.0.1:8000"]
allow_credentials = true
allow_methods = ["GET", "POST", "PUT", "DELETE"]
allow_headers = ["Authorization", "Content-Type"]
@@ -84,6 +76,3 @@ path = ["resources/app.mbank", "resources/plugins.mbank"]
[production.file_system]
type = "svfs"
path = "resources/application.vfs"
[cache]
default_ttl = 300
{{ cookiecutter.project_name }}/public/main.py
+4 -13
View File
@@ -5,9 +5,9 @@ FastAPI + Tortoise-ORM + {{ cookiecutter.project_name }} 示例
主入口文件，适配 {{ cookiecutter.project_name }} 打包系统
"""
import sys
import os
from utils.launch import launch
from multiprocessing import freeze_support
import server
def app_run(app_config: dict, default_config: dict):
@@ -19,7 +19,7 @@ def app_run(app_config: dict, default_config: dict):
default_config: 默认配置
"""
import uvicorn
from app.logger import setup_logging, get_logger # type: ignore
from app.core.logger import setup_logging, get_logger # type: ignore
setup_logging(app_config["logging"])
logger = get_logger("app")
@@ -47,10 +47,8 @@ def app_run(app_config: dict, default_config: dict):
logger.info(f"👥 Workers: {workers}")
logger.info("=" * 50)
# 关键修改：根据 reload 模式选择不同的启动方式
uvicorn.run(
"app:app",
"server:app",
host=host,
port=port,
reload=reload,
@@ -62,14 +60,7 @@ def app_run(app_config: dict, default_config: dict):
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")
launch(app_run, "production")
if __name__ == "__main__":
{{ cookiecutter.project_name }}/public/server.py
+19
View File
@@ -0,0 +1,19 @@
# server.py
import sys
from utils.launch import init_module_bank
from utils.config import ConfigLoader
config = ConfigLoader("config.toml")
init_module_bank(config)
# 从数据库导入真正的 app 模块
try:
import app # type: ignore # 此时会从 SQLite 数据库加载
app_instance = app.app # 获取 FastAPI 实例
except ImportError as e:
print(f"无法从数据库导入 app 模块: {e}")
sys.exit(1)
# 暴露 ASGI 应用变量，供 uvicorn 使用
app = app_instance
{{ cookiecutter.project_name }}/pyproject.toml
+11 -11
View File
@@ -9,20 +9,21 @@ description = "基于 modules-bank 的 fastapi 打包脚手架"
readme = "README.md"
requires-python = ">=3.8"
dependencies = [
"aiosqlite",
"bcrypt==4.0.1",
"fastapi[uvicorn]",
"module-bank",
"passlib[bcrypt]",
"pydantic-settings",
"pydantic[email]",
"sqlite-vfs",
"toml",
"tortoise-orm",
"uvicorn[standard]",
"sqlitedict",
"pyjwt",
"bcrypt==4.0.1",
"sqlite-vfs",
"sqlitedict",
"sqlmodel[aiosqlite]",
"tomli",
"typing-extensions",
"uvicorn",
]
[[tool.uv.index]]
url = "https://pypi.tuna.tsinghua.edu.cn/simple"
default = true
@@ -32,9 +33,8 @@ url = "https://mirrors.ustc.edu.cn/pypi/web/simple/"
[dependency-groups]
dev = [
"aerich",
"pyinstaller>=6.18.0",
"requests>=2.32.4",
"pyarmor",
"pyinstaller",
]
[tool.hatch.build.targets.wheel]
{{ cookiecutter.project_name }}/src/plugins/__init__.py
+2
View File
@@ -0,0 +1,2 @@
def hello():
print("Hello World")
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/__init__.py
+15 -81
View File
@@ -1,85 +1,19 @@
# main.py - 修复 reload 问题
#!/usr/bin/env python3
"""
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
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")
import uvicorn
from .app.core.config import load_config
from .app.core.logger import setup_logging, get_logger
def serve():
import asyncio
from .scripts import create_admin
config = load_config()
env = config["default"]["env"]
server_config = config[env]["server"]
log_config = config[env]["logging"]
setup_logging(log_config)
logget = get_logger("app")
logget.info(f"启动服务: {server_config}")
logget.info(f"启动日志: {log_config}")
uvicorn.run("{{ cookiecutter.project_name }}.app:app", **server_config)
print("🔍 正在执行 create_admin 脚本...")
success = asyncio.run(create_admin.main()) # type: ignore
if success:
print("✅ create_admin 脚本执行成功")
else:
print("⚠️ create_admin 脚本执行失败或跳过（例如，管理员已存在）")
main()
if __name__ == "__main__":
serve()
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/__init__.py
+67 -60
View File
@@ -1,90 +1,99 @@
"""
FastAPI 应用包
"""
__version__ = "1.0.0"
from contextlib import asynccontextmanager
"""
FastAPI 应用实例
"""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
from . import logger as _logger
from sqlmodel import SQLModel, text
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.exc import OperationalError
from .core.config import get_settings
from .core.database import register_db, init_db, close_db
from .core.config import get_config
from .core.jwt_cache import get_jwt_cache
from .core.logger import get_logger
from .api import api_router
# 获取配置
settings = get_settings()
logger = _logger.get_logger("app")
config = get_config()
logger = get_logger("app")
try:
env = config["default"]["env"]
except KeyError: # pragma: no cover
raise ValueError("配置不存在")
# 生命周期管理
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
应用生命周期管理
from plugins import hello
注意：在生产环境的模块银行中，数据库初始化可能失败
这里使用 try-except 确保应用能够正常启动
"""
logger.info(f"🚀 启动 {settings.app_name} v{settings.app_version}")
logger.info(f"🌍 环境: {settings.env}")
hello()
database = config["database"]
database_type = database["type"]
database_config = database.get(database_type, None)
if database_config is None:
raise ValueError("配置不存在")
# 初始化数据库（生产环境可能跳过）
# 启动时：创建引擎和会话工厂，初始化数据库表
engine = create_async_engine(database["url"], **database_config)
async_session_maker = async_sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
# 将资源存储到 app.state
app.state.engine = engine
app.state.async_session_maker = async_session_maker
# 创建表（异步方式）
if env == "development":
async with engine.begin() as conn:
if database_type == "sqlite":
await conn.execute(text("PRAGMA journal_mode=WAL"))
await conn.execute(text("PRAGMA synchronous=NORMAL"))
try:
await init_db(settings.env)
except Exception as e:
logger.warning(f"⚠️ 数据库初始化失败: {e}")
logger.warning("💡 提示：应用将继续运行，但数据库功能可能受限")
await conn.run_sync(
lambda conn: SQLModel.metadata.create_all(
bind=conn, checkfirst=True
)
)
except OperationalError as e:
if "already exists" in str(e):
raise
yield
yield # 应用运行期间
# 关闭数据库连接
try:
await close_db()
except Exception as e:
logger.warning(f"⚠️ 数据库关闭失败: {e}")
# 关闭时：释放引擎
await engine.dispose()
get_jwt_cache().close()
logger.info("👋 关闭应用")
# 创建 FastAPI 应用实例
cors = config[env].get("cors", {})
# 创建 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,
title=config["default"]["app"]["name"],
version=config["default"]["app"]["version"],
description=config["default"]["app"]["description"],
docs_url="/docs" if config[env]["debug"] else None,
redoc_url="/redoc" if config[env]["debug"] else None,
openapi_url="/openapi.json" if config[env]["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.add_middleware(CORSMiddleware, **cors)
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,
"message": f"欢迎使用 {config["default"]["app"]["name"]}",
"version": config["default"]["app"]["version"],
"docs": "/docs" if config[env]["debug"] else None,
"health": "/api/v1/health",
}
# 全局异常处理器
@app.exception_handler(Exception)
async def global_exception_handler(request, exc):
@@ -92,7 +101,5 @@ async def global_exception_handler(request, exc):
logger.error(f"全局异常: {exc}")
return {
"error": "内部服务器错误",
"detail": str(exc) if settings.debug else "请查看服务器日志"
"detail": str(exc) if config[env]["debug"] else "请查看服务器日志",
}
logger.info("✅ FastAPI 应用初始化完成")
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/__init__.py
+4 -4
View File
@@ -3,16 +3,16 @@ API 路由
"""
from fastapi import APIRouter
from .routers import health, items, users, auth
from .routers import todo, auth
# 创建主路由
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(todo.router, prefix="/todos", tags=["todo"])
api_router.include_router(auth.router, prefix="/auth", tags=["authentication"])
# 导出路由
__all__ = ["api_router"]
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/dependencies.py
-85
View File
@@ -1,85 +0,0 @@
"""
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:
# 检查数据库连接
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)}",
)
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
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/deps.py
+31
View File
@@ -0,0 +1,31 @@
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from ..service.auth import get_auth_service
from ..core.jwt_cache import TokenData
security = HTTPBearer()
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
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/__init__.py
-3
View File
@@ -1,3 +0,0 @@
"""
路由模块
"""
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/auth.py
+70 -71
View File
@@ -1,111 +1,107 @@
"""
认证路由 - 处理登录、注册、Token 刷新等
"""
from fastapi import APIRouter, Depends, HTTPException, status, Request
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from ...core.database import get_session
from ...core.jwt_cache import get_jwt_cache, TokenData
from ...core.security import verify_password
from ...service.user import UserService
from ...service.auth import get_auth_service
from ...model.user import UserCreate, UserRead, UserLogin
from ..deps import get_current_user
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="用户名已存在"
@router.post("/register", response_model=UserRead)
async def register(
user_data: UserCreate,
session: AsyncSession = Depends(get_session),
):
print(user_data.password)
existing = await UserService.get_user_by_email(user_data.email, session)
if existing:
raise HTTPException(status_code=400, detail="邮箱已被注册")
user = await UserService.create_user(
session=session,
email=user_data.email,
password=user_data.password,
username=user_data.username,
full_name=user_data.full_name,
)
# 检查邮箱是否已存在
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)
@router.post("/login")
async def login(
user_data: UserLogin,
request: Request,
session: AsyncSession = Depends(get_session),
):
user = await UserService.get_user_by_email(user_data.email, session)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名或密码错误",
headers={"WWW-Authenticate": "Bearer"},
)
# 检查用户是否激活
raise HTTPException(status_code=400, detail="邮箱或密码错误")
if not verify_password(user_data.password, user.hashed_password):
raise HTTPException(status_code=400, detail="邮箱或密码错误")
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="用户已被禁用"
raise HTTPException(status_code=400, detail="用户被禁用")
cache = get_jwt_cache()
user_data_dict = {
"sub": str(user.id),
"email": user.email,
"username": user.username,
}
token = cache.create_token(
user_id=str(user.id),
user_data=user_data_dict,
token_type="access",
device_id=None,
user_agent=request.headers.get("user-agent") if request else None,
ip_address=request.client.host if request else None, # type: ignore
)
# 创建令牌
tokens = await auth_service.create_tokens(
user, request, device_id=user_data.device_id
)
return tokens
return {"access_token": token, "token_type": "bearer"}
@router.post("/refresh", response_model=dict, summary="刷新访问令牌")
async def refresh_token(
request: Request, credentials: HTTPAuthorizationCredentials = Depends(security)
request: Request,
credentials: HTTPAuthorizationCredentials = Depends(security),
session: AsyncSession = Depends(get_session),
):
"""使用刷新令牌获取新的访问令牌"""
auth_service = get_auth_service()
refresh_token = credentials.credentials
new_access_token = await auth_service.refresh_access_token(refresh_token, request)
new_access_token = await auth_service.refresh_access_token(
refresh_token, session, 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,
}
return {"access_token": new_access_token, "token_type": "bearer"}
@router.post("/logout", summary="用户登出")
async def logout(credentials: HTTPAuthorizationCredentials = Depends(security)):
"""用户登出，使令牌失效"""
@router.post("/logout")
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="登出所有设备")
@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)
@@ -125,10 +121,13 @@ async def get_sessions(current_user: TokenData = Depends(get_current_user)):
}
@router.get("/me", response_model=UserResponse, summary="获取当前用户信息")
async def get_current_user_info(current_user: TokenData = Depends(get_current_user)):
@router.get("/me", response_model=UserRead, summary="获取当前用户信息")
async def get_current_user_info(
session: AsyncSession = Depends(get_session),
current_user: TokenData = Depends(get_current_user),
):
"""获取当前登录用户的信息"""
user = await UserService.get_user_by_id(int(current_user.user_id))
user = await UserService.get_user_by_id(int(current_user.user_id), session)
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="用户不存在")
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/health.py
-43
View File
@@ -1,43 +0,0 @@
"""
健康检查路由
"""
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,
}
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/items.py
-97
View File
@@ -1,97 +0,0 @@
"""
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/todo.py
+59
View File
@@ -0,0 +1,59 @@
from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from ...model.todo import TodoCreate, TodoUpdate, TodoRead
from ...core.database import get_session
from ...service.todo import TodoService
from ...model.user import User
from ..deps import get_current_user
router = APIRouter()
# -------------------- 路由 --------------------
@router.post("/", response_model=TodoRead, dependencies=[Depends(get_current_user)])
async def create_todo(
todo: TodoCreate,
session: AsyncSession = Depends(get_session),
):
return await TodoService.create_todo(todo, session)
@router.get(
"/", response_model=List[TodoRead], dependencies=[Depends(get_current_user)]
)
async def read_todos(
offset: int = 0, limit: int = 100, session: AsyncSession = Depends(get_session)
):
return await TodoService.get_todos(offset, limit, session)
@router.get(
"/{todo_id}", response_model=TodoRead, dependencies=[Depends(get_current_user)]
)
async def read_todo(todo_id: int, session: AsyncSession = Depends(get_session)):
todo = await TodoService.get_todo_by_id(todo_id, session)
if not todo:
raise HTTPException(status_code=404, detail="Todo not found")
return todo
@router.patch(
"/{todo_id}", response_model=TodoRead, dependencies=[Depends(get_current_user)]
)
async def update_todo(
todo_id: int, todo_update: TodoUpdate, session: AsyncSession = Depends(get_session)
):
todo = await TodoService.update_todo(todo_id, todo_update, session)
return todo
@router.delete("/{todo_id}", dependencies=[Depends(get_current_user)])
async def delete_todo(todo_id: int, session: AsyncSession = Depends(get_session)):
is_deleted = await TodoService.delete_todo(todo_id, session)
if not is_deleted:
raise HTTPException(status_code=404, detail="Todo not found")
return {"ok": True}
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/api/routers/users.py
-156
View File
@@ -1,156 +0,0 @@
"""
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,
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="获取用户列表",
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="创建用户",
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}' 已存在",
)
# 检查邮箱是否已存在
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("/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获取用户（需要管理员权限）"""
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="更新用户",
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} 不存在"
)
return user
@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} 不存在"
)
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
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/__init__.py
-3
View File
@@ -1,3 +0,0 @@
"""
核心模块
"""
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/config.py
+114 -207
View File
@@ -1,217 +1,124 @@
"""
配置管理模块 - 简化版本
基于 Pydantic Settings 和 TOML 文件
"""
from typing import Optional, List, Dict, Any
from pydantic_settings import BaseSettings
from pydantic import model_validator
import toml
import sys
from pathlib import Path
from typing import TypedDict, Literal, Optional, cast
try:
from ...utils.launch import args
except:
from utils.launch import args # type: ignore
if sys.version_info >= (3, 11):
import tomllib
from typing import Required, NotRequired
else:
import tomli as tomllib
from typing_extensions import Required, NotRequired
# 设置 env 为 development 或 production
class SQLiteConfig(BaseSettings):
"""SQLite 配置"""
engine: str = "tortoise.backends.sqlite"
file_path: str = "db.sqlite3"
journal_mode: str = "WAL"
busy_timeout: int = 10000
maxsize: int = 10
minsize: int = 1
cache_size: int = -2000
synchronous: str = "NORMAL"
temp_store: str = "MEMORY"
mmap_size: int = 1073741824
foreign_keys: bool = True
auto_vacuum: str = "INCREMENTAL"
max_cached_statements: int = 100
readonly: bool = False
@property
def tortoise_config(self) -> Dict[str, Any]:
return {
"default": {
"engine": self.engine,
"credentials": {
"file_path": self.file_path,
"journal_mode": self.journal_mode,
"busy_timeout": self.busy_timeout,
"maxsize": self.maxsize,
"minsize": self.minsize,
"cache_size": self.cache_size,
"synchronous": self.synchronous,
"temp_store": self.temp_store,
"mmap_size": self.mmap_size,
"foreign_keys": self.foreign_keys,
"auto_vacuum": self.auto_vacuum,
"max_cached_statements": self.max_cached_statements,
"readonly": self.readonly,
},
}
}
class DefaultApp(TypedDict):
name: str
version: str
description: str
debug: bool
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 = 1 # 开发环境默认1，避免与 reload 冲突
database_type: str = "sqlite"
# SQLite 配置
sqlite_config: SQLiteConfig = SQLiteConfig()
# 环境配置
env: str = "development"
debug: bool = False
# 数据库配置
generate_schemas: bool = False
# 日志配置
logging_level: str = "INFO"
logging_file: Optional[str] = None
# CORS 配置
cors_allow_origins: List[str] = ["*"]
cors_allow_credentials: bool = True
cors_allow_methods: List[str] = ["*"]
cors_allow_headers: List[str] = ["*"]
# 安全配置
secret_key: str = "your-secret-key-change-in-production"
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
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"
module_system_path: List[str] = ["plugins", "app"]
class Config:
env_file = ".env"
extra = "ignore"
@model_validator(mode="after")
def validate_settings(self):
"""验证和调整设置"""
# 如果启用 reload，workers 必须为 1
if self.reload and self.workers != 1:
print(f"⚠️ reload 启用，将 workers 从 {self.workers} 调整为 1")
self.workers = 1
return self
@classmethod
def from_toml(cls, config_path: Optional[str] = None) -> "Settings":
"""从 TOML 文件加载配置 - 简化版本"""
# 1. 确定配置文件路径
if config_path is None:
config_path = "config.toml"
config_file = Path(config_path)
if not config_file.exists():
print(f"⚠️ 配置文件不存在: {config_path}，使用默认配置")
return cls()
try:
# 2. 加载 TOML 文件
config_data = toml.load(config_file)
print(f"📄 加载配置文件: {config_path}")
# 3. 确定环境
default_config = config_data.get("default", {})
env = default_config.get("env", "development")
print(f"🌍 环境: {env}")
# 4. 获取环境特定配置
env_config = config_data.get(env, {})
# 5. 构建配置字典
config_dict = {}
# 5.1 处理默认配置
for key, value in default_config.items():
if key.startswith("app."):
config_dict[key.replace("app.", "app_")] = value
elif key.startswith("server."):
config_dict[key.replace("server.", "")] = value
elif key == "env":
config_dict["env"] = value
# 5.2 处理环境特定配置
for key, value in env_config.items():
if key.startswith("server."):
config_dict[key.replace("server.", "")] = value
elif key == "debug":
config_dict["debug"] = value
# 5.3 处理 database 配置
database_config = {}
if "database" in env_config:
database_type = env_config["database"].get("type", "sqlite")
database_config = env_config["database"].get(database_type, {})
config_dict["database_type"] = database_type
# 5.4 处理 generate_schemas
if "database" in env_config:
db_config = env_config["database"]
if isinstance(db_config, dict) and "generate_schemas" in db_config:
config_dict["generate_schemas"] = db_config["generate_schemas"]
if database_type == "sqlite":
if database_config:
config_dict["sqlite_config"] = SQLiteConfig(**database_config)
# 处理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:
print(f"❌ 加载配置失败: {e}")
import traceback
traceback.print_exc()
return cls()
class Default(TypedDict):
app: DefaultApp
env: Literal["production", "development"]
# 全局配置实例
_settings = None
class SqliteConnectionArgs(TypedDict, total=False):
timeout: int
check_same_thread: bool
def get_settings() -> Settings:
"""获取配置实例（单例模式）"""
global _settings
if _settings is None:
_settings = Settings.from_toml(args.config)
return _settings
class DatabaseSqlite(TypedDict, total=False):
echo: bool
pool_size: int
max_overflow: int
pool_timeout: int
pool_recycle: int
pool_pre_ping: bool
connect_args: SqliteConnectionArgs
class Database(TypedDict, total=False):
type: Required[Literal["sqlite", "mysql", "postgresql"]]
url: Required[str]
generate_schemas: bool
sqlite: DatabaseSqlite
mysql: dict
postgresql: dict
class Security(TypedDict):
secret_key: str
algorithm: Literal["HS256"]
access_token_expire_minutes: int
jwt_cache_db_path: str
refresh_token_expire_days: int
max_sessions_per_user: int
class Server(TypedDict, total=False):
host: Required[str]
port: Required[int]
reload: bool
workers: int
class Logging(TypedDict):
level: str
file: str
format: str
max_size: int
backup_count: int
class Cors(TypedDict, total=False):
allow_origins: list[str]
allow_credentials: bool
allow_methods: list[str]
allow_headers: list[str]
class ModuleSystem(TypedDict, total=False):
type: str
path: list[str]
class FileSystem(TypedDict, total=False):
type: str
path: list[str]
class Env(TypedDict, total=False):
debug: Required[bool]
server: Required[Server]
logging: Required[Logging]
cors: Cors
module_system: ModuleSystem
file_system: FileSystem
class Config(TypedDict):
default: Default
database: Database
security: Security
development: Env
production: Env
def load_config(config_path: Path = Path("config.toml")) -> Config:
"""从 TOML 文件加载配置"""
if not config_path.exists():
raise FileNotFoundError(f"配置文件不存在: {config_path}")
with open(config_path, "rb") as f:
return cast(Config, tomllib.load(f))
_config: Optional[Config] = None
def get_config(config_path: str = "config.toml") -> Config:
global _config
if _config is None:
_config = load_config(Path(config_path))
return _config
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/database.py
+8 -105
View File
@@ -1,108 +1,11 @@
"""
Tortoise-ORM 数据库配置
"""
from typing import AsyncGenerator
from tortoise import Tortoise
from tortoise.contrib.fastapi import register_tortoise
from .config import get_settings
from fastapi import Request
from sqlalchemy.ext.asyncio import AsyncSession
settings = get_settings()
async def init_db(run_mode: str):
"""初始化数据库连接"""
db_config = getattr(settings, f"{settings.database_type}_config").tortoise_config
if run_mode == "production":
db_config = {
"connections": db_config,
"apps": {
"models": {
"models": ["app.models", "aerich.models"],
"default_connection": "default",
}
},
"use_tz": False,
"timezone": "UTC",
}
elif run_mode == "development":
db_config = {
"connections": db_config,
"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"✅ 数据库连接成功: {db_config}")
async def close_db():
"""关闭数据库连接"""
await Tortoise.close_connections()
print("✅ 数据库连接已关闭")
def register_db(app, run_mode: str):
"""将数据库注册到 FastAPI 应用"""
modules_path = (
"app.models"
if run_mode == "production"
else "{{ cookiecutter.project_name }}.app.models"
)
file_path = getattr(settings, f"{settings.database_type}_config").file_path
register_tortoise(
app,
db_url=f"sqlite://{file_path}",
modules={"models": [modules_path]},
generate_schemas=settings.generate_schemas,
add_exception_handlers=True,
)
return app
# 为 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",
}
# 保存配置引用
TORTOISE_CONFIG = get_tortoise_config()
# 获取数据库会话的依赖
async def get_session(request: Request) -> AsyncGenerator[AsyncSession, None]:
async_session_maker = request.app.state.async_session_maker
async with async_session_maker() as session:
yield session
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/jwt_cache.py
+13 -18
View File
@@ -1,15 +1,17 @@
import jwt
import json
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from typing import Optional, Dict, Any, List, TypedDict
from sqlitedict import SqliteDict
from dataclasses import dataclass, asdict
from pydantic import BaseModel
from dataclasses import dataclass, asdict
# from pydantic import BaseModel
import hashlib
import threading
from pathlib import Path
from .config import get_settings
from .config import get_config
@dataclass
@@ -39,14 +41,10 @@ class JWTCache:
"""JWT Token 缓存管理器"""
def __init__(self, db_path: Optional[str] = None):
self.settings = get_settings()
self.config = get_config()
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_path = self.config["security"]["jwt_cache_db_path"]
db_file_path = Path(db_path)
db_file_path.mkdir(parents=True, exist_ok=True)
@@ -83,8 +81,8 @@ class JWTCache:
journal_mode="WAL",
)
self.secret_key = self.settings.secret_key
self.algorithm = self.settings.algorithm
self.secret_key = self.config["security"]["secret_key"]
self.algorithm = self.config["security"]["algorithm"]
def _json_encode(self, obj):
"""自定义 JSON 编码器，处理 datetime 对象"""
@@ -132,14 +130,10 @@ class JWTCache:
with self.lock:
# 计算过期时间
if token_type == "access":
expire_minutes = self.settings.access_token_expire_minutes
expire_minutes = self.config["security"]["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
)
expire_days = self.config["security"]["refresh_token_expire_days"]
expires_delta = timedelta(days=expire_days)
expire = datetime.utcnow() + expires_delta
@@ -210,7 +204,7 @@ class JWTCache:
def _enforce_session_limit(self, user_id: str):
"""强制用户会话限制"""
max_sessions = getattr(self.settings, "max_sessions_per_user", 5)
max_sessions = self.config["security"]["max_sessions_per_user"]
access_key = f"{user_id}:access"
refresh_key = f"{user_id}:refresh"
@@ -255,6 +249,7 @@ class JWTCache:
payload = jwt.decode(token, self.secret_key, algorithms=[self.algorithm])
# 检查 Token 类型
print(f"payload: {payload}")
if payload.get("type") != token_type:
return None
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/logger/__init__.py → {{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/logger.py
+16 -5
View File
@@ -1,5 +1,7 @@
import logging
from pathlib import Path
from .config import Logging
from logging.handlers import RotatingFileHandler
class ExcludeWatchFilesFilter(logging.Filter):
@@ -8,9 +10,9 @@ class ExcludeWatchFilesFilter(logging.Filter):
return not record.name.startswith("watchfiles")
def setup_logging(log_config: dict):
def setup_logging(log_config: Logging):
log_level = getattr(logging, log_config["level"], logging.INFO)
log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
log_format = log_config["format"]
# 创建文件夹
log_file_path = Path(log_config["file"])
log_file_dir = log_file_path.parent
@@ -27,8 +29,16 @@ def setup_logging(log_config: dict):
# 创建格式化器
formatter = logging.Formatter(log_format)
# 创建文件处理器并添加过滤器
file_handler = logging.FileHandler(log_config["file"], encoding="utf-8")
# 创建轮转文件处理器 - 限制文件大小为 10MB，保留 5 个备份文件
max_bytes = log_config.get("max_size", 10 * 1024 * 1024) # 默认 10MB
backup_count = log_config.get("backup_count", 5) # 默认保留 5 个备份
file_handler = RotatingFileHandler(
log_config["file"],
encoding="utf-8",
maxBytes=max_bytes,
backupCount=backup_count,
)
file_handler.setFormatter(formatter)
file_handler.addFilter(ExcludeWatchFilesFilter())
@@ -42,7 +52,8 @@ def setup_logging(log_config: dict):
root_logger.addHandler(console_handler)
print(
f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']}"
f"日志系统已配置 - 级别: {logging.getLevelName(log_level)}, 文件: {log_config['file']} "
f"(最大：{max_bytes / 1024 / 1024:.1f}MB, 备份：{backup_count})"
)
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/core/security.py
+14
View File
@@ -0,0 +1,14 @@
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""验证密码"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""密码哈希"""
return pwd_context.hash(password)
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/__init__.py → {{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/model/__init__.py
View File
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/model/todo.py
+28
View File
@@ -0,0 +1,28 @@
from typing import Optional
from sqlmodel import SQLModel, Field
# 定义模型
class TodoBase(SQLModel):
title: str
description: Optional[str] = None
completed: bool = False
class Todo(TodoBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
class TodoCreate(TodoBase):
pass
class TodoRead(TodoBase):
id: int
class TodoUpdate(SQLModel):
title: Optional[str] = None
description: Optional[str] = None
completed: Optional[bool] = None
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/model/user.py
+36
View File
@@ -0,0 +1,36 @@
from typing import Optional
from sqlmodel import SQLModel, Field
from pydantic import BaseModel
class UserBase(SQLModel):
email: str = Field(unique=True, index=True)
username: Optional[str] = Field(default=None, max_length=50)
full_name: Optional[str] = Field(default=None, max_length=100)
is_active: bool = Field(default=True)
is_superuser: bool = Field(default=False)
created_at: Optional[str] = Field(default=None) # 可自动填充，此处略
class User(UserBase, table=True):
"""用户模型"""
id: Optional[int] = Field(default=None, primary_key=True)
hashed_password: str
class UserCreate(UserBase):
"""用户创建模型"""
password: str
class UserRead(UserBase):
"""用户读取模型"""
id: int
class UserLogin(BaseModel):
email: str
password: str
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/__init__.py
-9
View File
@@ -1,9 +0,0 @@
"""
数据库模型
"""
from .item import Item
from .user import User
# 导出所有模型，供 Tortoise-ORM 自动发现
__all__ = ["Item", "User"]
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/item.py
-44
View File
@@ -1,44 +0,0 @@
"""
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: # type: ignore
table = "items"
table_description = "物品表"
ordering = ["-created_at"]
class PydanticMeta:
# Pydantic 模型配置
exclude = ["is_active"]
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/models/user.py
-38
View File
@@ -1,38 +0,0 @@
"""
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: # type: ignore
table = "users"
table_description = "用户表"
ordering = ["-created_at"]
class PydanticMeta:
# Pydantic 模型配置
exclude = ["hashed_password"]
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/__init__.py
-7
View File
@@ -1,7 +0,0 @@
"""
Pydantic 模型
"""
from .item import ItemCreate, ItemUpdate, ItemResponse, ItemInDB
__all__ = ["ItemCreate", "ItemUpdate", "ItemResponse", "ItemInDB"]
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/schemas/item.py
-50
View File
@@ -1,50 +0,0 @@
"""
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
-41
View File
@@ -1,41 +0,0 @@
# 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
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，用于多设备登录
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/__init__.py → {{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/service/__init__.py
View File
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/auth_service.py → {{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/service/auth.py
+19 -16
View File
@@ -2,16 +2,15 @@
认证服务 - 处理用户认证和 Token 管理
"""
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
from fastapi import HTTPException, status, Request
from typing import Optional
from fastapi import Request
from passlib.context import CryptContext
from sqlalchemy.ext.asyncio import AsyncSession
from ..models.user import User
from ..schemas.user import UserCreate, UserLogin
from .user_service import UserService
from ..model.user import User
from .user import UserService
from ..core.jwt_cache import get_jwt_cache, TokenResponse, TokenData
from ..core.config import get_settings
from ..core.config import get_config
class AuthService:
@@ -19,16 +18,17 @@ class AuthService:
def __init__(self):
self.jwt_cache = get_jwt_cache()
self.settings = get_settings()
self.settings = get_config()
self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
async def authenticate_user(self, username: str, password: str) -> Optional[User]:
async def authenticate_user(
self, email: str, password: str, session: AsyncSession
) -> Optional[User]:
"""验证用户凭据"""
user = await UserService.get_user_by_username(username)
user = await UserService.get_user_by_email(email, session)
if not user:
return None
if not UserService.verify_password(password, user.hashed_password):
if not self.pwd_context.verify(password, user.hashed_password):
return None
return user
@@ -78,7 +78,7 @@ class AuthService:
return TokenResponse(
access_token=access_token,
refresh_token=refresh_token,
expires_in=self.settings.access_token_expire_minutes * 60,
expires_in=self.settings["security"]["access_token_expire_minutes"] * 60,
refresh_expires_in=getattr(self.settings, "refresh_token_expire_days", 7)
* 24
* 60
@@ -99,17 +99,20 @@ class AuthService:
)
async def refresh_access_token(
self, refresh_token: str, request: Optional[Request] = None
self,
refresh_token: str,
session: AsyncSession,
request: Optional[Request] = None,
) -> Optional[str]:
"""使用刷新令牌获取新的访问令牌"""
validation = self.jwt_cache.validate_token(refresh_token, token_type="refresh")
validation = self.jwt_cache.validate_token(refresh_token, token_type="access")
if not validation:
return None
user_id = validation["user_id"]
# 获取用户
user = await UserService.get_user_by_id(int(user_id))
user = await UserService.get_user_by_id(int(user_id), session)
if not user:
return None
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/service/todo.py
+65
View File
@@ -0,0 +1,65 @@
from typing import Sequence, Optional
from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select
from ..model.todo import TodoCreate, Todo, TodoUpdate
class TodoService:
@staticmethod
async def create_todo(todo: TodoCreate, session: AsyncSession) -> Todo:
db_todo = Todo.model_validate(todo)
session.add(db_todo)
await session.commit()
await session.refresh(db_todo)
return db_todo
@staticmethod
async def get_todos(
offset: int,
limit: int,
session: AsyncSession,
) -> Sequence[Todo]:
result = await session.execute(select(Todo).offset(offset).limit(limit))
todos = result.scalars().all()
return todos
@staticmethod
async def get_todo_by_id(
todo_id: int,
session: AsyncSession,
) -> Optional[Todo]:
todo = await session.get(Todo, todo_id)
return todo
@staticmethod
async def update_todo(
todo_id: int,
todo_update: TodoUpdate,
session: AsyncSession,
) -> Optional[Todo]:
todo = await session.get(Todo, todo_id)
if not todo:
return None
update_data = todo_update.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(todo, key, value)
session.add(todo)
await session.commit()
await session.refresh(todo)
return todo
@staticmethod
async def delete_todo(
todo_id: int,
session: AsyncSession,
) -> bool:
todo = await session.get(Todo, todo_id)
if not todo:
return False
await session.delete(todo)
await session.commit()
return True
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/service/user.py
+28
View File
@@ -0,0 +1,28 @@
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional
from ..model.user import User
from ..core.security import get_password_hash
class UserService:
@staticmethod
async def get_user_by_email(email: str, session: AsyncSession) -> Optional[User]:
result = await session.execute(select(User).where(User.email == email)) # type: ignore
return result.scalar_one_or_none()
@staticmethod
async def get_user_by_id(user_id: int, session: AsyncSession) -> Optional[User]:
return await session.get(User, user_id)
@staticmethod
async def create_user(
session: AsyncSession, email: str, password: str, **kwargs
) -> User:
hashed = get_password_hash(password)
user = User(email=email, hashed_password=hashed, **kwargs)
session.add(user)
await session.commit()
await session.refresh(user)
return user
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/app/services/item_service.py
-45
View File
@@ -1,45 +0,0 @@
# 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
-88
View File
@@ -1,88 +0,0 @@
# 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
@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
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/build.py
+102 -67
View File
@@ -1,9 +1,11 @@
from module_bank import PythonToSQLite
from module_bank.encryption import Encryption
import os
from pathlib import Path
import shutil
import subprocess
import sys
import asyncio
def create_admin():
@@ -46,6 +48,31 @@ def create_admin():
print(f"异常: {e}")
def run_pyarmor_obfuscate():
# pyarmor obfuscate --output=obfuscated_script.py original_script.py
obfuscate_files = [
("main.py", "./"),
("server.py", "./"),
("utils/launch.py", "./utils"),
("utils/config.py", "./utils"),
]
for file, output in obfuscate_files:
cmd = [
"pyarmor",
"gen",
f"--output={output}",
file,
]
result = subprocess.run(cmd, cwd="./build", capture_output=True, text=True)
if result.returncode != 0:
print(f"打包文件: {file} 失败")
print(result.stderr)
return
else:
print(result.stdout)
print(f"打包文件: {file}")
def run_pyinstaller():
"""运行 PyInstaller 打包成可执行文件"""
try:
@@ -66,69 +93,37 @@ def run_pyinstaller():
"--clean",
# 隐藏导入
"--hidden-import",
"utils",
"--hidden-import",
"utils.launch",
"--hidden-import",
"utils.config",
"--hidden-import",
"server",
"--hidden-import",
"module_bank",
"--hidden-import",
"fastapi",
"--hidden-import",
"fastapi.middleware.cors",
"--hidden-import",
"uvicorn",
"--hidden-import",
"sqlmodel",
"--hidden-import",
"aiosqlite",
"--hidden-import",
"pyjwt",
"--hidden-import",
"sqlitedict",
"--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",
"PyJWT",
"--copy-metadata",
"sqlitedict",
# 收集整个包
"--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",
@@ -151,20 +146,48 @@ def run_pyinstaller():
print(f"PyInstaller 执行出错: {e}")
def main():
async def init_database():
from sqlalchemy.ext.asyncio import create_async_engine
from ..app.core.config import get_config
from sqlmodel import SQLModel, text
from sqlalchemy.exc import OperationalError
database = get_config()["database"]
database_type = database["type"]
database_config = database.get(database_type, None)
if database_config is None:
raise ValueError("配置不存在")
engine = create_async_engine(
"sqlite+aiosqlite:///./build/database.db", **database_config
)
async with engine.begin() as conn:
if database_type == "sqlite":
await conn.execute(text("PRAGMA journal_mode=WAL"))
await conn.execute(text("PRAGMA synchronous=NORMAL"))
try:
await conn.run_sync(
lambda conn: SQLModel.metadata.create_all(bind=conn, checkfirst=True)
)
except OperationalError as e:
if "already exists" in str(e):
raise
async def main():
# 打包模块
key = Encryption.generate_key()
resources_path = Path("resources")
if not resources_path.exists():
resources_path.mkdir()
app_packer = PythonToSQLite("./resources/app.mbank")
app_packer = PythonToSQLite("./resources/app.mbank", key)
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 = PythonToSQLite("./resources/plugins.mbank", key)
plugin_packer.pack_directory("src/plugins", "plugins")
plugin_packer.verify_package_structure()
plugin_packer.delete_source_code(None)
@@ -193,26 +216,38 @@ def main():
# 复制 main.py 到 build 文件夹
shutil.copy("public/main.py", "build")
shutil.copy("public/server.py", "build")
# 复制 config.toml 到 build 文件夹
shutil.copy("public/config.toml", "build")
# 复制 utils 文件夹到 build 文件夹
shutil.copytree("public/utils", "build/utils")
with open("./build/utils/launch.py", "r", encoding="utf-8") as f:
context = f.read()
context = context.replace("key = None", f"key = '{key}'")
with open("./build/utils/launch.py", "w", encoding="utf-8") as f:
f.write(context)
# 混淆
run_pyarmor_obfuscate()
# 运行 PyInstaller
run_pyinstaller()
# 创建管理员
create_admin()
# 清理临时文件
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)
shutil.rmtree(r"build/pyarmor_runtime_000000", ignore_errors=True)
os.remove(r"build/main.py")
os.remove(r"build/server.py")
# os.remove(r"build/config.toml")
print("构建完成")
# 初始化数据库
await init_database()
if __name__ == "__main__":
main()
def build():
asyncio.run(main())
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/code_content.py
+4 -2
View File
@@ -8,16 +8,18 @@ def generate_markdown_from_py_files(directory, output_file):
# 排除 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 != "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")
or file.endswith(".md")
):
file_path = os.path.join(root, file)
if file == "output.md":
continue
md_file.write(f"`{file_path}`\n")
md_file.write("```python\n")
with open(file_path, "r", encoding="utf-8") as py_file:
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/scripts/create_admin.py
-90
View File
@@ -1,90 +0,0 @@
#!/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())
{{ cookiecutter.project_name }}/src/{{ cookiecutter.project_name }}/utils/config.py
-82
View File
@@ -1,82 +0,0 @@
# 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
-55
View File
@@ -1,55 +0,0 @@
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="{{ 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()
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)