From 3fbd7764afbc289f413f73b4b0f5574adc09ac2d Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Sun, 26 Apr 2026 21:57:17 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20=E4=BB=A3=E7=A0=81=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- public/main.py | 27 +- public/server.py | 10 +- public/utils/config.py | 114 ++++++- public/utils/launch.py | 30 ++ src/chongming/__init__.py | 43 ++- src/chongming/app/__init__.py | 73 ++++- src/chongming/app/api/deps.py | 11 + src/chongming/app/api/routers/auth.py | 100 +++++- src/chongming/app/api/routers/todo.py | 58 ++++ src/chongming/app/api/routers/upload.py | 33 ++ src/chongming/app/core/cache.py | 215 +++++++++++-- src/chongming/app/core/config.py | 56 +++- src/chongming/app/core/logger.py | 80 ++++- src/chongming/app/core/security.py | 35 +- src/chongming/app/core/static_files.py | 42 ++- src/chongming/app/model/http_error.py | 20 ++ src/chongming/app/model/permission.py | 24 +- src/chongming/app/model/relationship_table.py | 22 +- src/chongming/app/model/role.py | 25 ++ src/chongming/app/model/todo.py | 47 +++ src/chongming/app/model/user.py | 80 ++++- src/chongming/app/service/auth.py | 98 +++++- src/chongming/app/service/permission.py | 81 +++++ src/chongming/app/service/role.py | 90 ++++++ src/chongming/app/service/todo.py | 56 ++++ src/chongming/app/service/uploads.py | 40 +++ src/chongming/app/service/user.py | 37 +++ src/chongming/app/task/__init__.py | 14 + src/chongming/app/task/dev_init_db.py | 50 +++ src/chongming/scripts/build.py | 104 +++++- src/plugins/jwt/jwt_cache.py | 176 ++++++++-- src/plugins/scheduler/scheduler.py | 302 +++++++++++++++++- 32 files changed, 2029 insertions(+), 164 deletions(-) diff --git a/public/main.py b/public/main.py index 1568db6..51a233a 100644 --- a/public/main.py +++ b/public/main.py @@ -12,13 +12,17 @@ import server use_shell = platform.system() == "Windows" + def app_run(app_config: dict, default_config: dict): """ 应用运行函数 - 适配模块银行环境 Args: - app_config: 应用配置 - default_config: 默认配置 + app_config: 应用配置字典，包含服务器配置、环境设置等 + default_config: 默认配置字典，包含应用名称、版本等基础信息 + + Returns: + None: 此函数不返回值，直接启动服务器并阻塞运行 """ from app.core.logger import get_logger # type: ignore @@ -48,8 +52,10 @@ def app_run(app_config: dict, default_config: dict): logger.info(f"👥 Workers: {workers}") logger.info("=" * 50) + # 根据操作系统选择不同的服务器启动方式 if use_shell: import uvicorn + uvicorn.run( "server:app", host=host, @@ -62,16 +68,27 @@ def app_run(app_config: dict, default_config: dict): ) else: from gunicorn.app.wsgiapp import run + sys.argv = [ "gunicorn", "server:app", - "-k", "uvicorn.workers.UvicornWorker", - "-w", str(workers), - "-b", f"{host}:{port}" + "-k", + "uvicorn.workers.UvicornWorker", + "-w", + str(workers), + "-b", + f"{host}:{port}", ] run() + def main(): + """ + 主函数入口 - 初始化多进程支持并启动应用 + + Returns: + None: 此函数不返回值，调用launch函数启动应用 + """ freeze_support() launch(app_run, "production") diff --git a/public/server.py b/public/server.py index 022f923..6e586ec 100644 --- a/public/server.py +++ b/public/server.py @@ -3,17 +3,19 @@ import sys from utils.launch import init_module_bank from utils.config import ConfigLoader +# 加载配置文件并初始化模块银行 config = ConfigLoader("config.toml") init_module_bank(config) -# 从数据库导入真正的 app 模块 +# 从数据库动态导入 app 模块 +# 该模块存储在 SQLite 数据库中，通过自定义导入机制加载 try: - import app # type: ignore # 此时会从 SQLite 数据库加载 + import app # type: ignore - app_instance = app.app # 获取 FastAPI 实例 + app_instance = app.app except ImportError as e: print(f"无法从数据库导入 app 模块: {e}") sys.exit(1) -# 暴露 ASGI 应用变量，供 uvicorn 使用 +# 暴露 ASGI 应用实例，供 uvicorn 服务器使用 app = app_instance diff --git a/public/utils/config.py b/public/utils/config.py index 0e3d837..bef1306 100644 --- a/public/utils/config.py +++ b/public/utils/config.py @@ -6,6 +6,7 @@ import sys from pathlib import Path from typing import Any, Dict, Optional +# 根据Python版本选择合适的TOML解析库 if sys.version_info >= (3, 11): import tomllib else: @@ -13,7 +14,15 @@ else: class ConfigLoader: - """TOML配置文件加载器""" + """TOML配置文件加载器 + + 提供配置文件的加载、访问和管理功能。支持自动查找配置文件、 + 点号分隔的嵌套键访问、配置节获取以及配置重载等功能。 + + Attributes: + config_path: 配置文件的路径 + _config: 存储解析后的配置数据 + """ def __init__(self, config_path: Optional[str] = None): """ @@ -21,6 +30,9 @@ class ConfigLoader: Args: config_path: 配置文件路径，如果为None则自动查找 + + Raises: + FileNotFoundError: 当无法找到配置文件时抛出异常 """ if config_path is None: # 自动查找配置文件 @@ -31,7 +43,21 @@ class ConfigLoader: self.load() def _find_config_file(self) -> str: - """自动查找配置文件""" + """ + 自动查找配置文件 + + 按照预定义的优先级顺序搜索配置文件： + 1. 当前目录下的config.toml + 2. config子目录下的config.toml + 3. 用户配置目录~/.config/myapp/config.toml + 4. 系统配置目录/etc/myapp/config.toml + + Returns: + str: 找到的第一个配置文件路径 + + Raises: + FileNotFoundError: 当所有可能的位置都未找到配置文件时抛出异常 + """ possible_paths = [ "config.toml", "config/config.toml", @@ -47,7 +73,15 @@ class ConfigLoader: raise FileNotFoundError("未找到配置文件") def load(self) -> None: - """加载配置文件""" + """ + 加载配置文件 + + 读取并解析TOML格式的配置文件，将配置数据存储到内部字典中。 + + Raises: + FileNotFoundError: 当配置文件不存在时抛出异常 + tomllib.TOMLDecodeError: 当配置文件格式不正确时抛出异常 + """ if not self.config_path.exists(): raise FileNotFoundError(f"配置文件不存在: {self.config_path}") @@ -55,7 +89,20 @@ class ConfigLoader: self._config = tomllib.loads(f.read()) def get(self, key: str, default: Any = None) -> Any: - """获取配置值，支持点号分隔的路径""" + """ + 获取配置值，支持点号分隔的路径 + + 通过点号分隔的键路径访问嵌套的配置值。例如： + - "database.host" 访问 {"database": {"host": "localhost"}} + - "server.port" 访问 {"server": {"port": 8080}} + + Args: + key: 配置键，支持点号分隔的嵌套路径 + default: 当键不存在时返回的默认值，默认为None + + Returns: + Any: 配置值，如果键不存在则返回默认值 + """ keys = key.split(".") value = self._config @@ -67,22 +114,71 @@ class ConfigLoader: return default def get_section(self, section: str) -> Dict[str, Any]: - """获取整个配置节""" + """ + 获取整个配置节 + + 返回指定配置节的完整字典内容，适用于需要访问某个配置节下 + 所有配置项的场景。 + + Args: + section: 配置节名称 + + Returns: + Dict[str, Any]: 配置节的内容字典，如果节不存在则返回空字典 + """ return self._config.get(section, {}) def reload(self) -> None: - """重新加载配置文件""" + """ + 重新加载配置文件 + + 从磁盘重新读取配置文件并更新内部配置数据。 + 适用于配置文件在运行时被修改的场景。 + + Raises: + FileNotFoundError: 当配置文件不存在时抛出异常 + """ self.load() def __getitem__(self, key: str) -> Any: - """支持字典式访问""" + """ + 支持字典式访问 + + 允许使用方括号语法访问配置值，如：config["database.host"] + + Args: + key: 配置键 + + Returns: + Any: 配置值 + + Raises: + KeyError: 当键不存在且无默认值时抛出异常（通过get方法处理） + """ return self.get(key) def __contains__(self, key: str) -> bool: - """检查配置是否存在""" + """ + 检查配置是否存在 + + 支持使用in操作符检查配置键是否存在，如："database.host" in config + + Args: + key: 要检查的配置键 + + Returns: + bool: 如果配置键存在且值不为None则返回True，否则返回False + """ return self.get(key) is not None @property def all(self) -> Dict[str, Any]: - """获取所有配置""" + """ + 获取所有配置 + + 返回完整的配置字典，适用于需要遍历或导出所有配置的场景。 + + Returns: + Dict[str, Any]: 包含所有配置的字典 + """ return self._config diff --git a/public/utils/launch.py b/public/utils/launch.py index fffc4c9..27be687 100644 --- a/public/utils/launch.py +++ b/public/utils/launch.py @@ -7,14 +7,29 @@ key = None def init_module_bank(config): + """ + 初始化模块银行系统，将指定路径下的Python模块打包并安装导入器。 + + 该函数会验证所有模块路径的有效性，创建PythonToSQLite打包器实例， + 列出所有可用模块，并为每个打包器安装导入器以支持模块的动态加载。 + + Args: + config (dict): 配置字典，必须包含 'production' -> 'module_system' -> 'path' 键， + 其值为模块路径列表 + + Raises: + ValueError: 当配置的模块路径中有任何一个不存在时抛出异常 + """ MODULE_BANK_PATH_LIST = config["production"]["module_system"]["path"] for module_path in MODULE_BANK_PATH_LIST: if not os.path.exists(module_path): raise ValueError(f"模块路径不存在: {module_path}") + # 为每个模块路径创建PythonToSQLite打包器实例 packer_list = [ PythonToSQLite(module_path, key=key) for module_path in MODULE_BANK_PATH_LIST ] + # 遍历所有打包器，列出模块信息并安装导入器 for packer in packer_list: modules = packer.list_modules() for module in modules: @@ -23,6 +38,21 @@ def init_module_bank(config): def launch(run_fun: Callable[[dict, dict], None], run_mode: str): + """ + 启动应用程序，加载配置并执行指定的运行函数。 + + 该函数从config.toml文件中加载配置，根据运行模式获取对应的应用配置和默认配置， + 然后调用传入的运行函数执行应用程序逻辑。 + + Args: + run_fun (Callable[[dict, dict], None]): 运行函数，接收两个参数： + - app_config (dict): 当前运行模式的配置 + - default_config (dict): 默认配置 + run_mode (str): 运行模式，用于从配置文件中获取对应模式的配置项 + + Returns: + None + """ config = ConfigLoader("config.toml") # 获取对应运行模式的配置 diff --git a/src/chongming/__init__.py b/src/chongming/__init__.py index 0833d01..37f052d 100644 --- a/src/chongming/__init__.py +++ b/src/chongming/__init__.py @@ -4,32 +4,57 @@ from .app.core.logger import get_logger def serve(): + """ + 使用 uvicorn 启动服务 + + 从配置文件中加载服务器配置和日志配置，然后使用 uvicorn 启动应用服务。 + 该函数适用于开发环境或单进程部署场景。 + + Returns: + None: 该函数不返回值，会阻塞运行直到服务停止 + """ config = load_config() env = config["default"]["env"] server_config = config[env]["server"] log_config = config[env]["logging"] - logget = get_logger("app") - logget.info(f"启动服务: {server_config}") - logget.info(f"启动日志: {log_config}") + logger = get_logger("app") + logger.info(f"启动服务: {server_config}") + logger.info(f"启动日志: {log_config}") uvicorn.run("chongming.app:app", **server_config) + def gunicorn_serve(): - """使用 gunicorn 启动服务""" + """ + 使用 gunicorn 启动服务 + + 通过 gunicorn 作为进程管理器启动应用，使用 UvicornWorker 作为 worker 类型， + 支持多进程部署，适用于生产环境。 + + 从配置文件中读取服务器配置，动态设置 gunicorn 的 worker 数量、绑定地址和端口等参数。 + + Returns: + None: 该函数不返回值，会阻塞运行直到服务停止 + """ import sys from gunicorn.app.wsgiapp import run + config = load_config() env = config["default"]["env"] server_config = config[env]["server"] - - # 设置 gunicorn 参数 + + # 构建 gunicorn 命令行参数并启动服务 sys.argv = [ "gunicorn", "chongming.app:app", - "-k", "uvicorn.workers.UvicornWorker", - "-w", f"{server_config.get('workers', 4)}", - "-b", f"{server_config.get('host', '0.0.0.0')}:{server_config.get('port', 8000)}" + "-k", + "uvicorn.workers.UvicornWorker", + "-w", + f"{server_config.get('workers', 4)}", + "-b", + f"{server_config.get('host', '0.0.0.0')}:{server_config.get('port', 8000)}", ] run() + if __name__ == "__main__": serve() diff --git a/src/chongming/app/__init__.py b/src/chongming/app/__init__.py index 0ee8f32..298e548 100644 --- a/src/chongming/app/__init__.py +++ b/src/chongming/app/__init__.py @@ -31,9 +31,28 @@ except KeyError: # pragma: no cover raise ValueError("配置不存在") -# 生命周期管理 @asynccontextmanager async def lifespan(app: FastAPI): + """ + 应用生命周期管理函数，负责应用的启动和关闭时的资源初始化与清理。 + + 在应用启动时： + - 初始化JWT缓存 + - 创建异步数据库引擎和会话工厂 + - 在开发环境下创建数据库表 + - 初始化定时任务服务 + + 在应用关闭时： + - 关闭定时任务服务 + - 释放数据库引擎资源 + - 关闭缓存连接 + + 参数: + app: FastAPI应用实例 + + Yields: + None: 应用运行期间保持上下文激活 + """ from plugins import hello from plugins.jwt.jwt_cache import get_jwt_cache @@ -46,17 +65,14 @@ async def lifespan(app: FastAPI): 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": @@ -72,22 +88,19 @@ async def lifespan(app: FastAPI): if "already exists" in str(e): raise - # 启动时：初始化任务服务 task_service = get_task_service_instance( config, get_logger("scheduler"), async_session_maker ) app.state.task_service = task_service await task_service.start(init_tasks_callback) - yield # 应用运行期间 + yield - # 关闭时：释放引擎 await task_service.shutdown(wait=True) await engine.dispose() cache.close() -# 创建 FastAPI 应用 cors = config[env].get("cors", {}) app = FastAPI( title=config["default"]["app"]["name"], @@ -99,15 +112,29 @@ app = FastAPI( lifespan=lifespan, ) -# 添加 CORS 中间件 app.add_middleware(CORSMiddleware, **cors) -# 添加路由 app.include_router(api_router, prefix=config["default"]["prefix"]) @app.middleware("http") async def access_log_middleware(request: Request, call_next): + """ + HTTP访问日志中间件，记录每个请求的详细信息。 + + 记录的信息包括： + - 请求方法和路径 + - 响应状态码 + - 请求处理时长（毫秒） + - 客户端IP地址 + + 参数: + request: FastAPI请求对象 + call_next: 下一个中间件或路由处理函数 + + 返回: + Response: 处理后的响应对象 + """ start = time.perf_counter() response = await call_next(request) duration = (time.perf_counter() - start) * 1000 @@ -116,15 +143,19 @@ async def access_log_middleware(request: Request, call_next): logger.info( f"{request.method} {request.url.path} status={response.status_code} " f"duration={duration:.2f}ms client={client}", - extra={"status_code": response.status_code}, # ← 关键 + extra={"status_code": response.status_code}, ) return response -# 根路由 @app.get("/", summary="根路径") async def root(): - """根路径，返回应用信息""" + """ + 根路径端点，返回应用的基本信息。 + + 返回: + dict: 包含应用名称、版本、文档地址和健康检查地址的字典 + """ return { "message": f"欢迎使用 {config['default']['app']['name']}", "version": config["default"]["app"]["version"], @@ -154,7 +185,6 @@ async def root(): # ) # return {"status": "success", "job_id": job.id} -# 挂载静态文件 config = get_config() IMAGE_DIR = Path(f"./{config['default']['upload_path']}/images") app.mount( @@ -163,17 +193,26 @@ app.mount( name="images", ) -# 挂载前端静态文件 if vfs_db_path := config[env].get("file_system", {}).get("path", None): logger.info("静态文件已存在，将使用静态文件服务") static_files_handler = SVFSStaticFiles(directory="/", vfs_db_path=vfs_db_path) app.mount("/static", static_files_handler, name="static") -# 全局异常处理器 @app.exception_handler(Exception) async def global_exception_handler(_: Request, exc: Exception): - """全局异常处理器""" + """ + 全局异常处理器，捕获并处理应用中未捕获的异常。 + + 在调试模式下返回详细的异常信息，在生产模式下返回通用错误提示。 + + 参数: + _: FastAPI请求对象（未使用） + exc: 捕获到的异常对象 + + 返回: + JSONResponse: 包含错误信息的JSON响应，状态码为500 + """ logger.error(f"全局异常: {exc}") return JSONResponse( status_code=500, diff --git a/src/chongming/app/api/deps.py b/src/chongming/app/api/deps.py index c3eaec4..28732be 100644 --- a/src/chongming/app/api/deps.py +++ b/src/chongming/app/api/deps.py @@ -15,12 +15,23 @@ async def get_current_user( 获取当前用户依赖项 验证访问令牌并返回用户信息 + + Args: + credentials: HTTP授权凭证，包含Bearer token + + Returns: + TokenData: 验证通过后的用户令牌数据 + + Raises: + HTTPException: 当访问令牌无效时抛出401未授权异常 """ 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, diff --git a/src/chongming/app/api/routers/auth.py b/src/chongming/app/api/routers/auth.py index 89bb380..ec24833 100644 --- a/src/chongming/app/api/routers/auth.py +++ b/src/chongming/app/api/routers/auth.py @@ -30,6 +30,21 @@ async def register( user_data: UserCreate, session: AsyncSession = Depends(get_session), ): + """ + 用户注册接口 + + 创建新用户账户，检查邮箱是否已被注册，如果邮箱未被注册则创建新用户。 + + Args: + user_data: 用户注册数据，包含邮箱、密码、用户名和全名等信息 + session: 数据库会话对象，用于执行数据库操作 + + Returns: + UserRead: 创建成功的用户信息，包含用户ID、邮箱、用户名等字段 + + Raises: + HTTPException: 当邮箱已被注册时抛出400错误 + """ existing = await UserService.get_user_by_email(user_data.email, session) if existing: raise HTTPException( @@ -63,6 +78,25 @@ async def login( request: Request, session: AsyncSession = Depends(get_session), ): + """ + 用户登录接口 + + 验证用户凭据并生成访问令牌和刷新令牌。依次检查邮箱是否存在、密码是否正确、用户是否被禁用。 + + Args: + user_data: 用户登录数据，包含邮箱和密码 + request: FastAPI请求对象，用于获取客户端信息 + session: 数据库会话对象，用于查询用户信息 + + Returns: + TokenResponse: 包含访问令牌(access_token)、刷新令牌(refresh_token)和令牌类型(token_type)的响应对象 + + Raises: + HTTPException: + - 400错误：当邮箱不存在时 + - 401错误：当密码错误时 + - 402错误：当用户账户被禁用时 + """ user = await UserService.get_user_by_email(user_data.email, session) if not user: raise HTTPException( @@ -93,6 +127,22 @@ async def refresh_token( credentials: HTTPAuthorizationCredentials = Depends(refresh_scheme), session: AsyncSession = Depends(get_session), ): + """ + 刷新访问令牌接口 + + 使用有效的刷新令牌获取新的访问令牌。验证刷新令牌的有效性，如果有效则生成新的访问令牌。 + + Args: + request: FastAPI请求对象，用于获取客户端信息 + credentials: HTTP认证凭证，从请求头中提取的刷新令牌 + session: 数据库会话对象，用于验证令牌状态 + + Returns: + RefreshTokenResponse: 包含新的访问令牌(access_token)和令牌类型(token_type)的响应对象 + + Raises: + HTTPException: 当刷新令牌无效或已过期时抛出401错误 + """ auth_service = get_auth_service() refresh_token = credentials.credentials @@ -112,6 +162,17 @@ async def refresh_token( async def logout( credentials: HTTPAuthorizationCredentials = Depends(access_scheme), ): + """ + 用户登出接口 + + 使当前的访问令牌失效，实现单设备登出功能。 + + Args: + credentials: HTTP认证凭证，从请求头中提取的访问令牌 + + Returns: + Userlogout: 登出成功响应，包含成功消息和状态码200 + """ auth_service = get_auth_service() token = credentials.credentials await auth_service.logout(token) @@ -120,6 +181,17 @@ async def logout( @router.post("/logout-all", summary="登出所有设备", response_model=Userlogout) async def logout_all(current_user: TokenData = Depends(get_current_user)): + """ + 登出所有设备接口 + + 使指定用户的所有活跃会话令牌失效，实现多设备同时登出功能。 + + Args: + current_user: 当前登录用户的令牌数据，包含用户ID等信息 + + Returns: + Userlogout: 登出成功响应，包含成功消息和状态码200 + """ auth_service = get_auth_service() await auth_service.logout_all(current_user.user_id) @@ -128,7 +200,17 @@ async def logout_all(current_user: TokenData = Depends(get_current_user)): @router.get("/sessions", summary="获取用户会话") async def get_sessions(current_user: TokenData = Depends(get_current_user)): - """获取用户的所有活跃会话""" + """ + 获取用户会话列表接口 + + 查询并返回指定用户的所有活跃会话信息，包括每个会话的设备信息和创建时间等。 + + Args: + current_user: 当前登录用户的令牌数据，包含用户ID等信息 + + Returns: + UserSession: 用户会话信息对象，包含用户ID、会话列表和会话总数 + """ auth_service = get_auth_service() sessions = await auth_service.get_user_sessions(current_user.user_id) @@ -147,7 +229,21 @@ async def get_current_user_info( session: AsyncSession = Depends(get_session), current_user: TokenData = Depends(get_current_user), ): - """获取当前登录用户的信息""" + """ + 获取当前用户信息接口 + + 根据当前登录用户的令牌信息，从数据库中查询并返回完整的用户资料。 + + Args: + session: 数据库会话对象，用于查询用户信息 + current_user: 当前登录用户的令牌数据，包含用户ID等信息 + + Returns: + UserRead: 用户详细信息，包含用户ID、邮箱、用户名、全名、激活状态等字段 + + Raises: + HTTPException: 当用户不存在时抛出404错误 + """ user = await UserService.get_user_by_id(UUID(current_user.user_id), session) if not user: raise HTTPException( diff --git a/src/chongming/app/api/routers/todo.py b/src/chongming/app/api/routers/todo.py index e7afc5b..0c70749 100644 --- a/src/chongming/app/api/routers/todo.py +++ b/src/chongming/app/api/routers/todo.py @@ -18,6 +18,16 @@ async def create_todo( todo: TodoCreate, session: AsyncSession = Depends(get_session), ): + """ + 创建新的待办事项 + + Args: + todo: 待办事项创建请求体，包含待办事项的详细信息 + session: 数据库会话对象，用于执行数据库操作 + + Returns: + TodoRead: 创建成功的待办事项对象，包含完整的待办事项信息 + """ return await TodoService.create_todo(todo, session) @@ -25,11 +35,35 @@ async def create_todo( async def read_todos( offset: int = 0, limit: int = 100, session: AsyncSession = Depends(get_session) ): + """ + 获取待办事项列表 + + Args: + offset: 分页偏移量，默认为0，用于指定从第几条记录开始查询 + limit: 每页返回的最大记录数，默认为100，用于限制返回结果的数量 + session: 数据库会话对象，用于执行数据库操作 + + Returns: + List[TodoRead]: 待办事项列表，包含指定范围内的所有待办事项 + """ return await TodoService.get_todos(offset, limit, session) @router.get("/{todo_id}", response_model=TodoRead) async def read_todo(todo_id: int, session: AsyncSession = Depends(get_session)): + """ + 根据ID获取单个待办事项 + + Args: + todo_id: 待办事项的唯一标识符 + session: 数据库会话对象，用于执行数据库操作 + + Returns: + TodoRead: 查询到的待办事项对象 + + Raises: + HTTPException: 当待办事项不存在时抛出404错误 + """ todo = await TodoService.get_todo_by_id(todo_id, session) if not todo: raise HTTPException(status_code=404, detail="Todo not found") @@ -40,12 +74,36 @@ async def read_todo(todo_id: int, session: AsyncSession = Depends(get_session)): async def update_todo( todo_id: int, todo_update: TodoUpdate, session: AsyncSession = Depends(get_session) ): + """ + 更新待办事项信息 + + Args: + todo_id: 待办事项的唯一标识符 + todo_update: 待办事项更新请求体，包含需要更新的字段信息 + session: 数据库会话对象，用于执行数据库操作 + + Returns: + TodoRead: 更新后的待办事项对象，包含最新的待办事项信息 + """ todo = await TodoService.update_todo(todo_id, todo_update, session) return todo @router.delete("/{todo_id}") async def delete_todo(todo_id: int, session: AsyncSession = Depends(get_session)): + """ + 删除待办事项 + + Args: + todo_id: 待办事项的唯一标识符 + session: 数据库会话对象，用于执行数据库操作 + + Returns: + dict: 包含操作结果的字典，{"ok": True} 表示删除成功 + + Raises: + HTTPException: 当待办事项不存在时抛出404错误 + """ is_deleted = await TodoService.delete_todo(todo_id, session) if not is_deleted: raise HTTPException(status_code=404, detail="Todo not found") diff --git a/src/chongming/app/api/routers/upload.py b/src/chongming/app/api/routers/upload.py index 6a64a8b..bd6fcc5 100644 --- a/src/chongming/app/api/routers/upload.py +++ b/src/chongming/app/api/routers/upload.py @@ -16,6 +16,8 @@ upload_image_service = UploadService( class UploadResponse(BaseModel): + """图片上传响应数据模型""" + url: str filename: str file_type: str @@ -23,13 +25,44 @@ class UploadResponse(BaseModel): @router.post("/image", response_model=UploadResponse) async def upload_image(file: UploadFile = File(...)): + """ + 上传图片文件接口 + 该接口用于接收用户上传的图片文件，进行格式验证后保存到服务器， + 并返回文件的访问URL、文件名和文件类型信息。 + + Args: + file (UploadFile): 上传的文件对象，必须为图片格式 + + Returns: + UploadResponse: 包含以下字段的响应对象： + - url (str): 上传文件的访问URL + - filename (str): 保存后的文件名 + - file_type (str): 文件扩展名（如 .jpg, .png 等） + + Raises: + HTTPException: + - 400: 当文件类型不是图片格式时 + - 400: 当文件名为空时 + - 400: 当文件扩展名不在允许列表中时 + + Note: + 支持的图片格式包括：.jpg, .jpeg, .png, .gif, .webp + 需要用户认证才能访问此接口 + """ + # 验证文件内容类型是否为图片 if not file.content_type or not file.content_type.startswith("image/"): raise HTTPException(status_code=400, detail="Invalid file type") + + # 验证文件名是否存在 if file.filename is None: raise HTTPException(status_code=400, detail="Invalid file name") + + # 检查文件扩展名是否在允许的范围内 ext = os.path.splitext(file.filename)[1].lower() if ext not in [".jpg", ".jpeg", ".png", ".gif", ".webp"]: raise HTTPException(status_code=400, detail="Invalid file type") + + # 执行文件上传操作 url, filename = await upload_image_service.upload(file) return UploadResponse(url=url, filename=filename, file_type=ext) diff --git a/src/chongming/app/core/cache.py b/src/chongming/app/core/cache.py index 41dca00..acfcc7b 100644 --- a/src/chongming/app/core/cache.py +++ b/src/chongming/app/core/cache.py @@ -1,3 +1,10 @@ +""" +磁盘缓存模块 + +基于 diskcache 实现的本地磁盘缓存系统，提供类似 Redis 的 API 接口。 +支持过期时间设置、TTL 查询、键模式匹配等功能。 +""" + import time from typing import Any, Optional @@ -7,7 +14,29 @@ from ..core.config import get_config class Cache: + """ + 磁盘缓存类 + + 封装 diskcache 提供统一的缓存操作接口，支持： + - 键值对存储与检索 + - 过期时间管理（秒/毫秒） + - TTL 查询 + - 键模式匹配 + - 批量操作 + + Attributes: + _cache: diskcache.Cache 实例，负责实际的磁盘缓存操作 + """ + def __init__(self): + """ + 初始化缓存实例 + + 从配置中读取缓存存储路径，创建 diskcache.Cache 实例。 + + Raises: + KeyError: 当配置中缺少 cache.cache_store_path 时 + """ config = get_config() cache_dir = config["cache"]["cache_store_path"] self._cache = diskcache.Cache(cache_dir) @@ -23,14 +52,23 @@ class Cache: """ 设置缓存 + 将键值对存储到缓存中，可选设置过期时间。内部使用元组 (value, expire_at) + 存储，以支持自定义的过期检查逻辑。 + Args: - key: 键 - value: 值 - ex: 过期时间（秒） - px: 过期时间（毫秒） + key: 缓存键，字符串类型 + value: 缓存值，可以是任意可序列化的 Python 对象 + ex: 过期时间（秒），与 px 互斥，优先使用 ex + px: 过期时间（毫秒），当 ex 为 None 时生效 + raw: 是否原始存储（当前未使用，保留参数兼容性） Returns: - bool: 是否设置成功 + bool: 设置成功返回 True，发生异常返回 False + + Note: + - ex 和 px 不能同时设置，ex 优先级更高 + - 过期时间在存储时计算绝对时间戳，便于后续检查 + - 异常会被捕获并打印错误信息，不会抛出异常 """ try: expire_at = None @@ -54,14 +92,21 @@ class Cache: def get(self, key: str, default: Any = None) -> Any: """ - 获取缓存 + 获取缓存值 + + 根据键获取缓存值，自动检查并清理已过期的缓存项。 + 如果键不存在或已过期，返回默认值。 Args: - key: 键 - default: 默认值 + key: 缓存键 + default: 当键不存在或已过期时的返回值，默认为 None Returns: - 缓存的值，如果key不存在或已过期返回default + Any: 缓存的值，如果key不存在或已过期返回default + + Note: + - 发现过期数据时会主动删除，避免脏数据累积 + - 异常情况下返回 default 值，保证调用安全性 """ try: stored = self._cache.get(key) @@ -81,13 +126,19 @@ class Cache: def delete(self, *keys: str) -> int: """ - 删除一个或多个key + 删除一个或多个缓存键 + + 批量删除指定的缓存键，返回成功删除的数量。 Args: - keys: 要删除的key + *keys: 可变参数，要删除的键列表 Returns: - 成功删除的数量 + int: 成功删除的键数量 + + Note: + - 逐个删除，某个键删除失败不影响其他键 + - 只统计实际存在的键的删除操作 """ deleted = 0 for key in keys: @@ -97,26 +148,36 @@ class Cache: def exists(self, key: str) -> bool: """ - 检查key是否存在且未过期 + 检查键是否存在且未过期 + + 通过尝试获取值来判断键的有效性，比直接检查更可靠， + 因为会触发过期检查和清理逻辑。 Args: - key: 键 + key: 要检查的缓存键 Returns: - bool: 是否存在 + bool: 键存在且未过期返回 True，否则返回 False """ return self.get(key, None) is not None def expire(self, key: str, _time: int) -> bool: """ - 设置过期时间 + 更新键的过期时间 + + 重新设置指定键的过期时间，保持原值不变。 + 如果键不存在，操作失败。 Args: - key: 键 - time: 过期时间（秒） + key: 缓存键 + _time: 新的过期时间（秒），从当前时间开始计算 Returns: - bool: 是否设置成功 + bool: 设置成功返回 True，键不存在或异常返回 False + + Note: + - 需要先读取原值，再重新存储以更新过期时间 + - 原子性不保证，高并发场景可能有问题 """ try: stored = self._cache.get(key) @@ -133,13 +194,22 @@ class Cache: def ttl(self, key: str) -> int: """ - 获取key的剩余存活时间 + 获取键的剩余生存时间（Time To Live） + + 计算键距离过期的剩余秒数。 Args: - key: 键 + key: 缓存键 Returns: - int: 剩余秒数，-1表示永不过期，-2表示key不存在 + int: 剩余秒数，返回值含义： + - >= 0: 剩余存活秒数 + - -1: 键存在但永不过期 + - -2: 键不存在 + + Note: + - 返回值为整数，向下取整 + - 即使剩余时间为负数也返回 0，表示即将过期 """ try: stored = self._cache.get(key) @@ -158,13 +228,21 @@ class Cache: def keys(self, pattern: Optional[str] = None) -> list: """ - 获取所有key，支持简单的通配符匹配（*） + 获取所有缓存键，支持通配符模式匹配 + + 遍历所有缓存键，可根据前缀模式过滤。 Args: - pattern: 匹配模式，如 "user:*" + pattern: 匹配模式，仅支持前缀通配，如 "user:*" 匹配所有 "user:" 开头的键 + None 或空字符串返回所有键 Returns: - list: key列表 + list: 匹配的键列表，按迭代顺序排列 + + Note: + - 仅支持简单的 "*" 后缀通配，不支持复杂模式 + - 大量键时性能较差，建议谨慎使用 + - pattern 中的 "*" 会被替换为空字符串作为前缀 """ all_keys = list(self._cache.iterkeys()) @@ -175,7 +253,17 @@ class Cache: return all_keys def clear(self) -> bool: - """清空所有缓存。""" + """ + 清空所有缓存 + + 删除缓存目录中的所有数据，包括未过期的项。 + + Returns: + bool: 清空成功返回 True，异常返回 False + + Warning: + 此操作不可逆，会删除所有缓存数据 + """ try: self._cache.clear() return True @@ -184,12 +272,36 @@ class Cache: return False def size(self) -> int: - """获取当前缓存数量（不含过期但尚未清理的项）。""" + """ + 获取当前有效缓存项数量 + + 统计当前未过期的缓存键数量。 + + Returns: + int: 有效缓存项的数量 + + Note: + - 不包含已过期但尚未清理的项 + - 需要遍历所有键，大数据量时可能有性能开销 + """ # diskcache 的 __len__ 会返回当前有效条目数 return len(list(self._cache.iterkeys())) def get_all(self) -> dict: - """获取所有未过期的缓存数据。""" + """ + 获取所有未过期的缓存数据 + + 遍历所有缓存键，返回有效的键值对字典。 + 自动过滤过期数据。 + + Returns: + dict: 所有有效缓存的键值对，格式为 {key: value} + + Note: + - 会触发每个键的过期检查 + - 大数据量时内存占用较高，谨慎使用 + - 返回的是快照，不保证原子性 + """ result = {} for key in self._cache.iterkeys(): value = self.get(str(key)) # get 内部会检查并清理过期 @@ -198,7 +310,15 @@ class Cache: return result def close(self): - """关闭缓存。""" + """ + 关闭缓存连接 + + 释放 diskcache 占用的资源，包括文件句柄等。 + 应在应用退出时调用。 + + Note: + 关闭后不应再使用该实例，除非重新初始化 + """ self._cache.close() @@ -206,6 +326,18 @@ _cache: Optional[Cache] = None def get_cache() -> Cache: + """ + 获取全局缓存单例 + + 使用懒加载模式创建缓存实例，确保整个应用中只有一个 Cache 实例。 + + Returns: + Cache: 全局缓存实例 + + Note: + - 线程不安全，多线程环境需要额外同步 + - 首次调用时初始化，后续调用返回同一实例 + """ global _cache if _cache is None: _cache = Cache() @@ -217,13 +349,30 @@ def cached(ttl: Optional[int] = None): """ 函数结果缓存装饰器 + 自动缓存函数的返回值，相同参数的调用直接从缓存返回， + 避免重复计算。适用于纯函数或副作用小的函数。 + Args: - ttl: 缓存时间（秒），None表示永不过期 + ttl: 缓存生存时间（秒），None 表示永不过期 + + Returns: + callable: 装饰器函数 Example: - @cached(ttl=60) - def get_user(user_id: int): - return {"id": user_id, "name": "test"} + >>> @cached(ttl=60) + ... def get_user(user_id: int): + ... return {"id": user_id, "name": "test"} + >>> + >>> # 第一次调用执行函数并缓存 + >>> get_user(1) + >>> # 60秒内再次调用直接返回缓存结果 + >>> get_user(1) + + Note: + - 缓存键由函数名 + 位置参数 + 关键字参数组成 + - 参数必须是可哈希的，否则会导致错误 + - 不适合缓存大型对象或频繁变化的数据 + - 不同参数组合会生成不同的缓存键 """ def decorator(func): diff --git a/src/chongming/app/core/config.py b/src/chongming/app/core/config.py index 46b682a..ff4aaad 100644 --- a/src/chongming/app/core/config.py +++ b/src/chongming/app/core/config.py @@ -3,6 +3,7 @@ from pathlib import Path from typing import TypedDict, Literal, Optional, cast, List +# 根据Python版本选择合适的TOML解析库和Required类型 if sys.version_info >= (3, 11): import tomllib from typing import Required @@ -14,6 +15,8 @@ else: class DefaultApp(TypedDict): + """应用默认配置结构""" + name: str version: str description: str @@ -21,6 +24,8 @@ class DefaultApp(TypedDict): class Default(TypedDict): + """系统默认配置结构，包含应用信息、环境、路径等基础配置""" + app: DefaultApp env: Literal["production", "development"] prefix: str @@ -28,11 +33,15 @@ class Default(TypedDict): class SqliteConnectionArgs(TypedDict, total=False): + """SQLite数据库连接参数配置""" + timeout: int check_same_thread: bool class DatabaseSqlite(TypedDict, total=False): + """SQLite数据库连接池及行为配置""" + echo: bool pool_size: int max_overflow: int @@ -43,6 +52,8 @@ class DatabaseSqlite(TypedDict, total=False): class Database(TypedDict, total=False): + """数据库配置结构，支持SQLite、MySQL、PostgreSQL三种数据库类型""" + type: Required[Literal["sqlite", "mysql", "postgresql"]] url: Required[str] generate_schemas: bool @@ -52,6 +63,8 @@ class Database(TypedDict, total=False): class Security(TypedDict): + """安全相关配置，包括JWT令牌和安全策略设置""" + secret_key: str algorithm: Literal["HS256"] access_token_expire_minutes: int @@ -60,6 +73,8 @@ class Security(TypedDict): class Server(TypedDict, total=False): + """服务器运行配置，包含主机、端口、工作进程等设置""" + host: Required[str] port: Required[int] reload: bool @@ -68,6 +83,8 @@ class Server(TypedDict, total=False): class Logging(TypedDict): + """日志系统配置，定义日志级别、输出格式和文件管理策略""" + level: str file: str format: str @@ -76,6 +93,8 @@ class Logging(TypedDict): class Cors(TypedDict, total=False): + """跨域资源共享(CORS)配置""" + allow_origins: List[str] allow_credentials: bool allow_methods: List[str] @@ -83,16 +102,22 @@ class Cors(TypedDict, total=False): class ModuleSystem(TypedDict, total=False): + """模块系统配置，定义模块加载类型和路径""" + type: str path: List[str] class FileSystem(TypedDict, total=False): + """文件系统配置，定义存储类型和路径""" + type: str path: str class Env(TypedDict, total=False): + """环境特定配置结构，用于开发和生产环境的差异化配置""" + debug: Required[bool] server: Required[Server] logging: Required[Logging] @@ -102,14 +127,20 @@ class Env(TypedDict, total=False): class Scheduler(TypedDict): + """任务调度器配置""" + job_store_path: str class Cache(TypedDict): + """缓存系统配置""" + cache_store_path: str class Config(TypedDict): + """完整配置结构，整合所有配置项，包含默认配置、数据库、调度器、缓存、安全和环境配置""" + default: Default database: Database scheduler: Scheduler @@ -120,17 +151,40 @@ class Config(TypedDict): def load_config(config_path: Path = Path("config.toml")) -> Config: - """从 TOML 文件加载配置""" + """ + 从TOML配置文件加载配置数据 + + Args: + config_path: 配置文件路径，默认为当前目录下的config.toml + + Returns: + Config: 解析后的配置字典对象 + + Raises: + FileNotFoundError: 当配置文件不存在时抛出异常 + """ 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: + """ + 获取应用程序配置（带缓存机制） + + 首次调用时加载配置文件并缓存，后续调用直接返回缓存的配置对象 + + Args: + config_path: 配置文件路径，默认为config.toml + + Returns: + Config: 配置字典对象 + """ global _config if _config is None: _config = load_config(Path(config_path)) diff --git a/src/chongming/app/core/logger.py b/src/chongming/app/core/logger.py index 770b384..2472958 100644 --- a/src/chongming/app/core/logger.py +++ b/src/chongming/app/core/logger.py @@ -14,7 +14,19 @@ log_config = config[env]["logging"] class ColoredFormatter(logging.Formatter): - """为 status_code 添加颜色的日志格式化器""" + """ + 为 HTTP 状态码添加颜色的日志格式化器 + + 根据 HTTP 状态码的首位数字为日志消息中的状态码添加不同颜色： + - 2xx: 绿色（成功） + - 3xx: 青色（重定向） + - 4xx: 黄色（客户端错误） + - 5xx: 红色（服务器错误） + + Attributes: + COLORS: 状态码首位数字到 ANSI 颜色代码的映射字典 + RESET: ANSI 重置颜色代码 + """ COLORS = { "2": "\033[32m", # 绿色 2xx @@ -25,6 +37,18 @@ class ColoredFormatter(logging.Formatter): RESET = "\033[0m" def format(self, record): + """ + 格式化日志记录，为状态码添加颜色 + + 如果日志记录包含 status_code 属性，则查找消息中的 "status=数字" 模式 + 并为其添加对应的颜色代码。格式化完成后恢复原始消息以避免影响其他 handler。 + + Args: + record: 日志记录对象，应包含 msg 属性和可选的 status_code 属性 + + Returns: + str: 格式化后的日志消息字符串，状态码部分可能包含 ANSI 颜色代码 + """ # 保存原始消息 original_msg = record.msg # 如果日志记录包含 status_code 属性，则为其添加颜色 @@ -44,7 +68,17 @@ class ColoredFormatter(logging.Formatter): class ExcludeWatchFilesFilter(logging.Filter): + """ + 日志过滤器，用于排除第三方库的日志输出 + + 过滤掉来自 watchfiles、aiosqlite、apscheduler、sqlalchemy 等库的日志记录， + 避免这些库的详细日志污染应用日志输出。 + """ + def __init__(self): + """ + 初始化过滤器，设置需要过滤的日志名称前缀列表 + """ super().__init__() self._filter_starts_with_tulpe = ( "watchfiles", @@ -55,8 +89,17 @@ class ExcludeWatchFilesFilter(logging.Filter): "sqlalchemy.engine.Engine", ) - # 重写过滤方法 def filter(self, record): + """ + 判断是否应该保留该日志记录 + + Args: + record: 日志记录对象，包含 name 属性表示日志来源 + + Returns: + bool: 如果日志名称不以过滤列表中的任何前缀开头则返回 True（保留）， + 否则返回 False（过滤掉） + """ return not record.name.startswith(self._filter_starts_with_tulpe) @@ -64,6 +107,27 @@ _configured_loggers = set() def setup_logging(logger: logging.Logger, log_name: str = "app"): + """ + 配置日志系统，包括文件和控制台输出 + + 为指定的 logger 配置以下功能： + 1. 设置日志级别和格式 + 2. 创建按日期命名的轮转文件处理器（支持文件大小限制和备份数量控制） + 3. 创建带颜色格式化的控制台处理器 + 4. 应用过滤器排除第三方库日志 + + 该函数具有防重复配置机制，同一个 log_name 只会配置一次。 + + Args: + logger: 需要配置的 logging.Logger 实例 + log_name: 日志名称标识，用于区分不同的日志配置和生成日志文件名，默认为 "app" + + Returns: + None + + Raises: + 无显式抛出异常，但可能因文件系统权限问题导致目录创建或文件写入失败 + """ if log_name in _configured_loggers: return _configured_loggers.add(log_name) @@ -110,6 +174,18 @@ def setup_logging(logger: logging.Logger, log_name: str = "app"): def get_logger(name: str) -> logging.Logger: + """ + 获取配置好的日志记录器实例 + + 根据名称获取或创建 logger 实例，并自动调用 setup_logging 进行配置。 + 如果该名称的 logger 已经配置过，则直接返回已配置的实例。 + + Args: + name: 日志记录器名称，通常使用模块名如 __name__ + + Returns: + logging.Logger: 已配置完成的日志记录器实例，可直接用于记录日志 + """ logger = logging.getLogger(name) setup_logging(logger, name) return logger diff --git a/src/chongming/app/core/security.py b/src/chongming/app/core/security.py index 9873361..23ce9ca 100644 --- a/src/chongming/app/core/security.py +++ b/src/chongming/app/core/security.py @@ -1,13 +1,36 @@ -# src/chongming/app/core/security.py import bcrypt + def get_password_hash(password: str) -> str: + """ + 对明文密码进行哈希加密 + + 使用bcrypt算法对密码进行加盐哈希处理,确保密码存储的安全性。 + + Args: + password: 需要加密的明文密码字符串 + + Returns: + 加密后的密码哈希字符串,采用UTF-8编码 + """ salt = bcrypt.gensalt() - hashed = bcrypt.hashpw(password.encode('utf-8'), salt) - return hashed.decode('utf-8') + hashed = bcrypt.hashpw(password.encode("utf-8"), salt) + return hashed.decode("utf-8") + def verify_password(plain_password: str, hashed_password: str) -> bool: + """ + 验证明文密码与哈希密码是否匹配 + + 将明文密码与已存储的哈希密码进行比对,用于用户登录时的密码验证。 + + Args: + plain_password: 用户输入的明文密码 + hashed_password: 数据库中存储的哈希密码 + + Returns: + 密码匹配返回True,否则返回False + """ return bcrypt.checkpw( - plain_password.encode('utf-8'), - hashed_password.encode('utf-8') - ) \ No newline at end of file + plain_password.encode("utf-8"), hashed_password.encode("utf-8") + ) diff --git a/src/chongming/app/core/static_files.py b/src/chongming/app/core/static_files.py index 3c6428a..0880cbe 100644 --- a/src/chongming/app/core/static_files.py +++ b/src/chongming/app/core/static_files.py @@ -5,18 +5,56 @@ from sqlite_vfs.core import SQLiteVFS class SVFSStaticFiles(StaticFiles): + """ + 基于 SQLite VFS 的静态文件服务类 + + 该类继承自 FastAPI 的 StaticFiles，使用 SQLite 虚拟文件系统来存储和提供静态文件。 + 支持单页应用（SPA）的路由回退机制，当请求的文件不存在时，会自动返回 index.html。 + """ + def __init__(self, directory, vfs_db_path="static.svfs"): + """ + 初始化 SVFSStaticFiles 实例 + + 参数: + directory (str): 静态文件的根目录路径 + vfs_db_path (str): SQLite VFS 数据库文件路径，默认为 "static.svfs" + + 返回值: + 无 + """ vfs_db_path = vfs_db_path self._vfs = SQLiteVFS(vfs_db_path, compress=False) super().__init__(directory=directory) async def get_response(self, path: str, scope): + """ + 根据请求路径获取对应的静态文件响应 + + 该方法处理静态文件的请求逻辑，包括： + 1. 路径标准化处理 + 2. 文件存在性检查 + 3. SPA 路由回退支持 + 4. 目录索引处理 + 5. MIME 类型自动识别 + + 参数: + path (str): 请求的文件路径 + scope: ASGI 作用域对象，包含请求上下文信息 + + 返回值: + Response: FastAPI 响应对象，包含文件内容和正确的 Content-Type + + 异常: + HTTPException: 当文件不存在且无法回退到 index.html 时，抛出 404 错误 + """ # 统一路径格式：以 / 开头，并将 Windows 反斜杠转为正斜杠 normalized_path = "/" + path.replace("\\", "/") # 获取文件信息 file_info = self._vfs.get_file_info(normalized_path) + # 处理文件不存在的情况，尝试 SPA 回退机制 if file_info is None: # SPA 回退：如果不是静态资源扩展名，尝试返回 index.html if not any( @@ -38,7 +76,7 @@ class SVFSStaticFiles(StaticFiles): return Response(content=content, media_type="text/html") raise HTTPException(status_code=404, detail="File not found") - # 如果是目录，也返回 index.html（某些情况） + # 处理目录请求，返回目录下的 index.html if file_info["is_directory"]: index_info = self._vfs.get_file_info(f"{normalized_path}/index.html") if index_info: @@ -46,7 +84,7 @@ class SVFSStaticFiles(StaticFiles): return Response(content=content, media_type="text/html") raise HTTPException(status_code=404, detail="Directory index not found") - # 读取文件内容 + # 读取文件内容并确定 MIME 类型 content = self._vfs.read_file(normalized_path) content_type = file_info.get("content_type") or "application/octet-stream" # 简单根据扩展名补充 MIME 类型（可选） diff --git a/src/chongming/app/model/http_error.py b/src/chongming/app/model/http_error.py index 4e3cd53..28f7937 100644 --- a/src/chongming/app/model/http_error.py +++ b/src/chongming/app/model/http_error.py @@ -3,9 +3,29 @@ from typing import Optional class Detail(BaseModel): + """ + HTTP错误详情模型 + + 用于封装HTTP错误的详细信息，包括错误消息和错误代码。 + + Attributes: + message (str): 错误描述信息 + code (int): 错误状态码或业务错误码 + """ + message: str code: int class HTTPError(BaseModel): + """ + HTTP错误响应模型 + + 用于标准化HTTP错误响应的数据结构，符合RESTful API规范。 + + Attributes: + detail (Optional[Detail]): 错误详情对象，包含具体的错误信息和错误码。 + 如果为None，表示没有详细的错误信息。 + """ + detail: Optional[Detail] = None diff --git a/src/chongming/app/model/permission.py b/src/chongming/app/model/permission.py index fb6f4d1..7e1f600 100644 --- a/src/chongming/app/model/permission.py +++ b/src/chongming/app/model/permission.py @@ -8,17 +8,29 @@ if TYPE_CHECKING: from .role import Role -# ---------- 权限表 ---------- class Permission(SQLModel, table=True): - """权限项，如 user:create, order:delete""" + """ + 权限模型类，用于定义系统中的权限项。 + + 权限采用资源:操作的格式进行标识，例如： + - user:create 表示创建用户的权限 + - order:delete 表示删除订单的权限 + + Attributes: + id: 权限的唯一标识符，主键，自增 + name: 权限标识符，唯一且建立索引，格式为"资源:操作" + description: 权限的描述信息，可选字段 + resource: 资源名称，建立索引，如 user、order、product 等 + action: 操作类型，如 create、read、update、delete 等 + roles: 关联的角色列表，通过 RolePermission 中间表实现多对多关系 + """ id: Optional[int] = Field(default=None, primary_key=True) - name: str = Field(unique=True, index=True) # 权限标识符 + name: str = Field(unique=True, index=True) description: Optional[str] = None - resource: str = Field(index=True) # 资源，如 user, order, product - action: str # 操作，如 create, read, update, delete + resource: str = Field(index=True) + action: str - # 多对多关联 roles: List["Role"] = Relationship( back_populates="permissions", link_model=RolePermission ) diff --git a/src/chongming/app/model/relationship_table.py b/src/chongming/app/model/relationship_table.py index f28de56..ba1eb80 100644 --- a/src/chongming/app/model/relationship_table.py +++ b/src/chongming/app/model/relationship_table.py @@ -4,14 +4,32 @@ from sqlmodel import SQLModel, Field # ---------- 关联表 ---------- class UserRole(SQLModel, table=True): - """用户角色关联表""" + """ + 用户角色关联表 + + 用于实现用户与角色之间的多对多关系。 + 每个记录表示一个用户拥有的一个角色。 + + Attributes: + user_id: 用户ID，外键关联到user表的id字段，作为复合主键的一部分 + role_id: 角色ID，外键关联到role表的id字段，作为复合主键的一部分 + """ user_id: UUID = Field(foreign_key="user.id", primary_key=True) role_id: int = Field(foreign_key="role.id", primary_key=True) class RolePermission(SQLModel, table=True): - """角色权限关联表""" + """ + 角色权限关联表 + + 用于实现角色与权限之间的多对多关系。 + 每个记录表示一个角色拥有的一项权限。 + + Attributes: + role_id: 角色ID，外键关联到role表的id字段，作为复合主键的一部分 + permission_id: 权限ID，外键关联到permission表的id字段，作为复合主键的一部分 + """ role_id: int = Field(foreign_key="role.id", primary_key=True) permission_id: int = Field(foreign_key="permission.id", primary_key=True) diff --git a/src/chongming/app/model/role.py b/src/chongming/app/model/role.py index f11a3b3..c730f50 100644 --- a/src/chongming/app/model/role.py +++ b/src/chongming/app/model/role.py @@ -12,6 +12,21 @@ if TYPE_CHECKING: # ---------- 角色表 ---------- class Role(SQLModel, table=True): + """ + 角色模型类，用于定义系统中的用户角色。 + + 该类继承自SQLModel，映射到数据库中的角色表。每个角色可以有多个用户和多个权限， + 通过中间表UserRole和RolePermission实现多对多关系。 + + Attributes: + id: 角色的唯一标识符，主键，自增 + name: 角色名称，具有唯一性约束和索引，例如：admin, manager, operator, viewer + description: 角色的描述信息，可选字段 + is_only_user: 标识该角色是否仅限单个用户使用，默认为False + users: 与该角色关联的用户列表，通过UserRole中间表建立多对多关系 + permissions: 与该角色关联的权限列表，通过RolePermission中间表建立多对多关系 + """ + id: Optional[int] = Field(default=None, primary_key=True) name: str = Field(unique=True, index=True) # admin, manager, operator, viewer description: Optional[str] = None @@ -23,5 +38,15 @@ class Role(SQLModel, table=True): class RoleUpdate(BaseModel): + """ + 角色更新请求模型类，用于接收角色更新时的参数。 + + 该类继承自Pydantic的BaseModel，所有字段均为可选，允许部分更新角色信息。 + + Attributes: + name: 更新后的角色名称，可选字段 + description: 更新后的角色描述信息，可选字段 + """ + name: Optional[str] = None description: Optional[str] = None diff --git a/src/chongming/app/model/todo.py b/src/chongming/app/model/todo.py index e53af47..7fa9b3c 100644 --- a/src/chongming/app/model/todo.py +++ b/src/chongming/app/model/todo.py @@ -5,24 +5,71 @@ from sqlmodel import SQLModel, Field # 定义模型 class TodoBase(SQLModel): + """ + Todo任务的基础模型类，包含任务的通用字段。 + + Attributes: + title: 任务标题，必填字段 + description: 任务描述，可选字段，默认为None + completed: 任务完成状态，布尔值，默认为False + """ + title: str description: Optional[str] = None completed: bool = False class Todo(TodoBase, table=True): + """ + Todo任务的数据库表模型类，继承自TodoBase。 + + 该类对应数据库中的todo表，用于存储和查询任务数据。 + + Attributes: + id: 任务唯一标识符，主键，由数据库自动生成 + """ + id: Optional[int] = Field(default=None, primary_key=True) class TodoCreate(TodoBase): + """ + 创建Todo任务时使用的请求模型类。 + + 继承自TodoBase，用于接收前端传入的新建任务数据。 + 不包含id字段，因为id由数据库自动生成。 + """ + pass class TodoRead(TodoBase): + """ + 读取Todo任务时使用的响应模型类。 + + 继承自TodoBase，用于向客户端返回任务数据。 + 包含完整的任务信息，包括自动生成的id字段。 + + Attributes: + id: 任务唯一标识符，整数类型，必填字段 + """ + id: int class TodoUpdate(SQLModel): + """ + 更新Todo任务时使用的请求模型类。 + + 所有字段均为可选，允许部分更新任务信息。 + 只有提供的字段会被更新，未提供的字段保持原值。 + + Attributes: + title: 任务标题，可选字段 + description: 任务描述，可选字段 + completed: 任务完成状态，可选字段 + """ + title: Optional[str] = None description: Optional[str] = None completed: Optional[bool] = None diff --git a/src/chongming/app/model/user.py b/src/chongming/app/model/user.py index d1fd151..4b15444 100644 --- a/src/chongming/app/model/user.py +++ b/src/chongming/app/model/user.py @@ -11,16 +11,41 @@ if TYPE_CHECKING: class UserBase(SQLModel): + """ + 用户基础模型 + + 定义用户的核心属性，作为其他用户相关模型的基类。 + 包含用户的基本信息和状态标识。 + + Attributes: + email: 用户邮箱地址，唯一且建立索引 + username: 用户名，可选，最大长度50字符 + full_name: 用户全名，可选，最大长度100字符 + is_active: 用户激活状态，默认为True + is_superuser: 是否为超级管理员，默认为False + created_at: 创建时间，可选字段 + """ + 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) # 可自动填充，此处略 + created_at: Optional[str] = Field(default=None) class User(UserBase, table=True): - """用户模型""" + """ + 用户数据库模型 + + 继承自UserBase，用于数据库表映射。 + 包含用户的完整信息以及与角色的多对多关系。 + + Attributes: + id: 用户唯一标识符，UUID类型，自动生成 + hashed_password: 加密后的密码 + roles: 用户关联的角色列表，通过UserRole中间表实现多对多关系 + """ id: UUID = Field(default_factory=uuid4, primary_key=True) hashed_password: str @@ -28,28 +53,75 @@ class User(UserBase, table=True): class UserCreate(UserBase): - """用户创建模型""" + """ + 用户创建请求模型 + + 用于创建新用户时的数据验证和传输。 + 在UserBase基础上增加了密码字段。 + + Attributes: + password: 用户明文密码，用于创建时传入 + """ password: str class UserRead(UserBase): - """用户读取模型""" + """ + 用户读取响应模型 + + 用于向客户端返回用户信息。 + 在UserBase基础上增加了用户ID字段。 + + Attributes: + id: 用户唯一标识符，UUID类型 + """ id: UUID class UserLogin(BaseModel): + """ + 用户登录请求模型 + + 用于处理用户登录时的数据验证。 + + Attributes: + email: 用户邮箱地址 + password: 用户密码 + """ + email: str password: str class Userlogout(BaseModel): + """ + 用户登出响应模型 + + 用于返回登出操作的结果。 + + Attributes: + code: 响应状态码，默认为200表示成功 + message: 响应消息，默认为"logout success" + """ + code: int = 200 message: str = "logout success" class UserSession(BaseModel): + """ + 用户会话信息模型 + + 用于存储和返回用户的会话相关信息。 + + Attributes: + user_id: 用户唯一标识符 + sessions: 会话列表，每个会话为字典类型，键值均为字符串或None + count: 会话总数 + """ + user_id: str sessions: List[Dict[str, Optional[str]]] count: int diff --git a/src/chongming/app/service/auth.py b/src/chongming/app/service/auth.py index c5c68f2..ff0e8ae 100644 --- a/src/chongming/app/service/auth.py +++ b/src/chongming/app/service/auth.py @@ -24,13 +24,24 @@ class AuthService: async def authenticate_user( self, email: str, password: str, session: AsyncSession ) -> Optional[User]: - """验证用户凭据""" + """ + 验证用户凭据 + + 通过邮箱和密码验证用户身份，使用 bcrypt 进行密码哈希比对。 + + Args: + email: 用户邮箱地址 + password: 用户明文密码 + session: 数据库异步会话对象 + + Returns: + 验证成功返回 User 对象，失败返回 None + """ user = await UserService.get_user_by_email(email, session) if not user: return None if not bcrypt.checkpw( - password.encode('utf-8'), - user.hashed_password.encode('utf-8') + password.encode("utf-8"), user.hashed_password.encode("utf-8") ): return None @@ -42,7 +53,20 @@ class AuthService: request: Optional[Request] = None, device_id: Optional[str] = None, ) -> TokenResponse: - """为用户创建访问和刷新令牌""" + """ + 为用户创建访问和刷新令牌 + + 根据用户信息和请求上下文生成 JWT access token 和 refresh token， + 同时记录设备信息用于安全审计。 + + Args: + user: 已验证的用户对象 + request: FastAPI 请求对象，用于提取客户端信息（可选） + device_id: 设备唯一标识符（可选） + + Returns: + TokenResponse 对象，包含 access_token、refresh_token 及过期时间 + """ # 获取用户数据 user_data = { @@ -89,7 +113,17 @@ class AuthService: ) async def validate_access_token(self, token: str) -> Optional[TokenData]: - """验证访问令牌""" + """ + 验证访问令牌 + + 校验 JWT access token 的有效性和完整性，提取用户身份信息。 + + Args: + token: JWT access token 字符串 + + Returns: + 验证成功返回 TokenData 对象（包含 user_id、username、email），失败返回 None + """ validation = self.jwt_cache.validate_token(token, token_type="access") if not validation: return None @@ -107,7 +141,20 @@ class AuthService: session: AsyncSession, request: Optional[Request] = None, ) -> Optional[str]: - """使用刷新令牌获取新的访问令牌""" + """ + 使用刷新令牌获取新的访问令牌 + + 验证 refresh token 的有效性，确认用户仍然存在，然后生成新的 access token。 + 保持原有的设备信息以确保会话连续性。 + + Args: + refresh_token: JWT refresh token 字符串 + session: 数据库异步会话对象 + request: FastAPI 请求对象，用于提取客户端信息（可选） + + Returns: + 验证成功返回新的 access token 字符串，失败返回 None + """ validation = self.jwt_cache.validate_token(refresh_token, token_type="refresh") if not validation: return None @@ -145,15 +192,39 @@ class AuthService: return new_access_token async def logout(self, token: str): - """用户登出 - 使令牌失效""" + """ + 用户登出 - 使令牌失效 + + 将指定的 token 加入黑名单，使其立即失效。 + + Args: + token: 需要失效的 JWT token 字符串 + """ self.jwt_cache.invalidate_token(token) async def logout_all(self, user_id: str): - """使用户的所有令牌失效""" + """ + 使用户的所有令牌失效 + + 强制注销用户的所有活跃会话，适用于密码修改或安全事件场景。 + + Args: + user_id: 用户 ID 字符串 + """ self.jwt_cache.invalidate_user_tokens(str(user_id)) async def get_user_sessions(self, user_id: str): - """获取用户的活跃会话""" + """ + 获取用户的活跃会话 + + 查询指定用户当前所有有效的 token 会话信息。 + + Args: + user_id: 用户 ID 字符串 + + Returns: + 用户活跃会话列表 + """ return self.jwt_cache.get_user_sessions(str(user_id)) @@ -162,7 +233,14 @@ _auth_service = None def get_auth_service() -> AuthService: - """获取认证服务实例（单例模式）""" + """ + 获取认证服务实例（单例模式） + + 确保整个应用中只有一个 AuthService 实例，避免重复初始化。 + + Returns: + AuthService 单例实例 + """ global _auth_service if _auth_service is None: _auth_service = AuthService() diff --git a/src/chongming/app/service/permission.py b/src/chongming/app/service/permission.py index 531b232..e247961 100644 --- a/src/chongming/app/service/permission.py +++ b/src/chongming/app/service/permission.py @@ -11,6 +11,8 @@ from .role import RoleService class PermissionService: + """权限服务类，提供权限的创建、查询、分配和解除绑定等功能""" + @classmethod async def create_permission( cls, @@ -20,6 +22,16 @@ class PermissionService: action: str, description: Optional[str] = None, ) -> Permission: + """ + 创建新的权限记录 + + :param session: 数据库会话对象 + :param name: 权限名称 + :param resource: 资源标识 + :param action: 操作类型 + :param description: 权限描述，可选 + :return: 创建的权限对象 + """ permission = Permission( name=name, resource=resource, action=action, description=description ) @@ -32,12 +44,26 @@ class PermissionService: async def get_permission_by_id( cls, session: AsyncSession, permission_id: int ) -> Optional[Permission]: + """ + 根据ID获取权限 + + :param session: 数据库会话对象 + :param permission_id: 权限ID + :return: 权限对象，如果不存在则返回None + """ return await session.get(Permission, permission_id) @classmethod async def get_permission_by_name( cls, session: AsyncSession, name: str ) -> Optional[Permission]: + """ + 根据名称获取权限 + + :param session: 数据库会话对象 + :param name: 权限名称 + :return: 权限对象，如果不存在则返回None + """ result = await session.execute( select(Permission).where(Permission.name == name) ) @@ -47,6 +73,14 @@ class PermissionService: async def assign_permission_to_role( cls, session: AsyncSession, role_id: int, permission_id: int ) -> RolePermission: + """ + 将权限分配给角色 + + :param session: 数据库会话对象 + :param role_id: 角色ID + :param permission_id: 权限ID + :return: 创建的角色权限关联对象 + """ role_permission = RolePermission(role_id=role_id, permission_id=permission_id) session.add(role_permission) await session.commit() @@ -60,21 +94,41 @@ class PermissionService: user_id: UUID, permission_id: int, ) -> User: + """ + 将权限分配给用户 + + 如果用户没有专属角色，则会为用户创建一个专属角色，然后将权限分配给该角色 + + :param session: 数据库会话对象 + :param user_id: 用户UUID + :param permission_id: 权限ID + :return: 更新后的用户对象 + :raises HTTPException: 当权限或用户不存在时抛出404错误，当角色或权限ID无效时抛出500错误 + """ + # 验证权限是否存在 permission = await cls.get_permission_by_id(session, permission_id) if not permission: raise HTTPException(status_code=404, detail="Permission not found") + + # 验证用户是否存在 user = await UserService.get_user_by_id(user_id, session) if user is None: raise HTTPException(status_code=404, detail="用户不存在") + + # 获取或创建用户的专属角色 role = await RoleService.get_role_by_name(session, str(user_id)) if role is None: role = await RoleService.create_role( session, str(user_id), is_only_user=True ) + + # 验证角色和权限ID的有效性 if role.id is None: raise HTTPException(status_code=500, detail="角色ID无效") if permission.id is None: raise HTTPException(status_code=500, detail="权限ID无效") + + # 将权限分配给角色 await cls.assign_permission_to_role(session, role.id, permission.id) user.roles.append(role) await session.commit() @@ -85,9 +139,21 @@ class PermissionService: async def unbind_permission_from_role( cls, session: AsyncSession, role_id: int, permission_id: int ) -> Role: + """ + 从角色中解除权限绑定 + + :param session: 数据库会话对象 + :param role_id: 角色ID + :param permission_id: 权限ID + :return: 更新后的角色对象 + :raises HTTPException: 当角色不存在时抛出404错误 + """ + # 验证角色是否存在 role = await RoleService.get_role_by_id(session, role_id) if role is None: raise HTTPException(status_code=404, detail="Role not found") + + # 从角色的权限列表中移除指定权限 role.permissions = [p for p in role.permissions if p.id != permission_id] await session.commit() await session.refresh(role) @@ -97,12 +163,27 @@ class PermissionService: async def unbind_permission_from_user( cls, session: AsyncSession, user_id: UUID, permission_id: int ): + """ + 从用户中解除权限绑定 + + 通过找到用户的专属角色，从该角色中移除指定权限 + + :param session: 数据库会话对象 + :param user_id: 用户UUID + :param permission_id: 权限ID + :raises HTTPException: 当用户或角色不存在时抛出404错误 + """ + # 验证用户是否存在 user = await UserService.get_user_by_id(user_id, session) if not user: raise HTTPException(status_code=404, detail="User not found") + + # 获取用户的专属角色 role = await RoleService.get_role_by_name(session, str(user.id)) if role is None: raise HTTPException(status_code=404, detail="Role not found") + + # 从角色权限列表中移除指定权限 role.permissions = [ permission for permission in role.permissions diff --git a/src/chongming/app/service/role.py b/src/chongming/app/service/role.py index 52f200a..0bb6aad 100644 --- a/src/chongming/app/service/role.py +++ b/src/chongming/app/service/role.py @@ -10,6 +10,8 @@ from ..service.user import UserService class RoleService: + """角色服务类，提供角色管理相关的业务逻辑""" + @staticmethod async def create_role( session: AsyncSession, @@ -17,6 +19,18 @@ class RoleService: description: Optional[str] = None, is_only_user: bool = False, ) -> Role: + """ + 创建新角色 + + Args: + session: 数据库会话对象 + name: 角色名称 + description: 角色描述，可选 + is_only_user: 是否仅用于用户，默认为False + + Returns: + Role: 创建成功的角色对象 + """ role = Role(name=name, description=description, is_only_user=is_only_user) session.add(role) await session.commit() @@ -30,6 +44,14 @@ class RoleService: user_id: UUID, role_id: int, ): + """ + 为用户分配角色 + + Args: + session: 数据库会话对象 + user_id: 用户UUID + role_id: 角色ID + """ user_role = UserRole(user_id=user_id, role_id=role_id) session.add(user_role) await session.commit() @@ -37,6 +59,16 @@ class RoleService: @classmethod async def get_role_by_name(cls, session: AsyncSession, name: str) -> Optional[Role]: + """ + 根据角色名称查询角色 + + Args: + session: 数据库会话对象 + name: 角色名称 + + Returns: + Optional[Role]: 找到的角色对象，未找到则返回None + """ result = await session.execute(select(Role).where(Role.name == name)) return result.scalar_one_or_none() @@ -44,6 +76,16 @@ class RoleService: async def get_role_by_id( cls, session: AsyncSession, role_id: int ) -> Optional[Role]: + """ + 根据角色ID查询角色 + + Args: + session: 数据库会话对象 + role_id: 角色ID + + Returns: + Optional[Role]: 找到的角色对象，未找到则返回None + """ result = await session.get(Role, role_id) return result @@ -53,9 +95,25 @@ class RoleService: user_id: UUID, role_id: int, ): + """ + 从用户解绑角色 + + Args: + session: 数据库会话对象 + user_id: 用户UUID + role_id: 角色ID + + Returns: + User: 更新后的用户对象 + + Raises: + HTTPException: 当用户不存在时抛出404错误 + """ user = await UserService.get_user_by_id(user_id, session) if user is None: raise HTTPException(status_code=404, detail="User not found") + + # 过滤掉指定角色，保留其他角色 user.roles = [role for role in user.roles if role.id != role_id] await session.commit() await session.refresh(user) @@ -66,14 +124,30 @@ class RoleService: role_id: int, session: AsyncSession, ) -> bool: + """ + 删除角色 + + Args: + role_id: 角色ID + session: 数据库会话对象 + + Returns: + bool: 删除成功返回True + + Raises: + HTTPException: 当角色被使用时抛出400错误，角色不存在时抛出404错误 + """ + # 检查角色是否被用户使用 result = await session.execute( select(UserRole).where(UserRole.role_id == role_id) ) if result.scalars().all(): raise HTTPException(status_code=400, detail="Role is not empty") + role = await RoleService.get_role_by_id(session, role_id) if role is None: raise HTTPException(status_code=404, detail="Role not found") + await session.delete(role) await session.commit() return True @@ -85,9 +159,25 @@ class RoleService: role_update: RoleUpdate, session: AsyncSession, ) -> Optional[Role]: + """ + 更新角色信息 + + Args: + role_id: 角色ID + role_update: 角色更新数据对象 + session: 数据库会话对象 + + Returns: + Optional[Role]: 更新后的角色对象 + + Raises: + HTTPException: 当角色不存在时抛出404错误 + """ role = await cls.get_role_by_id(session, role_id) if role is None: raise HTTPException(status_code=404, detail="Role not found") + + # 只更新提供的字段，保持原有值不变 role.name = role_update.name if role_update.name else role.name role.description = ( role_update.description if role_update.description else role.description diff --git a/src/chongming/app/service/todo.py b/src/chongming/app/service/todo.py index 18a29a5..d88fc38 100644 --- a/src/chongming/app/service/todo.py +++ b/src/chongming/app/service/todo.py @@ -7,8 +7,20 @@ from ..model.todo import TodoCreate, Todo, TodoUpdate class TodoService: + """待办事项服务类，提供待办事项的增删改查操作""" + @staticmethod async def create_todo(todo: TodoCreate, session: AsyncSession) -> Todo: + """ + 创建新的待办事项 + + Args: + todo: 待办事项创建数据对象，包含待办事项的初始信息 + session: 异步数据库会话对象，用于执行数据库操作 + + Returns: + Todo: 创建成功后的待办事项对象，包含生成的ID等信息 + """ db_todo = Todo.model_validate(todo) session.add(db_todo) await session.commit() @@ -21,6 +33,17 @@ class TodoService: limit: int, session: AsyncSession, ) -> Sequence[Todo]: + """ + 获取待办事项列表（支持分页） + + Args: + offset: 查询偏移量，指定从第几条记录开始查询 + limit: 查询限制数量，指定最多返回多少条记录 + session: 异步数据库会话对象，用于执行数据库操作 + + Returns: + Sequence[Todo]: 待办事项对象列表，按查询条件返回的结果集 + """ result = await session.execute(select(Todo).offset(offset).limit(limit)) todos = result.scalars().all() return todos @@ -30,6 +53,16 @@ class TodoService: todo_id: int, session: AsyncSession, ) -> Optional[Todo]: + """ + 根据ID获取单个待办事项 + + Args: + todo_id: 待办事项的唯一标识ID + session: 异步数据库会话对象，用于执行数据库操作 + + Returns: + Optional[Todo]: 如果找到则返回待办事项对象，否则返回None + """ todo = await session.get(Todo, todo_id) return todo @@ -39,9 +72,22 @@ class TodoService: todo_update: TodoUpdate, session: AsyncSession, ) -> Optional[Todo]: + """ + 更新待办事项信息 + + Args: + todo_id: 待办事项的唯一标识ID + todo_update: 待办事项更新数据对象，包含需要更新的字段信息 + session: 异步数据库会话对象，用于执行数据库操作 + + Returns: + Optional[Todo]: 如果更新成功则返回更新后的待办事项对象，如果待办事项不存在则返回None + """ 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(): @@ -57,6 +103,16 @@ class TodoService: todo_id: int, session: AsyncSession, ) -> bool: + """ + 删除待办事项 + + Args: + todo_id: 待办事项的唯一标识ID + session: 异步数据库会话对象，用于执行数据库操作 + + Returns: + bool: 如果删除成功返回True，如果待办事项不存在返回False + """ todo = await session.get(Todo, todo_id) if not todo: return False diff --git a/src/chongming/app/service/uploads.py b/src/chongming/app/service/uploads.py index a9b9e3b..bb8236b 100644 --- a/src/chongming/app/service/uploads.py +++ b/src/chongming/app/service/uploads.py @@ -13,7 +13,22 @@ config = get_config() class UploadService: + """ + 文件上传服务类 + + 提供文件上传功能，支持将上传的文件保存到指定目录， + 并生成唯一的文件名以避免冲突。 + """ + def __init__(self, upload_dir: str, chunk_size: int, file_type: str): + """ + 初始化上传服务 + + Args: + upload_dir: 文件上传的根目录路径 + chunk_size: 文件读取的块大小（字节），用于控制内存使用 + file_type: 文件类型标识，用于创建对应的子目录 + """ self.upload_dir = upload_dir self.file_type = file_type self.chunk_size = chunk_size @@ -21,12 +36,35 @@ class UploadService: UPLOAD_DIR.mkdir(parents=True, exist_ok=True) async def upload(self, file_data: UploadFile): + """ + 异步上传文件 + + 接收上传的文件数据，生成唯一文件名，并将文件分块写入到服务器磁盘。 + 上传成功后返回文件的访问URL和新文件名。 + + Args: + file_data: FastAPI的UploadFile对象，包含上传的文件数据和元信息 + + Returns: + tuple: 包含两个元素的元组 + - str: 文件的完整访问URL路径 + - str: 生成的新文件名（UUID + 原始扩展名） + + Raises: + HTTPException: + - 400: 当文件名为空或无效时抛出 + - 500: 当文件写入过程中发生异常时抛出 + """ if file_data.filename is None: raise HTTPException(status_code=400, detail="Invalid file name") + + # 提取文件扩展名并生成唯一的UUID文件名 ext = os.path.splitext(file_data.filename)[1].lower() new_filename = f"{uuid.uuid4().hex}{ext}" file_path = os.path.join(self.upload_dir, self.file_type, new_filename) + try: + # 分块读取上传文件并异步写入磁盘，避免大文件占用过多内存 async with aiofiles.open(file_path, "wb") as out_file: while chunk := await file_data.read(self.chunk_size): await out_file.write(chunk) @@ -34,6 +72,8 @@ class UploadService: raise HTTPException(status_code=500, detail=str(e)) finally: await file_data.close() + + # 构建并返回文件的访问URL和新文件名 return ( f"{config['default']['prefix']}/{self.file_type}/{new_filename}".replace( "\\", "/" diff --git a/src/chongming/app/service/user.py b/src/chongming/app/service/user.py index bff3658..b751680 100644 --- a/src/chongming/app/service/user.py +++ b/src/chongming/app/service/user.py @@ -9,19 +9,56 @@ from ..core.security import get_password_hash class UserService: + """用户服务类，提供用户相关的业务逻辑操作""" + @staticmethod async def get_user_by_email(email: str, session: AsyncSession) -> Optional[User]: + """ + 根据邮箱地址查询用户 + + Args: + email: 用户的邮箱地址 + session: 异步数据库会话 + + Returns: + 如果找到则返回用户对象，否则返回None + """ 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: UUID, session: AsyncSession) -> Optional[User]: + """ + 根据用户ID查询用户 + + Args: + user_id: 用户的UUID标识 + session: 异步数据库会话 + + Returns: + 如果找到则返回用户对象，否则返回None + """ return await session.get(User, user_id) @staticmethod async def create_user( session: AsyncSession, email: str, password: str, **kwargs ) -> User: + """ + 创建新用户 + + Args: + session: 异步数据库会话 + email: 用户的邮箱地址 + password: 用户的明文密码，将自动进行哈希加密 + **kwargs: 其他用户属性参数 + + Returns: + 创建成功的用户对象 + + Raises: + IntegrityError: 当邮箱已存在时可能抛出完整性错误 + """ hashed = get_password_hash(password) user = User(email=email, hashed_password=hashed, **kwargs) session.add(user) diff --git a/src/chongming/app/task/__init__.py b/src/chongming/app/task/__init__.py index a835188..77aee61 100644 --- a/src/chongming/app/task/__init__.py +++ b/src/chongming/app/task/__init__.py @@ -8,6 +8,20 @@ config = get_config() async def init_tasks_callback(task_service: TaskService): + """ + 初始化任务回调函数 + + 在开发环境下自动添加数据库初始化定时任务。 + + Args: + task_service (TaskService): 任务服务实例，用于管理和调度定时任务 + + Returns: + None: 此函数不返回任何值 + + Note: + 仅在开发环境（env == "development"）下执行数据库初始化任务 + """ if config["default"]["env"] == "development": await task_service.add_date_job( diff --git a/src/chongming/app/task/dev_init_db.py b/src/chongming/app/task/dev_init_db.py index 659eb86..09afda5 100644 --- a/src/chongming/app/task/dev_init_db.py +++ b/src/chongming/app/task/dev_init_db.py @@ -10,6 +10,23 @@ logger = get_logger("scheduler") async def dev_init_admin(session): + """ + 初始化系统管理员用户 + + 检查是否存在邮箱为"admin"的管理员用户，如果不存在则创建一个新的超级管理员账户。 + 该函数用于系统首次部署时的管理员账户初始化。 + + Args: + session: 数据库会话对象，用于执行数据库操作 + + Returns: + None + + Note: + - 默认管理员邮箱和密码均为"admin" + - 如果管理员已存在，则跳过创建并记录日志 + - 新创建的用户具有超级管理员权限（is_superuser=True） + """ admin_user = await UserService.get_user_by_email("admin", session) if admin_user: # type: ignore logger.info("管理员用户已存在") # type: ignore @@ -18,6 +35,24 @@ async def dev_init_admin(session): async def init_permission(session): + """ + 初始化系统权限数据 + + 遍历所有预定义的权限常量，检查数据库中是否已存在对应权限记录。 + 对于不存在的权限，创建新的权限记录并关联对应的资源和操作类型。 + + Args: + session: 数据库会话对象，用于执行数据库操作 + + Returns: + None + + Note: + - 权限格式为"resource:action"，例如"user:create" + - 使用冒号分隔资源和操作，如果缺少操作部分则默认为空字符串 + - 已存在的权限会被跳过，避免重复创建 + - 每个新创建的权限都会记录到日志中 + """ for permission in PermissionConstant.get_all(): parts = permission.split(":", 1) resource = parts[0] @@ -35,6 +70,21 @@ async def init_permission(session): async def init_db(): + """ + 执行数据库初始化流程 + + 协调执行所有数据库初始化任务，包括管理员账户创建和权限数据初始化。 + 该函数是系统启动时数据库初始化的入口点。 + + Returns: + None + + Note: + - 通过任务服务获取异步会话工厂 + - 如果会话工厂未初始化，则直接返回不执行任何操作 + - 按顺序执行：先初始化管理员，再初始化权限 + - 使用异步上下文管理器确保会话正确关闭 + """ task_service = get_task_service_instance() if task_service.async_session_maker is None: return diff --git a/src/chongming/scripts/build.py b/src/chongming/scripts/build.py index b1b2474..c0f43cb 100644 --- a/src/chongming/scripts/build.py +++ b/src/chongming/scripts/build.py @@ -1,21 +1,31 @@ -import platform - -from module_bank import PythonToSQLite -from module_bank.encryption import Encryption import os -from pathlib import Path +import platform import shutil import subprocess import sys import asyncio +from pathlib import Path + +from module_bank import PythonToSQLite +from module_bank.encryption import Encryption from sqlite_vfs.core import SQLiteVFS from sqlite_vfs.folder_packer import FolderPacker is_windows = platform.system() == "Windows" + def build_vue_web(): - """打包 Vue Web""" + """ + 打包 Vue Web 前端项目 + + 该函数执行以下操作: + 1. 运行 npm build 命令构建 Vue 项目 + 2. 将构建产物打包到 SQLite VFS 虚拟文件系统中 + + Returns: + None: 无返回值,通过打印信息反馈执行结果 + """ # 构建 Vue Web 命令 dist_dir = "./src/chongming-web/dist" output_db = "./build/static.svfs" @@ -43,12 +53,12 @@ def build_vue_web(): print("Vue Web 打包成功") if result.stdout: print(result.stdout) - + # 检查 dist 目录是否存在 if not os.path.exists(dist_dir): print(f"错误: dist 目录不存在: {dist_dir}") return - + vfs = SQLiteVFS(output_db, compress=False) packer = FolderPacker( sqlite_vfs=vfs, @@ -58,12 +68,27 @@ def build_vue_web(): try: packer.pack_folder(dist_dir) - print(f"✅ 打包完成，文件保存为 {output_db}") + print(f"✅ 打包完成,文件保存为 {output_db}") finally: vfs.close() def run_pyarmor_obfuscate(): + """ + 使用 PyArmor 对关键 Python 文件进行代码混淆 + + 该函数遍历预定义的文件列表,对每个文件执行 pyarmor gen 命令进行混淆处理, + 以保护源代码不被轻易反编译。 + + 处理的文件包括: + - main.py (主入口文件) + - server.py (服务器文件) + - utils/launch.py (启动工具) + - utils/config.py (配置工具) + + Returns: + None: 无返回值,遇到错误时提前返回 + """ # pyarmor obfuscate --output=obfuscated_script.py original_script.py obfuscate_files = [ ("main.py", "./"), @@ -89,7 +114,18 @@ def run_pyarmor_obfuscate(): def run_pyinstaller(): - """运行 PyInstaller 打包成可执行文件""" + """ + 使用 PyInstaller 将应用打包成单文件可执行程序 + + 该函数根据操作系统平台(Windows/非Windows)构建不同的 PyInstaller 命令, + 包含所有必要的隐藏导入以确保依赖库被正确打包。 + + Windows 和非 Windows 平台的主要区别: + - 非 Windows 平台额外包含 gunicorn 相关模块 + + Returns: + None: 无返回值,通过打印信息反馈执行结果 + """ try: # 构建 PyInstaller 命令 if is_windows: @@ -230,19 +266,36 @@ def run_pyinstaller(): result = subprocess.run(cmd, cwd="./build", capture_output=True, text=True) if result.returncode == 0: - print("PyInstaller 打包成功！") + print("PyInstaller 打包成功!") print(result.stdout) else: - print("PyInstaller 打包失败！") + print("PyInstaller 打包失败!") print("错误信息:", result.stderr) except FileNotFoundError: - print("错误：未找到 PyInstaller，请先安装 PyInstaller：pip install pyinstaller") + print("错误:未找到 PyInstaller,请先安装 PyInstaller:pip install pyinstaller") except Exception as e: print(f"PyInstaller 执行出错: {e}") async def init_database(): + """ + 异步初始化数据库 + + 该函数执行以下操作: + 1. 从配置中读取数据库连接信息 + 2. 创建异步数据库引擎和会话工厂 + 3. 设置 SQLite 优化参数(WAL 模式和同步策略) + 4. 创建所有 SQLModel 定义的数据库表 + 5. 初始化权限系统和默认管理员账户 + + Raises: + ValueError: 当数据库配置不存在时抛出 + OperationalError: 当数据库操作出现异常时抛出(已存在表的情况除外) + + Returns: + None: 无返回值 + """ from sqlalchemy.ext.asyncio import create_async_engine from sqlmodel import SQLModel, text from sqlalchemy.exc import OperationalError @@ -284,6 +337,23 @@ async def init_database(): async def main(): + """ + 主构建流程函数 + + 该函数协调整个应用的构建过程,按顺序执行以下步骤: + 1. 加密打包 Python 模块(app 和 plugins)到 SQLite 数据库文件 + 2. 创建并准备 build 目录结构 + 3. 复制必要的源文件和配置文件到 build 目录 + 4. 注入动态生成的加密密钥到配置文件 + 5. 对关键文件进行代码混淆 + 6. 使用 PyInstaller 打包成可执行文件 + 7. 清理临时构建文件 + 8. 初始化数据库 + 9. 构建 Vue 前端项目 + + Returns: + None: 无返回值 + """ # 打包模块 key = Encryption.generate_key() resources_path = Path("resources") @@ -367,4 +437,12 @@ async def main(): def build(): + """ + 构建入口函数 + + 该函数作为整个构建流程的入口点,通过 asyncio.run() 启动异步主构建流程。 + + Returns: + None: 无返回值 + """ asyncio.run(main()) diff --git a/src/plugins/jwt/jwt_cache.py b/src/plugins/jwt/jwt_cache.py index d02a034..318cd1e 100644 --- a/src/plugins/jwt/jwt_cache.py +++ b/src/plugins/jwt/jwt_cache.py @@ -26,6 +26,7 @@ class TokenCacheData: 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: @@ -33,9 +34,19 @@ class TokenCacheData: class JWTCache: - """JWT Token 缓存管理器""" + """JWT Token 缓存管理器 + + 负责 JWT Token 的创建、验证、失效和会话管理。 + 使用 Redis 缓存存储 Token 哈希和用户会话索引，支持多设备登录和会话限制。 + """ def __init__(self, config, cache): + """初始化 JWT 缓存管理器 + + Args: + config: 配置字典，包含 security 相关配置项 + cache: 缓存实例，提供 get/set/delete/exists/keys 等方法 + """ self.config = config self.lock = threading.RLock() @@ -46,7 +57,14 @@ class JWTCache: self.algorithm = self.config["security"]["algorithm"] def _json_encode(self, obj): - """自定义 JSON 编码器，处理 datetime 对象""" + """自定义 JSON 编码器，处理 datetime 对象 + + Args: + obj: 需要序列化的对象 + + Returns: + JSON 字符串 + """ def convert_datetime(o): if isinstance(o, datetime): @@ -58,11 +76,25 @@ class JWTCache: return json.dumps(obj, default=convert_datetime) def _json_decode(self, s): - """自定义 JSON 解码器""" + """自定义 JSON 解码器 + + Args: + s: JSON 字符串 + + Returns: + 解码后的 Python 对象 + """ return json.loads(s) def _hash_token(self, token: str) -> str: - """Token 哈希化（避免存储明文 Token）""" + """Token 哈希化（避免存储明文 Token） + + Args: + token: 原始 JWT Token 字符串 + + Returns: + SHA256 哈希值（十六进制字符串） + """ return hashlib.sha256(token.encode()).hexdigest() def create_token( @@ -74,16 +106,18 @@ class JWTCache: user_agent: Optional[str] = None, ip_address: Optional[str] = None, ) -> str: - """ - 创建并缓存 JWT Token + """创建并缓存 JWT Token + + 生成新的 JWT Token，将其哈希值和元数据存储到缓存中， + 并更新用户会话索引。如果超出会话限制，会自动清理最早的会话。 Args: user_id: 用户ID - user_data: 用户数据 + user_data: 用户数据，将嵌入到 Token payload 中 token_type: Token 类型（access 或 refresh） - device_id: 设备ID - user_agent: 用户代理 - ip_address: IP地址 + device_id: 设备ID，用于标识登录设备 + user_agent: 用户代理字符串，记录客户端信息 + ip_address: IP地址，记录登录来源 Returns: JWT Token 字符串 @@ -140,7 +174,11 @@ class JWTCache: return token def _generate_jti(self) -> str: - """生成唯一的 JWT ID""" + """生成唯一的 JWT ID + + Returns: + UUID v4 字符串，用作 Token 的唯一标识符 + """ import uuid return str(uuid.uuid4()) @@ -148,7 +186,16 @@ class JWTCache: def _update_user_session_index( self, user_id: str, token_hash: str, token_type: str ): - """更新用户会话索引""" + """更新用户会话索引 + + 维护每个用户每种 Token 类型的会话列表，限制最大数量为 100。 + 当超过限制时，保留最新的 100 个会话。 + + Args: + user_id: 用户ID + token_hash: Token 的 SHA256 哈希值 + token_type: Token 类型（access 或 refresh） + """ with self.lock: key = f"user:{user_id}:{token_type}" user_tokens = self.cache.get(key, []) @@ -164,7 +211,14 @@ class JWTCache: self.cache.set(key, user_tokens) def _enforce_session_limit(self, user_id: str): - """强制用户会话限制""" + """强制用户会话限制 + + 检查用户的 access 和 refresh token 数量，如果超过配置的最大会话数， + 则删除最早的会话以维持限制。 + + Args: + user_id: 用户ID + """ max_sessions = self.config["security"]["max_sessions_per_user"] access_key = f"user:{user_id}:access" @@ -195,15 +249,22 @@ class JWTCache: def validate_token( self, token: str, token_type: str = "access" ) -> Optional[Dict[str, Any]]: - """ - 验证 JWT Token + """验证 JWT Token + + 执行多层验证：JWT 签名验证、类型检查、缓存存在性检查、 + 活跃状态检查和过期时间检查。验证成功后更新最后使用时间。 Args: - token: JWT Token - token_type: 期望的 Token 类型 + token: JWT Token 字符串 + token_type: 期望的 Token 类型（access 或 refresh） Returns: - 验证成功返回用户数据，失败返回 None + 验证成功返回包含用户信息的字典，包含以下字段： + - user_id: 用户ID + - device_id: 设备ID + - payload: JWT payload 数据 + - cache_data: 缓存中的完整会话数据 + 验证失败返回 None """ try: # 解码验证 Token @@ -259,7 +320,14 @@ class JWTCache: return None def invalidate_token(self, token: str): - """使单个 Token 失效""" + """使单个 Token 失效 + + 通过标记缓存中的 Token 为非活跃状态来实现失效， + 而不是直接删除，以便保留审计信息。 + + Args: + token: 需要失效的 JWT Token 字符串 + """ with self.lock: token_hash = self._hash_token(token) @@ -270,7 +338,16 @@ class JWTCache: self.cache.set(f"token:{token_hash}", cache_data) def invalidate_user_tokens(self, user_id: str, token_type: Optional[str] = None): - """使用户的所有 Token 失效""" + """使用户的所有 Token 失效 + + 批量失效指定用户的所有 Token，可用于用户登出、密码修改等场景。 + 可以指定失效特定类型的 Token，或者同时失效 access 和 refresh token。 + + Args: + user_id: 用户ID + token_type: 可选，指定要失效的 Token 类型（access 或 refresh）。 + 如果不指定，则失效该用户的所有类型 Token + """ with self.lock: if token_type: keys = [f"user:{user_id}:{token_type}"] @@ -292,7 +369,24 @@ class JWTCache: self.cache.set(key, []) def get_user_sessions(self, user_id: str) -> List[Dict[str, Any]]: - """获取用户的所有活跃会话""" + """获取用户的所有活跃会话 + + 遍历用户的所有 Token 会话，过滤出当前活跃且未过期的会话， + 返回包含设备信息、时间戳等详细信息的会话列表。 + + Args: + user_id: 用户ID + + Returns: + 活跃会话列表，每个会话包含以下字段： + - token_type: Token 类型 + - device_id: 设备ID + - user_agent: 用户代理 + - ip_address: IP地址 + - created_at: 创建时间 + - last_used: 最后使用时间 + - expires_at: 过期时间 + """ with self.lock: sessions = [] @@ -325,7 +419,14 @@ class JWTCache: return sessions def cleanup_expired(self): - """清理过期数据""" + """清理过期数据 + + 扫描所有 Token 缓存，删除已过期的 Token 记录。 + 建议定期调用此方法以释放缓存空间。 + + Returns: + 清理的过期 Token 数量 + """ with self.lock: cleaned_count = 0 current_time = datetime.utcnow() @@ -345,7 +446,21 @@ _jwt_cache = None def get_jwt_cache(config=None, cache=None) -> JWTCache: - """获取 JWT 缓存实例（单例模式）""" + """获取 JWT 缓存实例（单例模式） + + 确保整个应用中只有一个 JWTCache 实例，避免重复初始化和资源浪费。 + 首次调用时必须提供 config 和 cache 参数。 + + Args: + config: 配置字典，仅在首次初始化时需要 + cache: 缓存实例，仅在首次初始化时需要 + + Returns: + JWTCache 单例实例 + + Raises: + ValueError: 首次调用时如果 config 或 cache 为空则抛出异常 + """ global _jwt_cache if _jwt_cache is None: if config is None or cache is None: @@ -356,7 +471,10 @@ def get_jwt_cache(config=None, cache=None) -> JWTCache: # Token 响应模型 class TokenResponse(BaseModel): - """Token 响应模型""" + """Token 响应模型 + + 用于登录接口返回 access token 和 refresh token + """ access_token: str refresh_token: str @@ -366,7 +484,10 @@ class TokenResponse(BaseModel): class RefreshTokenResponse(BaseModel): - """刷新 Token 响应模型""" + """刷新 Token 响应模型 + + 用于刷新 token 接口，仅返回新的 access token + """ access_token: str token_type: str = "bearer" @@ -374,7 +495,10 @@ class RefreshTokenResponse(BaseModel): # Token 数据模型 class TokenData(BaseModel): - """Token 数据模型""" + """Token 数据模型 + + 用于从 Token 中提取的用户身份信息 + """ user_id: str username: Optional[str] = None diff --git a/src/plugins/scheduler/scheduler.py b/src/plugins/scheduler/scheduler.py index 82e3b39..1cd3d3c 100644 --- a/src/plugins/scheduler/scheduler.py +++ b/src/plugins/scheduler/scheduler.py @@ -21,6 +21,17 @@ from apscheduler.triggers.date import DateTrigger def acquire_scheduler_lock(): + """ + 获取调度器分布式锁，确保只有一个进程实例运行调度器 + + 通过在临时目录创建锁文件并使用文件系统锁机制，防止多个进程同时启动调度器。 + 支持 Windows 和 Unix/Linux 系统的不同锁实现。 + + Returns: + tuple: 包含两个元素的元组 + - lock_file (file object or None): 锁文件对象，如果获取锁失败则为 None + - bool: 是否成功获取锁，成功为 True，失败为 False + """ lock_path = os.path.join(tempfile.gettempdir(), "chongming_scheduler.lock") lock_file = open(lock_path, "w") try: @@ -39,9 +50,33 @@ def acquire_scheduler_lock(): class TaskService: - """任务调度服务""" + """ + 任务调度服务类 + + 基于 APScheduler 实现的异步任务调度服务，支持多种触发器类型（间隔、Cron、一次性）。 + 使用 SQLite 数据库持久化存储任务信息，确保服务重启后任务不丢失。 + 通过文件锁机制保证多进程环境下只有一个实例执行调度任务。 + """ def __init__(self, config, logger, async_session_maker=None): + """ + 初始化任务调度服务 + + Args: + config (dict): 配置字典，包含调度器相关配置项 + - scheduler.job_store_path (str): 任务存储数据库路径，默认为 "scheduler_jobs.db" + logger (logging.Logger): 日志记录器实例，用于记录调度器运行日志 + async_session_maker (callable, optional): 异步会话工厂函数，用于数据库操作，默认为 None + + Attributes: + async_session_maker: 异步会话工厂函数 + job_store_db_path: 任务存储数据库文件路径 + scheduler: AsyncIOScheduler 调度器实例 + _started (bool): 调度器是否已启动的标志 + lock_file: 锁文件对象 + is_scheduler_worker (bool): 当前进程是否为调度器工作进程 + logger: 日志记录器实例 + """ self.async_session_maker = async_session_maker # 使用独立的 SQLite 数据库存储任务（与业务数据库分开） self.job_store_db_path = config.get("scheduler", {}).get( @@ -69,7 +104,20 @@ class TaskService: self.logger = logger async def start(self, init_tasks_callback: Optional[Callable] = None): - """启动调度器""" + """ + 启动任务调度器 + + 仅在当前进程成功获取调度器锁时启动调度器，避免多进程重复启动。 + 启动后可选择执行初始化任务回调函数。 + + Args: + init_tasks_callback (callable, optional): 初始化任务回调函数，接收 TaskService 实例作为参数。 + 用于在调度器启动后注册初始任务。默认为 None。 + + Note: + - 如果调度器已经启动，不会重复启动 + - 只有成功获取锁的进程才会启动调度器 + """ if self.is_scheduler_worker: if not self._started: self.scheduler.start() @@ -79,7 +127,20 @@ class TaskService: await init_tasks_callback(self) async def shutdown(self, wait: bool = True): - """关闭调度器""" + """ + 关闭任务调度器 + + 优雅地关闭调度器，包括暂停所有运行中的任务、关闭调度器、释放文件锁。 + + Args: + wait (bool): 是否等待正在执行的任务完成后再关闭。 + True 表示等待任务完成，False 表示立即关闭。默认为 True。 + + Note: + - 先暂停所有任务，再关闭调度器 + - 释放文件锁以允许其他进程获取锁 + - 使用 suppress 捕获所有异常，确保关闭过程不会抛出异常 + """ if self.lock_file: with suppress(Exception): for job in await self.get_jobs(): @@ -111,7 +172,28 @@ class TaskService: kwargs: Optional[Dict] = None, replace_existing: bool = True, ) -> Job: - """添加间隔执行任务""" + """ + 添加间隔执行任务 + + 创建按照固定时间间隔重复执行的任务。 + + Args: + func (callable): 要执行的任务函数 + seconds (int): 间隔秒数，默认为 0 + minutes (int): 间隔分钟数，默认为 0 + hours (int): 间隔小时数，默认为 0 + days (int): 间隔天数，默认为 0 + job_id (str, optional): 任务唯一标识符，如果未提供则自动生成 + args (list, optional): 传递给任务函数的位置参数列表，默认为空列表 + kwargs (dict, optional): 传递给任务函数的关键字参数字典，默认为空字典 + replace_existing (bool): 如果存在相同 ID 的任务，是否替换它。默认为 True + + Returns: + Job: 创建的任务对象 + + Example: + >>> await service.add_interval_job(my_func, minutes=5, job_id="cleanup_task") + """ trigger = IntervalTrigger( seconds=seconds, minutes=minutes, hours=hours, days=days ) @@ -142,7 +224,33 @@ class TaskService: kwargs: Optional[Dict] = None, replace_existing: bool = True, ) -> Job: - """添加 cron 表达式任务""" + """ + 添加 Cron 表达式任务 + + 创建按照 Cron 表达式规则执行的任务，支持精确的时间调度。 + + Args: + func (callable): 要执行的任务函数 + job_id (str, optional): 任务唯一标识符，如果未提供则自动生成 + year (int, optional): 年份，支持通配符和范围 + month (int, optional): 月份 (1-12) + day (int, optional): 日期 (1-31) + week (int, optional): 周数 (1-53) + day_of_week (int, optional): 星期几 (0-6, 0 表示周一) + hour (int, optional): 小时 (0-23) + minute (int, optional): 分钟 (0-59) + second (int, optional): 秒 (0-59) + args (list, optional): 传递给任务函数的位置参数列表，默认为空列表 + kwargs (dict, optional): 传递给任务函数的关键字参数字典，默认为空字典 + replace_existing (bool): 如果存在相同 ID 的任务，是否替换它。默认为 True + + Returns: + Job: 创建的任务对象 + + Example: + >>> # 每天凌晨 2 点执行 + >>> await service.add_cron_job(my_func, hour=2, minute=0, job_id="daily_task") + """ trigger = CronTrigger( year=year, month=month, @@ -173,7 +281,27 @@ class TaskService: kwargs: Optional[Dict] = None, replace_existing: bool = True, ) -> Job: - """添加一次性任务""" + """ + 添加一次性任务 + + 创建在指定时间点执行一次的任务，执行后自动移除。 + + Args: + func (callable): 要执行的任务函数 + run_date (datetime, optional): 任务执行的具体时间，如果未提供则立即执行 + job_id (str, optional): 任务唯一标识符，如果未提供则自动生成 + args (list, optional): 传递给任务函数的位置参数列表，默认为空列表 + kwargs (dict, optional): 传递给任务函数的关键字参数字典，默认为空字典 + replace_existing (bool): 如果存在相同 ID 的任务，是否替换它。默认为 True + + Returns: + Job: 创建的任务对象 + + Example: + >>> from datetime import datetime + >>> run_time = datetime(2024, 12, 25, 10, 0, 0) + >>> await service.add_date_job(my_func, run_date=run_time, job_id="christmas_task") + """ trigger = DateTrigger(run_date=run_date) job = self.scheduler.add_job( func, @@ -187,7 +315,21 @@ class TaskService: return job async def remove_job(self, job_id: str) -> bool: - """删除任务""" + """ + 删除任务 + + 从调度器中移除指定任务，任务将不再执行。 + + Args: + job_id (str): 要删除的任务的唯一标识符 + + Returns: + bool: 删除是否成功，成功返回 True，失败返回 False + + Note: + - 如果任务不存在或删除过程中发生异常，返回 False + - 会记录删除操作的日志 + """ try: self.scheduler.remove_job(job_id) self.logger.info(f"删除任务: {job_id}") @@ -197,7 +339,21 @@ class TaskService: return False async def pause_job(self, job_id: str) -> bool: - """暂停任务""" + """ + 暂停任务 + + 暂停指定任务的执行，任务保留在调度器中但不会触发。 + + Args: + job_id (str): 要暂停的任务的唯一标识符 + + Returns: + bool: 暂停是否成功，成功返回 True，失败返回 False + + Note: + - 暂停的任务可以通过 resume_job 恢复 + - 如果任务不存在或暂停过程中发生异常，返回 False + """ try: self.scheduler.pause_job(job_id) self.logger.info(f"暂停任务: {job_id}") @@ -207,7 +363,21 @@ class TaskService: return False async def resume_job(self, job_id: str) -> bool: - """恢复任务""" + """ + 恢复任务 + + 恢复之前被暂停的任务，使其继续按照原定计划执行。 + + Args: + job_id (str): 要恢复的任务的唯一标识符 + + Returns: + bool: 恢复是否成功，成功返回 True，失败返回 False + + Note: + - 只能恢复已被暂停的任务 + - 如果任务不存在或恢复过程中发生异常，返回 False + """ try: self.scheduler.resume_job(job_id) self.logger.info(f"恢复任务: {job_id}") @@ -217,21 +387,72 @@ class TaskService: return False async def get_job(self, job_id: str) -> Optional[Dict]: - """获取任务信息""" + """ + 获取单个任务信息 + + 查询指定任务的详细信息，包括下次执行时间、触发器等。 + + Args: + job_id (str): 要查询的任务的唯一标识符 + + Returns: + dict or None: 任务信息字典，如果任务不存在则返回 None。 + 字典包含以下字段： + - id (str): 任务 ID + - name (str): 任务名称 + - next_run_time (str or None): 下次执行时间的 ISO 格式字符串 + - trigger (str): 触发器描述 + - pending (bool): 任务是否处于待处理状态 + - args (list): 任务函数的位置参数 + - kwargs (dict): 任务函数的关键字参数 + """ job = self.scheduler.get_job(job_id) if job: return self._job_to_dict(job) return None async def get_jobs(self) -> List[Dict]: - """获取所有任务""" + """ + 获取所有任务信息 + + 查询调度器中所有任务的详细信息列表。 + + Returns: + list[dict]: 任务信息字典列表，每个字典的结构与 get_job 返回的相同。 + 如果没有任务，返回空列表。 + """ jobs = self.scheduler.get_jobs() return [self._job_to_dict(job) for job in jobs] async def reschedule_job( self, job_id: str, trigger_type: str, **trigger_args ) -> Optional[Job]: - """重新调度任务（修改触发时间）""" + """ + 重新调度任务（修改触发时间） + + 修改现有任务的触发器配置，改变任务的执行时间规则。 + + Args: + job_id (str): 要重新调度的任务的唯一标识符 + trigger_type (str): 触发器类型，可选值： + - "interval": 间隔触发器 + - "cron": Cron 表达式触发器 + - "date": 一次性触发器 + **trigger_args: 触发器参数，根据 trigger_type 不同而不同： + - interval: seconds, minutes, hours, days + - cron: year, month, day, week, day_of_week, hour, minute, second + - date: run_date + + Returns: + Job or None: 重新调度后的任务对象，如果任务不存在或重新调度失败则返回 None + + Raises: + ValueError: 当传入不支持的 trigger_type 时抛出 + + Example: + >>> # 将任务改为每 10 分钟执行一次 + >>> await service.reschedule_job("task_id", "interval", minutes=10) + """ try: job = self.scheduler.get_job(job_id) if not job: @@ -254,7 +475,25 @@ class TaskService: return None def _job_to_dict(self, job: Job) -> Dict: - """将 Job 对象转换为字典""" + """ + 将 Job 对象转换为字典 + + 内部辅助方法，将 APScheduler 的 Job 对象序列化为字典格式， + 便于 API 返回和数据传输。 + + Args: + job (Job): APScheduler 的 Job 对象 + + Returns: + dict: 包含任务信息的字典，字段包括： + - id (str): 任务 ID + - name (str): 任务名称 + - next_run_time (str or None): 下次执行时间的 ISO 8601 格式字符串 + - trigger (str): 触发器的字符串描述 + - pending (bool): 任务是否处于待处理状态 + - args (list): 任务函数的位置参数列表 + - kwargs (dict): 任务函数的关键字参数字典 + """ return { "id": job.id, "name": job.name, @@ -272,6 +511,28 @@ _task_service: Optional[TaskService] = None def get_task_service_instance(config=None, logger=None, async_session_maker=None): + """ + 获取任务调度服务单例实例 + + 使用全局变量实现单例模式，确保整个应用中只有一个 TaskService 实例。 + 首次调用时创建实例，后续调用直接返回已创建的实例。 + + Args: + config (dict, optional): 配置字典，仅在首次创建实例时需要 + logger (logging.Logger, optional): 日志记录器，仅在首次创建实例时需要 + async_session_maker (callable, optional): 异步会话工厂函数，仅在首次创建实例时需要 + + Returns: + TaskService: 任务调度服务单例实例 + + Raises: + ValueError: 如果首次调用时未提供 config 或 logger 参数 + + Note: + - 这是一个单例模式实现，确保全局只有一个调度器实例 + - 首次调用必须提供 config 和 logger 参数 + - 后续调用可以不提供参数，直接返回已创建的实例 + """ global _task_service if _task_service is None: if config is None or logger is None: @@ -281,4 +542,19 @@ def get_task_service_instance(config=None, logger=None, async_session_maker=None async def get_task_service(request: Request) -> AsyncGenerator: + """ + FastAPI 依赖注入函数，获取任务调度服务实例 + + 用于 FastAPI 路由函数的依赖注入，从应用状态中获取已初始化的任务调度服务。 + + Args: + request (Request): FastAPI 请求对象，从中获取应用实例 + + Returns: + TaskService: 从 request.app.state.task_service 获取的任务调度服务实例 + + Note: + - 此函数应在应用启动时通过 startup 事件初始化 task_service + - 适用于 FastAPI 的 Depends 依赖注入机制 + """ return request.app.state.task_service