litefs.handlers package
LiteFS 请求处理器包
提供 WSGI、ASGI 和 Socket 三种协议的请求处理器。
- class litefs.handlers.ASGIRequestHandler(app, scope, receive, send)[源代码]
-
ASGI 请求处理器,用于在 uvicorn、daphne 等 ASGI 服务器中运行
符合 ASGI 3.0 规范,处理 ASGI scope、receive 和 send 并返回标准响应
- property cookie
- class litefs.handlers.BaseRequestHandler(app, environ)[源代码]
基类:
RequestPropertiesMixin请求处理器基类,提供通用的请求处理功能
所有 WSGI/ASGI/Socket 请求处理器共享的属性和方法集中在此, 减少子类间的代码重复。
- class litefs.handlers.EnhancedRequestHandler(request_handler)[源代码]
基类:
object增强的请求处理器
提供分离的 query 和 post 参数,以及表单验证功能
- property body: str
获取请求体
- 返回:
请求体字符串
- property config
获取配置对象
- 返回:
配置对象
- property content_length: int
获取内容长度
- 返回:
内容长度
- property content_type: str | None
获取内容类型
- 返回:
内容类型
- property cookie
获取 Cookie
- 返回:
Cookie 对象
- property environ: Dict[str, Any]
获取环境变量
- 返回:
环境变量字典
- property files: Dict[str, Any]
获取上传的文件
- 返回:
文件字典
- get_file(key: str, default: Any | None = None) Any[源代码]
获取单个上传文件
- 参数:
key -- 文件字段名
default -- 默认值
- 返回:
文件对象
- get_post_param(key: str, default: Any | None = None) Any[源代码]
获取单个 POST 参数
- 参数:
key -- 参数名
default -- 默认值
- 返回:
参数值
- get_query_param(key: str, default: Any | None = None) Any[源代码]
获取单个查询参数
- 参数:
key -- 参数名
default -- 默认值
- 返回:
参数值
- property json: Dict[str, Any]
获取 JSON 请求体
- 返回:
JSON 数据字典
- property path_info: str
获取路径信息
- 返回:
路径信息
- property post: Dict[str, Any]
获取 POST 请求体参数
- 返回:
POST 参数字典
- property query: Dict[str, Any]
获取 URL 查询参数
- 返回:
查询参数字典
- property query_string: str
获取查询字符串
- 返回:
查询字符串
- property referer: str | None
获取来源页面
- 返回:
来源页面 URL
- property request_method: str
获取请求方法
- 返回:
请求方法(GET, POST, PUT, DELETE 等)
- property request_uri: str
获取请求 URI
- 返回:
请求 URI
- property session
获取会话对象
- 返回:
会话对象
- property session_id: str | None
获取会话 ID
- 返回:
会话 ID
- set_cookie(key: str, value: str, **kwargs)[源代码]
设置 Cookie
- 参数:
key -- Cookie 名称
value -- Cookie 值
**kwargs -- 其他 Cookie 参数
- start_response(status_code: int = 200, headers=None)[源代码]
开始响应
- 参数:
status_code -- HTTP 状态码
headers -- 响应头列表
- validate_all(query_rules: Dict[str, List] | None = None, post_rules: Dict[str, List] | None = None, file_rules: Dict[str, List] | None = None) Tuple[bool, Dict[str, Dict[str, List[str]]]][源代码]
验证所有参数
- 参数:
query_rules -- 查询参数验证规则
post_rules -- POST 参数验证规则
file_rules -- 文件验证规则
- 返回:
错误字典})
- 返回类型:
(是否验证成功, {参数类型
- validate_files(rules: Dict[str, List]) Tuple[bool, Dict[str, List[str]]][源代码]
验证上传文件
- 参数:
rules -- 验证规则字典 {字段名: [验证器列表]}
- 返回:
(是否验证成功, 错误字典)
- litefs.handlers.RequestHandler
- class litefs.handlers.Response(content=None, status_code=200, headers=None)[源代码]
基类:
object响应对象,提供更丰富的响应方法
- delete_cookie(key, path='/', domain=None)[源代码]
删除 Cookie
- 参数:
key -- Cookie 名称
path -- Cookie 路径
domain -- Cookie 域名
- classmethod file_stream(file_path, status_code=200)[源代码]
流式返回文件
- 参数:
file_path -- 文件路径
status_code -- HTTP 状态码
- 返回:
Response 对象
- set_cookie(key, value, max_age=None, expires=None, path='/', domain=None, secure=False, httponly=False, samesite='Lax')[源代码]
设置 Cookie
- 参数:
key -- Cookie 名称
value -- Cookie 值
max_age -- 最大存活时间(秒)
expires -- 过期时间
path -- Cookie 路径
domain -- Cookie 域名
secure -- 是否仅 HTTPS
httponly -- 是否禁止 JavaScript 访问
samesite -- SameSite 属性 (Strict, Lax, None)
- classmethod sse(content, status_code=200)[源代码]
返回 Server-Sent Events 流式响应
- 参数:
content -- 可迭代对象,如生成器、迭代器,用于流式返回 SSE 数据
status_code -- HTTP 状态码
- 返回:
Response 对象
- class litefs.handlers.SocketRequestHandler(app, rw, environ, request)[源代码]
基类:
SocketResponseMixin,SocketRequestPropertiesMixin,BaseRequestHandler原生 Socket 请求处理器
用于 litefs 内置 HTTP 服务器,直接操作 socket 进行请求处理
- default_headers = {'Content-Type': 'text/html; charset=utf-8', 'Server': 'litefs/0.10.0', 'X-Content-Type-Options': 'nosniff', 'X-Frame-Options': 'SAMEORIGIN', 'X-XSS-Protection': '1; mode=block'}
- class litefs.handlers.WSGIRequestHandler(app, environ)[源代码]
-
WSGI 请求处理器,用于在 gunicorn、uWSGI 等 WSGI 服务器中运行
符合 PEP 3333 规范,处理 WSGI environ 并返回标准响应
- property cookie
- litefs.handlers.double_slash_sub(repl, string, count=0)
Return the string obtained by replacing the leftmost non-overlapping occurrences of pattern in string by the replacement repl.
- litefs.handlers.parse_header(line)[源代码]
解析 HTTP 头部的 Content-Type 参数
- 参数:
line -- HTTP Content-Type 头部值
- 返回:
MIME 类型和参数字典
- 返回类型:
(mime_type, params_dict)
- litefs.handlers.parse_multipart_asgi(body, boundary, max_upload_size=52428800)[源代码]
解析 ASGI 环境下的 multipart 表单数据
- 参数:
body -- 请求体字节数据
boundary -- multipart boundary
max_upload_size -- 最大上传大小
- 返回:
表单字段字典和上传文件字典
- 返回类型:
(posts, files)
- litefs.handlers.parse_multipart_stream(rw, boundary, max_upload_size=52428800, DEFAULT_BUFFER_SIZE=8192)[源代码]
解析流式环境下的 multipart 表单数据
- 参数:
rw -- 读写流对象
boundary -- multipart boundary
max_upload_size -- 最大上传大小
DEFAULT_BUFFER_SIZE -- 默认缓冲区大小
- 返回:
表单字段字典和上传文件字典
- 返回类型:
(posts, files)
- litefs.handlers.parse_multipart_wsgi(wsgi_input, boundary, content_length, max_upload_size=52428800)[源代码]
解析 WSGI 环境下的 multipart 表单数据
- 参数:
wsgi_input -- WSGI 输入流
boundary -- multipart boundary
content_length -- 内容长度
max_upload_size -- 最大上传大小
- 返回:
表单字段字典和上传文件字典
- 返回类型:
(posts, files)
- litefs.handlers.startswith_dot_sub(repl, string, count=0)
Return the string obtained by replacing the leftmost non-overlapping occurrences of pattern in string by the replacement repl.
Submodules
litefs.handlers.base module
请求处理器基类
提供所有请求处理器共用的基础功能
- class litefs.handlers.base.BaseRequestHandler(app, environ)[源代码]
基类:
RequestPropertiesMixin请求处理器基类,提供通用的请求处理功能
所有 WSGI/ASGI/Socket 请求处理器共享的属性和方法集中在此, 减少子类间的代码重复。
litefs.handlers.wsgi module
WSGI 请求处理器
用于在 gunicorn、uWSGI 等 WSGI 服务器中运行 符合 PEP 3333 规范
litefs.handlers.asgi module
ASGI 请求处理器
用于在 uvicorn、daphne 等 ASGI 服务器中运行 符合 ASGI 3.0 规范
litefs.handlers.socket module
原生 Socket 请求处理器
用于 litefs 内置 HTTP 服务器 直接操作 socket 进行请求处理
- litefs.handlers.socket.RequestHandler
- class litefs.handlers.socket.SocketRequestHandler(app, rw, environ, request)[源代码]
基类:
SocketResponseMixin,SocketRequestPropertiesMixin,BaseRequestHandler原生 Socket 请求处理器
用于 litefs 内置 HTTP 服务器,直接操作 socket 进行请求处理
- default_headers = {'Content-Type': 'text/html; charset=utf-8', 'Server': 'litefs/0.10.0', 'X-Content-Type-Options': 'nosniff', 'X-Frame-Options': 'SAMEORIGIN', 'X-XSS-Protection': '1; mode=block'}
- litefs.handlers.socket.double_slash_sub(repl, string, count=0)
Return the string obtained by replacing the leftmost non-overlapping occurrences of pattern in string by the replacement repl.
- litefs.handlers.socket.startswith_dot_sub(repl, string, count=0)
Return the string obtained by replacing the leftmost non-overlapping occurrences of pattern in string by the replacement repl.
litefs.handlers.response module
HTTP 响应对象和响应构建工具
提供 Response 类和相关的响应构建方法
- class litefs.handlers.response.Response(content=None, status_code=200, headers=None)[源代码]
基类:
object响应对象,提供更丰富的响应方法
- delete_cookie(key, path='/', domain=None)[源代码]
删除 Cookie
- 参数:
key -- Cookie 名称
path -- Cookie 路径
domain -- Cookie 域名
- classmethod file_stream(file_path, status_code=200)[源代码]
流式返回文件
- 参数:
file_path -- 文件路径
status_code -- HTTP 状态码
- 返回:
Response 对象
- set_cookie(key, value, max_age=None, expires=None, path='/', domain=None, secure=False, httponly=False, samesite='Lax')[源代码]
设置 Cookie
- 参数:
key -- Cookie 名称
value -- Cookie 值
max_age -- 最大存活时间(秒)
expires -- 过期时间
path -- Cookie 路径
domain -- Cookie 域名
secure -- 是否仅 HTTPS
httponly -- 是否禁止 JavaScript 访问
samesite -- SameSite 属性 (Strict, Lax, None)
- classmethod sse(content, status_code=200)[源代码]
返回 Server-Sent Events 流式响应
- 参数:
content -- 可迭代对象,如生成器、迭代器,用于流式返回 SSE 数据
status_code -- HTTP 状态码
- 返回:
Response 对象
litefs.handlers.form module
表单数据解析模块
提供表单数据解析功能,支持 URL 编码表单和 multipart 表单
- litefs.handlers.form.form_dict_match(string, pos=0, endpos=9223372036854775807)
Matches zero or more characters at the beginning of the string.
- litefs.handlers.form.parse_form(query_string)[源代码]
解析表单数据,支持缓存
- 参数:
query_string -- 查询字符串
- 返回:
解析后的表单数据字典
- litefs.handlers.form.parse_header(line)[源代码]
解析 HTTP 头部的 Content-Type 参数
- 参数:
line -- HTTP Content-Type 头部值
- 返回:
MIME 类型和参数字典
- 返回类型:
(mime_type, params_dict)
- litefs.handlers.form.parse_multipart_asgi(body, boundary, max_upload_size=52428800)[源代码]
解析 ASGI 环境下的 multipart 表单数据
- 参数:
body -- 请求体字节数据
boundary -- multipart boundary
max_upload_size -- 最大上传大小
- 返回:
表单字段字典和上传文件字典
- 返回类型:
(posts, files)
litefs.handlers.enhanced module
增强的请求处理器
提供分离的 query 和 post 参数,以及表单验证功能
- class litefs.handlers.enhanced.EnhancedRequestHandler(request_handler)[源代码]
基类:
object增强的请求处理器
提供分离的 query 和 post 参数,以及表单验证功能
- property body: str
获取请求体
- 返回:
请求体字符串
- property config
获取配置对象
- 返回:
配置对象
- property content_length: int
获取内容长度
- 返回:
内容长度
- property content_type: str | None
获取内容类型
- 返回:
内容类型
- property cookie
获取 Cookie
- 返回:
Cookie 对象
- property environ: Dict[str, Any]
获取环境变量
- 返回:
环境变量字典
- property files: Dict[str, Any]
获取上传的文件
- 返回:
文件字典
- get_file(key: str, default: Any | None = None) Any[源代码]
获取单个上传文件
- 参数:
key -- 文件字段名
default -- 默认值
- 返回:
文件对象
- get_post_param(key: str, default: Any | None = None) Any[源代码]
获取单个 POST 参数
- 参数:
key -- 参数名
default -- 默认值
- 返回:
参数值
- get_query_param(key: str, default: Any | None = None) Any[源代码]
获取单个查询参数
- 参数:
key -- 参数名
default -- 默认值
- 返回:
参数值
- property json: Dict[str, Any]
获取 JSON 请求体
- 返回:
JSON 数据字典
- property path_info: str
获取路径信息
- 返回:
路径信息
- property post: Dict[str, Any]
获取 POST 请求体参数
- 返回:
POST 参数字典
- property query: Dict[str, Any]
获取 URL 查询参数
- 返回:
查询参数字典
- property query_string: str
获取查询字符串
- 返回:
查询字符串
- property referer: str | None
获取来源页面
- 返回:
来源页面 URL
- property request_method: str
获取请求方法
- 返回:
请求方法(GET, POST, PUT, DELETE 等)
- property request_uri: str
获取请求 URI
- 返回:
请求 URI
- property session
获取会话对象
- 返回:
会话对象
- property session_id: str | None
获取会话 ID
- 返回:
会话 ID
- set_cookie(key: str, value: str, **kwargs)[源代码]
设置 Cookie
- 参数:
key -- Cookie 名称
value -- Cookie 值
**kwargs -- 其他 Cookie 参数
- start_response(status_code: int = 200, headers=None)[源代码]
开始响应
- 参数:
status_code -- HTTP 状态码
headers -- 响应头列表
- validate_all(query_rules: Dict[str, List] | None = None, post_rules: Dict[str, List] | None = None, file_rules: Dict[str, List] | None = None) Tuple[bool, Dict[str, Dict[str, List[str]]]][源代码]
验证所有参数
- 参数:
query_rules -- 查询参数验证规则
post_rules -- POST 参数验证规则
file_rules -- 文件验证规则
- 返回:
错误字典})
- 返回类型:
(是否验证成功, {参数类型
- validate_files(rules: Dict[str, List]) Tuple[bool, Dict[str, List[str]]][源代码]
验证上传文件
- 参数:
rules -- 验证规则字典 {字段名: [验证器列表]}
- 返回:
(是否验证成功, 错误字典)