コンテンツにスキップ

pytilpack.flask

必要なextra

pip install pytilpack[flask]

pytilpack.flask

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

ResponseType = flask.Response | werkzeug.test.TestResponse module-attribute

レスポンスの型。

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

Bases: ProxyFix

リバースプロキシ対応。

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を確定する。 ただしenviron["SCRIPT_NAME"]はリクエストごとにX-Forwarded-Prefixヘッダーの値から 設定するため、ヘッダーが来ないリクエストではenviron["SCRIPT_NAME"]は更新されない。

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

x_prefixとx_hostはpytilpack側で完全管理し、親クラスへはx_prefix=0・ x_host=0を渡す。これにより検証で除外した値が親クラス側で再反映される経路を断つ。

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

ソースコード位置： pytilpack/flask/proxy_fix.py 
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
def __init__(
    self,
    flaskapp: flask.Flask,
    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,
):
    # x_prefix・x_hostは自前処理するため親クラスへは無効化して渡す
    super().__init__(
        flaskapp.wsgi_app,
        x_for=x_for,
        x_proto=x_proto,
        x_host=0,
        x_port=x_port,
        x_prefix=0,
    )
    self.flaskapp = flaskapp
    self._x_prefix = x_prefix
    self._x_host = x_host
    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

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

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

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 None

発生：

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

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

戻り値：

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

レスポンスボディ
ソースコード位置： pytilpack/flask/asserts.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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_body = response.get_data()
    _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)

レスポンスを検証して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/flask/asserts.py 
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
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_bytes = response.get_data()
    response_body = response_bytes.decode("utf-8")
    _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')

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

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 'application/json'

発生：

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

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

戻り値：

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

レスポンスのJSONデコード結果
ソースコード位置： pytilpack/flask/asserts.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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_body = response.get_data().decode("utf-8")
    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__')

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

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 '__default__'

発生：

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

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

戻り値：

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

レスポンスボディ
ソースコード位置： pytilpack/flask/asserts.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
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_body = response.get_data().decode("utf-8")
    _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)

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

引数：

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

レスポンス

 必須
status_code int

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

 200

発生：

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

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

戻り値：

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

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

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

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

    Returns:
        レスポンス

    """
    _core.assert_sse_core(response.status_code, response.content_type, status_code)
    return response

assert_response(response, status_code=200)

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

引数：

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

レスポンス

 必須
status_code int

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

 200

発生：

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

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

戻り値：

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

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

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

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

    Returns:
        レスポンス

    """
    _core.assert_response_core(response.status_code, status_code)
    return response

check_status_code(status_code, valid_status_code)

Deprecated.

ソースコード位置： pytilpack/flask/asserts.py 
183
184
185
186
187
188
189
190
def check_status_code(status_code: int, valid_status_code: int) -> None:
    """Deprecated."""
    warnings.warn(
        "pytilpack.flask_.check_status_code is deprecated. Use pytilpack.web.check_status_code instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    pytilpack.web.check_status_code(status_code, valid_status_code)

check_content_type(content_type, valid_content_types)

Deprecated.

ソースコード位置： pytilpack/flask/asserts.py 
193
194
195
196
197
198
199
200
def check_content_type(content_type: str, valid_content_types: str | typing.Iterable[str] | None) -> None:
    """Deprecated."""
    warnings.warn(
        "pytilpack.flask_.check_content_type is deprecated. Use pytilpack.web.check_content_type instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    pytilpack.web.check_content_type(content_type, valid_content_types)

generate_secret_key(cache_path)

Deprecated.

ソースコード位置： pytilpack/flask/misc.py 
36
37
38
39
40
41
42
43
def generate_secret_key(cache_path: str | pathlib.Path) -> bytes:
    """Deprecated."""
    warnings.warn(
        "pytilpack.flask_.generate_secret_key is deprecated. Use pytilpack.secrets_.generate_secret_key instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return pytilpack.secrets.generate_secret_key(cache_path)

data_url(data, mime_type)

小さい画像などのバイナリデータをURLに埋め込んだ文字列を生成して返す。

引数：

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

埋め込むデータ

 必須
mime_type str

例：'image/png'

 必須
ソースコード位置： pytilpack/flask/misc.py 
46
47
48
49
50
51
52
53
54
55
def data_url(data: bytes, mime_type: str) -> str:
    """小さい画像などのバイナリデータをURLに埋め込んだ文字列を生成して返す。

    Args:
        data: 埋め込むデータ
        mime_type: 例：'image/png'

    """
    b64 = base64.b64encode(data).decode("ascii")
    return f"data:{mime_type};base64,{b64}"

get_next_url()

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

ソースコード位置： pytilpack/flask/misc.py 
58
59
60
61
62
63
64
def get_next_url() -> str:
    """ログイン後遷移用のnextパラメータ用のURLを返す。"""
    return pytilpack.web.build_next_url(
        flask.request.script_root,
        flask.request.path,
        flask.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/flask/misc.py 
67
68
69
70
71
72
73
74
75
76
77
78
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(flask.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

flask.url_forに渡す追加の引数

 {}

戻り値：

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

静的ファイルのURL
ソースコード位置： pytilpack/flask/misc.py 
 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
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: flask.url_forに渡す追加の引数

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

get_safe_url(target, host_url, default_url)

Deprecated.

ソースコード位置： pytilpack/flask/misc.py 
113
114
115
116
117
118
119
120
def get_safe_url(target: str | None, host_url: str, default_url: str) -> str:
    """Deprecated."""
    warnings.warn(
        "pytilpack.flask_.get_safe_url is deprecated. Use pytilpack.web.get_safe_url instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return pytilpack.web.get_safe_url(target, host_url, default_url)

get_routes(app)

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

戻り値：

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

ルーティング情報のリスト。
ソースコード位置： pytilpack/flask/misc.py 
127
128
129
130
131
132
133
def get_routes(app: flask.Flask) -> list[RouteInfo]:
    """ルーティング情報を取得する。

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

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

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

ソースコード位置： pytilpack/flask/misc.py 
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
@contextlib.contextmanager
def run(app: flask.Flask, host: str = "localhost", port: int = 5000):
    """Flaskアプリを実行するコンテキストマネージャ。テストコードなど用。"""
    if not any(m.endpoint == "_pytilpack_flask_dummy" for m in app.url_map.iter_rules()):

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

    server = werkzeug.serving.make_server(host, port, app, threaded=True)
    ctx = app.app_context()
    ctx.push()
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        # サーバーが起動するまで待機
        while True:
            try:
                httpx.get(f"http://{host}:{port}/_pytilpack_flask_dummy").raise_for_status()
                break
            except Exception:
                pass
        # 制御を戻す
        yield
    finally:
        server.shutdown()
        thread.join()

asserts

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

ResponseType = flask.Response | werkzeug.test.TestResponse module-attribute

レスポンスの型。

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

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

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 None

発生：

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

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

戻り値：

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

レスポンスボディ
ソースコード位置： pytilpack/flask/asserts.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
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_body = response.get_data()
    _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)

レスポンスを検証して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/flask/asserts.py 
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
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_bytes = response.get_data()
    response_body = response_bytes.decode("utf-8")
    _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')

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

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 'application/json'

発生：

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

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

戻り値：

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

レスポンスのJSONデコード結果
ソースコード位置： pytilpack/flask/asserts.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
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_body = response.get_data().decode("utf-8")
    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__')

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

引数：

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

レスポンス

 必須
status_code int

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

 200
content_type str | Iterable[str] | None

期待するContent-Type

 '__default__'

発生：

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

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

戻り値：

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

レスポンスボディ
ソースコード位置： pytilpack/flask/asserts.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
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_body = response.get_data().decode("utf-8")
    _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)

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

引数：

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

レスポンス

 必須
status_code int

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

 200

発生：

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

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

戻り値：

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

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

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

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

    Returns:
        レスポンス

    """
    _core.assert_sse_core(response.status_code, response.content_type, status_code)
    return response

assert_response(response, status_code=200)

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

引数：

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

レスポンス

 必須
status_code int

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

 200

発生：

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

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

戻り値：

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

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

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

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

    Returns:
        レスポンス

    """
    _core.assert_response_core(response.status_code, status_code)
    return response

check_status_code(status_code, valid_status_code)

Deprecated.

ソースコード位置： pytilpack/flask/asserts.py 
183
184
185
186
187
188
189
190
def check_status_code(status_code: int, valid_status_code: int) -> None:
    """Deprecated."""
    warnings.warn(
        "pytilpack.flask_.check_status_code is deprecated. Use pytilpack.web.check_status_code instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    pytilpack.web.check_status_code(status_code, valid_status_code)

check_content_type(content_type, valid_content_types)

Deprecated.

ソースコード位置： pytilpack/flask/asserts.py 
193
194
195
196
197
198
199
200
def check_content_type(content_type: str, valid_content_types: str | typing.Iterable[str] | None) -> None:
    """Deprecated."""
    warnings.warn(
        "pytilpack.flask_.check_content_type is deprecated. Use pytilpack.web.check_content_type instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    pytilpack.web.check_content_type(content_type, valid_content_types)

misc

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

generate_secret_key(cache_path)

Deprecated.

ソースコード位置： pytilpack/flask/misc.py 
36
37
38
39
40
41
42
43
def generate_secret_key(cache_path: str | pathlib.Path) -> bytes:
    """Deprecated."""
    warnings.warn(
        "pytilpack.flask_.generate_secret_key is deprecated. Use pytilpack.secrets_.generate_secret_key instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return pytilpack.secrets.generate_secret_key(cache_path)

data_url(data, mime_type)

小さい画像などのバイナリデータをURLに埋め込んだ文字列を生成して返す。

引数：

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

埋め込むデータ

 必須
mime_type str

例：'image/png'

 必須
ソースコード位置： pytilpack/flask/misc.py 
46
47
48
49
50
51
52
53
54
55
def data_url(data: bytes, mime_type: str) -> str:
    """小さい画像などのバイナリデータをURLに埋め込んだ文字列を生成して返す。

    Args:
        data: 埋め込むデータ
        mime_type: 例：'image/png'

    """
    b64 = base64.b64encode(data).decode("ascii")
    return f"data:{mime_type};base64,{b64}"

get_next_url()

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

ソースコード位置： pytilpack/flask/misc.py 
58
59
60
61
62
63
64
def get_next_url() -> str:
    """ログイン後遷移用のnextパラメータ用のURLを返す。"""
    return pytilpack.web.build_next_url(
        flask.request.script_root,
        flask.request.path,
        flask.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/flask/misc.py 
67
68
69
70
71
72
73
74
75
76
77
78
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(flask.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

flask.url_forに渡す追加の引数

 {}

戻り値：

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

静的ファイルのURL
ソースコード位置： pytilpack/flask/misc.py 
 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
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: flask.url_forに渡す追加の引数

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

get_safe_url(target, host_url, default_url)

Deprecated.

ソースコード位置： pytilpack/flask/misc.py 
113
114
115
116
117
118
119
120
def get_safe_url(target: str | None, host_url: str, default_url: str) -> str:
    """Deprecated."""
    warnings.warn(
        "pytilpack.flask_.get_safe_url is deprecated. Use pytilpack.web.get_safe_url instead.",
        DeprecationWarning,
        stacklevel=2,
    )
    return pytilpack.web.get_safe_url(target, host_url, default_url)

get_routes(app)

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

戻り値：

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

ルーティング情報のリスト。
ソースコード位置： pytilpack/flask/misc.py 
127
128
129
130
131
132
133
def get_routes(app: flask.Flask) -> list[RouteInfo]:
    """ルーティング情報を取得する。

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

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

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

ソースコード位置： pytilpack/flask/misc.py 
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
@contextlib.contextmanager
def run(app: flask.Flask, host: str = "localhost", port: int = 5000):
    """Flaskアプリを実行するコンテキストマネージャ。テストコードなど用。"""
    if not any(m.endpoint == "_pytilpack_flask_dummy" for m in app.url_map.iter_rules()):

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

    server = werkzeug.serving.make_server(host, port, app, threaded=True)
    ctx = app.app_context()
    ctx.push()
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        # サーバーが起動するまで待機
        while True:
            try:
                httpx.get(f"http://{host}:{port}/_pytilpack_flask_dummy").raise_for_status()
                break
            except Exception:
                pass
        # 制御を戻す
        yield
    finally:
        server.shutdown()
        thread.join()

proxy_fix

リバースプロキシ対応。

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

Bases: ProxyFix

リバースプロキシ対応。

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を確定する。 ただしenviron["SCRIPT_NAME"]はリクエストごとにX-Forwarded-Prefixヘッダーの値から 設定するため、ヘッダーが来ないリクエストではenviron["SCRIPT_NAME"]は更新されない。

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

x_prefixとx_hostはpytilpack側で完全管理し、親クラスへはx_prefix=0・ x_host=0を渡す。これにより検証で除外した値が親クラス側で再反映される経路を断つ。

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

ソースコード位置： pytilpack/flask/proxy_fix.py 
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
def __init__(
    self,
    flaskapp: flask.Flask,
    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,
):
    # x_prefix・x_hostは自前処理するため親クラスへは無効化して渡す
    super().__init__(
        flaskapp.wsgi_app,
        x_for=x_for,
        x_proto=x_proto,
        x_host=0,
        x_port=x_port,
        x_prefix=0,
    )
    self.flaskapp = flaskapp
    self._x_prefix = x_prefix
    self._x_host = x_host
    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