コンテンツにスキップ

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 から参照される。

RouteInfo

Bases: NamedTuple

ルーティング情報を保持するクラス。

属性：

名前 タイプ デスクリプション
endpoint str

エンドポイント名
url_parts list[str]

URLのパーツのリスト
arg_names list[str]

URLパーツの引数名のリスト

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

リバースプロキシ対応。

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;

参考
ソースコード位置： pytilpack/quart/proxy_fix.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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,
):
    self.quartapp = quartapp
    self.asgi_app = quartapp.asgi_app
    self.x_for = x_for
    self.x_proto = x_proto
    self.x_host = x_host
    self.x_port = x_port
    self.x_prefix = x_prefix

__call__(scope, receive, send) async

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

ソースコード位置： pytilpack/quart/proxy_fix.py 
 43
 44
 45
 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
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"):
            forwarded_for = forwarded_for.split(",")[-1].strip()
            _, 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:
            host_val = forwarded_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)
            # update server tuple
            orig_server = scope.get("server") or (None, None)
            orig_port = orig_server[1]
            scope["server"] = (host, port or orig_port or 0)
            # rebuild Host header
            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 and forwarded_prefix != "/":
            prefix = forwarded_prefix.rstrip("/")
            scope["root_path"] = prefix
            self.quartapp.config["APPLICATION_ROOT"] = prefix
            self.quartapp.config["SESSION_COOKIE_PATH"] = prefix
            self.quartapp.config["QUART_AUTH_COOKIE_PATH"] = prefix
            # QuartAuthはinit_app時にコピーしてしまうので強制反映が必要…
            for extension in self.quartapp.extensions.get("QUART_AUTH", []):
                if isinstance(extension, quart_auth.QuartAuth):
                    extension.cookie_path = prefix

        scope["headers"] = headers

    await self.asgi_app(scope, receive, send)

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

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 None

発生：

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

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

戻り値：

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

レスポンスボディ
ソースコード位置： pytilpack/quart/asserts.py 
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
async def assert_bytes(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = None,
) -> 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)

    async with pytilpack.pytest.AssertBlock(response_body, suffix=".txt"):  # bin では開けないため txt として扱う
        # ステータスコードチェック
        pytilpack.web.check_status_code(response.status_code, status_code)

        # Content-Typeチェック
        pytilpack.web.check_content_type(response.content_type, content_type)

    return response_body

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

テストコード用。

html5libが必要なので注意。

引数：

名前 タイプ デスクリプション デフォルト
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
89
90
91
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:
    """テストコード用。

    html5libが必要なので注意。

    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)

    if content_type == "__default__":
        content_type = ["text/html", "application/xhtml+xml"]

    async with pytilpack.pytest.AssertBlock(response_body, suffix=".html", tmp_path=tmp_path):
        # ステータスコードチェック
        pytilpack.web.check_status_code(response.status_code, status_code)

        # Content-Typeチェック
        pytilpack.web.check_content_type(response.content_type, content_type)

        # HTMLのチェック
        pytilpack.web.check_html(response_bytes, strict=strict)

    return response_body

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

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 'application/json'

発生：

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

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

戻り値：

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

レスポンスのjson
ソースコード位置： pytilpack/quart/asserts.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
async def assert_json(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "application/json",
) -> typing.Any:
    """テストコード用。

    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)
    data: typing.Any

    async with pytilpack.pytest.AssertBlock(response_body, suffix=".json"):
        # ステータスコードチェック
        pytilpack.web.check_status_code(response.status_code, status_code)

        # Content-Typeチェック
        pytilpack.web.check_content_type(response.content_type, content_type)

        # JSONのチェック
        try:
            data = json.loads(response_body)
        except Exception as e:
            raise AssertionError(f"JSONエラー: {e}") from e

    return data

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

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 '__default__'

発生：

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

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

戻り値：

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

レスポンスのxml
ソースコード位置： pytilpack/quart/asserts.py 
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
async def assert_xml(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "__default__",
) -> str:
    """テストコード用。

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

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

    Returns:
        レスポンスのxml

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=True)

    if content_type == "__default__":
        content_type = ["text/xml", "application/xml"]

    async with pytilpack.pytest.AssertBlock(response_body, suffix=".xml"):
        # ステータスコードチェック
        pytilpack.web.check_status_code(response.status_code, status_code)

        # Content-Typeチェック
        pytilpack.web.check_content_type(response.content_type, content_type)

        # XMLのチェック
        try:
            _ = ET.fromstring(response_body)
        except Exception as e:
            raise AssertionError(f"XMLエラー: {e}") from e

    return response_body

assert_sse(response, status_code=200) async

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200

発生：

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

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

戻り値：

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

レスポンス
ソースコード位置： pytilpack/quart/asserts.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
async def assert_sse(
    response: ResponseType,
    status_code: int = 200,
) -> quart.Response:
    """テストコード用。

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

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

    Returns:
        レスポンス

    """
    response = await _get_response(response)
    pytilpack.web.check_status_code(response.status_code, status_code)
    pytilpack.web.check_content_type(response.content_type, "text/event-stream")
    return response

assert_response(response, status_code=200) async

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200

発生：

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

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

戻り値：

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

レスポンス
ソースコード位置： pytilpack/quart/asserts.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
async def assert_response(
    response: ResponseType,
    status_code: int = 200,
) -> quart.Response:
    """テストコード用。

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

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

    Returns:
        レスポンス

    """
    response = await _get_response(response)
    pytilpack.web.check_status_code(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 
11
12
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
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,
        )
        pytilpack.i18n.activate(i18n_state, locale)

set_max_concurrency(app, max_concurrency, timeout=3.0)

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

引数：

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

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

 必須
max_concurrency int

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

 必須
timeout float | None

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

 3.0
Notes
  • Hypercorn の --workers / --threads とは独立した アプリレベルの制御。1 ワーカー内のコルーチン数を絞る用途で使う。
ソースコード位置： pytilpack/quart/misc.py 
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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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 を返す。

    Notes:
        * Hypercorn の ``--workers`` / ``--threads`` とは独立した
        アプリレベルの制御。1 ワーカー内のコルーチン数を絞る用途で使う。
    """
    if max_concurrency < 1:
        raise ValueError("max_concurrency must be >= 1")

    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
        try:
            # テスト時にセマフォ/timeoutを一時変更できるようstateから読む
            sem = state.semaphore
            to = state.timeout
            if to is None:
                await sem.acquire()
            else:
                await asyncio.wait_for(sem.acquire(), timeout=to)
            quart.g.quart__concurrency_token = True
        except TimeoutError:
            logger.warning(f"Concurrency limit reached, aborting request: {quart.request.path}")
            quart.abort(
                503,
                description="サーバーが混みあっています。しばらく待ってから再度お試しください。",
            )

    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 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
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 
126
127
128
129
130
131
def get_next_url() -> str:
    """ログイン後遷移用のnextパラメータ用のURLを返す。"""
    path = quart.request.script_root + quart.request.path
    query_string = quart.request.query_string.decode("utf-8")
    next_ = f"{path}?{query_string}" if query_string else path
    return next_

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 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
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

    """
    accept_header = quart.request.headers.get("Accept", "")
    # text/htmlを先頭にすることで、同一quality時はHTMLを優先する（従来と同じ挙動）
    result = pytilpack.http.select_accept(accept_header, ["text/html", "text/markdown", "text/plain"])
    return result in ("text/markdown", "text/plain")

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 
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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
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
    """
    if not cache_busting:
        return quart.url_for("static", filename=filename, **kwargs)

    # スタティックファイルのパスを取得
    static_folder = quart.current_app.static_folder
    assert static_folder is not None, "static_folder is None"

    filepath = pathlib.Path(static_folder) / filename
    try:
        # ファイルの最終更新日時のキャッシュを利用するか否か
        if cache_timestamp is True or (cache_timestamp == "when_not_debug" and not quart.current_app.debug):
            # キャッシュを使う
            timestamp = _TIMESTAMP_CACHE.get(str(filepath))
            if timestamp is None:
                timestamp = int(filepath.stat().st_mtime)
                _TIMESTAMP_CACHE[str(filepath)] = timestamp
        else:
            # キャッシュを使わない
            timestamp = int(filepath.stat().st_mtime)

        # キャッシュバスティングありのURLを返す
        return quart.url_for("static", filename=filename, v=timestamp, **kwargs)
    except OSError:
        # ファイルが存在しない場合などは通常のURLを返す
        return quart.url_for("static", filename=filename, **kwargs)

get_routes(app)

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

戻り値：

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

ルーティング情報のリスト。
ソースコード位置： pytilpack/quart/misc.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def get_routes(app: quart.Quart) -> list[RouteInfo]:
    """ルーティング情報を取得する。

    Returns:
        ルーティング情報のリスト。
    """
    arg_regex = re.compile(r"<([^>]+)>")  # <name> <type:name> にマッチするための正規表現
    split_regex = re.compile(r"<[^>]+>")  # re.splitのためグループ無しにした版
    output: list[RouteInfo] = []
    for r in app.url_map.iter_rules():
        endpoint = str(r.endpoint)
        rule = (
            r.rule
            if app.config["APPLICATION_ROOT"] == "/" or not app.config["APPLICATION_ROOT"]
            else f"{app.config['APPLICATION_ROOT']}{r.rule}"
        )
        url_parts = [str(part) for part in split_regex.split(rule)]
        arg_names = [str(x.split(":")[-1]) for x in arg_regex.findall(rule)]
        output.append(RouteInfo(endpoint, url_parts, arg_names))
    return sorted(output, key=lambda x: len(x[2]), reverse=True)

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

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

ソースコード位置： pytilpack/quart/misc.py 
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
265
266
267
268
269
270
271
272
@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

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 None

発生：

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

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

戻り値：

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

レスポンスボディ
ソースコード位置： pytilpack/quart/asserts.py 
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
async def assert_bytes(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = None,
) -> 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)

    async with pytilpack.pytest.AssertBlock(response_body, suffix=".txt"):  # bin では開けないため txt として扱う
        # ステータスコードチェック
        pytilpack.web.check_status_code(response.status_code, status_code)

        # Content-Typeチェック
        pytilpack.web.check_content_type(response.content_type, content_type)

    return response_body

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

テストコード用。

html5libが必要なので注意。

引数：

名前 タイプ デスクリプション デフォルト
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
89
90
91
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:
    """テストコード用。

    html5libが必要なので注意。

    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)

    if content_type == "__default__":
        content_type = ["text/html", "application/xhtml+xml"]

    async with pytilpack.pytest.AssertBlock(response_body, suffix=".html", tmp_path=tmp_path):
        # ステータスコードチェック
        pytilpack.web.check_status_code(response.status_code, status_code)

        # Content-Typeチェック
        pytilpack.web.check_content_type(response.content_type, content_type)

        # HTMLのチェック
        pytilpack.web.check_html(response_bytes, strict=strict)

    return response_body

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

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 'application/json'

発生：

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

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

戻り値：

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

レスポンスのjson
ソースコード位置： pytilpack/quart/asserts.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
async def assert_json(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "application/json",
) -> typing.Any:
    """テストコード用。

    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)
    data: typing.Any

    async with pytilpack.pytest.AssertBlock(response_body, suffix=".json"):
        # ステータスコードチェック
        pytilpack.web.check_status_code(response.status_code, status_code)

        # Content-Typeチェック
        pytilpack.web.check_content_type(response.content_type, content_type)

        # JSONのチェック
        try:
            data = json.loads(response_body)
        except Exception as e:
            raise AssertionError(f"JSONエラー: {e}") from e

    return data

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

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 '__default__'

発生：

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

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

戻り値：

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

レスポンスのxml
ソースコード位置： pytilpack/quart/asserts.py 
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
async def assert_xml(
    response: ResponseType,
    status_code: int = 200,
    content_type: str | typing.Iterable[str] | None = "__default__",
) -> str:
    """テストコード用。

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

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

    Returns:
        レスポンスのxml

    """
    response = await _get_response(response)
    response_body = await response.get_data(as_text=True)

    if content_type == "__default__":
        content_type = ["text/xml", "application/xml"]

    async with pytilpack.pytest.AssertBlock(response_body, suffix=".xml"):
        # ステータスコードチェック
        pytilpack.web.check_status_code(response.status_code, status_code)

        # Content-Typeチェック
        pytilpack.web.check_content_type(response.content_type, content_type)

        # XMLのチェック
        try:
            _ = ET.fromstring(response_body)
        except Exception as e:
            raise AssertionError(f"XMLエラー: {e}") from e

    return response_body

assert_sse(response, status_code=200) async

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200

発生：

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

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

戻り値：

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

レスポンス
ソースコード位置： pytilpack/quart/asserts.py 
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
async def assert_sse(
    response: ResponseType,
    status_code: int = 200,
) -> quart.Response:
    """テストコード用。

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

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

    Returns:
        レスポンス

    """
    response = await _get_response(response)
    pytilpack.web.check_status_code(response.status_code, status_code)
    pytilpack.web.check_content_type(response.content_type, "text/event-stream")
    return response

assert_response(response, status_code=200) async

テストコード用。

引数：

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

レスポンス

 必須
status_code int

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

 200

発生：

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

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

戻り値：

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

レスポンス
ソースコード位置： pytilpack/quart/asserts.py 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
async def assert_response(
    response: ResponseType,
    status_code: int = 200,
) -> quart.Response:
    """テストコード用。

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

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

    Returns:
        レスポンス

    """
    response = await _get_response(response)
    pytilpack.web.check_status_code(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 
11
12
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
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,
        )
        pytilpack.i18n.activate(i18n_state, locale)

misc

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

ConcurrencyState(semaphore, max_concurrency, timeout) dataclass

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

RouteInfo

Bases: NamedTuple

ルーティング情報を保持するクラス。

属性：

名前 タイプ デスクリプション
endpoint str

エンドポイント名
url_parts list[str]

URLのパーツのリスト
arg_names list[str]

URLパーツの引数名のリスト

set_max_concurrency(app, max_concurrency, timeout=3.0)

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

引数：

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

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

 必須
max_concurrency int

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

 必須
timeout float | None

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

 3.0
Notes
  • Hypercorn の --workers / --threads とは独立した アプリレベルの制御。1 ワーカー内のコルーチン数を絞る用途で使う。
ソースコード位置： pytilpack/quart/misc.py 
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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
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 を返す。

    Notes:
        * Hypercorn の ``--workers`` / ``--threads`` とは独立した
        アプリレベルの制御。1 ワーカー内のコルーチン数を絞る用途で使う。
    """
    if max_concurrency < 1:
        raise ValueError("max_concurrency must be >= 1")

    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
        try:
            # テスト時にセマフォ/timeoutを一時変更できるようstateから読む
            sem = state.semaphore
            to = state.timeout
            if to is None:
                await sem.acquire()
            else:
                await asyncio.wait_for(sem.acquire(), timeout=to)
            quart.g.quart__concurrency_token = True
        except TimeoutError:
            logger.warning(f"Concurrency limit reached, aborting request: {quart.request.path}")
            quart.abort(
                503,
                description="サーバーが混みあっています。しばらく待ってから再度お試しください。",
            )

    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 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
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 
126
127
128
129
130
131
def get_next_url() -> str:
    """ログイン後遷移用のnextパラメータ用のURLを返す。"""
    path = quart.request.script_root + quart.request.path
    query_string = quart.request.query_string.decode("utf-8")
    next_ = f"{path}?{query_string}" if query_string else path
    return next_

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 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
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

    """
    accept_header = quart.request.headers.get("Accept", "")
    # text/htmlを先頭にすることで、同一quality時はHTMLを優先する（従来と同じ挙動）
    result = pytilpack.http.select_accept(accept_header, ["text/html", "text/markdown", "text/plain"])
    return result in ("text/markdown", "text/plain")

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 
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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
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
    """
    if not cache_busting:
        return quart.url_for("static", filename=filename, **kwargs)

    # スタティックファイルのパスを取得
    static_folder = quart.current_app.static_folder
    assert static_folder is not None, "static_folder is None"

    filepath = pathlib.Path(static_folder) / filename
    try:
        # ファイルの最終更新日時のキャッシュを利用するか否か
        if cache_timestamp is True or (cache_timestamp == "when_not_debug" and not quart.current_app.debug):
            # キャッシュを使う
            timestamp = _TIMESTAMP_CACHE.get(str(filepath))
            if timestamp is None:
                timestamp = int(filepath.stat().st_mtime)
                _TIMESTAMP_CACHE[str(filepath)] = timestamp
        else:
            # キャッシュを使わない
            timestamp = int(filepath.stat().st_mtime)

        # キャッシュバスティングありのURLを返す
        return quart.url_for("static", filename=filename, v=timestamp, **kwargs)
    except OSError:
        # ファイルが存在しない場合などは通常のURLを返す
        return quart.url_for("static", filename=filename, **kwargs)

get_routes(app)

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

戻り値：

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

ルーティング情報のリスト。
ソースコード位置： pytilpack/quart/misc.py 
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def get_routes(app: quart.Quart) -> list[RouteInfo]:
    """ルーティング情報を取得する。

    Returns:
        ルーティング情報のリスト。
    """
    arg_regex = re.compile(r"<([^>]+)>")  # <name> <type:name> にマッチするための正規表現
    split_regex = re.compile(r"<[^>]+>")  # re.splitのためグループ無しにした版
    output: list[RouteInfo] = []
    for r in app.url_map.iter_rules():
        endpoint = str(r.endpoint)
        rule = (
            r.rule
            if app.config["APPLICATION_ROOT"] == "/" or not app.config["APPLICATION_ROOT"]
            else f"{app.config['APPLICATION_ROOT']}{r.rule}"
        )
        url_parts = [str(part) for part in split_regex.split(rule)]
        arg_names = [str(x.split(":")[-1]) for x in arg_regex.findall(rule)]
        output.append(RouteInfo(endpoint, url_parts, arg_names))
    return sorted(output, key=lambda x: len(x[2]), reverse=True)

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

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

ソースコード位置： pytilpack/quart/misc.py 
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
265
266
267
268
269
270
271
272
@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)

リバースプロキシ対応。

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;

参考
ソースコード位置： pytilpack/quart/proxy_fix.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
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,
):
    self.quartapp = quartapp
    self.asgi_app = quartapp.asgi_app
    self.x_for = x_for
    self.x_proto = x_proto
    self.x_host = x_host
    self.x_port = x_port
    self.x_prefix = x_prefix
__call__(scope, receive, send) async

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

ソースコード位置： pytilpack/quart/proxy_fix.py 
 43
 44
 45
 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
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"):
            forwarded_for = forwarded_for.split(",")[-1].strip()
            _, 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:
            host_val = forwarded_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)
            # update server tuple
            orig_server = scope.get("server") or (None, None)
            orig_port = orig_server[1]
            scope["server"] = (host, port or orig_port or 0)
            # rebuild Host header
            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 and forwarded_prefix != "/":
            prefix = forwarded_prefix.rstrip("/")
            scope["root_path"] = prefix
            self.quartapp.config["APPLICATION_ROOT"] = prefix
            self.quartapp.config["SESSION_COOKIE_PATH"] = prefix
            self.quartapp.config["QUART_AUTH_COOKIE_PATH"] = prefix
            # QuartAuthはinit_app時にコピーしてしまうので強制反映が必要…
            for extension in self.quartapp.extensions.get("QUART_AUTH", []):
                if isinstance(extension, quart_auth.QuartAuth):
                    extension.cookie_path = prefix

        scope["headers"] = headers

    await self.asgi_app(scope, receive, send)