GitHub - TrumanDu/myboot: MyBoot - Python rapid development framework inspired by Spring Boot 类似 Spring Boot 的 Python 快速开发框架

MyBoot - 类似 Spring Boot 的 Python 快速开发框架

MyBoot 是一个功能丰富的 Python Web 框架，提供类似 Spring Boot 的自动配置和快速开发功能。它集成了 Web API、定时任务、日志管理、配置管理等核心功能，让您能够快速构建现代化的 Python 应用程序。

✨ 主要特性

• 🚀 快速启动 : 类似 Spring Boot 的自动配置和快速启动

• 🎯 约定优于配置 : 遵循约定，减少配置工作，自动发现和注册组件

• 🌐 Web API : 基于 FastAPI 的高性能 Web API 开发

• ⚡ 高性能服务器 : 默认使用 Hypercorn 服务器，支持 HTTP/2 和多进程

• ⏰ 定时任务 : 强大的任务调度系统，支持 Cron 表达式和间隔任务（详见 任务调度器使用说明 ）

• 📝 日志管理 : 基于 loguru 的强大日志系统，支持结构化日志和第三方库日志控制

• ⚙️ 配置管理 : 基于 Dynaconf 的强大配置系统，支持 YAML 配置、环境变量覆盖和远程配置（详见 配置管理使用说明 ）

• 🔧 中间件支持 : 丰富的中间件生态，包括 CORS、限流、安全等

• 📊 健康检查 : 内置健康检查、就绪检查和存活检查

• 🎯 依赖注入 : 简单的依赖注入和组件管理

• 🔄 优雅关闭 : 支持优雅关闭和资源清理

• 📚 自动文档 : 自动生成 API 文档和交互式界面

🚀 快速开始

安装

pip install myboot

命令行工具

MyBoot 提供了便捷的命令行工具用于初始化项目：

显示帮助信息

myboot --help

初始化新项目（交互式）

myboot init

使用指定模板初始化项目

myboot init --name my-app --template basic # 基础模板 myboot init --name my-app --template api # API 项目模板 myboot init --name my-app --template full # 完整项目模板

显示框架信息

myboot info

创建应用

使用 myboot init 初始化项目后，在 main.py 中创建应用：

"""main.py - 应用入口文件""" from myboot . core . application import create_app

创建应用实例

app = create_app ( name = "我的应用" )

运行应用

if __name__ == "__main__" : app . run ()

在 app/api/ 目录中定义路由（使用 @rest_controller 装饰器）：

"""app/api/routes.py""" from myboot . core . decorators import rest_controller , get , post

@ rest_controller ( '/api' ) class HelloController : """Hello 控制器"""

@ get ( '/' ) def hello ( self ): """Hello World 接口 - GET /api""" return { "message" : "Hello, MyBoot!" , "status" : "success" }

@ rest_controller ( '/api/users' ) class UserController : """用户控制器 - 支持依赖注入"""

def __init__ ( self , user_service : UserService ): self . user_service = user_service

@ get ( '/{user_id}' ) def get_user ( self , user_id : int ): """获取用户 - GET /api/users/{user_id}""" return self . user_service . get_user ( user_id )

运行应用

应用入口文件位于项目根目录的 main.py ：

直接运行

python main.py

启用自动重载（开发环境）

python main.py --reload

指定端口和主机

python main.py --host 0.0.0.0 --port 8080

访问 http://localhost:8000 查看您的应用！

🎯 约定优于配置

MyBoot 框架的核心设计理念是"约定优于配置"，让您能够快速开发而无需复杂的配置。

自动发现和注册

from myboot . core . decorators import service , rest_controller , get , cron , component

@ service () class UserService : """用户服务 - 自动注册为 'user_service'""" def get_user ( self , user_id ): return { "id" : user_id , "name" : f"用户 { user_id } " }

@ rest_controller ( '/api/users' ) class UserController : """用户控制器 - 支持依赖注入"""

def __init__ ( self , user_service : UserService ): self . user_service = user_service

@ get ( '/{user_id}' ) def get_user ( self , user_id : int ): """获取用户 - GET /api/users/{user_id}""" return self . user_service . get_user ( user_id )

@ component () class ScheduledJobs : """定时任务组件 - 使用 @component 装饰器定义定时任务"""

@ cron ( '0 */5 * * * *' ) def cleanup_task ( self ): """清理任务 - 自动注册定时任务""" print ( "执行清理任务" )

零配置启动

from myboot . core . application import Application

创建应用，自动发现和配置所有组件

app = Application ( name = "我的应用" , auto_configuration = True , # 启用自动配置 auto_discover_package = "app" # 自动发现 app 包 )

直接运行，无需手动注册

app . run ()

依赖注入和服务管理

MyBoot 提供了基于 dependency_injector 的强大依赖注入机制，支持自动依赖解析和注入，让您可以轻松管理服务之间的依赖关系。

自动依赖注入

框架会自动检测服务的依赖关系并自动注入，无需手动获取：

from myboot . core . decorators import service

@ service () class UserService : def __init__ ( self ): self . users = {}

@ service () class EmailService : def send_email ( self , to : str , subject : str ): print ( f"发送邮件到 { to } : { subject } " )

@ service () class OrderService :

自动注入 UserService 和 EmailService

def __init__ ( self , user_service : UserService , email_service : EmailService ): self . user_service = user_service self . email_service = email_service

def create_order ( self , user_id : int ): user = self . user_service . get_user ( user_id ) self . email_service . send_email ( user [ 'email' ], "订单创建" , "您的订单已创建" )

特性：

• ✅ 自动检测依赖关系

• ✅ 自动处理依赖顺序

• ✅ 支持多级依赖

• ✅ 支持可选依赖（ Optional[Type] ）

• ✅ 自动检测循环依赖

• ✅ 向后兼容，现有代码无需修改

更多信息： 查看 依赖注入使用指南

获取服务 (get_service)

服务是通过 @service() 装饰器自动注册的。 推荐方式：在控制器构造函数中通过类型注解自动注入。

方式一：依赖注入（推荐）

from myboot . core . decorators import rest_controller , get , service

@ service () class UserService : def get_user ( self , user_id : int ): return { "user_id" : user_id }

@ rest_controller ( '/api/users' ) class UserController : def __init__ ( self , user_service : UserService ):

通过构造函数自动注入服务

self . user_service = user_service

@ get ( '/{user_id}' ) def get_user ( self , user_id : int ): return self . user_service . get_user ( user_id )

方式二：通过全局函数（适用于非控制器场景）

from myboot . core . application import get_service

在启动钩子或其他地方获取服务

def some_function (): user_service = get_service ( 'user_service' ) return user_service . get_user ( 1 )

获取客户端 (get_client)

客户端是通过 @client() 装饰器自动注册的。 推荐方式：在控制器构造函数中通过类型注解自动注入。

方式一：依赖注入（推荐）

from myboot . core . decorators import rest_controller , get , client

@ client () class RedisClient : def get ( self , key : str ): return None

@ rest_controller ( '/api/products' ) class ProductController : def __init__ ( self , redis_client : RedisClient ):

通过构造函数自动注入客户端

self . redis_client = redis_client

@ get ( '/' ) def get_products ( self ): cache_data = self . redis_client . get ( 'products' ) return { "products" : cache_data or []}

方式二：通过全局函数（适用于非控制器场景）

from myboot . core . application import get_client

在启动钩子或其他地方获取客户端

def some_function (): redis_client = get_client ( 'redis_client' ) return redis_client . get ( 'products' )

完整示例

from myboot . core . decorators import service , client , rest_controller , get , post

定义服务

@ service () class UserService : """用户服务 - 自动注册为 'user_service'""" def get_user ( self , user_id : int ): return { "id" : user_id , "name" : f"用户 { user_id } " }

def create_user ( self , name : str , email : str ): return { "name" : name , "email" : email }

@ service () class EmailService : """邮件服务""" def send_email ( self , to : str , subject : str , body : str ): print ( f"发送邮件到 { to } " )

定义客户端

@ client ( 'redis_client' ) class RedisClient : """Redis 客户端 - 注册为 'redis_client'""" def get ( self , key : str ): return None

控制器中使用依赖注入

@ rest_controller ( '/api/users' ) class UserController : """用户控制器 - 自动注入服务和客户端"""

def __init__ ( self , user_service : UserService , email_service : EmailService , redis_client : RedisClient ): self . user_service = user_service self . email_service = email_service self . redis_client = redis_client

@ get ( '/{user_id}' ) def get_user ( self , user_id : int ): """获取用户"""

先检查缓存

cache_key = f"user: { user_id } " cached = self . redis_client . get ( cache_key ) if cached : return cached return self . user_service . get_user ( user_id )

@ post ( '/' ) def create_user ( self , name : str , email : str ): """创建用户""" user = self . user_service . create_user ( name , email ) self . email_service . send_email ( email , "欢迎" , f"欢迎 { name } " ) return { "message" : "用户创建成功" , "user" : user }

服务命名规则

• 默认命名 : 如果未指定名称，服务名会自动转换为类名的小写形式，并使用下划线分隔

• UserService → 'user_service'

• EmailService → 'email_service'

• DatabaseClient → 'database_client'

• RedisClient → 'redis_client'

• 自定义命名 : 可以通过装饰器参数指定名称

• @service('email_service') → 'email_service'

• @client('redis_client') → 'redis_client'

注意事项

• 推荐依赖注入 : 在控制器中推荐使用构造函数依赖注入，代码更清晰、可测试性更好

• 服务必须已注册 : 确保服务或客户端已经通过装饰器注册

• 全局函数适用场景 : get_service() 和 get_client() 适用于启动钩子、工具函数等非控制器场景

• 路由定义 : 所有路由必须在 @rest_controller 装饰的类中定义

约定规则

• 服务命名 : 类名自动转换为下划线分隔的小写形式作为服务名（如 UserService → user_service ）

• 路由映射 : 使用 @rest_controller 装饰器定义路由，方法装饰器 @get 、 @post 等定义具体端点

• 任务调度 : 在 @component 类中使用 @cron 、 @interval 、 @once 装饰器（详见 任务调度器使用说明 ）

• 组件扫描 : 自动扫描指定包中的所有组件

⚡ 高性能服务器

MyBoot 默认使用 Hypercorn 作为 ASGI 服务器，提供卓越的性能和特性：

服务器特性

• 高性能 : 基于 Hypercorn 的高性能 ASGI 服务器

• HTTP/2 支持 : 支持现代 HTTP 协议

• WebSocket 支持 : 支持实时通信

• 多进程支持 : 支持多工作进程，适合生产环境

• 自动重载 : 开发环境支持自动重载

• 优雅关闭 : 支持优雅关闭和资源清理

使用示例

from myboot . core . application import Application

创建应用

app = Application ( name = "我的应用" )

开发环境（单进程 + 自动重载）

app . run ( host = "0.0.0.0" , port = 8000 , reload = True , workers = 1 )

生产环境（多进程）

app . run ( host = "0.0.0.0" , port = 8000 , workers = 4 )

或者直接运行 main.py

python main.py --reload # 开发环境

python main.py --workers 4 # 生产环境

⚙️ 配置管理

MyBoot 使用 Dynaconf 提供强大的配置管理功能。完整说明（文件优先级、环境变量 __ 规则、常用配置项）见 配置管理使用说明 。

基本使用

from myboot . core . config import get_settings , get_config

直接使用 Dynaconf settings（自动查找配置文件）

settings = get_settings () app_name = settings . app . name server_port = settings . server . port

使用便捷函数

database_url = get_config ( 'database.url' , 'sqlite:///./app.db' ) debug_mode = get_config ( 'app.debug' , False )

指定配置文件路径

settings = get_settings ( 'custom_config.yaml' )

通过环境变量指定配置文件

export CONFIG_FILE=/path/to/config.yaml

或

export CONFIG_FILE=https://example.com/config.yaml

环境变量覆盖

环境变量可以直接覆盖配置值（使用 __ 作为分隔符），优先级高于所有配置文件：

使用环境变量覆盖配置值

export APP__NAME= " MyApp "

export SERVER__PORT=9000

export LOGGING__LEVEL=DEBUG

嵌套配置使用双下划线分隔

export SERVER__CORS__ALLOW_ORIGINS= ' ["http://localhost:3000"] '

注意 ：环境变量覆盖配置值的优先级最高，会覆盖所有配置文件中的对应值。

远程配置

from myboot . core . config import get_settings

从远程 URL 加载配置

settings = get_settings ( 'https://example.com/config.yaml' )

配置优先级

MyBoot 按照以下优先级查找和加载配置文件：

- 环境变量 CONFIG_FILE （最高优先级）

• 通过环境变量指定配置文件路径或 URL

export CONFIG_FILE=/path/to/config.yaml

或

export CONFIG_FILE=https://example.com/config.yaml

- 参数指定的配置文件

• 通过 create_app() 或 get_settings() 的 config_file 参数指定

app = create_app ( name = "我的应用" , config_file = "custom_config.yaml" )

- 项目根目录 /conf 目录下的配置文件

• 项目根目录/conf/config.yaml

• 项目根目录/conf/config.yml

- 项目根目录下的配置文件

• 项目根目录/config.yaml

• 项目根目录/config.yml

- 默认配置

• 内置的默认配置值

注意 ：环境变量还可以直接覆盖配置值（使用 __ 作为分隔符），优先级高于所有配置文件：

export APP__NAME= " MyApp "

export SERVER__PORT=9000

export LOGGING__LEVEL=DEBUG

📖 详细文档

• 📚 完整文档 - 文档中心

• ⚡ REST API 异步任务 - REST API 中使用异步任务指南

• 🔧 依赖注入 - 依赖注入使用指南

• ⏰ 任务调度器 - Cron / 间隔 / 一次性任务与配置说明

• ⚙️ 配置管理 - YAML、环境变量、.env 与远程配置优先级

• 🧩 多 Worker 模式 - 作用域、worker 钩子、primary-first 协调、任务级 all_workers

• 📊 Metrics 指南 - Prometheus 指标、多进程聚合、自定义指标 API

🐳 Docker 部署

myboot init （ api / full 模板）会自动生成 Dockerfile 、 .dockerignore 和 docker-compose.yaml ，开箱即可容器化：

docker build -t my-app .

docker run -p 8000:8000 my-app

或用 compose

docker compose up --build

Dockerfile 采用多阶段构建（uv 装依赖 → python:3.12-slim 运行）。容器中可用 环境变量覆盖配置（ __ 为层级分隔符），无需改镜像：

docker run -p 8080:8080 -e SERVER__PORT=8080 -e LOGGING__LEVEL=DEBUG my-app

🧩 多 Worker 能力（0.2.0+）

多 worker 模式下每个 worker 进程拥有独立的 service / client 实例（issue #11）。 完整说明见 多 Worker 模式指南 ，能力速览：

• 作用域 ： @service(scope="singleton"|"request"|"factory") 声明实例生命周期；

• Worker 钩子 ： @on_worker_start / @on_worker_stop 每个 worker 各触发一次；

• Client 自动清理 ：client 定义 close() 方法即可在 worker 停止时被框架自动调用，建连可直接写在 __init__ ；

• Primary-first 协调 ： run_primary_first(name, primary_fn, secondary_fn) ——primary 干重活（如下载模型），其余 worker 等完成后轻量加载；

• 任务级 all_workers ： @interval(hours=24, all_workers=True) 让单个定时任务在每个 worker 都执行（如刷新进程内缓存），其余任务保持仅 primary 执行一份。

📊 Prometheus Metrics（0.2.0+）

pip install myboot[metrics] 后配置 metrics.enabled: true 即自动获得

/metrics 端点与 HTTP 请求指标，多 worker 自动聚合；业务代码可用 get_counter / time_stage 记录自定义指标。详见 Metrics 指南 。

另外 0.2.0 起框架 自动加载项目根目录 .env （真实环境变量优先）， main.py 无需手动 load_dotenv() ，详见 配置管理 。

⬆️ 从 0.1.x 升级到 0.2.0

本次升级保持公开 API 兼容、不影响现有功能，仅以下两处为有意的破坏性变更：

cron_{函数名} 改为 cron_{模块名}.{限定名} （ interval_ / date_ 同理）。 不同类中的同名方法不再冲突。若你的代码按旧格式硬编码 ID 调用 remove_job / get_job_info ，需改用新格式（或在注册时显式传入 job_id ）。

• 定时任务默认 job_id 格式变更 ：未显式指定 job_id 时，默认 ID 从

myboot.exceptions 中的定义， myboot.web.exceptions 仍可 import（同一类）。 myboot.web.exceptions.ValidationError 旧有的 error_type 参数已移除， 以核心版签名 (message, field, value, details) 为准。

• 异常类收敛 ： ValidationError 、 ConfigurationError 统一为

其余修复（多 worker 实例独立化、内置配置默认值生效、 create_http_exception / 模块级函数路由 / get_job_info 等死路径修复、logger 同时支持 %s 与 {} 占位符）均为行为修复，不需要改动你的代码。

1. Web API 开发

重要 ：路由必须在 @rest_controller 装饰的类中定义，支持依赖注入。

REST 控制器（推荐方式）

from myboot . core . decorators import rest_controller , get , post , put , delete , service from myboot . web . models import BaseResponse

@ service () class UserService : """用户服务""" def get_users ( self ): return []

def get_user ( self , user_id : int ): return { "user_id" : user_id , "name" : f"用户 { user_id } " }

def create_user ( self , name : str , email : str ): return { "name" : name , "email" : email }

def update_user ( self , user_id : int , ** kwargs ): return { "user_id" : user_id , ** kwargs }

def delete_user ( self , user_id : int ): return { "user_id" : user_id }

@ rest_controller ( '/api/users' ) class UserController : """用户控制器 - 自动注入 UserService"""

def __init__ ( self , user_service : UserService ): self . user_service = user_service

@ get ( '/' ) def get_users ( self ): """获取用户列表 - GET /api/users""" users = self . user_service . get_users () return BaseResponse ( success = True , message = "获取用户列表成功" , data = { "users" : users })

@ get ( '/{user_id}' ) def get_user ( self , user_id : int ): """获取单个用户 - GET /api/users/{user_id}""" user = self . user_service . get_user ( user_id ) return BaseResponse ( success = True , message = "获取用户成功" , data = user )

@ post ( '/' ) def create_user ( self , name : str , email : str ): """创建用户 - POST /api/users""" user = self . user_service . create_user ( name , email ) return BaseResponse ( success = True , message = "用户创建成功" , data = user )

@ put ( '/{user_id}' ) def update_user ( self , user_id : int , name : str = None , email : str = None ): """更新用户 - PUT /api/users/{user_id}""" update_data = {} if name : update_data [ 'name' ] = name if email : update_data [ 'email' ] = email user = self .

user_service . update_user ( user_id , ** update_data ) return BaseResponse ( success = True , message = f"用户 { user_id } 更新成功" , data = user )

@ delete ( '/{user_id}' ) def delete_user ( self , user_id : int ): """删除用户 - DELETE /api/users/{user_id}""" user = self . user_service . delete_user ( user_id ) return BaseResponse ( success = True , message = f"用户 { user_id } 删除成功" , data = user )

约定优于配置说明：

1. 使用 @rest_controller 装饰器定义控制器类和基础路径

2. 使用 @get, @post, @put, @delete 装饰器定义路由方法

3. 构造函数参数自动进行依赖注入

4. 框架自动发现和注册控制器

5. 统一的响应格式和错误处理

REST 控制器

使用 @rest_controller 装饰器可以创建 REST 控制器类，为类中的方法提供统一的基础路径。类中的方法需要显式使用 @get 、 @post 、 @put 、 @delete 、 @patch 等装饰器才会生成路由。

基本用法：

from myboot . core . decorators import rest_controller , get , post , put , delete from myboot . web . models import BaseResponse

@ rest_controller ( '/api/users' ) class UserController : """用户控制器"""

def __init__ ( self ):

可以在这里初始化服务、客户端等

pass

@ get ( '/' ) def list_users ( self ): """获取用户列表 - GET /api/users""" return BaseResponse ( success = True , message = "获取用户列表成功" , data = { "users" : []} )

@ get ( '/{user_id}' ) def get_user ( self , user_id : int ): """获取单个用户 - GET /api/users/{user_id}""" return BaseResponse ( success = True , message = "获取用户成功" , data = { "user_id" : user_id , "name" : f"用户 { user_id } " } )

@ post ( '/' ) def create_user ( self , name : str , email : str ): """创建用户 - POST /api/users""" return BaseResponse ( success = True , message = "用户创建成功" , data = { "name" : name , "email" : email } )

@ put ( '/{user_id}' ) def update_user ( self , user_id : int , name : str = None , email : str = None ): """更新用户 - PUT /api/users/{user_id}""" return BaseResponse ( success = True , message = f"用户 { user_id } 更新成功" , data = { "user_id" : user_id , "name" : name , "email" : email } )

@ delete ( '/{user_id}' ) def delete_user ( self , user_id : int ): """删除用户 - DELETE /api/users/{user_id}""" return BaseResponse ( success = True , message = f"用户 { user_id } 删除成功" , data = { "user_id" : user_id } )

路径合并规则：

• 方法路径以 // 开头：作为绝对路径使用（去掉一个 / ）

• 方法路径以 / 开头：去掉开头的 / 后追加到基础路径

• 方法路径不以 / 开头：直接追加到基础路径

示例：

@ rest_controller ( '/api/reports' ) class ReportController : """报告控制器"""

@ post ( '/generate' ) # 最终路径: POST /api/reports/generate def create_report ( self , report_type : str ): return { "message" : "报告生成任务已创建" , "type" : report_type }

@ get ( '/status/{job_id}' ) # 最终路径: GET /api/reports/status/{job_id} def get_status ( self , job_id : str ): return { "status" : "completed" , "job_id" : job_id }

@ get ( '//health' ) # 最终路径: GET /health (绝对路径) def health_check ( self ): return { "status" : "ok" }

在控制器中使用依赖注入：

from myboot . core . decorators import rest_controller , get , post , service , client from myboot . web . models import BaseResponse

@ service () class ProductService : def get_all ( self ): return [] def create ( self , name : str , price : float ): return { "name" : name , "price" : price }

@ client () class RedisClient : def set ( self , key : str , value ): pass

@ rest_controller ( '/api/products' ) class ProductController : """产品控制器 - 使用依赖注入"""

def __init__ ( self , product_service : ProductService , redis_client : RedisClient ):

通过构造函数自动注入

self . product_service = product_service self . redis_client = redis_client

@ get ( '/' ) def list_products ( self ): """获取产品列表""" products = self . product_service . get_all () return BaseResponse ( success = True , data = { "products" : products })

@ post ( '/' ) def create_product ( self , name : str , price : float ): """创建产品""" product = self . product_service . create ( name , price ) self . redis_client .

set ( f"product: { product [ 'name' ] } " , product ) return BaseResponse ( success = True , data = { "product" : product })

注意事项：

• 显式装饰器 ：类中的方法必须显式使用 @get 、 @post 等装饰器才会生成路由

• 路径合并 ：方法路径会自动与基础路径合并，形成最终的路由路径

• 自动注册 ：控制器类会被自动发现和注册，无需手动配置

• 依赖注入 ：在构造函数中声明类型注解，框架自动注入服务和客户端

数据模型

from pydantic import BaseModel from typing import Optional from myboot . core . decorators import rest_controller , post from myboot . web . models import BaseResponse

class User ( BaseModel ): """用户数据模型""" id : Optional [ int ] = None name : str email : str age : Optional [ int ] = None

@ rest_controller ( '/api/users' ) class UserController : @ post ( '/' ) def create_user ( self , user : User ): """创建用户""" return BaseResponse ( success = True , message = "用户创建成功" , data = user . dict ())

分页处理

from myboot . core . decorators import rest_controller , get from myboot . web . models import BaseResponse from typing import Optional

@ rest_controller ( '/api/users' ) class UserController : @ get ( '/' ) def get_users ( self , page : int = 1 , size : int = 10 , search : Optional [ str ] = None ): """获取用户列表（分页） - GET /api/users"""

处理分页逻辑

return BaseResponse ( success = True , message = "获取用户列表成功" , data = { "users" : [], "total" : 0 , "page" : page , "size" : size } )

1. 定时任务

重要 ：定时任务必须在 @component 装饰的类中定义，支持依赖注入。完整说明见 任务调度器使用说明 。

Cron 表达式任务

from myboot . core . decorators import component , cron , interval , once from myboot . core . config import get_config

@ component () class ScheduledJobs : """定时任务组件"""

@ cron ( "0 0 * * * *" , enabled = True ) # 每小时执行 def hourly_task ( self ): print ( "每小时任务" )

从配置文件读取 enabled 状态

@ cron ( "0 0 2 * * *" , enabled = get_config ( 'jobs.cleanup_task.enabled' , True )) def daily_backup ( self ): """每天凌晨2点执行""" print ( "每日备份" )

间隔任务

@ component () class MonitorJobs : """监控任务组件"""

@ interval ( seconds = 30 , enabled = True ) # 每30秒执行 def heartbeat ( self ): print ( "心跳检测" )

@ interval ( minutes = 5 , enabled = get_config ( 'jobs.monitor.enabled' , True )) def monitor ( self ): """每5分钟执行""" print ( "系统监控" )

一次性任务

@ component () class OneTimeJobs : """一次性任务组件"""

@ once ( "2025-12-31 23:59:59" , enabled = True ) def new_year_task ( self ): """新年任务 - 过期后不再执行""" print ( "新年任务" )

带依赖注入的定时任务

from myboot . core . decorators import component , service , cron

@ service () class DataService : def sync_data ( self ): print ( "同步数据..." )

@ component () class DataSyncJobs : """数据同步任务 - 自动注入 DataService"""

def __init__ ( self , data_service : DataService ): self . data_service = data_service

@ cron ( "0 2 * * *" ) # 每天凌晨2点 def sync_daily ( self ): self . data_service . sync_data ()

1. 配置管理

文件合并顺序、环境变量命名与 API 说明见 配置管理使用说明 。

配置文件 (config.yaml)

应用配置

app : name : " 我的应用 " version : " 1.0.0 " debug : true

服务器配置

server : host : " 0.0.0.0 " port : 8000 reload : true

数据库配置

database : url : " sqlite:///./app.db " pool_size : 10

任务调度配置

scheduler : enabled : true # 是否启用调度器 timezone : " Asia/Shanghai " # 时区设置（可选，需要安装 pytz） max_workers : 10 # 最大工作线程数

任务启用配置（可选）

jobs : heartbeat : enabled : true cleanup_task : enabled : false monitor : enabled : true

日志配置

logging : level : " INFO " # 日志级别: DEBUG, INFO, WARNING, ERROR, CRITICAL format : " {time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message} " file : " logs/app.log " # 可选，如果配置会自动添加文件 handler

第三方库日志级别配置

third_party : urllib3 : " WARNING " requests : " WARNING " hypercorn : " WARNING "

配置使用

from myboot . core . config import get_settings , get_config , get_config_bool , get_config_str

方式一：使用 get_settings() 获取完整配置对象

settings = get_settings () port = settings . get ( "server.port" , 8000 ) debug = settings . get ( "app.debug" , False )

方式二：使用便捷函数（推荐）

port = get_config ( "server.port" , 8000 ) debug = get_config_bool ( "app.debug" , False ) db_url = get_config_str ( "database.url" , "sqlite:///./app.db" )

在应用实例中也可以直接使用

port = app . config . get ( "server.port" , 8000 )

调度器配置

Cron 表达式格式、星期字段含义及多 Worker 行为见 任务调度器使用说明 。

任务调度配置

scheduler : enabled : true # 是否启用调度器 timezone : " Asia/Shanghai " # 时区设置（需要安装 pytz） max_workers : 10 # 最大工作线程数

获取调度器配置

config = app . scheduler . get_config () print ( config ) # {'enabled': True, 'timezone': 'Asia/Shanghai', ...}

列出所有任务

jobs = app . scheduler . list_all_jobs () for job in jobs : print ( job )

获取单个任务信息

job_info = app . scheduler . get_job_info ( 'cron_app.jobs.ScheduledJobs.heartbeat' ) # 0.2.0 起默认 ID 为 cron_{模块名}.{限定名} print ( job_info )

任务启用控制

任务装饰器支持 enabled 参数，可以控制任务是否启用：

from myboot . core . decorators import component , cron , interval , once from myboot . core . config import get_config

@ component () class TaskControlDemo : """任务控制示例"""

方式一：直接指定

@ cron ( "0 */1 * * * *" , enabled = True ) # 启用 def enabled_task ( self ): print ( "启用状态" )

@ interval ( minutes = 2 , enabled = False ) # 禁用 def disabled_task ( self ): print ( "禁用状态" )

方式二：从配置文件读取

@ once ( "2025-01-01 00:00:00" , enabled = get_config ( 'jobs.my_task.enabled' , True )) def configurable_task ( self ): print ( "可配置任务" )

注意 ：

• 定时任务必须在 @component 装饰的类中定义

• 如果 enabled 为 None ，默认启用

• 一次性任务如果时间已过期，将自动标记为过期不再执行

• 已执行的一次性任务不会重复执行

1. 日志管理

MyBoot 使用 loguru 作为日志系统，提供强大的日志功能和优雅的 API。

基本使用

使用框架导出的 logger

from myboot . core . logger import logger

logger . info ( "应用启动" ) logger . error ( "发生错误" ) logger . debug ( "调试信息" ) logger . warning ( "警告信息" )

日志配置

日志系统会在应用启动时自动根据配置文件初始化，无需手动配置。

配置文件示例 (config.yaml):

日志配置

logging :

日志级别: DEBUG, INFO, WARNING, ERROR, CRITICAL

level : " INFO "

日志格式（支持 loguru 格式或标准 logging 格式，会自动转换）

如果设置了 json: true，此选项将被忽略

format : " {time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message} "

是否使用 JSON 格式输出（适用于日志聚合和分析工具）

设置为 true 时，日志以 JSON 格式输出，包含完整的结构化信息

json : false

日志文件路径（可选，如果配置会自动添加文件 handler，支持自动轮转）

file : " logs/app.log "

第三方库日志级别配置（用于控制第三方库的日志输出）

third_party : urllib3 : " WARNING " # 只显示 WARNING 及以上级别 requests : " WARNING " # 只显示 WARNING 及以上级别 hypercorn : " WARNING " # 只显示 WARNING 及以上级别 hypercorn.error : " WARNING " # hypercorn.error logger asyncio : " INFO " # 显示 INFO 及以上级别

JSON 格式日志

启用 JSON 格式后，日志会以结构化 JSON 格式输出，便于日志聚合和分析工具（如 ELK、Loki、Grafana 等）处理：

logging : level : " INFO " json : true # 启用 JSON 格式输出 file : " logs/app.log "

JSON 格式日志包含以下字段：

• text : 格式化的日志文本

• record : 完整的日志记录对象

• time : 时间戳

• level : 日志级别

• message : 日志消息

• name : logger 名称

• module : 模块名

• function : 函数名

• file : 文件名和路径

• line : 行号

• process : 进程信息

• thread : 线程信息

• exception : 异常信息（如果有）

日志格式说明

Loguru 格式（推荐）：

format : "{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}"

标准 logging 格式（会自动转换）：

format : "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

高级功能

from myboot . core . logger import logger

结构化日志

logger . info ( "用户登录" , user_id = 123 , username = "admin" )

异常日志（自动包含堆栈跟踪）

try : 1 / 0 except : logger . exception ( "发生错误" )

绑定上下文信息

logger . bind ( user_id = 123 ). info ( "用户操作" )

临时修改日志级别

with logger . contextualize ( level = "DEBUG" ): logger . debug ( "这是调试信息" )

添加自定义 handler（保留用户自定义 loguru 的能力）

from myboot . core . logger import logger logger . add ( "custom.log" , rotation = "100 MB" , retention = "30 days" )

手动初始化（可选）

如果需要手动初始化日志系统：

from myboot . core . logger import setup_logging

使用默认配置初始化

setup_logging ()

使用指定配置文件初始化

setup_logging ( "custom_config.yaml" )

配置参数说明

参数 类型 说明 默认值

logging.level str 日志级别 (DEBUG/INFO/WARNING/ERROR/CRITICAL) INFO

logging.format str 日志格式（支持 loguru 格式或标准 logging 格式，json 为 true 时忽略） loguru 默认格式

logging.json bool 是否使用 JSON 格式输出（适用于日志聚合和分析工具） false

logging.file str 日志文件路径，如果配置会自动添加文件 handler 无

logging.third_party.{library} str 第三方库日志级别，支持设置任意第三方库的日志级别 无

文件日志特性

如果配置了 logging.file ，loguru 会自动提供：

• 自动轮转 : 当日志文件达到 10MB 时自动轮转

• 自动压缩 : 旧日志文件自动压缩为 zip

• 自动清理 : 保留 7 天的日志文件

• 异常信息 : 自动包含完整的堆栈跟踪

第三方库日志控制

通过 logging.third_party 配置可以控制第三方库的日志输出级别：

logging : third_party : urllib3 : " WARNING " # 隐藏 urllib3 的 INFO 和 DEBUG 日志 requests : " WARNING " # 隐藏 requests 的 INFO 和 DEBUG 日志 hypercorn : " WARNING " # 隐藏 hypercorn 的 INFO 和 DEBUG 日志 asyncio : " INFO " # 只显示 asyncio 的 INFO 及以上级别

这样可以有效减少第三方库的噪音日志，让日志更加清晰。

1. 中间件

MyBoot 支持通过装饰器定义中间件，中间件会自动注册：

from myboot . core . decorators import middleware from fastapi import Request

@ middleware ( order = 1 , path_filter = '/api/*' ) def api_middleware ( request : Request , next_handler ): """API 中间件 - 只处理 /api/* 路径"""

前置处理

print ( f"处理请求: { request . method } { request . url } " )

调用下一个处理器

response = next_handler ( request )

后置处理

print ( f"响应状态: { response . status_code } " ) return response

@ middleware ( order = 2 , methods = [ 'POST' , 'PUT' ]) def post_middleware ( request : Request , next_handler ): """POST/PUT 中间件 - 只处理 POST 和 PUT 请求"""

可以在这里添加请求验证、日志记录等

return next_handler ( request )

中间件参数说明：

• order : 执行顺序，数字越小越先执行（默认 0）

• path_filter : 路径过滤，支持字符串、字符串列表或正则表达式，如 '/api/*' , ['/api/*', '/admin/*']

• methods : HTTP 方法过滤，如 ['GET', 'POST'] （默认 None，处理所有方法）

• condition : 条件函数，接收 request 对象，返回 bool 决定是否执行中间件

注意 ：CORS 中间件可以通过配置文件启用，无需手动添加。

1. 生命周期钩子

from myboot . core . application import create_app

app = create_app ( name = "我的应用" )

添加启动钩子

def startup_hook (): """应用启动时执行""" print ( "应用启动" )

app . add_startup_hook ( startup_hook )

添加关闭钩子

def shutdown_hook (): """应用关闭时执行""" print ( "应用关闭" )

app . add_shutdown_hook ( shutdown_hook )

📁 项目结构

标准项目结构（推荐）

使用 myboot init 命令创建的标准项目结构：

my-app/ ├── main.py # 应用入口（根目录） ├── pyproject.toml # 项目配置文件 ├── .gitignore # Git 忽略文件 ├── app/ # 应用代码 │ ├── api/ # API 路由 │ ├── service/ # 业务逻辑层 │ ├── model/ # 数据模型 │ ├── jobs/ # 定时任务 │ └── client/ # 客户端（第三方API调用等） ├── conf/ # 配置文件目录 │ └── config.yaml # 主配置文件 └── tests/ # 测试代码

目录说明

• main.py : 应用入口文件，位于项目根目录

• app/api/ : API 路由层，存放所有路由定义

• app/service/ : 业务逻辑层，存放业务服务类

• app/model/ : 数据模型层，存放 Pydantic 模型等

• app/jobs/ : 定时任务组件，存放使用 @component 装饰的类，类中方法可使用 @cron 、 @interval 等装饰器

• app/client/ : 客户端层，存放第三方服务客户端（如 Redis、HTTP 客户端等）

• conf/ : 配置文件目录，存放 YAML 配置文件

• tests/ : 测试代码目录

🔧 高级功能

1. 自定义中间件

使用装饰器定义中间件（推荐方式）：

from myboot . core . decorators import middleware from fastapi import Request

@ middleware ( order = 1 ) def custom_middleware ( request : Request , next_handler ): """自定义中间件"""

前置处理

print ( f"请求: { request . method } { request . url } " )

调用下一个处理器

response = next_handler ( request )

后置处理

print ( f"响应: { response . status_code } " ) return response

或者使用 FastAPI 的 BaseHTTPMiddleware：

from myboot . web . middleware import Middleware from starlette . middleware . base import BaseHTTPMiddleware from fastapi import Request

class CustomMiddleware ( BaseHTTPMiddleware ): async def dispatch ( self , request : Request , call_next ):

中间件逻辑

response = await call_next ( request ) return response

在应用初始化时添加

app . add_middleware ( Middle

— 本文由 AI 根据公开来源辅助整理，命令、版本与许可证请在使用前到原始页面复核。