中間件

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