litefs.middleware package

class litefs.middleware.AuthMiddleware(app, auth_header: str = 'Authorization', token_verifier: Callable[[str], bool] | None = None, jwt_manager=None)

基类：Middleware

认证中间件

基于 Bearer Token 的认证中间件，支持自定义验证函数和 JWT 集成

process_request(request_handler)

处理请求，检查认证信息

参数:

request_handler -- 请求处理器实例

返回:

如果认证失败，返回 401 响应 否则返回 None，继续处理请求

class litefs.middleware.CORSMiddleware(app, allow_origins: List[str] | None = None, allow_methods: List[str] | None = None, allow_headers: List[str] | None = None, expose_headers: List[str] | None = None, allow_credentials: bool = False, max_age: int = 86400)

基类：Middleware

CORS 中间件

处理跨域资源共享（CORS）请求，支持配置允许的来源、方法、头部等

process_request(request_handler)

处理请求，检查是否为预检请求

参数:

request_handler -- 请求处理器实例

返回:

如果是预检请求，返回预检响应 否则返回 None，继续处理请求

process_response(request_handler, response)

处理响应，添加 CORS 响应头

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

添加了 CORS 响应头的响应数据

class litefs.middleware.CSRFMiddleware(app, secret_key: str | None = None, token_name: str = 'csrf_token', header_name: str = 'X-CSRFToken', cookie_name: str = 'csrftoken', cookie_secure: bool = False, cookie_http_only: bool = True, cookie_same_site: str = 'Lax', exempt_methods: list | None = None)

基类：Middleware

CSRF 保护中间件

提供跨站请求伪造防护，通过生成和验证 CSRF 令牌来保护表单提交

process_request(request_handler)

处理请求，验证 CSRF 令牌

参数:

request_handler -- 请求处理器实例

返回:

如果 CSRF 验证失败，返回 403 响应 否则返回 None，继续处理请求

process_response(request_handler, response)

处理响应，设置 CSRF 令牌 cookie

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

添加了 CSRF 令牌 cookie 的响应数据

class litefs.middleware.EnhancedLoggingMiddleware(app, logger: Logger | None = None, structured: bool = False, sensitive_params: List[str] | None = None, exclude_paths: List[str] | None = None, log_request_body: bool = False, log_response_body: bool = False, max_body_length: int = 1000)

基类：Middleware

增强的日志中间件

功能特性： - 请求追踪：为每个请求生成唯一 ID - 结构化日志：支持 JSON 格式输出 - 性能监控：记录详细的性能指标 - 敏感信息过滤：自动过滤敏感参数 - 智能日志级别：根据响应状态码自动调整

SENSITIVE_PARAMS = ['password', 'passwd', 'pwd', 'token', 'access_token', 'refresh_token', 'secret', 'api_key', 'apikey', 'credit_card', 'card_number', 'ssn', 'social_security_number']

process_exception(request_handler, exception)

处理异常，记录异常信息

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常

返回类型:

None

process_request(request_handler)

处理请求，记录请求开始信息

参数:

request_handler -- 请求处理器实例

process_response(request_handler, response)

处理响应，记录请求处理结果

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

响应数据

class litefs.middleware.HealthCheck(app, path: str = '/health', ready_path: str = '/health/ready')

基类：Middleware

健康检查中间件

提供 /health 和 /health/ready 端点用于健康检查

add_check(name: str, check_func: Callable[[], bool])

添加健康检查

参数:

• name -- 检查名称

• check_func -- 检查函数，返回 True 表示健康，False 表示不健康

add_ready_check(name: str, check_func: Callable[[], bool])

添加就绪检查

参数:

• name -- 检查名称

• check_func -- 检查函数，返回 True 表示就绪，False 表示未就绪

process_request(request_handler)

处理请求，检查是否为健康检查端点

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应

返回类型:

None

class litefs.middleware.LoggingMiddleware(app, logger: Logger | None = None)

基类：Middleware

日志中间件

记录所有请求和响应的详细信息，包括请求时间、响应时间、状态码等

process_exception(request_handler, exception)

处理异常，记录异常信息

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常

返回类型:

None

process_request(request_handler)

处理请求，记录请求开始时间

参数:

request_handler -- 请求处理器实例

process_response(request_handler, response)

处理响应，记录请求处理时间

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

响应数据

class litefs.middleware.Middleware(app)

基类：object

中间件基类

所有中间件都应该继承此类并实现 process_request 和/或 process_response 方法。

支持同步和异步两种模式： - 同步模式：实现 process_request / process_response / process_exception - 异步模式：实现 async_process_request / async_process_response / async_process_exception

异步方法优先级高于同步方法。在 ASGI 路径中，如果中间件实现了异步方法， 将使用异步调用；否则回退到同步调用（通过 run_in_executor 避免阻塞事件循环）。

async async_process_exception(request_handler, exception)

处理异常（异步版本）

默认实现调用同步版本的 process_exception。 子类可以覆盖此方法来实现真正的异步逻辑。

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常 其他值: 返回该值作为响应，不抛出异常

返回类型:

None

async async_process_request(request_handler)

处理请求，在请求到达业务逻辑之前执行（异步版本）

默认实现调用同步版本的 process_request。 子类可以覆盖此方法来实现真正的异步逻辑。

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应，中断后续处理

返回类型:

None

async async_process_response(request_handler, response)

处理响应，在响应返回给客户端之前执行（异步版本）

默认实现调用同步版本的 process_response。 子类可以覆盖此方法来实现真正的异步逻辑。

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

修改后的响应数据

process_exception(request_handler, exception)

处理异常（同步版本）

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常 其他值: 返回该值作为响应，不抛出异常

返回类型:

None

process_request(request_handler)

处理请求，在请求到达业务逻辑之前执行（同步版本）

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应，中断后续处理

返回类型:

None

process_response(request_handler, response)

处理响应，在响应返回给客户端之前执行（同步版本）

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

修改后的响应数据

class litefs.middleware.MiddlewareManager

基类：object

中间件管理器

负责管理所有中间件的注册、加载和执行。 同时支持同步（WSGI）和异步（ASGI）调用路径。

add(middleware_class, **kwargs)

添加中间件

参数:

• middleware_class -- 中间件类

• **kwargs -- 传递给中间件构造函数的参数

返回:

支持链式调用

返回类型:

self

async async_process_exception(request_handler, exception)

按顺序执行所有中间件的异常处理（异步）

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常 其他值: 返回该值作为响应

返回类型:

None

async async_process_request(request_handler)

按顺序执行所有中间件的请求处理（异步）

对于实现了 async_process_request 的中间件，使用 await 调用； 对于仅有同步 process_request 的中间件，使用 run_in_executor 避免阻塞事件循环。

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应

返回类型:

None

async async_process_response(request_handler, response)

按逆序执行所有中间件的响应处理（异步）

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

处理后的响应数据

clear()

清空所有中间件

get_middleware_instances(app)

获取所有中间件实例

参数:

app -- Litefs 应用实例

返回:

中间件实例列表

process_exception(request_handler, exception)

按顺序执行所有中间件的 process_exception 方法（同步）

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常 其他值: 返回该值作为响应

返回类型:

None

process_request(request_handler)

按顺序执行所有中间件的 process_request 方法（同步）

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应

返回类型:

None

process_response(request_handler, response)

按逆序执行所有中间件的 process_response 方法（同步）

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

处理后的响应数据

remove(middleware_class)

移除中间件

参数:

middleware_class -- 中间件类

class litefs.middleware.PrometheusMiddleware(app, path: str = '/metrics', namespace: str = 'litefs', buckets: Sequence[float] | None = None, exclude_paths: Sequence[str] | None = None)

基类：Middleware

Prometheus 监控中间件

挂载方式:

app.add_middleware(PrometheusMiddleware)

指定暴露路径与命名空间:

app.add_middleware(PrometheusMiddleware, path='/internal/metrics',
                   namespace='myapp')

async async_process_exception(request_handler, exception)

异步版本的处理异常

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常

返回类型:

None

async async_process_request(request_handler)

异步版本的处理请求

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 三元组: 指标端点命中时直接返回响应

返回类型:

None

async async_process_response(request_handler, response)

异步版本的处理响应

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

未修改的响应数据

process_exception(request_handler, exception)

处理异常，补记一次错误指标

业务异常通常在路由执行阶段被框架捕获并转为 500 响应， 此处仅作为兜底（如 ASGI 的 HttpError），避免指标重复计数。

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常

返回类型:

None

process_request(request_handler)

处理请求，记录起始时间并响应指标端点

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 三元组: 指标端点命中时直接返回响应

返回类型:

None

process_response(request_handler, response)

处理响应，记录请求指标

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

未修改的响应数据

class litefs.middleware.RateLimitMiddleware(app, max_requests: int = 100, window_seconds: int = 60, key_func: callable | None = None, block_duration: int = 60)

基类：Middleware

限流中间件

基于令牌桶算法实现请求限流，支持按 IP 地址或用户限流

process_request(request_handler)

处理请求，检查是否超过限流

参数:

request_handler -- 请求处理器实例

返回:

如果超过限流，返回 429 响应 否则返回 None，继续处理请求

class litefs.middleware.RequestContextLogger(request_handler)

基类：object

请求上下文日志记录器

提供在请求上下文中记录日志的便捷方法

debug(message: str, **kwargs)

记录 DEBUG 级别日志

参数:

• message -- 日志消息

• **kwargs -- 额外的上下文数据

error(message: str, **kwargs)

记录 ERROR 级别日志

参数:

• message -- 日志消息

• **kwargs -- 额外的上下文数据

info(message: str, **kwargs)

记录 INFO 级别日志

参数:

• message -- 日志消息

• **kwargs -- 额外的上下文数据

warning(message: str, **kwargs)

记录 WARNING 级别日志

参数:

• message -- 日志消息

• **kwargs -- 额外的上下文数据

class litefs.middleware.SecurityMiddleware(app, x_frame_options: str = 'DENY', x_content_type_options: str = 'nosniff', x_xss_protection: str = '1; mode=block', strict_transport_security: str = 'max-age=31536000; includeSubDomains', content_security_policy: str | None = None, referrer_policy: str = 'strict-origin-when-cross-origin', permissions_policy: str | None = None)

基类：Middleware

安全中间件

添加各种安全相关的 HTTP 响应头，提高应用安全性

process_response(request_handler, response)

处理响应，添加安全响应头

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

添加了安全响应头的响应数据

class litefs.middleware.ThrottleMiddleware(app, min_interval: float = 0.1, key_func: callable | None = None)

基类：Middleware

节流中间件

控制请求的处理速率，防止服务器过载

process_request(request_handler)

处理请求，检查是否需要节流

参数:

request_handler -- 请求处理器实例

返回:

如果需要节流，返回 429 响应 否则返回 None，继续处理请求

litefs.middleware.log_async_performance(func)

异步性能日志装饰器

用于记录异步函数执行时间

参数:

func -- 被装饰的异步函数

返回:

装饰后的异步函数

litefs.middleware.log_performance(func)

性能日志装饰器

用于记录函数执行时间

参数:

func -- 被装饰的函数

返回:

装饰后的函数

Submodules

litefs.middleware.base module

class litefs.middleware.base.Middleware(app)

基类：object

中间件基类

所有中间件都应该继承此类并实现 process_request 和/或 process_response 方法。

支持同步和异步两种模式： - 同步模式：实现 process_request / process_response / process_exception - 异步模式：实现 async_process_request / async_process_response / async_process_exception

异步方法优先级高于同步方法。在 ASGI 路径中，如果中间件实现了异步方法， 将使用异步调用；否则回退到同步调用（通过 run_in_executor 避免阻塞事件循环）。

async async_process_exception(request_handler, exception)

处理异常（异步版本）

默认实现调用同步版本的 process_exception。 子类可以覆盖此方法来实现真正的异步逻辑。

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常 其他值: 返回该值作为响应，不抛出异常

返回类型:

None

async async_process_request(request_handler)

处理请求，在请求到达业务逻辑之前执行（异步版本）

默认实现调用同步版本的 process_request。 子类可以覆盖此方法来实现真正的异步逻辑。

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应，中断后续处理

返回类型:

None

async async_process_response(request_handler, response)

处理响应，在响应返回给客户端之前执行（异步版本）

默认实现调用同步版本的 process_response。 子类可以覆盖此方法来实现真正的异步逻辑。

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

修改后的响应数据

process_exception(request_handler, exception)

处理异常（同步版本）

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常 其他值: 返回该值作为响应，不抛出异常

返回类型:

None

process_request(request_handler)

处理请求，在请求到达业务逻辑之前执行（同步版本）

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应，中断后续处理

返回类型:

None

process_response(request_handler, response)

处理响应，在响应返回给客户端之前执行（同步版本）

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

修改后的响应数据

class litefs.middleware.base.MiddlewareManager

基类：object

中间件管理器

负责管理所有中间件的注册、加载和执行。 同时支持同步（WSGI）和异步（ASGI）调用路径。

add(middleware_class, **kwargs)

添加中间件

参数:

• middleware_class -- 中间件类

• **kwargs -- 传递给中间件构造函数的参数

返回:

支持链式调用

返回类型:

self

async async_process_exception(request_handler, exception)

按顺序执行所有中间件的异常处理（异步）

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常 其他值: 返回该值作为响应

返回类型:

None

async async_process_request(request_handler)

按顺序执行所有中间件的请求处理（异步）

对于实现了 async_process_request 的中间件，使用 await 调用； 对于仅有同步 process_request 的中间件，使用 run_in_executor 避免阻塞事件循环。

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应

返回类型:

None

async async_process_response(request_handler, response)

按逆序执行所有中间件的响应处理（异步）

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

处理后的响应数据

clear()

清空所有中间件

get_middleware_instances(app)

获取所有中间件实例

参数:

app -- Litefs 应用实例

返回:

中间件实例列表

process_exception(request_handler, exception)

按顺序执行所有中间件的 process_exception 方法（同步）

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常 其他值: 返回该值作为响应

返回类型:

None

process_request(request_handler)

按顺序执行所有中间件的 process_request 方法（同步）

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应

返回类型:

None

process_response(request_handler, response)

按逆序执行所有中间件的 process_response 方法（同步）

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

处理后的响应数据

remove(middleware_class)

移除中间件

参数:

middleware_class -- 中间件类

litefs.middleware.cors module

class litefs.middleware.cors.CORSMiddleware(app, allow_origins: List[str] | None = None, allow_methods: List[str] | None = None, allow_headers: List[str] | None = None, expose_headers: List[str] | None = None, allow_credentials: bool = False, max_age: int = 86400)

基类：Middleware

CORS 中间件

处理跨域资源共享（CORS）请求，支持配置允许的来源、方法、头部等

process_request(request_handler)

处理请求，检查是否为预检请求

参数:

request_handler -- 请求处理器实例

返回:

如果是预检请求，返回预检响应 否则返回 None，继续处理请求

process_response(request_handler, response)

处理响应，添加 CORS 响应头

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

添加了 CORS 响应头的响应数据

litefs.middleware.health_check module

class litefs.middleware.health_check.HealthCheck(app, path: str = '/health', ready_path: str = '/health/ready')

基类：Middleware

健康检查中间件

提供 /health 和 /health/ready 端点用于健康检查

add_check(name: str, check_func: Callable[[], bool])

添加健康检查

参数:

• name -- 检查名称

• check_func -- 检查函数，返回 True 表示健康，False 表示不健康

add_ready_check(name: str, check_func: Callable[[], bool])

添加就绪检查

参数:

• name -- 检查名称

• check_func -- 检查函数，返回 True 表示就绪，False 表示未就绪

process_request(request_handler)

处理请求，检查是否为健康检查端点

参数:

request_handler -- 请求处理器实例

返回:

继续处理请求 其他值: 直接返回该值作为响应

返回类型:

None

litefs.middleware.logging module

class litefs.middleware.logging.LoggingMiddleware(app, logger: Logger | None = None)

基类：Middleware

日志中间件

记录所有请求和响应的详细信息，包括请求时间、响应时间、状态码等

process_exception(request_handler, exception)

处理异常，记录异常信息

参数:

• request_handler -- 请求处理器实例

• exception -- 异常对象

返回:

继续抛出异常

返回类型:

None

process_request(request_handler)

处理请求，记录请求开始时间

参数:

request_handler -- 请求处理器实例

process_response(request_handler, response)

处理响应，记录请求处理时间

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

响应数据

litefs.middleware.rate_limit module

class litefs.middleware.rate_limit.RateLimitMiddleware(app, max_requests: int = 100, window_seconds: int = 60, key_func: callable | None = None, block_duration: int = 60)

基类：Middleware

限流中间件

基于令牌桶算法实现请求限流，支持按 IP 地址或用户限流

process_request(request_handler)

处理请求，检查是否超过限流

参数:

request_handler -- 请求处理器实例

返回:

如果超过限流，返回 429 响应 否则返回 None，继续处理请求

class litefs.middleware.rate_limit.ThrottleMiddleware(app, min_interval: float = 0.1, key_func: callable | None = None)

基类：Middleware

节流中间件

控制请求的处理速率，防止服务器过载

process_request(request_handler)

处理请求，检查是否需要节流

参数:

request_handler -- 请求处理器实例

返回:

如果需要节流，返回 429 响应 否则返回 None，继续处理请求

litefs.middleware.security module

class litefs.middleware.security.AuthMiddleware(app, auth_header: str = 'Authorization', token_verifier: Callable[[str], bool] | None = None, jwt_manager=None)

基类：Middleware

认证中间件

基于 Bearer Token 的认证中间件，支持自定义验证函数和 JWT 集成

process_request(request_handler)

处理请求，检查认证信息

参数:

request_handler -- 请求处理器实例

返回:

如果认证失败，返回 401 响应 否则返回 None，继续处理请求

class litefs.middleware.security.SecurityMiddleware(app, x_frame_options: str = 'DENY', x_content_type_options: str = 'nosniff', x_xss_protection: str = '1; mode=block', strict_transport_security: str = 'max-age=31536000; includeSubDomains', content_security_policy: str | None = None, referrer_policy: str = 'strict-origin-when-cross-origin', permissions_policy: str | None = None)

基类：Middleware

安全中间件

添加各种安全相关的 HTTP 响应头，提高应用安全性

process_response(request_handler, response)

处理响应，添加安全响应头

参数:

• request_handler -- 请求处理器实例

• response -- 响应数据

返回:

添加了安全响应头的响应数据