跳转至

Http

HTTP 处理器

在下文中,用于处理 HTTP 请求的可调用对象被称为 HTTP 处理器。

函数处理器

使用函数处理请求是很简单的。

1
2
3
4
5
6
7
8
from hintapi import HintAPI

app = HintAPI()


@app.router.http("/hello")
def hello():
    return "hello"

@app.router.http 装饰器将返回原始的函数,故而可以将同一个函数注册到多个路由下。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from hintapi import HintAPI, request

app = HintAPI()


@app.router.http("/hello", name="hello")
@app.router.http("/hello/{name}", name="hello-with-name")
def hello():
    if request.path_params:
        return f"hello {request.path_params['name']}"
    return "hello"

你还可以使用 required_method 来约束函数处理器仅接受指定的请求方法。

1
2
3
4
5
6
7
8
9
from hintapi import HintAPI, request, required_method

app = HintAPI()


@app.router.http("/hello")
@required_method("POST")
def need_post():
    return request.method

当你使用 required_method 对请求方法进行约束时,OPTIONS 方法将会被自动处理。

当你使用 required_method 允许 GET 方法时,HEAD 方法也会同时被允许。

类处理器

使用类处理多种请求十分简单。只需要继承 HttpView 并编写对应的方法,支持的方法有 "get""post""put""patch""delete""head""options""trace"

允许更多请求方法

在继承类时覆盖类属性 HTTP_METHOD_NAMES 即可。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
from hintapi import HintAPI, request, HttpView

app = HintAPI()


@app.router.http("/cat")
class Cat(HttpView):

    def get(self):
        return request.method

    def post(self):
        return request.method

    def put(self):
        return request.method

    def patch(self):
        return request.method

    def delete(self):
        return request.method

获取请求值

使用 from hintapi import request 语句获取全局变量 request,它是一个代理对象,可以读写删当前请求对应的 Request 对象的各个属性。一般来说这足以应付大部分需求,但如果你真的需要访问原始 Request 对象,可以使用 hintapi.requests.request_var.get()

以下是 hintapi.requests.Request 对象的常用属性与方法。

Method

通过 request.method 可以获取到请求方法,例如 GET/POST 等。

URL

通过 request.url 可以获取到请求路径。该属性是一个类似于字符串的对象,它公开了可以从URL中解析出的所有组件。

例如:request.url.path, request.url.port, request.url.scheme

Path Parameters

request.path_params 是一个字典,包含所有解析出的路径参数。

Headers

request.headers 是一个大小写无关的多值字典(multi-dict)。但通过 request.headers.keys()/request.headers.items() 取出来的 key 均为小写。

Accept

通过读取 request.accepted_types 属性你可以获取客户端接收的全部响应类型。

通过调用 request.accepts 函数你可以判断客户端接受什么样的响应类型。例如:request.accepts("text/html")

Content Type

你可以使用 request.content_type == "application/json" 之类的语句来判断请求类型是否满足条件。不必考虑现实中的 HTTP 请求头 Content-Type: application/json; charset=utf-8 尾部的 ; charset=utf-8 会影响判断的准确性,这类选项会被以字典形式解析到 request.content_type.options

Query Parameters

request.query_params 是一个不可变的多值字典(multi-dict)。

例如:request.query_params['search']

Client Address

request.client 是一个 namedtuple,定义为 namedtuple("Address", ["host", "port"])

获取客户端 hostname 或 IP 地址: request.client.host

获取客户端在当前连接中使用的端口: request.client.port

元组中任何一个元素都可能为 None。这受限于 ASGI 服务器传递的值。

Cookies

request.cookies 是一个标准字典,定义为 Dict[str, str]

例如:request.cookies.get('mycookie')

Body

有几种方法可以读到请求体内容:

  • request.body:返回一个 bytes

  • request.form:将 body 作为表单进行解析并返回结果(多值字典)。

  • request.json:将 body 作为 JSON 字符串解析并返回结果。

  • request.data():将 body 根据 content_type 提供的信息进行解析并返回。

你也可以使用 for 语法将 body 作为一个 bytes 流进行读取:

1
2
3
4
5
6
def post():
    ...
    body = b''
    for chunk in request.stream():
        body += chunk
    ...

如果你直接使用了 request.stream() 去读取数据,那么请求体将不会缓存在内存中。其后任何对 .body/.form/.json 的调用都将抛出错误。

在某些情况下,例如长轮询或流式响应,你可能需要确定客户端是否已断开连接。可以使用 disconnected = request.is_disconnected() 确定此状态。

Request Files

通过 request.form 可以解析通过 multipart/form-data 格式接收到的表单,包括文件。

文件将被包装为 starlette.datastructures.UploadFile 对象,它有如下属性:

  • filename: str: 被提交的原始文件名称 (例如 myimage.jpg).
  • content_type: str: 文件类型 (MIME type / media type) (例如 image/jpeg).
  • file: tempfile.SpooledTemporaryFile: 存储文件内容的临时文件(可以直接读写这个对象,但最好不要)。

UploadFile 还有四个方法:

  • write(data: Union[str, bytes]): 写入数据到文件中。
  • read(size: int): 从文件中读取数据。
  • seek(offset: int): 文件指针跳转到指定位置。
  • close(): 关闭文件。

下面是一个读取原始文件名称和内容的例子:

1
2
3
form = request.form
filename = form["upload_file"].filename
contents = form["upload_file"].read()

State

某些情况下需要储存一些额外的自定义信息到 request 中,可以使用 request.state 用于存储。

1
2
3
4
5
request.state.user = User(name="Alice")  # 写

user_name = request.state.user.name  # 读

del request.state.user  # 删

返回响应值

对于任何正常处理的 HTTP 请求都必须返回一个 HttpResponse 对象或者是它的子类对象。

HttpResponse

签名:HttpResponse(status_code: int = 200, headers: Mapping[str, str] = None)

  • status_code - HTTP 状态码。
  • headers - 字符串字典。

HttpResponse 提供 set_cookie 方法以允许你设置 cookies。

签名:HttpResponse.set_cookie(key, value="", max_age=None, expires=None, path="/", domain=None, secure=False, httponly=False, samesite="lax")

  • key: str,将成为 Cookie 的键。
  • value: str = "",将是 Cookie 的值。
  • max_age: Optional[int],以秒为单位定义 Cookie 的生存期。非正整数会立即丢弃 Cookie。
  • expires: Optional[int],它定义 Cookie 过期之前的秒数。
  • path: str = "/",它指定 Cookie 将应用到的路由的子集。
  • domain: Optional[str],用于指定 Cookie 对其有效的域。
  • secure: bool = False,指示仅当使用 HTTPS 协议发出请求时,才会将 Cookie 发送到服务器。
  • httponly: bool = False,指示无法通过 Javascript 通过 Document.cookie 属性、XMLHttpRequestRequest 等 API 来访问 Cookie。
  • samesite: str = "lax",用于指定 Cookie 的相同网站策略。有效值为 "lax""strict""none"

HttpResponse 也提供了 delete_cookie 方法指定已设置的 Cookie 过期。

签名: HttpResponse.delete_cookie(key, path='/', domain=None)

PlainTextResponse

接受 strbytes 并返回纯文本响应。

1
2
3
4
5
from hintapi import PlainTextResponse


def return_plaintext():
    return PlainTextResponse('Hello, world!')

HTMLResponse

接受 strbytes 并返回 HTML 响应。

1
2
3
4
5
from hintapi import HTMLResponse


def return_html():
    return HTMLResponse('<html><body><h1>Hello, world!</h1></body></html>')

JSONResponse

接受一些 Python 对象并返回一个 application/json 编码的响应。

1
2
3
4
5
from hintapi import JSONResponse


def return_json():
    return JSONResponse({'hello': 'world'})

JSONResponse 以关键词参数的形式暴露出全部 json.dumps 的选项以供自定义。例如在很多时候,Python 内置的 json 转换器无法满足实际项目的序列化需求,可以通过覆盖 default 方法来自定义无法序列化的对象该如何处理。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
import json
import decimal
import datetime

from hintapi import JSONResponse


def custom_convert(obj):
    if isinstance(obj, datetime.datetime):
        return obj.strftime("%Y-%m-%d %H:%M:%S")

    if isinstance(obj, datetime.date):
        return obj.strftime("%Y-%m-%d")

    if isinstance(obj, decimal.Decimal):
        return str(obj)

    raise TypeError(f'Object of type {obj.__class__.__name__} is not JSON serializable')


def return_json():
    return JSONResponse({'hello': 'world'}, default=custom_convert)

RedirectResponse

返回 HTTP 重定向。默认情况下使用 307 状态代码。

1
2
3
4
5
from hintapi import RedirectResponse


def return_redirect():
    return RedirectResponse('/')

StreamResponse

接受一个生成器,流式传输响应主体。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import time

from hintapi import StreamResponse


def slow_numbers(minimum, maximum):
    yield('<html><body><ul>')
    for number in range(minimum, maximum + 1):
        yield '<li>%d</li>' % number
        time.sleep(0.5)
    yield('</ul></body></html>')


def return_stream(scope, receive, send):
    generator = slow_numbers(1, 10)
    return StreamResponse(generator, content_type='text/html')

FileResponse

传输文件作为响应。

与其他响应类型相比,它采用不同的参数进行实例化:

  • filepath - 要流式传输的文件的文件路径。
  • headers - 与 Response 中的 headers 参数的作用相同。
  • content_type - 文件的 MIME 媒体类型。如果未设置,则文件名或路径将用于推断媒体类型。
  • download_name - 如果设置此参数,它将包含在响应的 Content-Disposition 中。
  • stat_result - 接受一个 os.stat_result 对象,如果不传入则会自动使用 os.stat(filepath) 的结果。

FileResponse 将自动设置适当的 Content-LengthLast-ModifiedETag 标头。并且无需任何额外的处理即可支持文件范围请求

TemplateResponse

TemplateResponseapp.templates.TemplateResponse 的一个快捷方式。

Jinja2 模板引擎

hintapi 内置了对 Jinja2 模板的支持,只要你安装了 jinja2 模块,就能从 hintapi.templates 中导出 Jinja2Templates。以下是一个简单的使用样例,访问 "/" 它将从项目根目录下的 templates 目录寻找 homepage.html 文件进行渲染。

1
2
3
4
5
6
7
8
9
from hintapi import HintAPI, TemplateResponse
from hintapi.http.templates import Jinja2Templates

app = HintAPI(templates=Jinja2Templates("templates"))


@app.router.http("/")
def homepage():
    return TemplateResponse("homepage.html")

如果你要使用某个模块下的指定文件夹中的模板文件,可以使用 Jinja2Templates("module_name:dirname")。你还可以传递多个目录让 Jinja2 按照顺序依次查找,直到找到第一个可用的模板,例如:Jinja2Templates("templates", "module_name:dirname")

其他模板引擎

通过继承 hintapi.templates.BaseTemplates 并实现 TemplateResponse 方法,你可以实现自己的模板引擎类。

SendEventResponse

通过 SendEventResponse 可以返回一个 Server-sent Events 响应,这是一种 HTTP 长连接响应,可应用于服务器实时推送数据到客户端等场景。

SendEventResponse 除了可以接受诸如 status_codeheaders 等常规参数外,还需要自行传入一个用于生成消息的生成器。传入的生成器 yield 的每一条消息都需要为合规的 Server-Sent Event 消息(hintapi.ServerSentEvent 类型)。

如下是一个每隔一秒发送一条 hello 消息、一共发送一百零一条消息的样例。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
import time

from hintapi import HintAPI, SendEventResponse

app = HintAPI()


@app.router.http("/message")
def message():

    def message_gen():
        for i in range(101):
            time.sleep(1)
            yield {"id": i, "data": "hello"}

    return SendEventResponse(message_gen())

响应的简化写法

为了方便使用,hintapi 允许自定义一些函数来处理 HTTP 处理器返回的非 HttpResponse 对象。它的原理是拦截响应,通过响应值的类型来自动选择处理函数,把非 HttpResponse 对象转换为 HttpResponse 对象。

如果需要手动把函数的返回值转换为 HttpResponse 对象,则可以使用 hintapi.responses.convert_response

在下例中,视图函数返回一个 dict 对象,但客户端接收到的却是一个 JSON。这是因为 hintapi 内置了一些处理函数用于处理常见的类型:

  • dict | tuple | list:自动转换为 JSONResponse
  • str | bytes:自动转换为 PlainTextResponse
  • typing.AsyncGeneratorType:自动转换为 SendEventResponse
  • pathlib.Path:自动转换为 FileResponse
  • baize.datastructures.URL:自动转换为 RedirectResponse
1
2
def get_detail():
    return {"key": "value"}

你还可以返回多个值来自定义 HTTP Status 和 HTTP Headers:

1
2
3
4
5
6
def not_found():
    return {"message": "Not found"}, 404


def no_content():
    return "", 301, {"location": "https://hintapi.aber.sh"}

同样的,你也可以自定义响应值的简化写法以统一项目的响应规范(哪怕有 TypedDict,Python 的 Dict 约束依旧很弱,但 dataclass 则有效得多),如下例所示,当你在视图函数里返回 Error 对象时,它都会自动被转换为 JSONResponse,并且状态码默认为 400

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
from dataclasses import dataclass, asdict
from typing import Mapping

from hintapi import HintAPI, HttpResponse, JSONResponse

app = HintAPI()


@dataclass
class Error:
    code: int = 0
    title: str = ""
    message: str = ""


@app.response_convertor.register(Error)
def _error_json(error: Error, status: int = 400, headers: Mapping[str, str] = None) -> HttpResponse:
    return JSONResponse(asdict(error), status, headers)

或者你想覆盖默认的 tuple/list/dict 所对应的 JSONResponse

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from typing import Mapping
from hintapi import HintAPI, HttpResponse

app = HintAPI()


@app.response_convertor.register(tuple)
@app.response_convertor.register(list)
@app.response_convertor.register(dict)
def _more_json(body, status: int = 200, headers: Mapping[str, str] = None) -> HttpResponse:
    return CustomizeJSONResponse(body, status, headers)

异常处理

HTTPException

其参数签名是:HTTPException(status_code: int, headers: dict = None, content: typing.Any = None)

你可以通过抛出 HTTPException 来返回一个 HTTP 响应(不必担心它变成一个真正的异常抛出,hintapi 会将它变成一个普通的响应对象)。如果你没有给出一个类型为 bytesstrcontent 值,那么它将使用 Python 标准库中的 http.HTTPStatus(status_code).description 作为最终结果。

1
2
3
4
5
6
7
from hintapi import HTTPException


def endpoint():
    ...
    raise HTTPException(400)
    ...

有时候也许你想返回更多的信息,可以像使用 HttpResponse 一样为它传递 contentheaders 参数来控制最终实际的响应对象。下面是一个简单的例子。

1
2
3
4
5
6
7
from hintapi import HTTPException


def endpoint():
    ...
    raise HTTPException(405, headers={"Allow": "HEAD, GET, POST"})
    ...

如果你想在 lambda 函数里抛出 HTTPException,则可以使用 baize.exceptions.abort

自定义异常处理

对于一些故意抛出的异常,hintapi 提供了方法进行统一处理。

你可以捕捉指定的 HTTP 状态码,那么在应对包含对应 HTTP 状态码的 HTTPException 异常时,hintapi 会使用你定义的函数而不是默认行为。你也可以捕捉其他继承自 Exception 的异常,通过自定义函数,返回指定的内容给客户端。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
from hintapi import HintAPI, HTTPException, HttpResponse, PlainTextResponse

app = HintAPI()


@app.exception_handler(404)
def not_found(exc: HTTPException) -> HttpResponse:
    return PlainTextResponse("what do you want to do?", status_code=404)


@app.exception_handler(ValueError)
def value_error(exc: ValueError) -> HttpResponse:
    return PlainTextResponse("Something went wrong with the server.", status_code=500)

除了装饰器注册,你同样可以使用列表式的注册方式,下例与上例等价:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from hintapi import HintAPI, HTTPException, HttpResponse, PlainTextResponse


def not_found(exc: HTTPException) -> HttpResponse:
    return PlainTextResponse("what do you want to do?", status_code=404)


def value_error(exc: ValueError) -> HttpResponse:
    return PlainTextResponse("Something went wrong with the server.", status_code=500)


app = HintAPI(exception_handlers={
    404: not_found,
    ValueError: value_error,
})

内置中间件

CORS

在现代浏览器中解决跨域问题一般使用 Cross-Origin Resource Sharing,在 hintapi 使用如下代码即可快速配置 API 允许跨域。

1
2
3
4
5
from hintapi import Routes
from hintapi.middlewares import CORSMiddleware


routes = Routes(..., http_middlewares=[CORSMiddleware()])

CORSMiddleware 有如下选项:

  • allow_origins: Iterable[Pattern]:允许的 Origin,需要 re.compile 预编译后的 Pattern 对象;默认值为 (re.compile(".*"), )
  • allow_methods: Iterable[str]:允许的请求方法;默认值为 ("GET","POST","PUT","PATCH","DELETE","HEAD","OPTIONS","TRACE")
  • allow_headers: Iterable[str]:允许的请求头,对应 Access-Control-Allow-Headers
  • expose_headers: Iterable[str]:能在响应中列出的请求头,对应 Access-Control-Expose-Headers
  • allow_credentials: bool:为真时则允许跨域请求携带 Cookies,反之不允许;默认为 False
  • max_age: int:预请求的缓存时间;默认为 600 秒。
Version: master
master
stable
0.22