litefs.handlers package

LiteFS 请求处理器包

提供 WSGI、ASGI 和 Socket 三种协议的请求处理器。

class litefs.handlers.ASGIRequestHandler(app, scope, receive, send)

基类：BaseRequestHandler

ASGI 请求处理器，用于在 uvicorn、daphne 等 ASGI 服务器中运行

符合 ASGI 3.0 规范，处理 ASGI scope、receive 和 send 并返回标准响应

property cookie

async handler()

异步处理请求

class litefs.handlers.BaseRequestHandler(app, environ)

基类：object

请求处理器基类，提供通用的请求处理功能

所有 WSGI/ASGI/Socket 请求处理器共享的属性和方法集中在此， 减少子类间的代码重复。

property app

获取应用实例

返回:

应用实例

property body

property charset

property config

property content_length

property content_type

property data

property environ

property files

property form

表单数据，与 post 属性相同

返回:

表单数据字典

property get

handle_response(result)

处理响应结果，支持 Response 对象

property headers

获取所有请求头（已缓存，O(1) 访问）

返回:

包含所有请求头的字典

property json

property method

property params

property path_info

property post

property query_string

property referer

render_template(template_name, **kwargs)

渲染模板

模板上下文中会自动注入翻译函数 _ / gettext / ngettext。 未启用国际化（或当前无请求 locale）时，这些函数为恒等函数， 渲染结果与未注入前完全一致。

参数:

• template_name -- 模板文件名

• **kwargs -- 模板变量

返回:

渲染后的 HTML 字符串

property request_method

property request_uri

property server_protocol

property session

property session_id

set_cookie(key, value, **kwargs)

start_response(status_code=200, headers=None)

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 参数

set_form_validator(validator: FormValidator)

设置表单验证器

参数:

validator -- 表单验证器实例

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 -- 验证规则字典 {字段名: [验证器列表]}

返回:

(是否验证成功, 错误字典)

validate_post(rules: Dict[str, List]) → Tuple[bool, Dict[str, List[str]]]

验证 POST 参数

参数:

rules -- 验证规则字典 {字段名: [验证器列表]}

返回:

(是否验证成功, 错误字典)

validate_query(rules: Dict[str, List]) → Tuple[bool, Dict[str, List[str]]]

验证查询参数

参数:

rules -- 验证规则字典 {字段名: [验证器列表]}

返回:

(是否验证成功, 错误字典)

litefs.handlers.RequestHandler

SocketRequestHandler 的别名

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 error(status_code, message=None, headers=None)

返回错误响应

classmethod file(file_path, status_code=200, headers=None)

返回文件响应

classmethod file_stream(file_path, status_code=200)

流式返回文件

参数:

• file_path -- 文件路径

• status_code -- HTTP 状态码

返回:

Response 对象

classmethod html(content, status_code=200, headers=None)

返回 HTML 响应

classmethod json(data, status_code=200, headers=None)

返回 JSON 响应

classmethod redirect(url, status_code=302, headers=None)

返回重定向响应

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 对象

classmethod stream(content, status_code=200, headers=None)

返回流式响应

参数:

• content -- 可迭代对象，如生成器、迭代器，用于流式返回数据

• status_code -- HTTP 状态码

• headers -- 响应头列表

返回:

Response 对象

classmethod text(content, status_code=200, headers=None)

返回纯文本响应

class litefs.handlers.SocketRequestHandler(app, rw, environ, request)

基类：BaseRequestHandler

原生 Socket 请求处理器

用于 litefs 内置 HTTP 服务器，直接操作 socket 进行请求处理

property body

property charset

property config

property content_length

property content_type

property cookie

property data

default_headers = {'Content-Type': 'text/html; charset=utf-8', 'Server': 'litefs/0.8.3', 'X-Content-Type-Options': 'nosniff', 'X-Frame-Options': 'SAMEORIGIN', 'X-XSS-Protection': '1; mode=block'}

property environ

property files

finish(content)

完成响应并写入 socket

handle_response(result)

处理响应结果，支持 Response 对象

handler()

处理请求的主方法

property json

property method

property params

property path_info

property query_string

redirect(url=None)

property referer

property request_method

property request_uri

property server_protocol

property session

property session_id

set_cookie(name, value, **options)

class litefs.handlers.WSGIRequestHandler(app, environ)

基类：BaseRequestHandler

WSGI 请求处理器，用于在 gunicorn、uWSGI 等 WSGI 服务器中运行

符合 PEP 3333 规范，处理 WSGI environ 并返回标准响应

property cookie

handler()

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.imap(func, iterable)

延迟版本的 map

litefs.handlers.is_bytes(s)

检查对象是否为字节类型

litefs.handlers.parse_form(query_string)

解析表单数据，支持缓存

参数:

query_string -- 查询字符串

返回:

解析后的表单数据字典

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)

基类：object

请求处理器基类，提供通用的请求处理功能

所有 WSGI/ASGI/Socket 请求处理器共享的属性和方法集中在此， 减少子类间的代码重复。

property app

获取应用实例

返回:

应用实例

property body

property charset

property config

property content_length

property content_type

property data

property environ

property files

property form

表单数据，与 post 属性相同

返回:

表单数据字典

property get

handle_response(result)

处理响应结果，支持 Response 对象

property headers

获取所有请求头（已缓存，O(1) 访问）

返回:

包含所有请求头的字典

property json

property method

property params

property path_info

property post

property query_string

property referer

render_template(template_name, **kwargs)

渲染模板

模板上下文中会自动注入翻译函数 _ / gettext / ngettext。 未启用国际化（或当前无请求 locale）时，这些函数为恒等函数， 渲染结果与未注入前完全一致。

参数:

• template_name -- 模板文件名

• **kwargs -- 模板变量

返回:

渲染后的 HTML 字符串

property request_method

property request_uri

property server_protocol

property session

property session_id

set_cookie(key, value, **kwargs)

start_response(status_code=200, headers=None)

litefs.handlers.wsgi module

WSGI 请求处理器

用于在 gunicorn、uWSGI 等 WSGI 服务器中运行 符合 PEP 3333 规范

class litefs.handlers.wsgi.WSGIRequestHandler(app, environ)

基类：BaseRequestHandler

WSGI 请求处理器，用于在 gunicorn、uWSGI 等 WSGI 服务器中运行

符合 PEP 3333 规范，处理 WSGI environ 并返回标准响应

property cookie

handler()

litefs.handlers.asgi module

ASGI 请求处理器

用于在 uvicorn、daphne 等 ASGI 服务器中运行 符合 ASGI 3.0 规范

class litefs.handlers.asgi.ASGIRequestHandler(app, scope, receive, send)

基类：BaseRequestHandler

ASGI 请求处理器，用于在 uvicorn、daphne 等 ASGI 服务器中运行

符合 ASGI 3.0 规范，处理 ASGI scope、receive 和 send 并返回标准响应

property cookie

async handler()

异步处理请求

litefs.handlers.socket module

原生 Socket 请求处理器

用于 litefs 内置 HTTP 服务器 直接操作 socket 进行请求处理

litefs.handlers.socket.RequestHandler

SocketRequestHandler 的别名

class litefs.handlers.socket.SocketRequestHandler(app, rw, environ, request)

基类：BaseRequestHandler

原生 Socket 请求处理器

用于 litefs 内置 HTTP 服务器，直接操作 socket 进行请求处理

property body

property charset

property config

property content_length

property content_type

property cookie

property data

default_headers = {'Content-Type': 'text/html; charset=utf-8', 'Server': 'litefs/0.8.3', 'X-Content-Type-Options': 'nosniff', 'X-Frame-Options': 'SAMEORIGIN', 'X-XSS-Protection': '1; mode=block'}

property environ

property files

finish(content)

完成响应并写入 socket

handle_response(result)

处理响应结果，支持 Response 对象

handler()

处理请求的主方法

property json

property method

property params

property path_info

property query_string

redirect(url=None)

property referer

property request_method

property request_uri

property server_protocol

property session

property session_id

set_cookie(name, value, **options)

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.imap(func, iterable)

延迟版本的 map

litefs.handlers.socket.is_bytes(s)

检查对象是否为字节类型

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 error(status_code, message=None, headers=None)

返回错误响应

classmethod file(file_path, status_code=200, headers=None)

返回文件响应

classmethod file_stream(file_path, status_code=200)

流式返回文件

参数:

• file_path -- 文件路径

• status_code -- HTTP 状态码

返回:

Response 对象

classmethod html(content, status_code=200, headers=None)

返回 HTML 响应

classmethod json(data, status_code=200, headers=None)

返回 JSON 响应

classmethod redirect(url, status_code=302, headers=None)

返回重定向响应

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 对象

classmethod stream(content, status_code=200, headers=None)

返回流式响应

参数:

• content -- 可迭代对象，如生成器、迭代器，用于流式返回数据

• status_code -- HTTP 状态码

• headers -- 响应头列表

返回:

Response 对象

classmethod text(content, status_code=200, headers=None)

返回纯文本响应

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.form.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.form.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.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 参数

set_form_validator(validator: FormValidator)

设置表单验证器

参数:

validator -- 表单验证器实例

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 -- 验证规则字典 {字段名: [验证器列表]}

返回:

(是否验证成功, 错误字典)

validate_post(rules: Dict[str, List]) → Tuple[bool, Dict[str, List[str]]]

验证 POST 参数

参数:

rules -- 验证规则字典 {字段名: [验证器列表]}

返回:

(是否验证成功, 错误字典)

validate_query(rules: Dict[str, List]) → Tuple[bool, Dict[str, List[str]]]

验证查询参数

参数:

rules -- 验证规则字典 {字段名: [验证器列表]}

返回:

(是否验证成功, 错误字典)