microdot --- 极简 HTTP 框架

Microdot 是一个小巧的、受 Flask 启发的 MicroPython HTTP 框架。它运行于 asyncio 之上,无需线程即可处理多个并发客户端,并提供熟悉的路由装饰器 API。一个最简应用看起来如下:

from microdot import Microdot

app = Microdot()

@app.route('/')
async def index(request):
    return 'Hello, world!'

app.run(host='0.0.0.0', port=80)

class Microdot

class microdot.Microdot

一个 HTTP 应用实例。通常在脚本顶部附近构造一个实例,并用路由处理器对其进行装饰;调用 run() 或 await start_server() 即开始提供服务。

路由注册

route(url_pattern: str, methods: list | None = None) Callable

装饰器,为 url_pattern 在所列出的 HTTP 方法(默认 ['GET'])下注册一个处理器。返回一个装饰器,当应用于某函数时,将其注册为处理器并原样返回该函数。

url_pattern

路径模式。支持静态片段(/users)以及用 < / > 括起来的动态片段。动态片段可接受一个用 : 分隔的可选类型前缀 -- <int:id><path:rest><re:[0-9a-f]+:hex>。默认类型为 string

methods

HTTP 方法名称列表。如果省略,则只匹配 GET

调用处理器时,第一个参数是请求对象,随后是以关键字参数形式传入的所有捕获到的动态片段。处理器的返回值将成为 HTTP 响应:一个 Response、一个字符串、一个元组 (body, status_code[, headers]),或一个 dict / list(作为 JSON 发送)。

get(url_pattern: str) Callable

route(url_pattern, methods=['GET']) 的便捷别名 -- 将被装饰的函数注册为 url_patternGET 处理器。返回该装饰器。

post(url_pattern: str) Callable

route(url_pattern, methods=['POST']) 的便捷别名 -- 将被装饰的函数注册为 url_patternPOST 处理器。返回该装饰器。

put(url_pattern: str) Callable

route(url_pattern, methods=['PUT']) 的便捷别名 -- 将被装饰的函数注册为 url_patternPUT 处理器。返回该装饰器。

patch(url_pattern: str) Callable

route(url_pattern, methods=['PATCH']) 的便捷别名 -- 将被装饰的函数注册为 url_patternPATCH 处理器。返回该装饰器。

delete(url_pattern: str) Callable

route(url_pattern, methods=['DELETE']) 的便捷别名 -- 将被装饰的函数注册为 url_patternDELETE 处理器。返回该装饰器。

生命周期钩子

before_request(f: Callable) Callable

装饰器,注册 f 使其在每个请求之前运行。f 接收请求对象;其返回值通常被忽略。若要短路该请求,可返回一个 Response(或一个可转换为响应的值)-- 此时流水线的其余部分将被跳过,并发送该响应。原样返回 f

after_request(f: Callable) Callable

装饰器,注册 f 使其在每个成功的请求之后运行。f 接收 (request, response),并且必须返回(可能已修改的)响应对象。原样返回 f

after_error_request(f: Callable) Callable

装饰器,注册 f 使其在 Microdot 生成错误响应(404、500、抛出的异常等)之后运行。f 接收 (request, response),并且必须返回(可能已修改的)响应对象。原样返回 f

errorhandler(status_code_or_exception_class) Callable

装饰器,为某个 HTTP 状态码或某个 Python 异常类注册自定义处理器。对于状态码,处理器只接收请求;对于异常类,则接收 (request, exception)。返回该装饰器。

挂载与中止

mount(subapp: Microdot, url_prefix: str = '', local: bool = False) None

将另一个 Microdot 实例的路由挂载到 url_prefix 下。当 localFalse(默认)时,子应用的 before / after / error 处理器也会被挂载到父应用上。当为 True 时,这些处理器只对子应用的路由运行。返回 None

static abort(status_code: int, reason: str | None = None) None

抛出 HTTPException 以使用给定的状态码中止当前请求。永不正常返回 -- 框架会捕获该异常并将其转换为相应的错误响应。在路由函数体内使用很方便:

from microdot import abort

@app.get('/users/<int:id>')
def get_user(request, id):
    user = lookup(id)
    if user is None:
        abort(404)
    return user.to_dict()

运行服务器

run(host: str = '0.0.0.0', port: int = 5000, debug: bool = False, ssl=None) None

阻塞调用线程,对 start_server() 运行 asyncio.run()。仅当 shutdown() 被调用且监听循环退出后才返回 None。这是启动独立应用最简单的方式:

app.run(host='0.0.0.0', port=80)
async start_server(host: str = '0.0.0.0', port: int = 5000, debug: bool = False, ssl=None, start_serving: bool = True) None

以协程方式启动服务器。当需要与已有的 asyncio 事件循环及其他任务集成时使用此方法。该协程直到 shutdown() 被调用前不会返回:

async def main():
    await asyncio.gather(
        app.start_server(port=80),
        capture_loop(),
    )
host

要监听的网络接口。'0.0.0.0'(默认)表示所有接口;'127.0.0.1' 表示仅环回。

port

要监听的 TCP 端口。默认 5000 -- 对于普通 HTTP 选择 80,对于 HTTPS 选择 443

debug

若为 True,则记录每个请求并将回溯信息转储到 stdout。

ssl

用于将传入连接包装到 TLS 中的 ssl.SSLContextNone(默认)表示普通 HTTP。

start_serving

仅在 CPython 上相关;在 MicroPython 上被忽略。

shutdown() None

请求服务器优雅关闭。从路由处理器中调用是安全的 -- 当前请求会在循环退出前完成。立即返回;实际的关闭会在处理中的请求结束后发生。

属性

url_map: list

已注册路由的列表,以 (methods, URLPattern, handler, url_prefix, subapp) 元组形式表示。

before_request_handlers: list

通过 before_request() 注册的可调用对象列表,按注册顺序排列。每个都在路由处理器之前以请求对象运行。直接改变此列表也有效,但很少需要 -- 建议使用装饰器。

after_request_handlers: list

通过 after_request() 注册的可调用对象列表,按注册顺序排列。每个都在请求成功后以 (request, response) 运行,并且必须返回(可能已修改的)响应。

after_error_request_handlers: list

通过 after_error_request() 注册的可调用对象列表,按注册顺序排列。每个都在生成错误响应(由框架或由应用错误处理器生成)后以 (request, response) 运行,并且必须返回该响应。

error_handlers: dict

错误键到处理器可调用对象的映射,由 errorhandler() 填充。键要么是 HTTP 状态码(int),要么是 Python 异常类;值是已注册的处理器。状态码处理器接收 (request);异常类处理器接收 (request, exception)

debug: bool

当服务器以 debug=True 运行时为 True

class Request

class microdot.Request

一个传入的 HTTP 请求。实例作为第一个位置参数传递给路由处理器。应用不会直接构造 Request

类属性

max_content_length: int

拒绝其 Content-Length 超过此字节数的请求,并返回 413 响应。默认 16 KB。

max_body_length: int

缓冲到内存中并通过 body 暴露的最大请求体。更大的请求体(直至 max_content_length)会留在套接字上,必须通过 stream 读取。默认 16 KB。

max_readline: int

单个请求行 / 头部行的最大长度,以字节为单位。默认 2 KB。

实例属性

app: Microdot

处理该请求的 Microdot 实例。

client_addr: tuple

客户端地址,以 (host, port) 形式表示。

method: str

HTTP 方法字符串('GET''POST',...)。

scheme: str

'http''https'

url: str

完整的请求 URL 路径和查询字符串(主机之后的全部内容)。

path: str

仅路径部分。

query_string: str | None

原始查询字符串部分,或 None

args: MultiDict

解析后的查询字符串,以 MultiDict 形式表示。

headers: NoCaseDict

请求头部,以 NoCaseDict 形式表示(不区分大小写的查找)。

cookies: dict

解析后的 Cookie 头部,以 dict 形式表示。

content_length: int

整数 Content-Length 值,若不存在则为 0。

content_type: str | None

Content-Type 头部值,或 None

g: object

一个自由格式的每请求容器(一个裸对象)。可向其赋值属性,以在钩子和处理器之间传递值(request.g.user = ...)。

route: Callable

匹配该请求的处理器函数。

url_prefix: str

路由挂载所在的前缀,或 ''

subapp: Microdot | None

挂载的子应用实例,或 None

请求体访问

body: bytes

完整请求体,以 bytes 形式表示。当请求体正被流式传输时为空(参见 stream)。

stream: object

一个异步流对象,对请求体暴露 read()。对于大于 max_body_length 的请求体使用此对象。

json: dict | list | str | int | float | bool | None

解析为 JSON 的请求体(一个 dictliststrintfloatbool -- 取决于载荷的编码内容),若 Content-Type 不是 application/json 则为 None

form: MultiDict | None

URL 编码的表单字段,以 MultiDict 形式表示,或 None。对于 multipart/form-data,请用 microdot.multipart.with_form_data() 装饰该路由。

files: dict | None

上传的文件,以 {name: FileUpload} 形式表示,由 microdot.multipart.with_form_data() 填充。在该装饰器运行之前为 None

after_request(f: Callable) Callable

注册一个请求级的 after-request 钩子 -- 在应用级 after-request 处理器之后运行,且仅在成功时运行。f 接收 (request, response),并且必须返回(可能已修改的)响应对象。原样返回 f

class Response

class microdot.Response(body=b'', status_code: int = 200, headers: dict | None = None, reason: str | None = None)

一个 HTTP 响应。大多数处理器返回的值会被 Microdot 自动转换为 Response;当需要自定义头部或特定状态码时直接构造一个。

body

响应体。str 以 UTF-8 编码;dict / list 以 JSON 编码,并相应地设置 Content-Typebytes 原样发送;文件类对象或异步生成器则进行流式传输。

status_code

数值 HTTP 状态码。默认 200。

headers

响应头部字典(不区分大小写)。

reason

自定义原因短语。对 200 默认为 "OK",否则默认为 "N/A"

类属性

default_content_type: str

当未显式设置时使用的 Content-Type。默认 'text/plain'

default_send_file_max_age: int | None

当未给出 max_age 时,send_file() 使用的 Cache-Control: max-age 值。None(默认)表示不设 Cache-Control 头部。

send_file_buffer_size: int

send_file() 流式传输的块大小。默认 1024。

already_handled: None

由已经直接写出响应的处理器返回的哨兵值(用于 WebSocket 升级)。始终为 None;重要的是其标识身份。

types_map: dict

send_file() 用于内容类型推断的扩展名到 MIME 类型的映射。映射 cssgifhtmljpgjsjsonpngtxtsvg

方法

添加一个 Set-Cookie 头部。expires 可以是预格式化的字符串,或一个带有 timetuple()datetime 类对象。返回 None;该头部就地追加到此响应上。

设置一个使给定 cookie 立即过期的 Set-Cookiekwargs 接受与 set_cookie() 相同的选项(pathdomainsecurehttp_onlypartitioned);expiresmax_age 被忽略。返回 None;该头部就地追加到此响应上。

classmethod redirect(location: str, status_code: int = 302) Response

返回一个重定向响应:

@app.get('/old')
def old(request):
    return Response.redirect('/new')
classmethod send_file(filename: str, status_code: int = 200, content_type: str | None = None, stream=None, max_age: int | None = None, compressed: bool | str = False, file_extension: str = '') Response

从文件系统流式传输一个文件作为响应体。若未给出 content_type,则通过 types_map 从扩展名推断。compressed=True 会设置 Content-Encoding: gzip(文件必须已经过压缩)。

警告

filename 会被直接打开。切勿传入未经清理的用户提供路径 -- 这样做会允许任意文件泄露。

异常

exception microdot.HTTPException

abort() 抛出,用特定状态码短路一个请求。Microdot 会捕获它并将其转换为相应的错误响应。

status_code: int

要返回的数值 HTTP 状态码 -- 即传递给 abort() 的值。

reason: str

要包含在错误响应中的原因短语。如果未传给 abort(),则默认为 "<status_code> error"(例如 "404 error")。

class URLPattern

class microdot.URLPattern(url_pattern: str)

路由 URL 模式的编译形式。由 Microdot.route() 自动构造;应用很少直接实例化它。

classmethod register_type(type_name: str, pattern: str = '[^/]+', parser: Callable | None = None) None

注册一个新的动态片段类型以在路由模式中使用。返回 None;该类型被添加到类级别的类型注册表中。例如,要添加一个 <uuid:...> 类型:

URLPattern.register_type('uuid', '[0-9a-f-]{36}')

parser 是一个可选的可调用对象,它在捕获到的字符串到达处理器之前对其进行转换。

match(path: str) dict | None

path 与该模式进行匹配,并返回一个包含捕获组的 dict,若不匹配则返回 None

辅助类

class microdot.MultiDict(initial_dict: dict | None = None)

一个 dict 子类,可为每个键存储多个值。用于查询字符串和 URL 编码的表单体,在这些场景中同一个键可以出现不止一次(?tag=a&tag=b)。

get(key, default=None, type: Callable | None = None)

返回 key 的第一个值,可选地用 type(一个可调用对象)进行转换。如果键缺失或转换失败,则返回 default;否则返回(可选地经过转换的)值。

getlist(key, type: Callable | None = None) list

以列表形式返回 key 的所有值,可选地用 type 转换每个值。如果 key 不存在,则返回空列表。

class microdot.NoCaseDict

一个具有不区分大小写字符串键的 dict 子类。用于 Request.headersResponse.headers

模块级函数

microdot.abort(status_code: int, reason: str | None = None) None

Microdot.abort() 的快捷方式。永不正常返回 -- 抛出 HTTPException。可通过 from microdot import abort 导入。

microdot.redirect(location: str, status_code: int = 302) Response

Response.redirect() 的快捷方式。可通过 from microdot import redirect 导入。

microdot.send_file(filename: str, **kwargs) Response

Response.send_file() 的快捷方式。

microdot.urlencode(s: str) str

对一个 URL 组件进行百分号编码。将在 URL 中具有保留含义的字符(/?&=#、空格,...)替换为它们的 %xx 十六进制转义,以便结果可以安全地放在路径片段或查询值内部。返回编码后的 str

microdot.urldecode(s: str) str

对一个 URL 组件进行百分号解码 -- 即 urlencode() 的逆操作。%xx 转义被替换为它所编码的字节,+ 被转换为空格(历史上的查询字符串约定)。返回解码后的 str

子模块