1. リクエストとレスポンスの制御

Flaskでは、クライアントからのリクエスト情報はグローバルな request オブジェクトを介して取得します。このオブジェクトはスレッドセーフであり、現在のリクエストコンテキストに即したデータを提供します。

from flask import Flask, request, make_response, jsonify

app = Flask(__name__)

@app.route('/data', methods=['GET', 'POST'])
def handle_data():
    # メソッドの確認
    print(f"Method: {request.method}")
    
    # クエリパラメータ (GET)
    user_id = request.args.get('id')
    
    # フォームデータ (POST)
    user_name = request.form.get('name')
    
    # JSONデータ
    json_body = request.json
    
    # ヘッダーとCookie
    user_agent = request.headers.get('User-Agent')
    token = request.cookies.get('session_token')
    
    # パス情報
    full_url = request.url
    path_info = request.path

    return "Processing complete"

レスポンスオブジェクトのカスタマイズ

通常の文字列や辞書の返却以外に、ヘッダーやCookieを明示的に操作する場合は make_response を使用します。

@app.route('/set-info')
def set_info():
    content = "Cookie and header update"
    response = make_response(content)
    
    # カスタムヘッダーの設定
    response.headers['X-Server-Version'] = '1.0.2'
    
    # Cookieの設定 (有効期限やドメインの指定が可能)
    response.set_cookie('user_pref', 'dark_mode', max_age=3600)
    
    return response

2. セッション管理の仕組み

Flaskのデフォルトセッションは、データをシリアライズして暗号化し、クライアント側のCookieに保存する「クライアントサイドセッション」です。

from flask import session

app.secret_key = 'your_secret_key_here'

@app.route('/login')
def login():
    session['user_role'] = 'admin'
    return "Session stored"

@app.route('/dashboard')
def dashboard():
    role = session.get('user_role')
    return f"Current role: {role}"

内部動作としては、session_interface が open_session でCookieから復号し、save_session でレスポンス時に暗号化してCookieへ書き戻します。

3. メッセージの「フラッシュ」機能

あるリクエストで発生したメッセージを、次のリクエスト（リダイレクト先など）で一度だけ表示するために flash を利用します。

from flask import flash, get_flashed_messages

@app.route('/submit')
def submit():
    flash("保存が完了しました", category="success")
    flash("権限が不足しています", category="error")
    return redirect('/status')

@app.route('/status')
def status():
    # カテゴリ別にメッセージを取得可能
    success_msgs = get_flashed_messages(category_filter=['success'])
    return jsonify(messages=success_msgs)

4. 非同期処理への対応

Flask 2.0以降、async def による非同期ビュー関数がサポートされています。ただし、内部で呼び出すライブラリ（DBドライバやRedisクライアント）も非同期対応（例: aiomysql, aioredis）である必要があります。

import asyncio
import aioredis

async def fetch_redis_val(key):
    redis = await aioredis.from_url("redis://localhost")
    val = await redis.get(key)
    return val

@app.route('/async-data')
async def get_async_data():
    result = await fetch_redis_val('site_notice')
    return f"Notice: {result}"

5. リクエスト拡張（フック関数）

Djangoのミドルウェアに近い機能として、特定のタイミングでコードを実行するデコレータが用意されています。

• before_request: ビュー関数の実行前に処理。値を返すとビューは実行されません。
• after_request: ビュー関数の実行後に処理。レスポンスオブジェクトを受け取り、返す必要があります。
• teardown_request: リクエスト終了時に必ず実行（例外発生時も含む）。ログ記録等に利用。
• errorhandler: 特定のステータスコード発生時の挙動を定義。

@app.before_request
def check_auth():
    if not session.get('logged_in') and request.path != '/login':
        return redirect('/login')

@app.after_request
def add_security_headers(response):
    response.headers['X-Content-Type-Options'] = 'nosniff'
    return response

6. Blueprintによるモジュール化

大規模なアプリケーションでは、機能ごとにルーティングやテンプレートを分割するために Blueprint を使用します。

# auth.py
from flask import Blueprint
auth_bp = Blueprint('auth', __name__, url_prefix='/auth')

@auth_bp.route('/login')
def login_page():
    return "Login UI"

# app.py
app.register_blueprint(auth_bp)

7. g オブジェクト

g は1回のリクエスト内だけで有効なグローバル一時ストレージです。リクエストを跨いで保持されることはありません。

from flask import g

@app.before_request
def load_user():
    # DBから取得したユーザー情報を格納
    g.current_user = "Taro"

def some_internal_logic():
    # 引数で渡さなくてもgから取得可能
    user = g.current_user
    print(user)

8. データベース接続プール (DBUtils)

リクエストごとにDB接続を生成すると負荷が高くなるため、接続プールを利用するのが一般的です。

from dbutils.pooled_db import PooledDB
import pymysql

DB_POOL = PooledDB(
    creator=pymysql,
    maxconnections=10,
    mincached=2,
    host='127.0.0.1',
    user='root',
    password='password',
    database='my_db'
)

@app.route('/users')
def list_users():
    conn = DB_POOL.connection()
    cursor = conn.cursor()
    cursor.execute('SELECT name FROM users')
    data = cursor.fetchall()
    conn.close() # プールへ戻す
    return jsonify(data)

9. シグナル (Signals)

Blinkerライブラリを使用して、Flask内のイベント発生時に特定の関数をトリガーできます。コードの疎結合化に役立ちます。

from flask import signals

def log_template_render(sender, template, context, **extra):
    print(f"Rendering: {template.name} with context {context}")

signals.before_render_template.connect(log_template_render)

10. threading.localとコンテキストの仕組み

Flaskは threading.local を拡張した仕組みを利用して、リクエストごとに独立したデータを保持します。以下のコードは、スレッドやコルーチン（Greenlet）を識別してデータを隔離する簡易的な実装例です。

try:
    from greenlet import getcurrent as get_ident
except ImportError:
    from threading import get_ident

class CustomLocal:
    def __init__(self):
        # 識別子ごとにデータを保持する辞書
        object.__setattr__(self, "__storage__", {})

    def __setattr__(self, name, value):
        ident = get_ident()
        storage = self.__storage__
        if ident in storage:
            storage[ident][name] = value
        else:
            storage[ident] = {name: value}

    def __getattr__(self, name):
        ident = get_ident()
        try:
            return self.__storage__[ident][name]
        except KeyError:
            raise AttributeError(name)

# 各スレッドで異なる値を保持できる
local_data = CustomLocal()

Flaskの request や session は LocalProxy というプロキシオブジェクトであり、内部的には _request_ctx_stack というスタックから現在の実行コンテキストに対応するオブジェクトを取得して操作を転送しています。