【Python】Flask・RQ・Redis・PillowでPNG画像処理ジョブWebアプリを作ってみた
公開：
PNG画像をアップロードすると、裏側でリサイズとグレースケール化が走り、終わったら処理済み画像をダウンロードできます。そんな小さなジョブ処理WebアプリをFlask・RQ・Redis・Pillowで作ってみました。画面はアップロードフォームとジョブ一覧だけのシンプル構成で、重い画像処理はキュー越しにワーカーへ渡します。
通常運用はRQ_ASYNC=1でワーカーによる非同期実行、隔離した動作検証のときだけRQ_ASYNC=0で同じジョブ関数を同期実行できるようにしました。実装のポイントとつまずきやすい箇所を、コードの断片ごとに追いかけていきます。
RQの基本概念、要件定義、実装、動作確認までを順番に学べる構成です。動画は目次から確認したい場面へ移動でき、本文だけでも手順と考え方が完結します。
動画の内容をテキストで確認する
オープニング。FlaskとRQ・Redis・Pillowを使ってPNG画像処理ジョブWebアプリを作るカリキュラムを始めます。概要紹介。
FlaskとRQ・Redis・Pillowの役割と使い方を学ぶPNG画像処理ジョブWebアプリの要件を整理する 完成コードと実行結果を確認する 最後に完成した画面を実際に操作して確かめる 具体的にやること。
アップロードフォームとジョブ状態一覧をトップ画面に表示する 受け取ったPNGをジョブID付きの名前でuploadsへ保存する 保存したパスを渡してRQキューへ画像処理ジョブを投入するPillowで幅160pxへリサイズしグレースケール化して保存する ジョブ詳細画面でRedisが保持する状態を表示する 実装環境・必須アプリ。
OS：Windows 11 Pro Python：3.13.3シェル：PowerShell 5.1必須アプリ：コードエディター、Webブラウザ、ターミナル パッケージ：pip、flask、rq、redis、pillow、fakeredis FlaskとRQ・Redis・Pillowとは。
RQ：RQは、時間のかかる処理を後回しにしてキューへ積み、別プロセスのワーカーへ実行させるPython向けジョブキューライブラリFlask：Flaskは、URLと関数を結び付けて画面やJSONを返す、小さく始められるPython製Webフレームワーク Flask・RQで作る画像処理ジョブアプリの要点。
保存先はuploadsディレクトリ 出力先はresultsディレクトリ 投入直後はジョブ詳細へリダイレクト PNG画像処理ジョブWebアプリの要件定義。
トップ画面にアップロードフォームとジョブ状態の表が出るPNGを投入するとジョブ詳細画面へリダイレクトされる 同期実行では状態がfinishedになる 詳細画面に縮小前後のサイズとmode=Lが並ぶ 一覧の完了ジョブにダウンロードリンクが出る ダウンロードでジョブID付きの処理済みPNGを受け取れる INTRO: Monaco EditorでPNG画像処理ジョブWebアプリを実装。
コードを1行ずつ入力し、補完と自動インデントを使いながら実行結果を確認します。LINE 001: モジュール全体の説明。このファイルがFlask・RQ・Redis・Pillowを組み合わせた画像処理ジョブWebアプリであることを示すドキュメント文字列です。
ファイルの先頭でアプリ全体の役割を説明しています。LINE 004: 型ヒント記法の将来機能を有効化。Python 3.10未満でも新しい型ヒントの書き方が使えるようにするための宣言です。
ファイルの一番上に書くことで以降のコード全体に適用されます。LINE 006: OS操作用モジュールの読み込み。環境変数の取得などOSに関する機能を使うためにosモジュールを読み込んでいます。
後段でREDIS_URLなどの設定値取得に利用します。LINE 007: 一時ディレクトリ機能の読み込み。OS標準の一時フォルダのパスを取得するためにtempfileモジュールを読み込んでいます。
作業用フォルダの作成場所を決める際に使います。LINE 008: 一意なID生成機能の読み込み。ジョブごとに重複しない識別子を作るためにuuidモジュールを読み込んでいます。
後でジョブIDの生成に利用します。LINE 009: パス操作クラスの読み込み。ファイルパスを扱いやすくするためにpathlibからPathクラスを読み込んでいます。
ディレクトリ作成やファイル名操作で使います。LINE 011: Redis接続ライブラリの読み込み。実際のRedisサーバーに接続するためにredisライブラリを読み込んでいます。
非同期実行時の接続処理で利用します。LINE 012: Flask関連機能のまとめて読み込み開始。Webアプリを作るために必要なFlaskの機能をまとめて読み込む記述の始まりです。
以下の行で個別の機能名を指定しています。LINE 013: Flaskアプリ本体クラス。Webアプリケーションそのものを表すFlaskクラスを読み込んでいます。
この後でアプリのインスタンスを作る際に使います。LINE 014: 404エラー表示機能。存在しないジョブなどにアクセスされた際にエラーを返すためのabort関数を読み込んでいます。
処理中断に利用します。LINE 015: JSON応答生成機能。APIの応答をJSON形式で返すためのjsonify関数を読み込んでいます。
ジョブ状態取得APIで使います。LINE 016: リダイレクト機能。別のURLへ画面遷移させるためのredirect関数を読み込んでいます。
アップロード後の画面遷移などで利用します。LINE 017: HTMLテンプレート描画機能。文字列で書いたHTMLテンプレートをその場でレンダリングするためのrender_template_string関数を読み込んでいます。
画面表示処理で使います。LINE 018: リクエスト情報取得機能。送信されたフォームデータやファイルにアクセスするためのrequestオブジェクトを読み込んでいます。
アップロード処理で利用します。LINE 019: ファイル送信機能。処理済み画像ファイルをレスポンスとして返すためのsend_file関数を読み込んでいます。
ダウンロード処理で使います。LINE 020: URL生成機能。ルート名から実際のURLを組み立てるためのurl_for関数を読み込んでいます。
テンプレート内のリンク生成で利用します。LINE 021: Flaskインポート文の終了。ここまでの一連のFlaskインポート記述をまとめる括弧を閉じています。
複数機能を1つのimport文で読み込むための区切りです。LINE 022: Pillow画像処理機能の読み込み。画像を開いて処理するためにImageクラスと、白黒変換などを行うImageOpsをPillowから読み込んでいます。
process_image関数内で使用します。LINE 023: RQキュークラスの読み込み。バックグラウンドジョブを積むためのQueueクラスをrqライブラリから読み込んでいます。
ジョブ投入処理で使用します。LINE 024: 存在しないジョブ用の例外。指定したジョブIDが見つからない場合に発生する例外NoSuchJobErrorを読み込んでいます。
ジョブ取得処理でエラー処理に利用します。LINE 025: ジョブ管理クラスの読み込み。投入したジョブの状態や結果を扱うためのJobクラスをrqライブラリから読み込んでいます。
ジョブ状態確認などで使用します。LINE 026: 安全なファイル名変換機能。アップロードされたファイル名から危険な文字を取り除くsecure_filename関数をwerkzeugから読み込んでいます。
ファイル保存時の安全対策に使います。LINE 031: 作業用ベースディレクトリの決定。OS標準の一時ディレクトリの下にアプリ専用の作業フォルダのパスを組み立てています。
以降のアップロード先や保存先の基準パスとして使われます。LINE 032: アップロード保存先パスの定義。ベースディレクトリの下にアップロード画像を保存するためのフォルダパスを定義しています。
画像アップロード時にここへファイルが保存されます。LINE 033: 処理結果保存先パスの定義。ベースディレクトリの下に処理済み画像を保存するためのフォルダパスを定義しています。
process_image関数の出力先として使われます。LINE 034: アップロード用フォルダの作成。アップロード保存先フォルダが存在しない場合に自動で作成しています。
既に存在してもエラーにならないよう設定されています。LINE 035: 結果保存用フォルダの作成。処理結果保存先フォルダが存在しない場合に自動で作成しています。
既に存在してもエラーにならないよう設定されています。LINE 037: Redis接続先URLの取得。環境変数REDIS_URLから接続先を取得し、未設定時はローカルの標準アドレスを使うようにしています。
Redis接続処理で参照される設定値です。LINE 038: 非同期実行モードの判定。環境変数RQ_ASYNCが1のときのみ非同期実行を有効にする真偽値を作成しています。
この値によって実Redisかインメモリかの接続先が切り替わります。LINE 039: キュー名の定義。RQに積むジョブのキュー名を定数として定義しています。
Queueオブジェクトの作成時に使用されます。LINE 040: ジョブ一覧管理キーの定義。Redis上でジョブIDの一覧を保持するためのキー名を定義しています。
ジョブ投入時と一覧取得時の両方で使われます。LINE 041: リサイズ後の幅の定義。画像をリサイズする際の目標幅をピクセル単位の定数として定義しています。
process_image関数のデフォルト値として使われます。LINE 042: ジョブ結果保持時間の定義。完了したジョブの結果をRedis上に保持する秒数を定数として定義しています。
ジョブ投入時のresult_ttlに使われます。RUN 1/8: 設定値と作業ディレクトリを確認する。定数の読み込みが終わった時点で、接続先や実行モード、作業用フォルダが用意できているかを表示して確かめます。
CHECK 1/8: 途中実行に成功。Redis URL: redis://127.0.0.1:6379/0 非同期モード: False キュー名: image-jobs / 目標幅: 160 作業フォルダの有無: True True RETURN 01: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 046: 画像処理ジョブ関数の定義。元画像パスと保存先パス、目標幅を受け取り処理結果の辞書を返す関数を定義しています。RQワーカーが実際に呼び出すジョブ本体です。
LINE 047: 関数の説明文。この関数がPNG画像のリサイズとグレースケール化を行って保存することを説明するドキュメント文字列です。処理内容を簡潔に示しています。
LINE 048: 画像ファイルを開く。指定された元画像パスからPillowで画像を開いています。withブロックを使うことで処理後に自動的にファイルを閉じるようにしています。
LINE 049: RGBモードへの変換。開いた画像をRGBモードに変換しています。透過情報などによる後続処理の不具合を防ぐための下準備です。
LINE 050: 元画像サイズの取得。変換後の画像から幅と高さを取得しています。取得した値は結果表示とリサイズ後の高さ計算に使われます。
LINE 051: リサイズ後の高さ計算。目標幅に合わせて縦横比を保つように新しい高さを計算しています。最低でも1ピクセルになるよう調整しています。
LINE 053: 画像のリサイズ実行。計算した幅と高さを使って画像を高品質なLANCZOS法でリサイズしています。この結果が次のグレースケール変換の元になります。
LINE 054: グレースケール変換。リサイズ後の画像をImageOpsを使って白黒のグレースケール画像に変換しています。変換結果は保存とサイズ情報の取得に使われます。
LINE 055: 処理結果画像の保存。グレースケール変換した画像をPNG形式で指定した保存先パスに書き込んでいます。この保存によりダウンロードできるファイルが生成されます。
LINE 056: 処理結果情報の返却開始。処理結果の詳細情報をまとめた辞書を作成し呼び出し元に返す処理の始まりです。RQのジョブ結果としてこの辞書が保持されます。
LINE 057: 元画像サイズの記録。元画像の幅と高さを文字列形式で結果辞書に格納しています。画面表示時に変換前後のサイズを比較できるようにするためです。
LINE 058: 処理後画像サイズの記録。グレースケール画像の幅と高さを文字列形式で結果辞書に格納しています。画面表示時にリサイズ後のサイズを確認できるようにするためです。
LINE 059: 画像モードの記録。グレースケール画像のカラーモードを結果辞書に格納しています。白黒変換が正しく行われたことを確認できる情報として使われます。
LINE 060: 結果ファイル名の記録。保存先パスからファイル名だけを取り出して結果辞書に格納しています。ダウンロードリンク表示などで参照される情報です。
LINE 061: 結果辞書定義の終了。process_image関数が返す結果辞書の定義をここで閉じています。この辞書がジョブの戻り値としてRedisに保持されます。
RUN 2/8: 画像処理関数を単体で動かす。ジョブ本体となる関数が書けた時点で、サンプル画像を作って直接呼び出し、戻り値と出力ファイルを確認します。CHECK 2/8: 途中実行に成功。
処理結果: {'original_size': '600x300', 'result_size': '160x80', 'mode': 'L', 'result_name': 'checkpoint_sample_processed.png'} 出力ファイルの有無: True RETURN 02: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 065: Redis接続作成関数の定義。実行モードに応じて適切なRedis接続を作成して返す関数を定義しています。
この関数の戻り値がキューの接続先として使われます。LINE 066: 関数の説明文。この関数が非同期モードなら実Redis、そうでなければ検証用のインメモリRedisへ接続することを説明するドキュメント文字列です。
LINE 067: 非同期モードかどうかの判定。RQ_ASYNCの値を確認し、非同期実行が有効かどうかで接続方法を分岐させています。この判定によって次の行の実Redis接続の可否が決まります。
LINE 068: 実Redisへの接続。環境変数から取得したREDIS_URLを使って本物のRedisサーバーに接続しています。非同期モードのときにのみ実行される処理です。
LINE 069: インメモリRedisライブラリの読み込み。同期モードで動作確認するために本物のRedisを使わずに済むfakeredisライブラリをここで読み込んでいます。必要なときだけ読み込む形にしています。
LINE 071: インメモリRedisへの接続。fakeredisを使って実際のRedisサーバーなしで動作するインメモリの接続を作成して返しています。同期モード時にジョブ管理用として使われます。
LINE 074: Redis接続の作成と保持。create_connection関数を呼び出して実際のRedis接続を作成し、connectionという名前で保持しています。この接続はキュー作成やジョブ取得で共通して使われます。
LINE 075: RQキューオブジェクトの作成。作成したRedis接続を使ってジョブを積むためのQueueオブジェクトを作成しています。以降のジョブ投入処理でこのqueueが使われます。
RUN 3/8: Redis接続とキューの状態を見る。接続とキューを作った直後に、どの接続方式で動いているかと待機中のジョブ数を表示して確認します。CHECK 3/8: 途中実行に成功。
接続方式: FakeStrictRedis キュー名: image-jobs / 非同期: False 待機ジョブ数: 0 RETURN 03: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 079: ジョブ状態取得関数の定義。
Jobオブジェクトを受け取り、状態を文字列で返す関数を定義しています。画面表示やAPI応答で状態を表示する際に使われます。LINE 080: 関数の説明文。
この関数がRQの状態をqueuedやfinishedなどの文字列に変換することを説明するドキュメント文字列です。LINE 081: 最新状態の取得。Redisから最新のジョブ状態を取得するようrefreshを指定して呼び出しています。
処理中に状態が変わっても正しい値を取得できるようにしています。LINE 082: 状態値の文字列化。取得した状態オブジェクトから文字列表現を取り出して返しています。
状態の型によらず一貫した文字列として扱えるようにしています。LINE 085: ジョブ取得関数の定義。ジョブIDを受け取り、対応するJobオブジェクトを取得する関数を定義しています。
詳細画面やダウンロード処理などで共通して使われます。LINE 086: 例外処理の開始。ジョブ取得時に存在しないジョブIDが指定された場合に備えて例外処理を開始しています。
LINE 087: Redisからのジョブ取得。指定されたジョブIDと接続情報を使ってRedis上のジョブ情報を取得しています。取得に成功すればそのままJobオブジェクトを返します。
LINE 088: 存在しないジョブの例外捕捉。指定したジョブIDがRedis上に見つからない場合の例外を捕捉しています。この後Noneを返すことで呼び出し元がエラー処理を行えるようにしています。
LINE 089: ジョブ未検出時の戻り値。NoSuchJobErrorが発生した場合にNoneを返し、指定したジョブIDが存在しないことを呼び出し元へ伝えます。LINE 092: ジョブ一覧取得関数の定義。
limit件数を上限としてジョブの状態一覧を取得するlist_jobs関数を定義しています。デフォルトでは10件まで取得します。LINE 093: 関数の説明文。
Redisに積んだジョブIDを使ってジョブ状態確認を行い、各ジョブの状態を読み出すことをdocstringで説明しています。LINE 094: 結果格納用リストの初期化。取得したジョブ情報を格納するための空のリストjobsを用意しています。
LINE 095: RedisからジョブIDを取得。connection.lrangeでRedisのリストからジョブIDを新しい順にlimit件だけ取り出し、順番に処理していきます。LINE 096: ジョブIDのデコードと取得。
バイト列で取得したジョブIDを文字列にデコードし、fetch_jobを使って対応するジョブオブジェクトを取得します。LINE 097: ジョブ存在チェック。ジョブが取得できなかった場合にNoneかどうかを確認し、次の処理へ進むかどうかを判定しています。
LINE 098: 存在しないジョブのスキップ。ジョブが見つからなかった場合はcontinueで処理をスキップし、一覧に含めないようにしています。LINE 099: ジョブ状態の取得。
status_text関数を使って現在のジョブの状態をqueuedやfinishedなどの文字列として取得しています。LINE 100: ジョブ情報の追加開始。取得したジョブの情報を辞書としてjobsリストへ追加する処理を開始しています。
LINE 101: 辞書オブジェクトの開始。1件分のジョブ情報をまとめる辞書リテラルの開始を表しています。LINE 102: ジョブIDの格納。
辞書のidキーにジョブオブジェクトのidを格納し、画面表示やリンク生成に利用できるようにしています。LINE 103: 元ファイル名の格納。job.metaからsource_nameを取り出し、見つからない場合は「-」を代わりに格納しています。
LINE 104: 状態文字列の格納。先に取得したstate変数をstatusキーへ格納し、一覧画面で状態を表示できるようにしています。LINE 105: 完了判定フラグの格納。
stateが「finished」と一致するかどうかを判定し、finishedキーに真偽値として格納しています。LINE 106: 辞書リテラルの終了。1件分のジョブ情報をまとめた辞書の定義を閉じています。
LINE 107: append呼び出しの終了。jobs.appendの呼び出し括弧を閉じ、辞書をリストへ追加する処理を完了させています。LINE 108: ジョブ一覧の返却。
作成したjobsリストを呼び出し元へ返し、画面表示に使えるようにしています。RUN 4/8: ジョブ一覧の読み出しを試す。一覧用の関数まで書けた時点で、Redisに積まれたジョブIDの数と一覧の取得件数を表示します。
CHECK 4/8: 途中実行に成功。インデックスキー: image-jobs:index 積まれたジョブID数: 0 一覧取得件数: 0 RETURN 04: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 112: 一覧画面テンプレートの開始。INDEX_HTML変数にHTML文書の先頭であるdoctype宣言を含めて定義を開始しています。LINE 113: html要素の開始。
言語属性jaを指定したhtml要素を開始し、日本語ページであることを示しています。LINE 114: head要素の開始。ページのメタ情報やスタイルをまとめるhead要素を開始しています。
LINE 115: 文字コードの指定。metaタグで文字コードをutf-8に指定し、日本語が正しく表示されるようにしています。LINE 116: ページタイトルの設定。
ブラウザのタブに表示されるタイトルを「画像処理ジョブ | Flask + RQ」として設定しています。LINE 117: スタイル定義の開始。ページの見た目を整えるstyleタグを開始しています。
LINE 118: 全体のフォントと余白設定。bodyのフォントを見やすいsans-serifにし、周囲に余白と文字色を指定しています。LINE 119: テーブル表示の基本設定。
テーブルの罫線を1本にまとめ、上に少し余白を付けています。LINE 120: セルの見た目設定。見出しとデータのセルに枠線と余白、文字サイズを指定して読みやすくしています。
LINE 121: 見出しセルの背景色。テーブルの見出しセルに薄い背景色を付け、データ行と区別しやすくしています。LINE 123: ボタンの余白設定。
ボタン要素に上下左右の余白を付けて押しやすい見た目に整えています。LINE 124: スタイル定義の終了。styleタグを閉じ、見た目の設定を終えています。
LINE 125: head要素の終了。ページのメタ情報部分であるhead要素を閉じています。LINE 126: body要素の開始。
画面に表示される本文部分であるbody要素を開始しています。LINE 127: 見出しの表示。h1見出しとして「画像処理ジョブWebアプリ」というページタイトルを表示しています。
LINE 128: 実行モードの表示。async_modeの値によって非同期または同期のどちらで動作しているかをテキストで切り替えて表示しています。LINE 129: エラーメッセージの表示。
errorが渡された場合のみエラーメッセージ用の段落を表示するようにしています。LINE 130: アップロード見出しの表示。h2見出しとして「画像アップロード」という文言を表示しています。
LINE 131: アップロードフォームの開始。POSTメソッドでenqueue_jobエンドポイントへ送信するファイルアップロード用フォームを開始しています。LINE 132: ファイル選択欄の設置。
PNG画像を選択できるファイル入力欄をフォーム内に配置しています。LINE 133: 送信ボタンの設置。「ジョブを投入」というラベルの送信ボタンをフォーム内に配置しています。
LINE 134: フォームの終了。アップロードフォームの定義を閉じています。LINE 135: ジョブ状態見出しの表示。
h2見出しとして「ジョブ状態」という文言を表示しています。LINE 136: ジョブ一覧テーブルの開始。idにjob-listを持つテーブル要素を開始し、ジョブ一覧を表示する枠を用意しています。
LINE 137: テーブル見出し行の定義。ジョブID・元ファイル・状態・処理済み画像という4つの列見出しを定義しています。LINE 138: ジョブ一覧の繰り返し開始。
jobsリストの各要素を1件ずつ取り出して行を作るループの開始を表しています。LINE 139: テーブル行の開始。1件分のジョブ情報を表示する行trの開始を表しています。
LINE 140: ジョブ詳細へのリンク表示。ジョブIDをリンクとして表示し、クリックするとjob_detailページへ遷移できるようにしています。LINE 141: 元ファイル名の表示。
アップロード時のファイル名をセルとして表示しています。LINE 142: ジョブ状態の表示。現在のジョブの状態文字列をセルとして表示しています。
LINE 143: ダウンロードリンクの表示切り替え。finishedがtrueならダウンロードリンクを表示し、そうでなければ「処理中」と表示するよう切り替えています。LINE 144: テーブル行の終了。
1件分のジョブ情報を表示する行trを閉じています。LINE 145: ループのelse節開始。jobsリストが空だった場合に実行されるelse節の開始を表しています。
LINE 146: ジョブなし時の表示。ジョブが1件も無い場合に「まだジョブはありません」というメッセージを4列分結合して表示しています。LINE 147: 繰り返し処理の終了。
ジョブ一覧を表示するforループを閉じています。LINE 148: テーブルの終了。ジョブ一覧を表示していたテーブル要素を閉じています。
LINE 149: body要素の終了。ページ本文部分であるbody要素を閉じています。LINE 150: html要素の終了。
ページ全体を囲んでいたhtml要素を閉じています。LINE 151: テンプレート文字列の終了。INDEX_HTML変数として定義していたHTML文字列の定義を終えています。
RUN 5/8: 一覧画面のテンプレートを確かめる。トップ画面用のテンプレート文字列を書き終えた時点で、必要な文言が含まれているかを確認します。CHECK 5/8: 途中実行に成功。
テンプレートの文字数: 1400 送信ボタンの文言あり: True 空表示の文言あり: True RETURN 05: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。LINE 153: 詳細画面テンプレートの開始。
DETAIL_HTML変数にジョブ詳細ページのHTML文書を定義し始めています。LINE 154: html要素の開始。言語属性jaを指定したhtml要素を開始し、日本語ページであることを示しています。
LINE 155: head要素の開始。ページのメタ情報やスタイルをまとめるhead要素を開始しています。LINE 156: 文字コードの指定。
metaタグで文字コードをutf-8に指定し、日本語が正しく表示されるようにしています。LINE 157: ページタイトルの設定。ブラウザのタブに表示されるタイトルを「ジョブ詳細 | Flask + RQ」として設定しています。
LINE 158: スタイル定義の開始。ページの見た目を整えるstyleタグを開始しています。LINE 159: 全体のフォントと余白設定。
bodyのフォントを見やすいsans-serifにし、周囲に余白と文字色を指定しています。LINE 161: 段落の行間指定。p要素の行間を1.8倍に広げるスタイルです。
ジョブ詳細画面の文章を読みやすくするための見た目の調整になります。LINE 162: styleタグの終了。ここまで書いてきたCSSの定義を閉じるタグです。
これより後は見た目の指定ではなくHTMLの本体部分になります。LINE 163: headタグの終了。文書のメタ情報やスタイルをまとめていたhead部分を閉じています。
次からはbody部分でページの中身を表示します。LINE 164: 本文領域の開始。ここから実際に画面に表示される本文が始まります。
この中にジョブの詳細情報を並べていきます。LINE 165: 見出しの表示。ジョブ詳細ページの大見出しとして「ジョブ詳細」という文字列を表示します。
ユーザーが今どの画面を見ているか分かりやすくしています。LINE 166: ジョブIDの表示。job_idというテンプレート変数を使って、対象ジョブの識別子を画面に表示します。
どのジョブの詳細を見ているか確認できるようにしています。LINE 167: 元ファイル名の表示。アップロード時に保存したsource_nameを表示し、どの画像ファイルを処理しているかが分かるようにしています。
LINE 168: ジョブ状態の表示。Redisから取得したstatus変数の値を表示し、ジョブが待機中・実行中・完了などどの状態にあるかを画面上で確認できるようにしています。LINE 169: 処理結果有無の分岐開始。
result変数が存在するかどうかで表示内容を切り替えるためのJinjaの条件分岐を開始しています。処理が終わっているかどうかで表示を変えます。LINE 170: リサイズ結果の表示。
処理結果resultに含まれる元画像サイズと変換後サイズ、グレースケールのモードを組み合わせて1行の文章として表示しています。LINE 171: ダウンロードリンクの表示。download_resultへのURLを生成し、処理済み画像をダウンロードできるリンクを表示しています。
ユーザーはここをクリックして結果画像を取得できます。LINE 172: 分岐のelse部分。resultが存在しない場合、つまりジョブがまだ完了していない場合の表示内容に切り替える部分です。
LINE 173: 未完了時のメッセージ表示。処理がまだ完了していないことを伝えるメッセージを表示します。結果がないときに空白にならないよう配慮しています。
LINE 174: 条件分岐の終了。result変数の有無による表示切り替えのif文をここで閉じています。これ以降は共通の表示内容に戻ります。
LINE 175: リンク段落の開始。再確認リンクと戻るリンクをまとめて表示するための段落タグを開始しています。LINE 176: 状態再確認リンク。
同じジョブ詳細ページへのリンクを表示し、クリックすると最新のジョブ状態を再取得して表示できるようにしています。LINE 177: アップロード画面へ戻るリンク。index関数へのURLを生成し、アップロード画面へ戻るリンクを表示しています。
ユーザーが別の画像を投入したいときに使う導線です。LINE 178: リンク段落の終了。再確認リンクと戻るリンクをまとめていた段落タグを閉じています。
LINE 179: bodyタグの終了。ページ本文の内容をここまでで終え、body要素を閉じています。LINE 180: htmlタグの終了。
文書全体を表していたhtml要素を閉じ、ページ全体の構造を終了しています。LINE 181: テンプレート文字列の終了。DETAIL_HTMLという複数行文字列の終わりを示す三重引用符です。
ここまでがジョブ詳細画面のテンプレート定義になります。LINE 185: Flaskアプリの生成。Flaskクラスをもとにアプリケーションのインスタンスを作成しています。
これ以降のルーティング定義はこのappオブジェクトに登録されます。LINE 186: アップロード上限の設定。アップロードできるファイルサイズの上限を8メガバイトに設定しています。
これより大きなファイルが送られた場合はFlaskが自動的にエラーにします。RUN 6/8: Flaskアプリの設定を確認する。アプリを作って上限サイズを設定した直後に、アプリ名とアップロード上限、登録済みルート数を表示します。
CHECK 6/8: 途中実行に成功。アプリ名: lesson アップロード上限バイト: 8388608 登録済みルート数: 1 RETURN 06: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 189: トップページのルート定義。HTTPのGETメソッドでルートパスにアクセスされたときにindex関数を呼び出すよう登録しています。LINE 190: index関数の定義開始。
トップページの処理内容を記述する関数を定義しています。アップロードフォームとジョブ一覧を表示する役割を持ちます。LINE 191: index関数の説明文。
この関数がアップロードフォームとジョブ状態一覧を表示するものであることを説明するドキュメント文字列です。LINE 192: テンプレート描画の開始。INDEX_HTMLのテンプレート文字列をもとに、実際に画面へ返すHTMLを生成する処理を開始しています。
LINE 193: 使用するテンプレートの指定。render_template_stringに渡す最初の引数として、トップページ用のHTMLテンプレートを指定しています。LINE 194: ジョブ一覧データの渡し込み。
list_jobs関数を呼び出して取得したジョブ一覧をテンプレートのjobs変数として渡し、画面に表として表示できるようにしています。LINE 195: 実行モードの渡し込み。RQ_ASYNCの値をasync_mode変数としてテンプレートに渡し、非同期実行か同期実行かを画面上に表示できるようにしています。
LINE 196: エラーメッセージの渡し込み。リクエストのクエリパラメータからerrorという値を取り出し、テンプレートに渡してエラー文言を表示できるようにしています。LINE 197: render_template_string呼び出しの終了。
テンプレート描画に必要な引数の指定をここで終え、生成されたHTMLをレスポンスとして返します。RUN 7/8: トップ画面の表示を確認する。最初のルートが書けた時点でテストクライアントからトップ画面を開き、見出しと空一覧の表示を確かめます。
CHECK 7/8: 途中実行に成功。トップ画面のステータス: 500 見出しあり: False 空一覧の表示あり: False RETURN 07: エディターへ戻りました。入力済みのコードを維持したまま、次の実装へ進みます。
LINE 200: ジョブ投入ルートの定義。HTTPのPOSTメソッドで/jobsにアクセスされたときにenqueue_job関数を呼び出すよう登録しています。LINE 201: enqueue_job関数の定義開始。
画像アップロードを受け付けてジョブを投入する処理を記述する関数を定義しています。LINE 202: enqueue_job関数の説明文。この関数が画像アップロードの受付・保存とRQキューへのジョブ投入を行うものであることを説明するドキュメント文字列です。
LINE 203: アップロードファイルの取得。リクエストに含まれるファイルの中からimageという名前で送られたファイルを取り出し、変数uploadedに格納しています。LINE 204: ファイル未選択の判定。
uploadedがNoneであるか、ファイル名が空である場合、つまり画像が選択されていない場合を判定する条件文です。LINE 205: 未選択時のリダイレクト。画像が選択されていなかった場合に、エラーメッセージ付きでトップページへリダイレクトさせています。
LINE 207: 安全なファイル名への変換。アップロードされたファイル名をsecure_filenameで安全な形式に変換し、万一取得できない場合はupload.pngという既定名を使うようにしています。LINE 208: 拡張子の判定。
ファイル名の末尾が.pngであるかどうかを小文字化して判定し、PNG以外の画像がアップロードされていないか確認しています。LINE 209: PNG以外の場合のリダイレクト。PNG形式でないファイルがアップロードされた場合に、エラーメッセージ付きでトップページへ戻すリダイレクト処理です。
LINE 211: ジョブIDの生成。uuid4を使ってランダムな識別子を生成し、先頭12文字を切り出してこのジョブ専用のIDとして使用します。LINE 212: アップロード保存先パスの作成。
ジョブIDとファイル名を組み合わせて、アップロードされた画像を保存する先のパスを組み立てています。LINE 213: 処理結果保存先パスの作成。ジョブIDを使って、画像処理後の結果ファイルを保存する先のパスをあらかじめ用意しています。
LINE 214: アップロードファイルの保存。uploaded.saveを呼び出し、アップロードされた画像ファイルを先ほど組み立てたsource_pathへ実際に書き込んで保存しています。LINE 217: ジョブ投入処理の開始。
queue.enqueueを呼び出し、process_image関数をRQのジョブとしてキューへ登録する処理を開始しています。LINE 218: 実行対象関数の指定。ジョブとして実行してほしい関数としてprocess_imageを指定しています。
この関数がワーカー側で後から呼び出されます。LINE 219: 元画像パスの引数渡し。process_imageに渡す引数として、保存済みのアップロード画像パスを文字列に変換して渡しています。
LINE 220: 結果画像パスの引数渡し。process_imageに渡す引数として、処理結果を保存する先のパスを文字列に変換して渡しています。LINE 221: 目標幅の引数渡し。
リサイズ後の画像幅としてTARGET_WIDTHの値をprocess_imageに渡し、統一されたサイズで処理させるようにしています。LINE 222: ジョブIDの指定。先ほど生成したjob_idをキー引数として渡し、このジョブをRedis上で識別できるIDとして登録しています。
LINE 223: 結果保持期間の指定。RESULT_TTLの値を渡し、ジョブの実行結果をRedisに保持しておく秒数を指定しています。LINE 224: メタ情報の付与。
元ファイル名と結果ファイルパスをmeta引数として渡し、後からジョブ詳細画面やダウンロード処理でこれらの情報を参照できるようにしています。LINE 225: enqueue呼び出しの終了。queue.enqueueに渡す引数の指定をここで終え、実際にジョブがキューへ登録され、戻り値としてjob変数に格納されます。
LINE 226: ジョブ一覧への登録。connection.lpushを使って生成されたジョブIDをRedisのリストの先頭に追加し、一覧表示で参照できるようにしています。LINE 227: ジョブ詳細画面へのリダイレクト。
投入したジョブのIDを使ってjob_detailのURLを生成し、ジョブ詳細画面へリダイレクトさせています。LINE 230: ジョブ詳細ルートの定義。HTTPのGETメソッドで/jobs/<job_id>にアクセスされたときにjob_detail関数を呼び出すよう登録しています。
LINE 231: job_detail関数の定義開始。URLパスから受け取ったjob_idを引数として、ジョブの詳細情報を表示する処理を記述する関数を定義しています。LINE 232: job_detail関数の説明文。
この関数がRedisから1件分のジョブ状態と処理結果を読み出して画面表示するものであることを説明するドキュメント文字列です。LINE 233: ジョブオブジェクトの取得。fetch_job関数を呼び出し、指定されたjob_idに対応するジョブオブジェクトをRedisから取得しています。
LINE 234: ジョブ未検出の判定。取得したjobがNoneであるかどうか、つまり該当するジョブが存在しなかったかどうかを判定する条件文です。LINE 235: 404エラーの発生。
ジョブが見つからなかった場合に、abort関数を呼び出して404エラー画面を返すようにしています。LINE 236: ジョブ状態文字列の取得。status_text関数を呼び出し、ジョブの現在の状態をqueuedやfinishedなどの文字列として取得しています。
LINE 237: 処理結果の取得判定。statusがfinishedである場合のみjob.return_valueを呼び出して処理結果を取得し、それ以外の場合はNoneとしています。LINE 238: 詳細画面テンプレートの描画開始。
render_template_stringを呼び出して、ジョブ詳細画面のHTMLを生成する処理を開始します。続く行で渡す変数がテンプレート内で利用されます。LINE 239: 詳細画面用テンプレートの指定。
表示に使うテンプレートとして、先に定義しておいたDETAIL_HTML文字列を指定しています。これによりジョブ詳細用の画面レイアウトが使われます。LINE 240: ジョブIDをテンプレートへ渡す。
job.idをjob_idという名前でテンプレートに渡し、画面上にジョブIDを表示できるようにしています。LINE 241: ジョブ状態をテンプレートへ渡す。先ほど取得したstatus変数をそのままテンプレートに渡し、画面にジョブの現在の状態を表示します。
LINE 242: 処理結果をテンプレートへ渡す。resultをテンプレートへ渡すことで、処理が完了していればリサイズ結果などの情報を画面に表示できるようにしています。LINE 243: 元ファイル名をテンプレートへ渡す。
job.metaに保存しておいた元ファイル名を取り出し、source_nameとしてテンプレートへ渡して画面に表示します。LINE 244: render_template_string呼び出しの終了。render_template_stringへの引数指定を閉じ、これまで渡した変数を使って詳細画面のHTMLを完成させ返却します。
LINE 247: ダウンロード用ルートの定義。「/jobs/<job_id>/download」というURLに対してGETリクエストが来たときに、この後の関数が呼ばれるように設定しています。LINE 248: ダウンロード処理関数の定義。
URLに含まれるjob_idを引数として受け取り、処理済み画像をダウンロードさせるための関数download_resultを定義しています。LINE 249: 関数の役割を説明するdocstring。この関数が処理済み画像をPNGファイルとして添付形式で送信する役割を持つことを説明する文字列です。
LINE 250: ジョブ情報の取得。job_idを使ってRedisからジョブ情報を取得し、変数jobに格納します。ここでジョブが見つかるかどうかを後続の処理で確認します。
LINE 251: ジョブ未検出の判定。取得したjobがNoneかどうか、つまり指定されたジョブIDが存在するかどうかを確認する条件分岐です。LINE 252: 404エラーの返却。
ジョブが見つからなかった場合に、404エラーを返してリクエストを終了させています。LINE 253: 保存先パスの取得。ジョブのメタ情報からresult_pathを取り出し、Pathオブジェクトへ変換して結果画像のファイルパスを組み立てています。
LINE 254: 結果ファイルの存在確認。result_pathで指定したファイルが実際に存在するかどうかを確認する条件分岐です。処理未完了などでファイルがない場合を判定します。
LINE 255: ファイル未存在時の404返却。結果ファイルが存在しない場合に404エラーを返し、不正なダウンロード要求を防いでいます。LINE 256: ファイル送信処理の開始。
send_file関数を呼び出し、処理済み画像ファイルをクライアントへ送信する処理を開始しています。LINE 257: 送信対象ファイルの指定。send_fileに対して、実際に送信するファイルのパスとしてresult_pathを渡しています。
LINE 258: MIMEタイプの指定。送信するファイルの種類をimage/pngとして明示し、ブラウザが画像として正しく扱えるようにしています。LINE 259: 添付ファイルとしての送信指定。
as_attachment=Trueを指定することで、ブラウザ上で画像を表示するのではなくダウンロードとして扱わせています。LINE 260: ダウンロード時のファイル名指定。download_nameにresult_path.nameを指定し、ユーザーがダウンロードする際のファイル名を分かりやすく設定しています。
LINE 261: send_file呼び出しの終了。send_file関数への引数指定を閉じ、これまでの設定に基づいて処理済み画像ファイルの送信処理を完了させます。RUN 8/8: 投入から詳細表示とダウンロードまで通す。
ダウンロード用のルートまで書けた時点で、画像の投入、詳細画面の表示、処理済みPNGの受け取りを一続きで確認します。CHECK 8/8: 途中実行に成功。詳細画面に完了表示あり: True ダウンロードリンクあり: True ダウンロード応答: 200 image/png RETURN 08: エディターへ戻りました。
入力済みのコードを維持したまま、次の実装へ進みます。LINE 264: API用ルートの定義。「/api/jobs/<job_id>」というURLに対するGETリクエストを受け付けるルーティングを設定しています。
LINE 265: ジョブ状態API関数の定義。job_idを受け取り、ジョブの状態をJSON形式で返すための関数job_status_apiを定義しています。LINE 266: API関数の役割を説明するdocstring。
この関数がポーリング用にジョブ状態をJSONで返すためのものであることを説明する文字列です。LINE 267: ジョブ情報の取得。job_idを使ってRedisからジョブ情報を取得し、jobという変数に格納しています。
LINE 268: ジョブ未検出の判定。取得したjobがNoneかどうかを確認し、指定したジョブIDが存在しない場合の処理を分岐させています。LINE 269: エラーレスポンスの返却。
ジョブが見つからない場合に、エラーメッセージを含むJSONと404ステータスコードを合わせて返しています。LINE 270: ジョブ状態文字列の取得。status_text関数を使ってジョブの現在の状態を文字列として取得し、statusに格納しています。
LINE 271: JSONレスポンス生成の開始。jsonify関数を呼び出し、ジョブの情報をJSON形式のレスポンスとして返す処理を開始しています。LINE 272: レスポンス辞書の開始。
jsonifyへ渡す辞書の定義を開始し、この後の各キーがJSONレスポンスの項目になります。LINE 273: ジョブIDの項目設定。レスポンスにjob_idというキーでジョブのIDを含め、どのジョブに関する情報かを明示しています。
LINE 274: 状態情報の項目設定。レスポンスにstatusというキーで先ほど取得したジョブの状態文字列を含めています。LINE 275: 処理結果情報の項目設定。
ジョブの状態がfinishedであれば処理結果を、そうでなければNoneをresultというキーに設定してレスポンスへ含めています。LINE 276: レスポンス辞書の終了。jsonifyへ渡す辞書の定義を閉じ、これまで設定した項目を持つJSONオブジェクトの内容を確定させます。
LINE 277: jsonify呼び出しの終了。jsonify関数への引数指定を閉じ、組み立てた辞書からJSON形式のレスポンスを生成して返します。LINE 281: 動作確認対象ルートの一覧定義。
動作確認したい画面のURLとして「/」を含むリストROUTESを定義し、確認対象のパスを管理しています。PREVIEW: 完成したPNG画像処理ジョブWebアプリを操作。入力・送信・結果表示を実際のブラウザ画面で確認します。
UPLOAD: 実際の入力データでアプリを操作。PNG画像をアップロードしてRQジョブを投入し、ジョブ状態がfinishedになったうえで処理済み画像のダウンロードリンクが出ることを確認します 確認1/5: PNG画像を投入してジョブを作る。サンプルのPNG画像を作ってアップロードフォームへ送り、ジョブが作られてジョブ詳細画面へ移動することを確認します。
確認2/5: ジョブ詳細で処理結果を確認する。投入したジョブの詳細画面を開き、Redisが保持する状態と、リサイズ後のサイズおよびグレースケール化の結果が表示されることを確認します。確認3/5: 一覧で処理済み画像のリンクを確認する。
トップ画面のジョブ一覧に元ファイル名、完了状態、ダウンロードリンクが並ぶことを実画面で確認します。確認4/5: PNG以外のファイルを送って拒否を確認する。拡張子がpngでないファイルを送信し、トップ画面へ戻ってエラーメッセージと実行モードが表示されることを確認します。
確認5/5: 存在しないジョブIDのAPI応答を確認する。存在しないジョブIDで状態確認APIを呼び、404とエラー内容がJSONで返ることを実ブラウザから確認します。学習内容のまとめ。
キュー名はimage-jobsフォーム描画はrender_template_string保存先はuploadsディレクトリ 作業用フォルダをOS標準の一時ディレクトリ配下に作成 小さく実行確認しながら完成状態まで段階的に組み立てる エンディング。Python研修はCodeCampでご確認ください。
FlaskとRQ・Redis・Pillowとは
今回使用する主要なライブラリについて、役割と使い分けを順番に確認します。
非同期処理を担うRQとRedisの役割
RQは、時間のかかる処理を後回しにしてキューへ積み、別プロセスのワーカーへ実行させるPython向けジョブキューライブラリです。積んだジョブの内容や状態を保存する場所がRedisで、メモリ上でキーと値を高速に読み書きするデータストアです。今回のアプリでは画像処理をRQへ預けるため、アップロード直後の画面はすぐに応答を返せます。
ジョブが今どの段階にあるかはRedisに残るので、あとから同じジョブIDを指定して状態を読み直せる点がこの構成の利点です。RQ_ASYNC=1なら実際のRedisとワーカーで動き、0なら隔離した検証向けにfakeredisへ切り替わります。
本アプリでRQとRedisが受け持つ処理と、コードで指定している設定値の一覧です
- キュー名はimage-jobs
- ジョブ本体はprocess_image関数
- ジョブIDはuuid4の先頭12文字を指定
- 結果はresult_ttl=3600秒で保持
- 投入したジョブIDはimage-jobs:indexへlpush
- 状態文字列はqueued/started/finished/failed
画面を作るFlaskと画像を加工するPillowの担当範囲
Flaskは、URLと関数を結び付けて画面やJSONを返す、小さく始められるPython製Webフレームワークです。アップロードされた画像そのものを加工する担当はPillowで、リサイズや色変換を扱う画像処理ライブラリです。Flask側はフォームの受け取り、ファイル保存、リダイレクトといった入口と出口だけを引き受けます。
Pillowは開いた画像を幅160pxへ縮め、グレースケールへ変換してPNGとして書き出す役目に専念する形です。役割を分けておくと、画面の見た目を変えるときも画像処理のコードへ手を入れずに済みます。
FlaskとPillowが実際に受け持っている処理と、使っている関数の一覧です
- フォーム描画はrender_template_string
- リンク生成はurl_for
- 存在しないジョブはabortで404
- 画像の縮小はImage.resize
- 色変換はImageOps.grayscale
- 保存形式はPNG固定
Python・RQで開発する場合の環境構築
この記事のセットアップ手順と掲載コードは、Windows 11 Pro、PowerShell 5.1、Python 3.13.3で動作確認しています。仮想環境を有効化せず、その中のPythonを直接指定するため、以下のコマンドはPowerShellとコマンドプロンプト（cmd）の両方で使えます。
python -m venv .venv
.\.venv\Scripts\python.exe -m pip install --upgrade pip
.\.venv\Scripts\python.exe -m pip install flask rq redis pillow fakeredismacOS・Linuxでは仮想環境内のPythonパスが異なります。今回の動作確認環境とは異なるため、以下は環境差分を補う参考手順です。
python3 -m venv .venv
./.venv/bin/python -m pip install --upgrade pip
./.venv/bin/python -m pip install flask rq redis pillow fakeredis- RQ_ASYNC=1で非同期実行を試す場合は、別途Redisサーバ(DockerやWSL上のredis-server)を起動し、REDIS_URLで接続先を指定します。
- RQ_ASYNC=0のときはfakeredisのインメモリ接続を使うため、Redisサーバがなくても同じジョブ関数を同期実行して動作確認できます。
- 起動は .\.venv\Scripts\python.exe -m flask --app image_job_app runを使い、ワーカーは同じフォルダで .\.venv\Scripts\rq.exe worker image-jobsを実行します。
PNG画像処理ジョブWebアプリの要件定義
目的は、PNG画像のアップロードからRQキューへのジョブ投入、Pillowによるリサイズとグレースケール化、処理済み画像のダウンロードまでをFlaskの画面で確認できるようにすることです。
対象者として、PythonとFlaskの基本を触ったことがあり、RQとRedisを使った非同期ジョブ処理の組み立て方を学びたい人を想定しています。
完成物は、Flask・RQ・Redis・Pillowで作り、アップロードしたPNGを幅160pxのグレースケール画像へ変換し、ジョブ状態一覧とダウンロードを画面から確認できる画像処理ジョブWebアプリです。
実装へ入る前に、機能・品質・受け入れ条件を分けて確認します。
機能要件
- アップロードフォームとジョブ状態一覧をトップ画面に表示する
- 受け取ったPNGをジョブID付きの名前でuploadsへ保存する
- 保存したパスを渡してRQキューへ画像処理ジョブを投入する
- Pillowで幅160pxへリサイズしグレースケール化して保存する
- ジョブ詳細画面でRedisが保持する状態を表示する
- 完了時に元サイズと結果サイズとmodeを表示する
- 処理済みPNGを添付形式でダウンロードさせる
- ジョブ状態をJSONで返すポーリング用APIを提供する
- 画面上部に非同期か同期かの実行モードを表示する
- 最新10件のジョブを一覧に並べる
非機能要件
- 作業用フォルダをOS標準の一時ディレクトリ配下に作成
- アップロード上限を8MBに設定
- secure_filenameでファイル名を無害化
- 拡張子が.png以外のアップロードを拒否
- ファイル未選択時はエラー付きでトップへ戻す
- RQ_ASYNC=1で実Redis接続と非同期実行
- RQ_ASYNC=0でfakeredisによる同期実行
- ジョブ結果の保持期間を3600秒に設定
- 存在しないジョブIDは404で応答
- ジョブIDはuuid4の先頭12文字
実装方針
今回はFlaskとRQ・Redis・Pillowの基本動作を追いやすくするため、PNG画像処理ジョブWebアプリ本体を1つのPythonファイルへまとめます。
入力、判定、結果表示の役割を分け、実行結果を確認しながら機能を積み上げます。
PNG画像処理ジョブWebアプリを安全に組み立てるための実装方針は次のとおりです。
- 作業用フォルダをOS標準の一時ディレクトリ配下に作成
- アップロード上限を8MBに設定
- secure_filenameでファイル名を無害化
- 拡張子が.png以外のアップロードを拒否
- ファイル未選択時はエラー付きでトップへ戻す
- RQ_ASYNC=1で実Redis接続と非同期実行
- RQ_ASYNC=0でfakeredisによる同期実行
- ジョブ結果の保持期間を3600秒に設定
- 存在しないジョブIDは404で応答
- ジョブIDはuuid4の先頭12文字
完成と判断する条件
- トップ画面にアップロードフォームとジョブ状態の表が出る
- PNGを投入するとジョブ詳細画面へリダイレクトされる
- 同期実行では状態がfinishedになる
- 詳細画面に縮小前後のサイズとmode=Lが並ぶ
- 一覧の完了ジョブにダウンロードリンクが出る
- ダウンロードでジョブID付きの処理済みPNGを受け取れる
- PNG以外を送るとエラーメッセージ付きでトップへ戻る
- 状態確認APIがstatusと結果を含むJSONを返す
- 未登録のジョブIDでは404が返る
PNG画像処理ジョブWebアプリを作る際の重要ポイント
このアプリの中心は、1つのジョブIDでファイルと状態をつなぎ続ける流れです。フォームから届いたPNGは、ジョブID付きの名前でuploadsへ保存します。保存したパスとジョブIDをそのままRQキューへ渡す作りです。
投入が終わるとジョブ詳細画面へ移動し、Redisが保持する状態をその場で読み出します。処理が完了していれば戻り値から縮小前後のサイズとmodeを取り出せるので、画面上で結果を見比べられます。ダウンロードはmetaへ控えた結果パスを使う設計です。
ジョブ投入から画像を受け取るまでに通る画面と処理の順番の一覧です
- 保存先はuploadsディレクトリ
- 出力先はresultsディレクトリ
- 投入直後はジョブ詳細へリダイレクト
- 詳細では元サイズと結果サイズを並べて表示
- 完了後にダウンロードリンクが有効化
- ポーリング用にJSON APIも用意
ジョブIDでファイルと状態を結び付ける設計
ジョブIDを1つ決めて、保存名にも結果名にも同じIDを使うと、後から対応関係を追いやすくなります。RQへもjob_idを明示して渡すため、URLに現れるIDでそのままジョブを引ける形です。
結果ファイルの場所はmetaへ控えておき、ダウンロード時に読み直します。
1つのジョブIDが使われる場所の一覧です
- uuid4の先頭12文字をジョブIDに採用
- 元画像はジョブIDと元ファイル名で保存
- 結果画像は<job_id>_processed.png
- 詳細画面のURLは/jobs/<job_id>
- metaのresult_pathをダウンロードで再利用
一覧画面と詳細画面での状態の見せ方
一覧では、直近に投入したジョブから順に状態を並べます。完了したものだけダウンロード欄をリンクへ変えると、処理中のジョブを押してしまう混乱を避けられる形です。
詳細画面には再確認用のリンクを置いたので、同じURLを開き直して状態の変化を追えます。
2つの画面が表示する項目と、表示を切り替える条件の一覧です
- 一覧は最新10件まで表示
- 列はジョブIDと元ファイルと状態と処理済み画像
- finished以外は処理中と表示
- ジョブが無いときは空表示用の行を1行だけ出す
- 詳細では状態を再確認リンクで再読み込み
処理済み画像を配信するときの取り扱い
ダウンロードは、ブラウザで開くのではなくファイルとして受け取れると扱いやすくなります。そこでPNGのMIMEタイプを指定し、添付として送る形にしました。
結果ファイルが見つからない場合は404を返すため、壊れたリンクをそのまま開かせずに済みます。
送信処理で指定している設定と、例外時の扱いの一覧です
- send_fileでファイルを送信
- mimetypeへimage/pngを指定
- as_attachment=Trueで添付ダウンロード
- download_nameへ結果ファイル名
- 結果ファイルが無ければ404
PythonでPNG画像処理ジョブWebアプリの完成コード
コードは1ファイル構成で、上から設定、ジョブ本体のprocess_image、Redis接続とキュー生成、ジョブ状態を扱うヘルパーの順に並べました。
Flaskのルートはこの後ろに続き、アップロードを受けてenqueueするPOST /jobsや、Job.fetchの結果を描画する詳細画面がぶら下がります。
全行を追うより、構造が分かる要所だけ抜き出したほうが理解が早いはずです。ここでは7か所に絞って見ていきます。
このセクションの用語
- Image.LANCZOS
- 縮小時の画質を保ちやすい補間方式の指定です。写真の縮小によく使われます。
- ImageOps.grayscale
-
Pillowのヘルパーで、画像を白黒(モードL)へ変換します。 - get_status
-
RQのジョブが今どの段階かを返すメソッドです。refresh=TrueでRedisの最新値を読み直します。 - NoSuchJobError
- 指定したジョブIDが
Redisに見つからないときにRQが送出する例外です。 - tempfile.gettempdir()
- OSが用意した一時ファイル置き場のパスを返す標準ライブラリの関数です。
"""FlaskとRQ・Redis・Pillowで作る画像処理ジョブWebアプリ。"""
# ライブラリの読み込み
from __future__ import annotations
import os
import tempfile
import uuid
from pathlib import Path
import redis
from flask import (
Flask,
abort,
jsonify,
redirect,
render_template_string,
request,
send_file,
url_for,
)
from PIL import Image, ImageOps
from rq import Queue
from rq.exceptions import NoSuchJobError
from rq.job import Job
from werkzeug.utils import secure_filename
# アプリ設定(RQ_ASYNC=1でRQワーカーによる非同期実行、0で同期実行)
# 実行環境によってはスクリプト設置先が読み取り専用になるため、
# 書き込み可能なOS標準の一時ディレクトリ配下に作業用フォルダを作る
BASE_DIR = Path(tempfile.gettempdir()) / "flask_rq_image_jobs"
UPLOAD_DIR = BASE_DIR / "uploads"
RESULT_DIR = BASE_DIR / "results"
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
RESULT_DIR.mkdir(parents=True, exist_ok=True)
REDIS_URL = os.environ.get("REDIS_URL", "redis://127.0.0.1:6379/0")
RQ_ASYNC = os.environ.get("RQ_ASYNC", "0") == "1"
QUEUE_NAME = "image-jobs"
JOB_INDEX_KEY = "image-jobs:index"
TARGET_WIDTH = 160
RESULT_TTL = 3600
# Pillowによる画像処理(RQワーカーが実行するジョブ本体)
def process_image(source_path: str, result_path: str, target_width: int = TARGET_WIDTH) -> dict:
"""PNG画像リサイズ(縦横比を保った幅160pxへの変換)とグレースケール化を行って保存する。"""
with Image.open(source_path) as image:
source = image.convert("RGB")
width, height = source.size
new_height = max(1, round(height * target_width / width))
# 画像リサイズ: 幅をtarget_widthに合わせ、縦横比を保って高さを算出する
resized = source.resize((target_width, new_height), Image.LANCZOS)
grayscale = ImageOps.grayscale(resized)
grayscale.save(result_path, format="PNG")
return {
"original_size": f"{width}x{height}",
"result_size": f"{grayscale.width}x{grayscale.height}",
"mode": grayscale.mode,
"result_name": Path(result_path).name,
}
# Redis接続とRQキューの準備
def create_connection():
"""RQ_ASYNC=1なら実Redisへ、0なら隔離検証用のインメモリRedisへ接続する。"""
if RQ_ASYNC:
return redis.Redis.from_url(REDIS_URL)
import fakeredis
return fakeredis.FakeStrictRedis()
connection = create_connection()
queue = Queue(QUEUE_NAME, connection=connection, is_async=RQ_ASYNC)
# ジョブ状態確認のためRedisに保持された状態を取り出す処理
def status_text(job: Job) -> str:
"""RQのジョブ状態を queued / started / finished / failed の文字列にする。"""
status = job.get_status(refresh=True)
return getattr(status, "value", str(status))
def fetch_job(job_id: str):
try:
return Job.fetch(job_id, connection=connection)
except NoSuchJobError:
return None
def list_jobs(limit: int = 10) -> list[dict]:
"""Redisのリストに積んだジョブIDを使ってジョブ状態確認を行い、各ジョブの状態を読み出す。"""
jobs = []
for raw_id in connection.lrange(JOB_INDEX_KEY, 0, limit - 1):
job = fetch_job(raw_id.decode("utf-8"))
if job is None:
continue
state = status_text(job)
jobs.append(
{
"id": job.id,
"source_name": job.meta.get("source_name", "-"),
"status": state,
"finished": state == "finished",
}
)
return jobs
# 画面テンプレート
INDEX_HTML = """<!doctype html>
<html lang="ja">
<head>
<meta charset="utf-8">
<title>画像処理ジョブ | Flask + RQ</title>
<style>
body { font-family: sans-serif; margin: 2rem; color: #222; }
table { border-collapse: collapse; margin-top: 0.5rem; }
th, td { border: 1px solid #ccc; padding: 0.4rem 0.9rem; font-size: 0.95rem; }
th { background: #f2f4f7; }
#error-message { color: #b00020; font-weight: bold; }
button { padding: 0.4rem 1.2rem; }
</style>
</head>
<body>
<h1>画像処理ジョブWebアプリ</h1>
<p id="run-mode">実行モード: {% if async_mode %}非同期(RQワーカーが処理){% else %}同期(RQ_ASYNC=0){% endif %}</p>
{% if error %}<p id="error-message">{{ error }}</p>{% endif %}
<h2>画像アップロード</h2>
<form id="upload-form" method="post" action="{{ url_for('enqueue_job') }}" enctype="multipart/form-data">
<input id="image-file" type="file" name="image" accept="image/png">
<button id="submit-job" type="submit">ジョブを投入</button>
</form>
<h2>ジョブ状態</h2>
<table id="job-list">
<tr><th>ジョブID</th><th>元ファイル</th><th>状態</th><th>処理済み画像</th></tr>
{% for job in jobs %}
<tr>
<td><a href="{{ url_for('job_detail', job_id=job.id) }}">{{ job.id }}</a></td>
<td>{{ job.source_name }}</td>
<td>{{ job.status }}</td>
<td>{% if job.finished %}<a href="{{ url_for('download_result', job_id=job.id) }}">ダウンロード</a>{% else %}処理中{% endif %}</td>
</tr>
{% else %}
<tr><td colspan="4">まだジョブはありません</td></tr>
{% endfor %}
</table>
</body>
</html>
"""
DETAIL_HTML = """<!doctype html>
<html lang="ja">
<head>
<meta charset="utf-8">
<title>ジョブ詳細 | Flask + RQ</title>
<style>
body { font-family: sans-serif; margin: 2rem; color: #222; }
#job-status { font-weight: bold; }
p { line-height: 1.8; }
</style>
</head>
<body>
<h1>ジョブ詳細</h1>
<p>ジョブID: <span id="job-id">{{ job_id }}</span></p>
<p>元ファイル: <span id="source-name">{{ source_name }}</span></p>
<p>Redisが保持する状態: <span id="job-status">{{ status }}</span></p>
{% if result %}
<p id="job-result">リサイズ: {{ result.original_size }} → {{ result.result_size }} / グレースケール: mode={{ result.mode }}</p>
<p><a id="download-link" href="{{ url_for('download_result', job_id=job_id) }}">処理済み画像をダウンロード</a></p>
{% else %}
<p id="job-result">処理結果はまだありません</p>
{% endif %}
<p>
<a id="refresh-status" href="{{ url_for('job_detail', job_id=job_id) }}">状態を再確認</a> /
<a id="back-link" href="{{ url_for('index') }}">アップロード画面へ戻る</a>
</p>
</body>
</html>
"""
# Flaskアプリとルーティング
app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 8 * 1024 * 1024
@app.get("/")
def index():
"""画像アップロードフォームとジョブ状態一覧を表示する。"""
return render_template_string(
INDEX_HTML,
jobs=list_jobs(),
async_mode=RQ_ASYNC,
error=request.args.get("error"),
)
@app.post("/jobs")
def enqueue_job():
"""画像アップロードを受け付けて保存し、RQキューへのジョブ投入を行う。"""
uploaded = request.files.get("image")
if uploaded is None or not uploaded.filename:
return redirect(url_for("index", error="PNG画像を選択してください"))
filename = secure_filename(uploaded.filename) or "upload.png"
if not filename.lower().endswith(".png"):
return redirect(url_for("index", error="アップロードできるのはPNG画像だけです"))
job_id = uuid.uuid4().hex[:12]
source_path = UPLOAD_DIR / f"{job_id}_{filename}"
result_path = RESULT_DIR / f"{job_id}_processed.png"
uploaded.save(source_path)
# ジョブ投入: 保存したファイルパスを渡してRQキューへ画像処理ジョブを積む
job = queue.enqueue(
process_image,
str(source_path),
str(result_path),
TARGET_WIDTH,
job_id=job_id,
result_ttl=RESULT_TTL,
meta={"source_name": filename, "result_path": str(result_path)},
)
connection.lpush(JOB_INDEX_KEY, job.id)
return redirect(url_for("job_detail", job_id=job.id))
@app.get("/jobs/<job_id>")
def job_detail(job_id: str):
"""ジョブ状態確認としてRedisから1件の状態と処理結果を読み出して表示する。"""
job = fetch_job(job_id)
if job is None:
abort(404)
status = status_text(job)
result = job.return_value() if status == "finished" else None
return render_template_string(
DETAIL_HTML,
job_id=job.id,
status=status,
result=result,
source_name=job.meta.get("source_name", "-"),
)
@app.get("/jobs/<job_id>/download")
def download_result(job_id: str):
"""処理済み画像ダウンロードとしてPNGファイルを添付形式で送信する。"""
job = fetch_job(job_id)
if job is None:
abort(404)
result_path = Path(job.meta.get("result_path", ""))
if not result_path.is_file():
abort(404)
return send_file(
result_path,
mimetype="image/png",
as_attachment=True,
download_name=result_path.name,
)
@app.get("/api/jobs/<job_id>")
def job_status_api(job_id: str):
"""ジョブ状態確認用にジョブ状態をJSONで返す(ポーリング用)。"""
job = fetch_job(job_id)
if job is None:
return jsonify({"error": "job not found"}), 404
status = status_text(job)
return jsonify(
{
"job_id": job.id,
"status": status,
"result": job.return_value() if status == "finished" else None,
}
)
# 動作確認したい画面
ROUTES = ["/"]
コード全文は上の折り畳みに入れてあるので、全部を上から読む必要はありません。ここでは特に重要な部分だけを抜き出して、何をしているのか順番に見ていきます。
一時ディレクトリに保存先を用意する
BASE_DIR = Path(tempfile.gettempdir()) / "flask_rq_image_jobs"
UPLOAD_DIR = BASE_DIR / "uploads"
RESULT_DIR = BASE_DIR / "results"
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
RESULT_DIR.mkdir(parents=True, exist_ok=True)スクリプトの置き場所が書き込み不可でも動くよう、OS標準の一時ディレクトリの下に作業フォルダを作っています。mkdir(parents=True, exist_ok=True)は親フォルダごと作りつつ、既に存在してもエラーにしない指定です。アップロード元と処理結果をフォルダで分けておくと、後片付けもしやすくなります。
環境変数で接続先とRQ_ASYNCを決める
REDIS_URL = os.environ.get("REDIS_URL", "redis://127.0.0.1:6379/0")
RQ_ASYNC = os.environ.get("RQ_ASYNC", "0") == "1"
QUEUE_NAME = "image-jobs"
JOB_INDEX_KEY = "image-jobs:index"
TARGET_WIDTH = 160接続先と実行モードを環境変数から読み、RQ_ASYNCは文字列の1のときだけTrueになります。QUEUE_NAMEはワーカーが監視するキュー名、JOB_INDEX_KEYはジョブIDを積むRedisのキー名です。TARGET_WIDTHが160なので、リサイズ後の幅は160pxに揃います。
process_imageのリサイズとグレースケール化
width, height = source.size
new_height = max(1, round(height * target_width / width))
# 画像リサイズ: 幅をtarget_widthに合わせ、縦横比を保って高さを算出する
resized = source.resize((target_width, new_height), Image.LANCZOS)
grayscale = ImageOps.grayscale(resized)
grayscale.save(result_path, format="PNG")元画像の幅と高さから、幅を160pxにしたときの高さを比例計算しています。max(1, ...)で高さが0にならないよう保険をかけ、resizeのあとImageOps.grayscaleで白黒へ変換しました。最後にsaveでPNGとして書き出します。
create_connectionでRedisを切り替える
if RQ_ASYNC:
return redis.Redis.from_url(REDIS_URL)
import fakeredis
return fakeredis.FakeStrictRedis()非同期のときはREDIS_URLの実Redisへ、同期検証のときはfakeredisのインメモリRedisへつなぎ分けます。import fakeredisを関数の中に置いているので、非同期運用ではfakeredisが入っていなくても読み込まれません。
Queueのis_asyncで実行方式を決める
connection = create_connection()
queue = Queue(QUEUE_NAME, connection=connection, is_async=RQ_ASYNC)create_connectionが返した接続でQueueを作り、is_asyncへそのままRQ_ASYNCを渡しています。ここがFalseだとenqueueした時点で関数がその場で実行され、Trueならワーカー待ちのキューへ入ります。運用と検証の挙動が1行で切り替わるのが利点です。
status_textでジョブ状態を文字列にする
status = job.get_status(refresh=True)
return getattr(status, "value", str(status))get_status(refresh=True)はRedisから最新の状態を読み直します。RQのバージョンによって戻り値が列挙型か文字列かで揺れるため、getattr(status, "value", str(status))で表示用の文字列に揃えました。
fetch_jobでNoSuchJobErrorを受け止める
try:
return Job.fetch(job_id, connection=connection)
except NoSuchJobError:
return None存在しないジョブIDをJob.fetchに渡すとNoSuchJobErrorが送出されます。ここで捕まえてNoneを返しておくと、呼び出し側はジョブが無い場合を素直に404などへ振り分けられます。
RQ_ASYNCでPythonの同期と非同期を切り替える
同じジョブ関数を、通常運用では非同期に、検証ではその場で同期に動かしたい場面はよくあります。今回はRQ_ASYNCという環境変数ひとつで両方を切り替えられるようにしました。
RQ_ASYNC=1のときはredis.Redis.from_url(REDIS_URL)で実Redisへつなぎ、投入したジョブは別プロセスのワーカーが拾って実行します。
RQ_ASYNC=0のときはfakeredis.FakeStrictRedis()のインメモリRedisにつなぎ、is_async=Falseによってenqueueした瞬間に処理が走ります。外部のRedisサーバやワーカーを立てずに、同じprocess_imageまで含めた動きを決定的に確かめられました。
同期・非同期の切り替えで押さえておきたいポイントは次の2点です。
このセクションの用語
- is_async
-
Queueの引数で、ジョブをワーカーに任せるか即時実行するかを決めます。Falseで同期実行になります。 - fakeredis
-
Redisの振る舞いをPythonのメモリ上で真似るライブラリです。サーバを起動せずに検証できます。 - 環境変数
- OS側から与える設定値です。コードを書き換えずに接続先や動作モードを変えられます。
| 設定 | つなぐRedis | ジョブの実行 |
|---|---|---|
| RQ_ASYNC=1 | REDIS_URLで指定した実Redis | キューに積まれ、rq workerが取り出して非同期実行 |
| RQ_ASYNC=0 | fakeredisのインメモリRedis | enqueue時にその場で同期実行され、戻り値がすぐ入る |
参考：
©RQ公式ドキュメントFor testing purposes, you can enqueue jobs without delegating the actual execution to a worker. To do this, pass the is_async=False argument into the Queue constructor.
検証時：fakeredisで代替
非同期時：実Redisで実行
PythonでPNG画像処理ジョブWebアプリのエラー対処
ここからは、同じ構成を組むときに一般に遭遇しやすいエラーを挙げます。今回の実行で必ず起きたものではなく、原因の見当をつけるための一覧です。
多くはRedisへの接続、ワーカーの起動状態、Pillowが開けないファイルのどれかに分類できます。切り分けの順番を決めておくと調査が早く進みます。
このセクションの用語
- rq worker
- キューを監視してジョブを実行する
RQのコマンドです。引数に監視するキュー名を渡します。 - RESULT_TTL
- ジョブの結果を
Redisに保持する秒数の設定です。今回は3600を定数として置いています。
| エラー例 | 原因 | 対処 |
|---|---|---|
| redis.exceptions.ConnectionError | RQ_ASYNC=1なのにRedisサーバが未起動、またはREDIS_URLの指定違い | Redisを起動するかREDIS_URLを見直す。切り分けならRQ_ASYNC=0で起動する |
| ジョブ状態がqueuedのまま変わらない | ワーカー未起動、または別のキュー名を監視している | rq worker image-jobsのようにQUEUE_NAMEと同じ名前で起動する |
| ModuleNotFoundError: No module named 'fakeredis' | RQ_ASYNC=0の経路で使うfakeredisが未インストール | fakeredisを追加するか、実Redisを用意してRQ_ASYNC=1で実行する |
| PIL.UnidentifiedImageError | PNG以外や壊れたファイルをImage.openに渡した | アップロード時に拡張子と内容を確認し、開けない場合は画面側でエラー表示にする |
| rq.exceptions.NoSuchJobErrorや結果ファイルが無い | ジョブ情報の保持期限切れ、または一時ディレクトリが消えた | fetch_jobのNone分岐で404を返し、残したい結果は永続ディレクトリへ保存する |
PNG画像処理ジョブWebアプリで注意したい点
動かない原因の多くは、画像処理そのものよりジョブの受け渡し周りにあります。RQ_ASYNC=1ではワーカーを別に起動しておく必要があり、忘れると状態がqueuedのまま止まって見えます。
保存先も見落としがちです。今回はtempfile.gettempdir()配下にuploadsとresultsを作るため、環境によっては再起動やクリーンアップでファイルが消えることがあります。
ジョブ状態は画面を開いた瞬間の値なので、投入直後にfinishedでないからといって失敗とは限りません。ポイントとしては、ワーカーの起動、保存先の性質、状態確認のタイミングを押さえておくことです。
ワーカー起動：enqueueだけでは動かない
保存先：一時ディレクトリは消える前提
ジョブ状態：finishedまで再読込
PythonアプリでPNG投入とRQジョブ状態を確認
まず操作前の画面をキャプチャし、アップロードフォームとジョブ一覧がHTTP 200で表示される状態を確認しました。
続いてPNG画像をアップロードしてRQジョブを投入し、ジョブ状態がfinishedになったうえで処理済み画像のダウンロードリンクが出ることを確認しています。
実ブラウザ操作のキャプチャも取得したので、フォーム送信から結果表示までの流れが画面として追えます。表に出す集計データは扱っていないため、確認は画面の状態が中心です。
このセクションの用語
- finished
-
RQのジョブ状態のひとつで、処理が最後まで終わったことを示します。 - HTTP 200
- リクエストが正常に処理されたことを表すステータスコードです。
- send_file
-
Flaskでサーバ上のファイルをレスポンスとして返す関数です。添付ダウンロードにも使えます。
キャプチャで押さえた確認ポイントは次の3点です。
- 操作前:アップロードフォームとジョブ一覧が表示される(HTTP 200)
- 投入後:ジョブ状態がfinishedになる
- 詳細画面:処理済み画像のダウンロードリンクが出る
PythonでFlask・RQの画像処理ジョブが役立つ場面
この構成の要点は、重い処理をWebリクエストから切り離し、状態と成果物をあとから取りに行けることです。画像以外にも応用が利きます。
一般に、時間のかかる変換処理や、失敗したときに作り直したい処理と相性が良いといえます。
このセクションの用語
- サムネイル
- 一覧表示用に小さく作り直した画像です。元画像より軽く、表示が速くなります。
- 再投入
- 失敗したジョブを同じ引数でもう一度キューへ入れ直すことです。原因を直したあとの復旧に使います。
| 使える場面 | 具体的な使い方 |
|---|---|
| 投稿画像のサムネイル生成 | アップロード直後にenqueueし、幅160pxのような規格サイズへ縦横比維持で変換して一覧表示に使う |
| 解析前の画像前処理 | ImageOps.grayscaleでの白黒化やリサイズをワーカー側にまとめ、Web側は投入と結果取得だけを担当する |
| 社内ツールの一括変換 | 複数ファイルをまとめてenqueueし、image-jobs:indexのようなインデックスで一覧管理して失敗分だけ再投入する |
| 時間のかかる帳票やレポート生成 | 生成処理をジョブ化し、ジョブ状態画面で進行を見せて完了後にsend_fileでダウンロードさせる |
| 動作検証や自動テスト | RQ_ASYNC=0の同期実行で、Redisサーバやワーカーを立てずに投入から結果表示までの画面遷移を確かめる |
PNG画像処理ジョブWebアプリ開発のまとめ
Flaskでアップロードを受け、queue.enqueue(process_image, ...)でRQへ投げ、Redisに残った状態をJob.fetchで読み、Pillowが幅160pxのグレースケールPNGを作ります。役割がはっきり分かれているので、1ファイルでも見通しは悪くありませんでした。
実際にPNG画像をアップロードしてジョブを投入し、状態がfinishedになったうえでダウンロードリンクが出るところまで確認しています。
RQ_ASYNCで非同期と同期を切り替えられる形にしておくと、運用と検証で同じジョブ関数を使い回せます。まずはRQ_ASYNC=0で画面の流れを掴み、慣れてから実Redisとワーカーへ進むのが安全な進め方です。
参考にした一次情報
- ^ Flask Documentation — Uploading Files. https://flask.palletsprojects.com/en/stable/patterns/fileuploads/, (参照26-08-18).
- ^ RQ: Simple job queues for Python — Queues. https://python-rq.org/docs/, (参照26-08-18).
- ^ RQ Documentation — Testing (is_async=False). https://python-rq.org/docs/testing/, (参照26-08-18).
- ^ Pillow Documentation — ImageOps Module. https://pillow.readthedocs.io/en/stable/reference/ImageOps.html, (参照26-08-18).
※内容は執筆時点のものです。ライブラリやサイトの仕様は変わる可能性があるため、公式ドキュメントもあわせてご確認ください。
新着記事
AI時代の採用競争力を高める「採用LLMOパッケージサービス」を提供
アルパカ、CFTC登録の先物取引業者（FCM）として認可を取得
オルタナティブ投資プラットフォーム「オルタナバンク」、『【3ヶ月毎分配】インド貸金事業者中長期支援ファンド ID1145』を公開
「理想の姿勢」を測る時代の終わりに ── こころ整体院グループとタンクル、AI姿勢分析システムを共同開発
