コンテンツにスキップ

pytilpack.quart

必要なextra

pip install pytilpack[quart]

pytilpack.quart

Quart関連のユーティリティ。

ResponseType = quart.Response | typing.Awaitable[quart.Response] module-attribute

レスポンスの型。

ConcurrencyState(semaphore, max_concurrency, timeout) dataclass

set_max_concurrency の内部状態。exhaust_concurrency から参照される。

ProxyFix(quartapp, x_for=1, x_proto=1, x_host=0, x_port=0, x_prefix=1, static_prefix=None, trusted_hosts=None)

リバースプロキシ対応。

nginx.conf設定例::

proxy_set_header X-Forwarded-For $remote_addr;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header X-Forwarded-Port $server_port;
proxy_set_header X-Forwarded-Prefix $http_x_forwarded_prefix;
参考
設計意図

prefixはpin方式で管理する。初回有効リクエスト時に一度だけapp.configへ反映し、 以降は変更しない。これにより攻撃者制御のX-Forwarded-Prefix値で Cookie発行パス等のアプリ全体設定が書き換わる経路を断つ。

運用時にprefixが既知であれば、static_prefix引数で起動時に確定させることを推奨する。 static_prefixを指定した場合、初期化時にapp.configを確定する。 ただしscope["root_path"]はリクエストごとにX-Forwarded-Prefixヘッダーの値から 設定するため、ヘッダーが来ないリクエストではscope["root_path"]は更新されない。

x_host > 0でX-Forwarded-Hostを有効化する場合は、trusted_hosts引数で 許可ホストリストを指定することを推奨する。指定がない場合は値検証（制御文字・ //始まり等の排除）のみ行い追加警告は出力しない。

旧挙動（リクエストごとにapp.configを上書き）を期待していた利用者は、 static_prefix指定方式、またはpin後の挙動（初回リクエストでpinされる）へ移行すること。

ソースコード位置： pytilpack/quart/proxy_fix.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(
    self,
    quartapp: quart.Quart,
    x_for: int = 1,
    x_proto: int = 1,
    x_host: int = 0,
    x_port: int = 0,
    x_prefix: int = 1,
    static_prefix: str | None = None,
    trusted_hosts: typing.Iterable[str] | None = None,
):
    self.quartapp = quartapp
    self.asgi_app = quartapp.asgi_app
    self.x_for = x_for
    self.x_proto = x_proto
    self.x_port = x_port
    self._x_host = x_host
    self._x_prefix = x_prefix
    self.trusted_hosts = list(trusted_hosts) if trusted_hosts is not None else None

    self._pin_lock = threading.Lock()
    self._pinned_prefix: str | None = None
    self._prefix_pinned = False

    if static_prefix is not None:
        validated = pytilpack.web.validate_forwarded_prefix(static_prefix)
        if validated is None:
            raise ValueError(f"static_prefixが不正: {static_prefix!r}")
        self._apply_prefix_to_config(validated)
        self._pinned_prefix = validated
        self._prefix_pinned = True

__call__(scope, receive, send) async

ASGIアプリケーションとしての処理。

ソースコード位置： pytilpack/quart/proxy_fix.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
async def __call__(
    self,
    scope: hypercorn.typing.Scope,
    receive: hypercorn.typing.ASGIReceiveCallable,
    send: hypercorn.typing.ASGISendCallable,
) -> None:
    """ASGIアプリケーションとしての処理。"""
    if scope["type"] in ("http", "websocket"):
        scope = typing.cast(hypercorn.typing.HTTPScope, copy.deepcopy(scope))
        headers = list(scope["headers"])

        # X-Forwarded-For → client
        forwarded_for = self._get_trusted_value(b"x-forwarded-for", headers, self.x_for)
        if forwarded_for and scope.get("client"):
            _, orig_port = scope.get("client") or (None, None)
            scope["client"] = (forwarded_for, orig_port or 0)

        # X-Forwarded-Proto → scheme
        forwarded_proto = self._get_trusted_value(b"x-forwarded-proto", headers, self.x_proto)
        if forwarded_proto:
            scope["scheme"] = forwarded_proto

        # X-Forwarded-Host → server & Host header
        forwarded_host = self._get_trusted_value(b"x-forwarded-host", headers, self._x_host)
        if forwarded_host:
            validated_host = pytilpack.web.validate_forwarded_host(forwarded_host)
            if validated_host is None:
                logger.warning(f"X-Forwarded-Hostに不正な値が含まれています: {forwarded_host!r}")
            elif self.trusted_hosts is not None and not pytilpack.web.is_host_in_trusted(
                validated_host, self.trusted_hosts
            ):
                logger.warning(f"X-Forwarded-Hostが許可リスト外です: {validated_host!r}")
            else:
                host_val = validated_host
                host, port = host_val, None
                if ":" in host_val and not host_val.startswith("["):
                    h, p = host_val.rsplit(":", 1)
                    if p.isdigit():
                        host, port = h, int(p)
                orig_server = scope.get("server") or (None, None)
                orig_port = orig_server[1]
                scope["server"] = (host, port or orig_port or 0)
                headers = [(hn, hv) for hn, hv in headers if hn.lower() != b"host"]
                host_hdr = host if port is None else f"{host}:{port}"
                headers.append((b"host", host_hdr.encode("utf-8", errors="replace")))

        # X-Forwarded-Port → server port & Host header
        forwarded_port = self._get_trusted_value(b"x-forwarded-port", headers, self.x_port)
        if forwarded_port and forwarded_port.isdigit():
            port_int = int(forwarded_port)
            orig_server = scope.get("server") or (None, None)
            orig_host = str(orig_server[0])
            scope["server"] = (orig_host, port_int)
            headers = [(hn, hv) for hn, hv in headers if hn.lower() != b"host"]
            headers.append((b"host", f"{orig_host}:{port_int}".encode()))

        # X-Forwarded-Prefix → root_path + config
        forwarded_prefix = self._get_trusted_value(b"x-forwarded-prefix", headers, self._x_prefix)
        if forwarded_prefix:
            prefix = pytilpack.web.validate_forwarded_prefix(forwarded_prefix)
            if prefix is None:
                logger.warning(f"X-Forwarded-Prefixに不正な値が含まれています: {forwarded_prefix!r}")
            else:
                scope["root_path"] = prefix
                if not self._prefix_pinned:
                    with self._pin_lock:
                        if not self._prefix_pinned:
                            self._apply_prefix_to_config(prefix)
                            self._pinned_prefix = prefix
                            self._prefix_pinned = True
                        elif self._pinned_prefix != prefix:
                            logger.warning(
                                f"X-Forwarded-Prefixがpin済みの値と異なります: "
                                f"pinned={self._pinned_prefix!r}, received={prefix!r}"
                            )
                elif self._pinned_prefix != prefix:
                    logger.warning(
                        f"X-Forwarded-Prefixがpin済みの値と異なります: pinned={self._pinned_prefix!r}, received={prefix!r}"
                    )

        scope["headers"] = headers

    await self.asgi_app(scope, receive, send)

assert_bytes(response, status_code=200, content_type=None) async

レスポンスのステータスコードとContent-Typeを検証してボディをbytesで返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200
content_type str | Iterable[str] | None

期待するContent-Type

 None

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合など

戻り値：

タイプ デスクリプション
bytes

レスポンスボディ
ソースコード位置： pytilpack/quart/asserts.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
async def assert_bytes(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = None,
) -> bytes:
    """レスポンスのステータスコードとContent-Typeを検証してボディをbytesで返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード
        content_type: 期待するContent-Type

    Raises:
        AssertionError: ステータスコードが異なる場合など

    Returns:
        レスポンスボディ

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=False)
    _core.assert_bytes_core(response_body, response.status_code, response.content_type, status_code, content_type)
    return response_body

assert_html(response, status_code=200, content_type='__default__', strict=False, tmp_path=None) async

レスポンスを検証してHTMLボディを文字列で返す。html5libが必要。

strict・tmp_pathはキーワード引数での指定を推奨する。flask/quart/fastapi間で 位置引数順を揃えているが、将来の引数追加時の後方互換性を保つためである。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200
content_type str | Iterable[str] | None

期待するContent-Type

 '__default__'
strict bool

Trueの場合、HTML5の仕様に従ったパースを行う

 False
tmp_path Path | None

一時ファイルを保存するディレクトリ

 None

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合など

戻り値：

タイプ デスクリプション
str

レスポンスボディ (bs4.BeautifulSoup)
ソースコード位置： pytilpack/quart/asserts.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def assert_html(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "__default__",
    strict: bool = False,
    tmp_path: pathlib.Path | None = None,
) -> str:
    """レスポンスを検証してHTMLボディを文字列で返す。html5libが必要。

    strict・tmp_pathはキーワード引数での指定を推奨する。flask/quart/fastapi間で
    位置引数順を揃えているが、将来の引数追加時の後方互換性を保つためである。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード
        content_type: 期待するContent-Type
        strict: Trueの場合、HTML5の仕様に従ったパースを行う
        tmp_path: 一時ファイルを保存するディレクトリ

    Raises:
        AssertionError: ステータスコードが異なる場合など

    Returns:
        レスポンスボディ (bs4.BeautifulSoup)

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=True)
    response_bytes = await response.get_data(as_text=False)
    _core.assert_html_core(
        response_body,
        response_bytes,
        response.status_code,
        response.content_type,
        status_code,
        content_type,
        strict,
        tmp_path,
    )
    return response_body

assert_json(response, status_code=200, content_type='application/json') async

レスポンスを検証してJSONをデコードして返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200
content_type str | Iterable[str] | None

期待するContent-Type

 'application/json'

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合など

戻り値：

タイプ デスクリプション
Any

レスポンスのJSONデコード結果
ソースコード位置： pytilpack/quart/asserts.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def assert_json(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "application/json",
) -> typing.Any:
    """レスポンスを検証してJSONをデコードして返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード
        content_type: 期待するContent-Type

    Raises:
        AssertionError: ステータスコードが異なる場合など

    Returns:
        レスポンスのJSONデコード結果

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=True)
    return _core.assert_json_core(response_body, response.status_code, response.content_type, status_code, content_type)

assert_xml(response, status_code=200, content_type='__default__') async

レスポンスを検証してXMLボディを文字列で返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200
content_type str | Iterable[str] | None

期待するContent-Type

 '__default__'

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合など

戻り値：

タイプ デスクリプション
str

レスポンスボディ
ソースコード位置： pytilpack/quart/asserts.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def assert_xml(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "__default__",
) -> str:
    """レスポンスを検証してXMLボディを文字列で返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード
        content_type: 期待するContent-Type

    Raises:
        AssertionError: ステータスコードが異なる場合など

    Returns:
        レスポンスボディ

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=True)
    _core.assert_xml_core(response_body, response.status_code, response.content_type, status_code, content_type)
    return response_body

assert_sse(response, status_code=200) async

レスポンスのステータスコードとSSE用Content-Typeを検証して返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合、またはContent-Typeが異なる場合

戻り値：

タイプ デスクリプション
Response

レスポンス
ソースコード位置： pytilpack/quart/asserts.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
async def assert_sse(
    response: ResponseType,
    status_code: int = 200,
) -> quart.Response:
    """レスポンスのステータスコードとSSE用Content-Typeを検証して返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード

    Raises:
        AssertionError: ステータスコードが異なる場合、またはContent-Typeが異なる場合

    Returns:
        レスポンス

    """
    response = await _get_response(response)
    _core.assert_sse_core(response.status_code, response.content_type, status_code)
    return response

assert_response(response, status_code=200) async

レスポンスのステータスコードを検証して返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合

戻り値：

タイプ デスクリプション
Response

レスポンス
ソースコード位置： pytilpack/quart/asserts.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
async def assert_response(
    response: ResponseType,
    status_code: int = 200,
) -> quart.Response:
    """レスポンスのステータスコードを検証して返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード

    Raises:
        AssertionError: ステータスコードが異なる場合

    Returns:
        レスポンス

    """
    response = await _get_response(response)
    _core.assert_response_core(response.status_code, status_code)
    return response

init_app(app, locale_dir, domain='messages', supported_locales=None, default_locale='en', fallback=True)

Quartアプリにi18nを統合する。

before_requestでAccept-Languageからロケールを自動設定し、 Jinja2テンプレートに翻訳関数を登録する。

引数：

名前 タイプ デスクリプション デフォルト
app Quart

Quartアプリ

 必須
locale_dir str | Path

localeディレクトリのパス

 必須
domain str

gettextドメイン名

 'messages'
supported_locales list[str] | None

サポートするロケール一覧

 None
default_locale str

デフォルトロケール

 'en'
fallback bool

フォールバック有無

 True
ソースコード位置： pytilpack/quart/i18n.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def init_app(
    app: quart.Quart,
    locale_dir: str | pathlib.Path,
    domain: str = "messages",
    supported_locales: list[str] | None = None,
    default_locale: str = "en",
    fallback: bool = True,
) -> None:
    """Quartアプリにi18nを統合する。

    before_requestでAccept-Languageからロケールを自動設定し、
    Jinja2テンプレートに翻訳関数を登録する。

    Args:
        app: Quartアプリ
        locale_dir: localeディレクトリのパス
        domain: gettextドメイン名
        supported_locales: サポートするロケール一覧
        default_locale: デフォルトロケール
        fallback: フォールバック有無

    """
    # I18nStateを作成してextensionsに格納
    state = pytilpack.i18n.I18nState(
        locale_dir=locale_dir,
        domain=domain,
        locales=supported_locales,
        default_locale=default_locale,
        fallback=fallback,
    )
    app.extensions["pytilpack_i18n"] = state

    # Jinja2テンプレートに翻訳関数を登録
    app.jinja_env.globals["_"] = pytilpack.i18n.gettext_func
    app.jinja_env.globals["ngettext"] = pytilpack.i18n.ngettext

    @app.before_request
    async def _set_locale_from_request() -> None:
        i18n_state: pytilpack.i18n.I18nState = app.extensions["pytilpack_i18n"]
        # Accept-Languageからロケールを決定
        accept_lang = quart.request.headers.get("Accept-Language", "")
        locale = pytilpack.http.select_accept_language(
            accept_lang,
            i18n_state.supported_locales,
            default=i18n_state.default_locale,
        )
        # Tokenタプルをgに保存してteardown_requestでdeactivateする
        quart.g.pytilpack_i18n_tokens = pytilpack.i18n.activate(i18n_state, locale)

    @app.teardown_request
    async def _teardown_locale(exc: BaseException | None) -> None:
        del exc  # noqa
        if hasattr(quart.g, "pytilpack_i18n_tokens"):
            pytilpack.i18n.deactivate(quart.g.pytilpack_i18n_tokens)

set_max_concurrency(app, max_concurrency, timeout=3.0)

Quart アプリ全体の最大同時リクエスト数を制限する。

引数：

名前 タイプ デスクリプション デフォルト
app Quart

対象の Quart アプリケーション。

 必須
max_concurrency int

許可する同時リクエスト数の上限。

 必須
timeout float | None

最大待機秒数。タイムアウト時は 503 Service Unavailable を返す。

 3.0

発生：

タイプ デスクリプション
ValueError

max_concurrency が 1 未満の場合。
RuntimeError

同一アプリに対して二度目に呼び出した場合。 before_request/teardown_request が重複登録されてセマフォが二重管理されるため、 同一アプリへの複数回呼び出しは禁止する。
Notes
  • Hypercorn の --workers / --threads とは独立した アプリレベルの制御。1 ワーカー内のコルーチン数を制限する用途で使う。
ソースコード位置： pytilpack/quart/misc.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def set_max_concurrency(app: quart.Quart, max_concurrency: int, timeout: float | None = 3.0) -> None:
    """Quart アプリ全体の最大同時リクエスト数を制限する。

    Args:
        app: 対象の Quart アプリケーション。
        max_concurrency: 許可する同時リクエスト数の上限。
        timeout: 最大待機秒数。タイムアウト時は 503 Service Unavailable を返す。

    Raises:
        ValueError: max_concurrency が 1 未満の場合。
        RuntimeError: 同一アプリに対して二度目に呼び出した場合。
            before_request/teardown_request が重複登録されてセマフォが二重管理されるため、
            同一アプリへの複数回呼び出しは禁止する。

    Notes:
        * Hypercorn の ``--workers`` / ``--threads`` とは独立した
        アプリレベルの制御。1 ワーカー内のコルーチン数を制限する用途で使う。
    """
    if max_concurrency < 1:
        raise ValueError("max_concurrency must be >= 1")
    if "pytilpack_concurrency" in app.extensions:
        raise RuntimeError("set_max_concurrency is already configured on this app")

    semaphore = asyncio.Semaphore(max_concurrency)
    state = ConcurrencyState(
        semaphore=semaphore,
        max_concurrency=max_concurrency,
        timeout=timeout,
    )
    app.extensions["pytilpack_concurrency"] = state

    async def _acquire() -> None:  # before_request
        # テスト時にセマフォ/timeoutを一時変更できるようstateから読む
        sem = state.semaphore
        to = state.timeout
        try:
            if to is None:
                await sem.acquire()
            else:
                await asyncio.wait_for(sem.acquire(), timeout=to)
        except TimeoutError:
            logger.warning(f"Concurrency limit reached, aborting request: {quart.request.path}")
            quart.abort(
                503,
                description="サーバーが混みあっています。しばらく待ってから再度お試しください。",
            )
        # asyncio.CancelledError はここでは捕捉しない。
        # - to is None 経路（await sem.acquire() 直接）: acquire 待機中にキャンセルされると
        #   セマフォ未取得のまま CancelledError が伝播するため、漏洩は発生しない。
        # - to is not None 経路（asyncio.wait_for 経由）: wait_for がキャンセルされた場合、
        #   CPython の保証により acquire 完了済みであれば内部で release 済み。
        #   未完了であれば未取得のまま伝播する。
        # いずれの経路でも、この時点では未取得状態が保証されるため保護は不要。

        # acquire 完了。トークン設定前にキャンセルが発生してもセマフォを解放するよう保護する。
        try:
            quart.g.quart__concurrency_token = True
        except BaseException:
            sem.release()
            raise

    async def _release(_: typing.Any) -> None:
        if hasattr(quart.g, "quart__concurrency_token"):
            state.semaphore.release()
            del quart.g.quart__concurrency_token

    app.before_request(_acquire)
    app.teardown_request(_release)

exhaust_concurrency(app) async

テスト用: セマフォを枯渇させて503を発生させるコンテキストマネージャ。

使用例::

async with pytilpack.quart.exhaust_concurrency(app):
    response = await client.get("/any-endpoint")
    assert response.status_code == 503
ソースコード位置： pytilpack/quart/misc.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@contextlib.asynccontextmanager
async def exhaust_concurrency(app: quart.Quart):
    """テスト用: セマフォを枯渇させて503を発生させるコンテキストマネージャ。

    使用例::

        async with pytilpack.quart.exhaust_concurrency(app):
            response = await client.get("/any-endpoint")
            assert response.status_code == 503
    """
    state: ConcurrencyState = app.extensions["pytilpack_concurrency"]
    original_semaphore = state.semaphore
    original_timeout = state.timeout

    # スロット0のセマフォに差し替えて即座にタイムアウト→503を返すようにする
    state.semaphore = asyncio.Semaphore(0)
    state.timeout = 0.01

    try:
        yield
    finally:
        state.semaphore = original_semaphore
        state.timeout = original_timeout

run_sync(func)

同期関数を非同期に実行するデコレーター。

quart.utils.run_syncの型ヒントがいまいちなので用意。

ソースコード位置： pytilpack/quart/misc.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def run_sync[**P, R](
    func: typing.Callable[P, R],
) -> typing.Callable[P, typing.Awaitable[R]]:
    """同期関数を非同期に実行するデコレーター。

    quart.utils.run_syncの型ヒントがいまいちなので用意。

    """

    @functools.wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        result = await quart.utils.run_sync(func)(*args, **kwargs)
        return typing.cast(R, result)

    return wrapper

get_next_url()

ログイン後遷移用のnextパラメータ用のURLを返す。

ソースコード位置： pytilpack/quart/misc.py 
158
159
160
161
162
163
164
def get_next_url() -> str:
    """ログイン後遷移用のnextパラメータ用のURLを返す。"""
    return pytilpack.web.build_next_url(
        quart.request.script_root,
        quart.request.path,
        quart.request.query_string.decode("utf-8"),
    )

prefer_markdown()

AcceptヘッダーでmarkdownがHTMLより優先されているかを返す。

参考: https://vercel.com/blog/making-agent-friendly-pages-with-content-negotiation

(CDNやプロキシがAcceptヘッダーを書き換える場合があるという話もあるが…。)

戻り値：

タイプ デスクリプション
bool

markdownがHTMLより優先されている場合True、そうでなければFalse
ソースコード位置： pytilpack/quart/misc.py 
167
168
169
170
171
172
173
174
175
176
177
178
def prefer_markdown() -> bool:
    """AcceptヘッダーでmarkdownがHTMLより優先されているかを返す。

    参考: <https://vercel.com/blog/making-agent-friendly-pages-with-content-negotiation>

    (CDNやプロキシがAcceptヘッダーを書き換える場合があるという話もあるが…。)

    Returns:
        markdownがHTMLより優先されている場合True、そうでなければFalse

    """
    return pytilpack.web.is_prefer_markdown(quart.request.headers.get("Accept", ""))

static_url_for(filename, cache_busting=True, cache_timestamp='when_not_debug', **kwargs)

静的ファイルのURLを生成する。

引数：

名前 タイプ デスクリプション デフォルト
filename str

静的ファイルの名前

 必須
cache_busting bool

キャッシュバスティングを有効にするかどうか (デフォルト: True)

 True
cache_timestamp bool | Literal['when_not_debug']

キャッシュバスティングするときのファイルの最終更新日時をプロセス単位でキャッシュするか否か。 - True: プロセス単位でキャッシュする。プロセスの再起動やSIGHUPなどをしない限り更新されない。 - False: キャッシュしない。常に最新を参照する。 - "when_not_debug": デバッグモードでないときのみキャッシュする。

 'when_not_debug'
**kwargs Any

その他の引数 (quart.url_forに渡される)

 {}

戻り値：

タイプ デスクリプション
str

静的ファイルのURL
ソースコード位置： pytilpack/quart/misc.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def static_url_for(
    filename: str,
    cache_busting: bool = True,
    cache_timestamp: bool | typing.Literal["when_not_debug"] = "when_not_debug",
    **kwargs: typing.Any,
) -> str:
    """静的ファイルのURLを生成する。

    Args:
        filename: 静的ファイルの名前
        cache_busting: キャッシュバスティングを有効にするかどうか (デフォルト: True)
        cache_timestamp: キャッシュバスティングするときのファイルの最終更新日時をプロセス単位でキャッシュするか否か。
            - True: プロセス単位でキャッシュする。プロセスの再起動やSIGHUPなどをしない限り更新されない。
            - False: キャッシュしない。常に最新を参照する。
            - "when_not_debug": デバッグモードでないときのみキャッシュする。
        **kwargs: その他の引数 (quart.url_forに渡される)

    Returns:
        静的ファイルのURL
    """
    return pytilpack.web.resolve_static_url(
        quart.current_app.static_folder,
        filename,
        bool(quart.current_app.debug),
        cache_busting,
        cache_timestamp,
        lambda **kw: quart.url_for("static", **kw),
        _TIMESTAMP_CACHE,
        **kwargs,
    )

get_routes(app)

ルーティング情報を取得する。

戻り値：

タイプ デスクリプション
list[RouteInfo]

ルーティング情報のリスト。
ソースコード位置： pytilpack/quart/misc.py 
217
218
219
220
221
222
223
def get_routes(app: quart.Quart) -> list[RouteInfo]:
    """ルーティング情報を取得する。

    Returns:
        ルーティング情報のリスト。
    """
    return pytilpack.web.build_routes(app.url_map.iter_rules(), app.config.get("APPLICATION_ROOT"))

run(app, host='localhost', port=5000) async

Quartアプリを実行するコンテキストマネージャ。テストコードなど用。

ソースコード位置： pytilpack/quart/misc.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
@contextlib.asynccontextmanager
async def run(app: quart.Quart, host: str = "localhost", port: int = 5000):
    """Quartアプリを実行するコンテキストマネージャ。テストコードなど用。"""
    # ダミーエンドポイントが存在しない場合は追加
    if not any(rule.endpoint == "_pytilpack_quart_dummy" for rule in app.url_map.iter_rules()):

        @app.route("/_pytilpack_quart_dummy")
        async def _pytilpack_quart_dummy():
            return "OK"

    # Uvicornサーバーの設定
    config = uvicorn.Config(app=app, host=host, port=port)
    server = uvicorn.Server(config)

    # 別スレッドでサーバーを起動
    def run_server():
        asyncio.run(server.serve())

    thread = threading.Thread(target=run_server, daemon=True)
    thread.start()

    try:
        # サーバーが起動するまで待機
        async with httpx.AsyncClient() as client:
            while True:
                try:
                    response = await client.get(f"http://{host}:{port}/_pytilpack_quart_dummy")
                    response.raise_for_status()
                    break
                except Exception:
                    await asyncio.sleep(0.1)  # 少し待機

        # 制御を戻す
        yield

    finally:
        # サーバーを停止
        server.should_exit = True
        thread.join(timeout=5.0)  # タイムアウトを設定

asserts

Quartのテストコード用アサーション関数。

ResponseType = quart.Response | typing.Awaitable[quart.Response] module-attribute

レスポンスの型。

assert_bytes(response, status_code=200, content_type=None) async

レスポンスのステータスコードとContent-Typeを検証してボディをbytesで返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200
content_type str | Iterable[str] | None

期待するContent-Type

 None

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合など

戻り値：

タイプ デスクリプション
bytes

レスポンスボディ
ソースコード位置： pytilpack/quart/asserts.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
async def assert_bytes(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = None,
) -> bytes:
    """レスポンスのステータスコードとContent-Typeを検証してボディをbytesで返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード
        content_type: 期待するContent-Type

    Raises:
        AssertionError: ステータスコードが異なる場合など

    Returns:
        レスポンスボディ

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=False)
    _core.assert_bytes_core(response_body, response.status_code, response.content_type, status_code, content_type)
    return response_body

assert_html(response, status_code=200, content_type='__default__', strict=False, tmp_path=None) async

レスポンスを検証してHTMLボディを文字列で返す。html5libが必要。

strict・tmp_pathはキーワード引数での指定を推奨する。flask/quart/fastapi間で 位置引数順を揃えているが、将来の引数追加時の後方互換性を保つためである。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200
content_type str | Iterable[str] | None

期待するContent-Type

 '__default__'
strict bool

Trueの場合、HTML5の仕様に従ったパースを行う

 False
tmp_path Path | None

一時ファイルを保存するディレクトリ

 None

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合など

戻り値：

タイプ デスクリプション
str

レスポンスボディ (bs4.BeautifulSoup)
ソースコード位置： pytilpack/quart/asserts.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
async def assert_html(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "__default__",
    strict: bool = False,
    tmp_path: pathlib.Path | None = None,
) -> str:
    """レスポンスを検証してHTMLボディを文字列で返す。html5libが必要。

    strict・tmp_pathはキーワード引数での指定を推奨する。flask/quart/fastapi間で
    位置引数順を揃えているが、将来の引数追加時の後方互換性を保つためである。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード
        content_type: 期待するContent-Type
        strict: Trueの場合、HTML5の仕様に従ったパースを行う
        tmp_path: 一時ファイルを保存するディレクトリ

    Raises:
        AssertionError: ステータスコードが異なる場合など

    Returns:
        レスポンスボディ (bs4.BeautifulSoup)

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=True)
    response_bytes = await response.get_data(as_text=False)
    _core.assert_html_core(
        response_body,
        response_bytes,
        response.status_code,
        response.content_type,
        status_code,
        content_type,
        strict,
        tmp_path,
    )
    return response_body

assert_json(response, status_code=200, content_type='application/json') async

レスポンスを検証してJSONをデコードして返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200
content_type str | Iterable[str] | None

期待するContent-Type

 'application/json'

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合など

戻り値：

タイプ デスクリプション
Any

レスポンスのJSONデコード結果
ソースコード位置： pytilpack/quart/asserts.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def assert_json(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "application/json",
) -> typing.Any:
    """レスポンスを検証してJSONをデコードして返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード
        content_type: 期待するContent-Type

    Raises:
        AssertionError: ステータスコードが異なる場合など

    Returns:
        レスポンスのJSONデコード結果

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=True)
    return _core.assert_json_core(response_body, response.status_code, response.content_type, status_code, content_type)

assert_xml(response, status_code=200, content_type='__default__') async

レスポンスを検証してXMLボディを文字列で返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200
content_type str | Iterable[str] | None

期待するContent-Type

 '__default__'

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合など

戻り値：

タイプ デスクリプション
str

レスポンスボディ
ソースコード位置： pytilpack/quart/asserts.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
async def assert_xml(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "__default__",
) -> str:
    """レスポンスを検証してXMLボディを文字列で返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード
        content_type: 期待するContent-Type

    Raises:
        AssertionError: ステータスコードが異なる場合など

    Returns:
        レスポンスボディ

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=True)
    _core.assert_xml_core(response_body, response.status_code, response.content_type, status_code, content_type)
    return response_body

assert_sse(response, status_code=200) async

レスポンスのステータスコードとSSE用Content-Typeを検証して返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合、またはContent-Typeが異なる場合

戻り値：

タイプ デスクリプション
Response

レスポンス
ソースコード位置： pytilpack/quart/asserts.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
async def assert_sse(
    response: ResponseType,
    status_code: int = 200,
) -> quart.Response:
    """レスポンスのステータスコードとSSE用Content-Typeを検証して返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード

    Raises:
        AssertionError: ステータスコードが異なる場合、またはContent-Typeが異なる場合

    Returns:
        レスポンス

    """
    response = await _get_response(response)
    _core.assert_sse_core(response.status_code, response.content_type, status_code)
    return response

assert_response(response, status_code=200) async

レスポンスのステータスコードを検証して返す。

引数：

名前 タイプ デスクリプション デフォルト
response ResponseType

レスポンス

 必須
status_code int

期待するステータスコード

 200

発生：

タイプ デスクリプション
AssertionError

ステータスコードが異なる場合

戻り値：

タイプ デスクリプション
Response

レスポンス
ソースコード位置： pytilpack/quart/asserts.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
async def assert_response(
    response: ResponseType,
    status_code: int = 200,
) -> quart.Response:
    """レスポンスのステータスコードを検証して返す。

    Args:
        response: レスポンス
        status_code: 期待するステータスコード

    Raises:
        AssertionError: ステータスコードが異なる場合

    Returns:
        レスポンス

    """
    response = await _get_response(response)
    _core.assert_response_core(response.status_code, status_code)
    return response

i18n

Quart用i18n統合。

init_app(app, locale_dir, domain='messages', supported_locales=None, default_locale='en', fallback=True)

Quartアプリにi18nを統合する。

before_requestでAccept-Languageからロケールを自動設定し、 Jinja2テンプレートに翻訳関数を登録する。

引数：

名前 タイプ デスクリプション デフォルト
app Quart

Quartアプリ

 必須
locale_dir str | Path

localeディレクトリのパス

 必須
domain str

gettextドメイン名

 'messages'
supported_locales list[str] | None

サポートするロケール一覧

 None
default_locale str

デフォルトロケール

 'en'
fallback bool

フォールバック有無

 True
ソースコード位置： pytilpack/quart/i18n.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def init_app(
    app: quart.Quart,
    locale_dir: str | pathlib.Path,
    domain: str = "messages",
    supported_locales: list[str] | None = None,
    default_locale: str = "en",
    fallback: bool = True,
) -> None:
    """Quartアプリにi18nを統合する。

    before_requestでAccept-Languageからロケールを自動設定し、
    Jinja2テンプレートに翻訳関数を登録する。

    Args:
        app: Quartアプリ
        locale_dir: localeディレクトリのパス
        domain: gettextドメイン名
        supported_locales: サポートするロケール一覧
        default_locale: デフォルトロケール
        fallback: フォールバック有無

    """
    # I18nStateを作成してextensionsに格納
    state = pytilpack.i18n.I18nState(
        locale_dir=locale_dir,
        domain=domain,
        locales=supported_locales,
        default_locale=default_locale,
        fallback=fallback,
    )
    app.extensions["pytilpack_i18n"] = state

    # Jinja2テンプレートに翻訳関数を登録
    app.jinja_env.globals["_"] = pytilpack.i18n.gettext_func
    app.jinja_env.globals["ngettext"] = pytilpack.i18n.ngettext

    @app.before_request
    async def _set_locale_from_request() -> None:
        i18n_state: pytilpack.i18n.I18nState = app.extensions["pytilpack_i18n"]
        # Accept-Languageからロケールを決定
        accept_lang = quart.request.headers.get("Accept-Language", "")
        locale = pytilpack.http.select_accept_language(
            accept_lang,
            i18n_state.supported_locales,
            default=i18n_state.default_locale,
        )
        # Tokenタプルをgに保存してteardown_requestでdeactivateする
        quart.g.pytilpack_i18n_tokens = pytilpack.i18n.activate(i18n_state, locale)

    @app.teardown_request
    async def _teardown_locale(exc: BaseException | None) -> None:
        del exc  # noqa
        if hasattr(quart.g, "pytilpack_i18n_tokens"):
            pytilpack.i18n.deactivate(quart.g.pytilpack_i18n_tokens)

misc

Quart関連のその他のユーティリティ。

ConcurrencyState(semaphore, max_concurrency, timeout) dataclass

set_max_concurrency の内部状態。exhaust_concurrency から参照される。

set_max_concurrency(app, max_concurrency, timeout=3.0)

Quart アプリ全体の最大同時リクエスト数を制限する。

引数：

名前 タイプ デスクリプション デフォルト
app Quart

対象の Quart アプリケーション。

 必須
max_concurrency int

許可する同時リクエスト数の上限。

 必須
timeout float | None

最大待機秒数。タイムアウト時は 503 Service Unavailable を返す。

 3.0

発生：

タイプ デスクリプション
ValueError

max_concurrency が 1 未満の場合。
RuntimeError

同一アプリに対して二度目に呼び出した場合。 before_request/teardown_request が重複登録されてセマフォが二重管理されるため、 同一アプリへの複数回呼び出しは禁止する。
Notes
  • Hypercorn の --workers / --threads とは独立した アプリレベルの制御。1 ワーカー内のコルーチン数を制限する用途で使う。
ソースコード位置： pytilpack/quart/misc.py 
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def set_max_concurrency(app: quart.Quart, max_concurrency: int, timeout: float | None = 3.0) -> None:
    """Quart アプリ全体の最大同時リクエスト数を制限する。

    Args:
        app: 対象の Quart アプリケーション。
        max_concurrency: 許可する同時リクエスト数の上限。
        timeout: 最大待機秒数。タイムアウト時は 503 Service Unavailable を返す。

    Raises:
        ValueError: max_concurrency が 1 未満の場合。
        RuntimeError: 同一アプリに対して二度目に呼び出した場合。
            before_request/teardown_request が重複登録されてセマフォが二重管理されるため、
            同一アプリへの複数回呼び出しは禁止する。

    Notes:
        * Hypercorn の ``--workers`` / ``--threads`` とは独立した
        アプリレベルの制御。1 ワーカー内のコルーチン数を制限する用途で使う。
    """
    if max_concurrency < 1:
        raise ValueError("max_concurrency must be >= 1")
    if "pytilpack_concurrency" in app.extensions:
        raise RuntimeError("set_max_concurrency is already configured on this app")

    semaphore = asyncio.Semaphore(max_concurrency)
    state = ConcurrencyState(
        semaphore=semaphore,
        max_concurrency=max_concurrency,
        timeout=timeout,
    )
    app.extensions["pytilpack_concurrency"] = state

    async def _acquire() -> None:  # before_request
        # テスト時にセマフォ/timeoutを一時変更できるようstateから読む
        sem = state.semaphore
        to = state.timeout
        try:
            if to is None:
                await sem.acquire()
            else:
                await asyncio.wait_for(sem.acquire(), timeout=to)
        except TimeoutError:
            logger.warning(f"Concurrency limit reached, aborting request: {quart.request.path}")
            quart.abort(
                503,
                description="サーバーが混みあっています。しばらく待ってから再度お試しください。",
            )
        # asyncio.CancelledError はここでは捕捉しない。
        # - to is None 経路（await sem.acquire() 直接）: acquire 待機中にキャンセルされると
        #   セマフォ未取得のまま CancelledError が伝播するため、漏洩は発生しない。
        # - to is not None 経路（asyncio.wait_for 経由）: wait_for がキャンセルされた場合、
        #   CPython の保証により acquire 完了済みであれば内部で release 済み。
        #   未完了であれば未取得のまま伝播する。
        # いずれの経路でも、この時点では未取得状態が保証されるため保護は不要。

        # acquire 完了。トークン設定前にキャンセルが発生してもセマフォを解放するよう保護する。
        try:
            quart.g.quart__concurrency_token = True
        except BaseException:
            sem.release()
            raise

    async def _release(_: typing.Any) -> None:
        if hasattr(quart.g, "quart__concurrency_token"):
            state.semaphore.release()
            del quart.g.quart__concurrency_token

    app.before_request(_acquire)
    app.teardown_request(_release)

exhaust_concurrency(app) async

テスト用: セマフォを枯渇させて503を発生させるコンテキストマネージャ。

使用例::

async with pytilpack.quart.exhaust_concurrency(app):
    response = await client.get("/any-endpoint")
    assert response.status_code == 503
ソースコード位置： pytilpack/quart/misc.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@contextlib.asynccontextmanager
async def exhaust_concurrency(app: quart.Quart):
    """テスト用: セマフォを枯渇させて503を発生させるコンテキストマネージャ。

    使用例::

        async with pytilpack.quart.exhaust_concurrency(app):
            response = await client.get("/any-endpoint")
            assert response.status_code == 503
    """
    state: ConcurrencyState = app.extensions["pytilpack_concurrency"]
    original_semaphore = state.semaphore
    original_timeout = state.timeout

    # スロット0のセマフォに差し替えて即座にタイムアウト→503を返すようにする
    state.semaphore = asyncio.Semaphore(0)
    state.timeout = 0.01

    try:
        yield
    finally:
        state.semaphore = original_semaphore
        state.timeout = original_timeout

run_sync(func)

同期関数を非同期に実行するデコレーター。

quart.utils.run_syncの型ヒントがいまいちなので用意。

ソースコード位置： pytilpack/quart/misc.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def run_sync[**P, R](
    func: typing.Callable[P, R],
) -> typing.Callable[P, typing.Awaitable[R]]:
    """同期関数を非同期に実行するデコレーター。

    quart.utils.run_syncの型ヒントがいまいちなので用意。

    """

    @functools.wraps(func)
    async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        result = await quart.utils.run_sync(func)(*args, **kwargs)
        return typing.cast(R, result)

    return wrapper

get_next_url()

ログイン後遷移用のnextパラメータ用のURLを返す。

ソースコード位置： pytilpack/quart/misc.py 
158
159
160
161
162
163
164
def get_next_url() -> str:
    """ログイン後遷移用のnextパラメータ用のURLを返す。"""
    return pytilpack.web.build_next_url(
        quart.request.script_root,
        quart.request.path,
        quart.request.query_string.decode("utf-8"),
    )

prefer_markdown()

AcceptヘッダーでmarkdownがHTMLより優先されているかを返す。

参考: https://vercel.com/blog/making-agent-friendly-pages-with-content-negotiation

(CDNやプロキシがAcceptヘッダーを書き換える場合があるという話もあるが…。)

戻り値：

タイプ デスクリプション
bool

markdownがHTMLより優先されている場合True、そうでなければFalse
ソースコード位置： pytilpack/quart/misc.py 
167
168
169
170
171
172
173
174
175
176
177
178
def prefer_markdown() -> bool:
    """AcceptヘッダーでmarkdownがHTMLより優先されているかを返す。

    参考: <https://vercel.com/blog/making-agent-friendly-pages-with-content-negotiation>

    (CDNやプロキシがAcceptヘッダーを書き換える場合があるという話もあるが…。)

    Returns:
        markdownがHTMLより優先されている場合True、そうでなければFalse

    """
    return pytilpack.web.is_prefer_markdown(quart.request.headers.get("Accept", ""))

static_url_for(filename, cache_busting=True, cache_timestamp='when_not_debug', **kwargs)

静的ファイルのURLを生成する。

引数：

名前 タイプ デスクリプション デフォルト
filename str

静的ファイルの名前

 必須
cache_busting bool

キャッシュバスティングを有効にするかどうか (デフォルト: True)

 True
cache_timestamp bool | Literal['when_not_debug']

キャッシュバスティングするときのファイルの最終更新日時をプロセス単位でキャッシュするか否か。 - True: プロセス単位でキャッシュする。プロセスの再起動やSIGHUPなどをしない限り更新されない。 - False: キャッシュしない。常に最新を参照する。 - "when_not_debug": デバッグモードでないときのみキャッシュする。

 'when_not_debug'
**kwargs Any

その他の引数 (quart.url_forに渡される)

 {}

戻り値：

タイプ デスクリプション
str

静的ファイルのURL
ソースコード位置： pytilpack/quart/misc.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def static_url_for(
    filename: str,
    cache_busting: bool = True,
    cache_timestamp: bool | typing.Literal["when_not_debug"] = "when_not_debug",
    **kwargs: typing.Any,
) -> str:
    """静的ファイルのURLを生成する。

    Args:
        filename: 静的ファイルの名前
        cache_busting: キャッシュバスティングを有効にするかどうか (デフォルト: True)
        cache_timestamp: キャッシュバスティングするときのファイルの最終更新日時をプロセス単位でキャッシュするか否か。
            - True: プロセス単位でキャッシュする。プロセスの再起動やSIGHUPなどをしない限り更新されない。
            - False: キャッシュしない。常に最新を参照する。
            - "when_not_debug": デバッグモードでないときのみキャッシュする。
        **kwargs: その他の引数 (quart.url_forに渡される)

    Returns:
        静的ファイルのURL
    """
    return pytilpack.web.resolve_static_url(
        quart.current_app.static_folder,
        filename,
        bool(quart.current_app.debug),
        cache_busting,
        cache_timestamp,
        lambda **kw: quart.url_for("static", **kw),
        _TIMESTAMP_CACHE,
        **kwargs,
    )

get_routes(app)

ルーティング情報を取得する。

戻り値：

タイプ デスクリプション
list[RouteInfo]

ルーティング情報のリスト。
ソースコード位置： pytilpack/quart/misc.py 
217
218
219
220
221
222
223
def get_routes(app: quart.Quart) -> list[RouteInfo]:
    """ルーティング情報を取得する。

    Returns:
        ルーティング情報のリスト。
    """
    return pytilpack.web.build_routes(app.url_map.iter_rules(), app.config.get("APPLICATION_ROOT"))

run(app, host='localhost', port=5000) async

Quartアプリを実行するコンテキストマネージャ。テストコードなど用。

ソースコード位置： pytilpack/quart/misc.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
@contextlib.asynccontextmanager
async def run(app: quart.Quart, host: str = "localhost", port: int = 5000):
    """Quartアプリを実行するコンテキストマネージャ。テストコードなど用。"""
    # ダミーエンドポイントが存在しない場合は追加
    if not any(rule.endpoint == "_pytilpack_quart_dummy" for rule in app.url_map.iter_rules()):

        @app.route("/_pytilpack_quart_dummy")
        async def _pytilpack_quart_dummy():
            return "OK"

    # Uvicornサーバーの設定
    config = uvicorn.Config(app=app, host=host, port=port)
    server = uvicorn.Server(config)

    # 別スレッドでサーバーを起動
    def run_server():
        asyncio.run(server.serve())

    thread = threading.Thread(target=run_server, daemon=True)
    thread.start()

    try:
        # サーバーが起動するまで待機
        async with httpx.AsyncClient() as client:
            while True:
                try:
                    response = await client.get(f"http://{host}:{port}/_pytilpack_quart_dummy")
                    response.raise_for_status()
                    break
                except Exception:
                    await asyncio.sleep(0.1)  # 少し待機

        # 制御を戻す
        yield

    finally:
        # サーバーを停止
        server.should_exit = True
        thread.join(timeout=5.0)  # タイムアウトを設定

proxy_fix

リバースプロキシ対応。

ProxyFix(quartapp, x_for=1, x_proto=1, x_host=0, x_port=0, x_prefix=1, static_prefix=None, trusted_hosts=None)

リバースプロキシ対応。

nginx.conf設定例::

proxy_set_header X-Forwarded-For $remote_addr;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header X-Forwarded-Port $server_port;
proxy_set_header X-Forwarded-Prefix $http_x_forwarded_prefix;
参考
設計意図

prefixはpin方式で管理する。初回有効リクエスト時に一度だけapp.configへ反映し、 以降は変更しない。これにより攻撃者制御のX-Forwarded-Prefix値で Cookie発行パス等のアプリ全体設定が書き換わる経路を断つ。

運用時にprefixが既知であれば、static_prefix引数で起動時に確定させることを推奨する。 static_prefixを指定した場合、初期化時にapp.configを確定する。 ただしscope["root_path"]はリクエストごとにX-Forwarded-Prefixヘッダーの値から 設定するため、ヘッダーが来ないリクエストではscope["root_path"]は更新されない。

x_host > 0でX-Forwarded-Hostを有効化する場合は、trusted_hosts引数で 許可ホストリストを指定することを推奨する。指定がない場合は値検証（制御文字・ //始まり等の排除）のみ行い追加警告は出力しない。

旧挙動（リクエストごとにapp.configを上書き）を期待していた利用者は、 static_prefix指定方式、またはpin後の挙動（初回リクエストでpinされる）へ移行すること。

ソースコード位置： pytilpack/quart/proxy_fix.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def __init__(
    self,
    quartapp: quart.Quart,
    x_for: int = 1,
    x_proto: int = 1,
    x_host: int = 0,
    x_port: int = 0,
    x_prefix: int = 1,
    static_prefix: str | None = None,
    trusted_hosts: typing.Iterable[str] | None = None,
):
    self.quartapp = quartapp
    self.asgi_app = quartapp.asgi_app
    self.x_for = x_for
    self.x_proto = x_proto
    self.x_port = x_port
    self._x_host = x_host
    self._x_prefix = x_prefix
    self.trusted_hosts = list(trusted_hosts) if trusted_hosts is not None else None

    self._pin_lock = threading.Lock()
    self._pinned_prefix: str | None = None
    self._prefix_pinned = False

    if static_prefix is not None:
        validated = pytilpack.web.validate_forwarded_prefix(static_prefix)
        if validated is None:
            raise ValueError(f"static_prefixが不正: {static_prefix!r}")
        self._apply_prefix_to_config(validated)
        self._pinned_prefix = validated
        self._prefix_pinned = True
__call__(scope, receive, send) async

ASGIアプリケーションとしての処理。

ソースコード位置： pytilpack/quart/proxy_fix.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
async def __call__(
    self,
    scope: hypercorn.typing.Scope,
    receive: hypercorn.typing.ASGIReceiveCallable,
    send: hypercorn.typing.ASGISendCallable,
) -> None:
    """ASGIアプリケーションとしての処理。"""
    if scope["type"] in ("http", "websocket"):
        scope = typing.cast(hypercorn.typing.HTTPScope, copy.deepcopy(scope))
        headers = list(scope["headers"])

        # X-Forwarded-For → client
        forwarded_for = self._get_trusted_value(b"x-forwarded-for", headers, self.x_for)
        if forwarded_for and scope.get("client"):
            _, orig_port = scope.get("client") or (None, None)
            scope["client"] = (forwarded_for, orig_port or 0)

        # X-Forwarded-Proto → scheme
        forwarded_proto = self._get_trusted_value(b"x-forwarded-proto", headers, self.x_proto)
        if forwarded_proto:
            scope["scheme"] = forwarded_proto

        # X-Forwarded-Host → server & Host header
        forwarded_host = self._get_trusted_value(b"x-forwarded-host", headers, self._x_host)
        if forwarded_host:
            validated_host = pytilpack.web.validate_forwarded_host(forwarded_host)
            if validated_host is None:
                logger.warning(f"X-Forwarded-Hostに不正な値が含まれています: {forwarded_host!r}")
            elif self.trusted_hosts is not None and not pytilpack.web.is_host_in_trusted(
                validated_host, self.trusted_hosts
            ):
                logger.warning(f"X-Forwarded-Hostが許可リスト外です: {validated_host!r}")
            else:
                host_val = validated_host
                host, port = host_val, None
                if ":" in host_val and not host_val.startswith("["):
                    h, p = host_val.rsplit(":", 1)
                    if p.isdigit():
                        host, port = h, int(p)
                orig_server = scope.get("server") or (None, None)
                orig_port = orig_server[1]
                scope["server"] = (host, port or orig_port or 0)
                headers = [(hn, hv) for hn, hv in headers if hn.lower() != b"host"]
                host_hdr = host if port is None else f"{host}:{port}"
                headers.append((b"host", host_hdr.encode("utf-8", errors="replace")))

        # X-Forwarded-Port → server port & Host header
        forwarded_port = self._get_trusted_value(b"x-forwarded-port", headers, self.x_port)
        if forwarded_port and forwarded_port.isdigit():
            port_int = int(forwarded_port)
            orig_server = scope.get("server") or (None, None)
            orig_host = str(orig_server[0])
            scope["server"] = (orig_host, port_int)
            headers = [(hn, hv) for hn, hv in headers if hn.lower() != b"host"]
            headers.append((b"host", f"{orig_host}:{port_int}".encode()))

        # X-Forwarded-Prefix → root_path + config
        forwarded_prefix = self._get_trusted_value(b"x-forwarded-prefix", headers, self._x_prefix)
        if forwarded_prefix:
            prefix = pytilpack.web.validate_forwarded_prefix(forwarded_prefix)
            if prefix is None:
                logger.warning(f"X-Forwarded-Prefixに不正な値が含まれています: {forwarded_prefix!r}")
            else:
                scope["root_path"] = prefix
                if not self._prefix_pinned:
                    with self._pin_lock:
                        if not self._prefix_pinned:
                            self._apply_prefix_to_config(prefix)
                            self._pinned_prefix = prefix
                            self._prefix_pinned = True
                        elif self._pinned_prefix != prefix:
                            logger.warning(
                                f"X-Forwarded-Prefixがpin済みの値と異なります: "
                                f"pinned={self._pinned_prefix!r}, received={prefix!r}"
                            )
                elif self._pinned_prefix != prefix:
                    logger.warning(
                        f"X-Forwarded-Prefixがpin済みの値と異なります: pinned={self._pinned_prefix!r}, received={prefix!r}"
                    )

        scope["headers"] = headers

    await self.asgi_app(scope, receive, send)