中间件

Starlette 直接提供了几个可用的中间件。

阅读有关它们的更多信息，请访问 FastAPI 中间件文档。

fastapi.middleware.cors.CORSMiddleware

CORSMiddleware(
    app,
    allow_origins=(),
    allow_methods=("GET"),
    allow_headers=(),
    allow_credentials=False,
    allow_origin_regex=None,
    expose_headers=(),
    max_age=600,
)

参数	描述
app	类型: ASGIApp
allow_origins	类型: Sequence[str] 默认值: ()
allow_methods	类型: Sequence[str] 默认值: ('GET')
allow_headers	类型: Sequence[str] 默认值: ()
allow_credentials	类型: bool 默认值: False
allow_origin_regex	类型: str | None 默认值: None
expose_headers	类型: Sequence[str] 默认值: ()
max_age	类型: int 默认值: 600

starlette/middleware/cors.py 中的源代码

def __init__(
    self,
    app: ASGIApp,
    allow_origins: typing.Sequence[str] = (),
    allow_methods: typing.Sequence[str] = ("GET",),
    allow_headers: typing.Sequence[str] = (),
    allow_credentials: bool = False,
    allow_origin_regex: str | None = None,
    expose_headers: typing.Sequence[str] = (),
    max_age: int = 600,
) -> None:
    if "*" in allow_methods:
        allow_methods = ALL_METHODS

    compiled_allow_origin_regex = None
    if allow_origin_regex is not None:
        compiled_allow_origin_regex = re.compile(allow_origin_regex)

    allow_all_origins = "*" in allow_origins
    allow_all_headers = "*" in allow_headers
    preflight_explicit_allow_origin = not allow_all_origins or allow_credentials

    simple_headers = {}
    if allow_all_origins:
        simple_headers["Access-Control-Allow-Origin"] = "*"
    if allow_credentials:
        simple_headers["Access-Control-Allow-Credentials"] = "true"
    if expose_headers:
        simple_headers["Access-Control-Expose-Headers"] = ", ".join(expose_headers)

    preflight_headers = {}
    if preflight_explicit_allow_origin:
        # The origin value will be set in preflight_response() if it is allowed.
        preflight_headers["Vary"] = "Origin"
    else:
        preflight_headers["Access-Control-Allow-Origin"] = "*"
    preflight_headers.update(
        {
            "Access-Control-Allow-Methods": ", ".join(allow_methods),
            "Access-Control-Max-Age": str(max_age),
        }
    )
    allow_headers = sorted(SAFELISTED_HEADERS | set(allow_headers))
    if allow_headers and not allow_all_headers:
        preflight_headers["Access-Control-Allow-Headers"] = ", ".join(allow_headers)
    if allow_credentials:
        preflight_headers["Access-Control-Allow-Credentials"] = "true"

    self.app = app
    self.allow_origins = allow_origins
    self.allow_methods = allow_methods
    self.allow_headers = [h.lower() for h in allow_headers]
    self.allow_all_origins = allow_all_origins
    self.allow_all_headers = allow_all_headers
    self.preflight_explicit_allow_origin = preflight_explicit_allow_origin
    self.allow_origin_regex = compiled_allow_origin_regex
    self.simple_headers = simple_headers
    self.preflight_headers = preflight_headers

app 实例属性

app = app

allow_origins 实例属性

allow_origins = allow_origins

allow_methods 实例属性

allow_methods = allow_methods

allow_headers 实例属性

allow_headers = [lower() for h in allow_headers]

allow_all_origins 实例属性

allow_all_origins = allow_all_origins

allow_all_headers 实例属性

allow_all_headers = allow_all_headers

preflight_explicit_allow_origin 实例属性

preflight_explicit_allow_origin = (
    preflight_explicit_allow_origin
)

allow_origin_regex 实例属性

allow_origin_regex = compiled_allow_origin_regex

simple_headers 实例属性

simple_headers = simple_headers

preflight_headers 实例属性

preflight_headers = preflight_headers

is_allowed_origin

is_allowed_origin(origin)

参数	描述
origin	类型: str

starlette/middleware/cors.py 中的源代码

def is_allowed_origin(self, origin: str) -> bool:
    if self.allow_all_origins:
        return True

    if self.allow_origin_regex is not None and self.allow_origin_regex.fullmatch(origin):
        return True

    return origin in self.allow_origins

preflight_response

preflight_response(request_headers)

参数	描述
request_headers	类型: Headers

starlette/middleware/cors.py 中的源代码

def preflight_response(self, request_headers: Headers) -> Response:
    requested_origin = request_headers["origin"]
    requested_method = request_headers["access-control-request-method"]
    requested_headers = request_headers.get("access-control-request-headers")

    headers = dict(self.preflight_headers)
    failures = []

    if self.is_allowed_origin(origin=requested_origin):
        if self.preflight_explicit_allow_origin:
            # The "else" case is already accounted for in self.preflight_headers
            # and the value would be "*".
            headers["Access-Control-Allow-Origin"] = requested_origin
    else:
        failures.append("origin")

    if requested_method not in self.allow_methods:
        failures.append("method")

    # If we allow all headers, then we have to mirror back any requested
    # headers in the response.
    if self.allow_all_headers and requested_headers is not None:
        headers["Access-Control-Allow-Headers"] = requested_headers
    elif requested_headers is not None:
        for header in [h.lower() for h in requested_headers.split(",")]:
            if header.strip() not in self.allow_headers:
                failures.append("headers")
                break

    # We don't strictly need to use 400 responses here, since its up to
    # the browser to enforce the CORS policy, but its more informative
    # if we do.
    if failures:
        failure_text = "Disallowed CORS " + ", ".join(failures)
        return PlainTextResponse(failure_text, status_code=400, headers=headers)

    return PlainTextResponse("OK", status_code=200, headers=headers)

simple_response 异步

simple_response(scope, receive, send, request_headers)

参数	描述
scope	类型: Scope
receive	类型: Receive
send	类型: Send
request_headers	类型: Headers

starlette/middleware/cors.py 中的源代码

async def simple_response(self, scope: Scope, receive: Receive, send: Send, request_headers: Headers) -> None:
    send = functools.partial(self.send, send=send, request_headers=request_headers)
    await self.app(scope, receive, send)

send 异步

send(message, send, request_headers)

参数	描述
message	类型: Message
send	类型: Send
request_headers	类型: Headers

starlette/middleware/cors.py 中的源代码

async def send(self, message: Message, send: Send, request_headers: Headers) -> None:
    if message["type"] != "http.response.start":
        await send(message)
        return

    message.setdefault("headers", [])
    headers = MutableHeaders(scope=message)
    headers.update(self.simple_headers)
    origin = request_headers["Origin"]
    has_cookie = "cookie" in request_headers

    # If request includes any cookie headers, then we must respond
    # with the specific origin instead of '*'.
    if self.allow_all_origins and has_cookie:
        self.allow_explicit_origin(headers, origin)

    # If we only allow specific origins, then we have to mirror back
    # the Origin header in the response.
    elif not self.allow_all_origins and self.is_allowed_origin(origin=origin):
        self.allow_explicit_origin(headers, origin)

    await send(message)

allow_explicit_origin 静态方法

allow_explicit_origin(headers, origin)

参数	描述
headers	类型: MutableHeaders
origin	类型: str

starlette/middleware/cors.py 中的源代码

@staticmethod
def allow_explicit_origin(headers: MutableHeaders, origin: str) -> None:
    headers["Access-Control-Allow-Origin"] = origin
    headers.add_vary_header("Origin")

可以从fastapi导入

from fastapi.middleware.cors import CORSMiddleware

fastapi.middleware.gzip.GZipMiddleware

GZipMiddleware(app, minimum_size=500, compresslevel=9)

参数	描述
app	类型: ASGIApp
minimum_size	类型: int 默认值: 500
compresslevel	类型: int 默认值: 9

源代码在starlette/middleware/gzip.py中

def __init__(self, app: ASGIApp, minimum_size: int = 500, compresslevel: int = 9) -> None:
    self.app = app
    self.minimum_size = minimum_size
    self.compresslevel = compresslevel

app 实例属性

app = app

minimum_size 实例属性

minimum_size = minimum_size

compresslevel 实例属性

compresslevel = compresslevel

可以从fastapi导入

from fastapi.middleware.gzip import GZipMiddleware

fastapi.middleware.httpsredirect.HTTPSRedirectMiddleware

HTTPSRedirectMiddleware(app)

参数	描述
app	类型: ASGIApp

源代码在starlette/middleware/httpsredirect.py中

def __init__(self, app: ASGIApp) -> None:
    self.app = app

app 实例属性

app = app

可以从fastapi导入

from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

fastapi.middleware.trustedhost.TrustedHostMiddleware

TrustedHostMiddleware(
    app, allowed_hosts=None, www_redirect=True
)

参数	描述
app	类型: ASGIApp
allowed_hosts	类型: Sequence[str] | None 默认值: None
www_redirect	类型: bool 默认值: True

源代码在starlette/middleware/trustedhost.py中

def __init__(
    self,
    app: ASGIApp,
    allowed_hosts: typing.Sequence[str] | None = None,
    www_redirect: bool = True,
) -> None:
    if allowed_hosts is None:
        allowed_hosts = ["*"]

    for pattern in allowed_hosts:
        assert "*" not in pattern[1:], ENFORCE_DOMAIN_WILDCARD
        if pattern.startswith("*") and pattern != "*":
            assert pattern.startswith("*."), ENFORCE_DOMAIN_WILDCARD
    self.app = app
    self.allowed_hosts = list(allowed_hosts)
    self.allow_any = "*" in allowed_hosts
    self.www_redirect = www_redirect

app 实例属性

app = app

allowed_hosts 实例属性

allowed_hosts = list(allowed_hosts)

allow_any 实例属性

allow_any = '*' in allowed_hosts

www_redirect 实例属性

www_redirect = www_redirect

可以从fastapi导入

from fastapi.middleware.trustedhost import TrustedHostMiddleware

fastapi.middleware.wsgi.WSGIMiddleware

WSGIMiddleware(app)

参数	描述
app	类型: Callable[..., Any]

源代码在starlette/middleware/wsgi.py中

def __init__(self, app: typing.Callable[..., typing.Any]) -> None:
    self.app = app

app 实例属性

app = app

可以从fastapi导入

from fastapi.middleware.wsgi import WSGIMiddleware