Python

はじめに Pythonで大量のAPI呼び出しやファイル操作を行う際、処理時間がボトルネックになることは珍しくありません。同期的な処理では、一つの操作が完了するまで次の処理を開始できないため、I/O待ち時間が積み重なってしまいます。 本記事では、Pythonのasyncioモジュールを使った非同期プログラミングについて、基礎概念から実践的なパターンまでを体系的に解説します。実際のプロジェクトで使える具体的なコード例を通じて、処理速度を劇的に改善する方法を学んでいきましょう。 asyncioの基本概念 イベントループとは asyncioの中心にあるのがイベントループです。イベントループは、非同期タスクの実行を管理し、I/O操作の完了を監視して、適切なタイミングでタスクを再開させる役割を担います。 1 2 3 4 5 6 7 8 9 import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") # Python 3.7以降の推奨方法 asyncio.run(main()) async/awaitの仕組み async defで定義された関数はコルーチン関数となり、呼び出すとコルーチンオブジェクトを返します。awaitキーワードを使うことで、そのコルーチンの完了を待機しつつ、その間に他のタスクを実行できるようになります。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 import asyncio import aiohttp async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict: """URLからデータを非同期で取得""" async with session.get(url) as response: return { "url": url, "status": response.status, "content_length": len(await response.text()) } async def main(): urls = [ "https://api.github.com", "https://api.stripe.com", "https://api.openai.com" ] async with aiohttp.ClientSession() as session: # 全URLを並行して取得 tasks = [fetch_url(session, url) for url in urls] results = await asyncio.gather(*tasks) for result in results: print(f"{result['url']}: {result['status']}") asyncio.run(main()) 実践パターン1：大量のAPI呼び出しを高速化 問題：同期処理での遅延 例えば、1000件のユーザーデータをAPIから取得する場合を考えます。1リクエストあたり100msかかるとすると、同期処理では100秒もの時間が必要です。 ...

FastAPI + SQLAlchemy性能改善プレイブック: 遅いAPIを計測ベースで高速化する FastAPIの初期実装は非常に快適です。しかし運用フェーズに入ると、次のような症状が出てきます。 一覧APIのレスポンスが急に遅くなる 同時接続が増えるとp95が跳ねる CPUは余っているのにタイムアウトが増える DB接続数が上限に張り付く こうした問題の多くは「Pythonが遅い」のではなく、SQLAlchemyの使い方とDBアクセス設計 に起因します。 本記事では、FastAPI + SQLAlchemy + PostgreSQL構成を前提に、実際の改善手順を計測ベースで整理します。 1. 最初に測るべき指標 最適化は、体感ではなく数値で進めます。最低限、以下を可視化します。 APIのp50/p95/p99レイテンシ エンドポイント別SQL発行回数 1リクエストあたりのDB滞在時間 connection pool待ち時間 slow query件数（200ms以上など） OpenTelemetryやNew Relicを使っているなら、アプリspanとDB spanを必ず紐付けてください。これだけでボトルネック特定速度が上がります。 2. N+1問題を最優先で潰す 最も頻出するのがN+1です。例えばユーザー一覧でプロフィールを参照すると、ユーザー数分の追加クエリが発行されます。 2.1 悪い例 1 2 3 4 5 6 7 8 users = session.query(User).limit(100).all() result = [] for u in users: result.append({ "id": u.id, "name": u.name, "profile": u.profile.bio, }) 2.2 改善例（joinedload/selectinload） 1 2 3 4 5 6 7 8 from sqlalchemy.orm import selectinload users = ( session.query(User) .options(selectinload(User.profile)) .limit(100) .all() ) joinedload と selectinload はデータ量で使い分けます。 ...

FastAPI + Celery信頼性設計: 非同期ジョブを本番で壊さないための実装パターン FastAPIでAPIを作ると、重い処理はすぐに非同期ジョブへ逃がしたくなります。画像変換、レポート生成、外部API連携、メール配信など、Celeryは非常に便利です。ですが、本番で問題になるのは「動くかどうか」ではなく、失敗したときに壊れないか です。 同じジョブが二重実行される 一時障害で永遠にリトライしてキューが詰まる ワーカー再起動で中途半端な状態が残る 完了通知が先に飛んで実データがない 本記事では FastAPI + Celery + Redis 構成を前提に、再実行安全性（idempotency）と運用信頼性を上げる実装手順をまとめます。 1. まず守るべき設計原則 非同期基盤の事故は、ほぼ次の4原則で防げます。 At-least-once前提（同一タスク再実行は必ず起こる） 副作用は冪等化（何回実行されても結果が壊れない） 状態遷移を明示（PENDING/RUNNING/SUCCEEDED/FAILED） 失敗を可観測化（リトライ回数・死活・滞留時間を計測） この原則を外すと、障害時に「何が完了して何が未完了か」が追えなくなります。 2. 参照アーキテクチャ API: FastAPI Queue Broker: Redis Worker: Celery Result Store: PostgreSQL（業務状態） Monitoring: Flower + Prometheus + Sentry ポイントは、業務上重要な状態はRedis結果バックエンドに依存しない ことです。Redisは一時的に使い、真実の状態はRDBに持たせます。 3. 実装の土台: タスク受付API 3.1 受け付け時に idempotency_key を必須化 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 from fastapi import FastAPI, HTTPException from pydantic import BaseModel from sqlalchemy import select app = FastAPI() class JobRequest(BaseModel): idempotency_key: str report_type: str user_id: str @app.post("/reports") def create_report(req: JobRequest): existing = find_job_by_key(req.idempotency_key) if existing: return {"job_id": existing.id, "status": existing.status} job = create_job_record( idempotency_key=req.idempotency_key, status="PENDING", report_type=req.report_type, user_id=req.user_id, ) generate_report.delay(job.id) return {"job_id": job.id, "status": "PENDING"} これでクライアント再送が来てもジョブ多重作成を防げます。 ...

FastAPI認証・認可の本番設計：JWT運用、権限制御、監査ログまで含めた実装パターン FastAPI は実装が速い反面、認証・認可を最小構成のまま本番に出してしまい、後からセキュリティ事故に発展するケースが少なくありません。特に「JWT を入れたから安全」という誤解は危険です。 本記事では、開発速度を落とさずに本番で耐える認証基盤を作るための設計を、コード例と運用手順込みで解説します。 1. 認証と認可を分離して設計する 最初に押さえるべきは責務分離です。 認証（Authentication）: 誰かを確認する 認可（Authorization）: 何をしてよいか判定する この2つを混ぜると、実装も監査も破綻します。FastAPI では dependency を分け、get_current_user と require_permission を独立させるのが基本です。 2. JWT は「短命 + リフレッシュ + 失効管理」で使う アクセストークンを長寿命にすると、漏えい時の被害が大きくなります。実運用では以下が標準です。 Access Token: 5〜15分 Refresh Token: 7〜30日 Refresh Token は DB 保存し、ローテーション時に旧トークンを失効 sub だけでなく、jti（トークンID）や scope を持たせると管理しやすくなります。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 from datetime import datetime, timedelta, timezone import jwt ALGORITHM = "HS256" def create_access_token(user_id: str, scopes: list[str], secret: str) -> str: now = datetime.now(timezone.utc) payload = { "sub": user_id, "scope": " ".join(scopes), "iat": int(now.timestamp()), "exp": int((now + timedelta(minutes=10)).timestamp()), "jti": "generated-uuid" } return jwt.encode(payload, secret, algorithm=ALGORITHM) 3. 鍵管理とローテーション 秘密鍵を .env に固定して数年運用するのは典型的な事故パターンです。最低限、次を実施します。 ...

GitHub Actions高速化実践：Matrix戦略・依存キャッシュ・失敗切り分けの設計ガイド GitHub Actions は便利ですが、プロジェクトが成長すると「遅い」「不安定」「原因が分かりにくい」という三重苦になりがちです。特に monorepo や複数ランタイム対応（Node/Python/Go など）では、ワークフローの設計次第で CI 時間が 2〜3 倍変わります。 本記事では、実行時間を短くしながら失敗時の調査コストも下げるために、matrix 設計・キャッシュ設計・障害時の確認順序を具体的に整理します。 1. まず「何を並列化するか」を決める Actions の高速化は、いきなりキャッシュ最適化から入るより、先にジョブ分解を決める方が効きます。原則は次の通りです。 並列化すべき: 独立テスト（OS/バージョン別、サービス別） 直列にすべき: デプロイ、DB マイグレーション、本番反映 依存を分ける: lint/typecheck/test/build を一つに詰め込まない 悪い例は、1ジョブに全部詰め込み、失敗時に最初から再実行するパターンです。良い設計では「lint は通るが test だけ失敗」のように切り分けできます。 2. matrix を作るときの実践ルール matrix は便利ですが、組み合わせ爆発で逆に遅くなることがあります。例えば os x runtime x db をすべて直積にすると、不要なジョブが大量発生します。そこで include/exclude を活用します。 1 2 3 4 5 6 7 8 9 10 11 12 strategy: fail-fast: false matrix: os: [ubuntu-latest, macos-latest] node: [20, 22] include: - os: ubuntu-latest node: 22 coverage: true exclude: - os: macos-latest node: 20 ポイントは次です。 ...

Python asyncioバックプレッシャー設計：落ちない非同期バッチを作る実装パターン asyncio は速く作れる一方で、負荷が上がった瞬間に崩壊する設計を作りやすいという側面があります。特に「処理待ちが無限に積み上がる」「外部API遅延で全体が詰まる」「リトライ嵐でさらに遅くなる」は典型的です。 本記事では、非同期ワーカーを本番運用する前提で、バックプレッシャーを実装に落とす方法を解説します。単なる概念ではなく、すぐ使えるコード断片を中心に進めます。 1. なぜバックプレッシャーが必要か バックプレッシャーは「これ以上は受けない」仕組みです。これがない設計は、ピーク時に次の順で壊れます。 入力が処理速度を超える キューが無限増加してメモリ圧迫 GC増加でスループット低下 タイムアウト増加→リトライ増加 システム全体が雪崩れる つまり、受けすぎないことは性能ではなく可用性の話です。 2. 基本設計：3つの制限を必ず入れる 2-1. キュー上限（bounded queue） 1 2 3 4 import asyncio QUEUE_MAX = 1000 queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=QUEUE_MAX) maxsize なしは原則禁止です。業務要件で「捨てられない」場合でも、無限キューより「受け付け停止 + 明示エラー」のほうが復旧可能です。 2-2. 同時実行数上限（semaphore） 1 2 3 4 5 6 CONCURRENCY = 20 semaphore = asyncio.Semaphore(CONCURRENCY) async def guarded_call(fn, *args, **kwargs): async with semaphore: return await fn(*args, **kwargs) CPU でも I/O でも、同時実行数に上限を持たせると遅延の尾が短くなります。 ...

FastAPI本番運用ハードニング完全ガイド：セキュリティ・性能・障害対応を実装で固める FastAPI は開発速度が高く、PoC から本番まで一気に進めやすいフレームワークです。しかし、早く作れることと安全に運用できることは別問題です。実際の障害は、コードの正しさよりも運用の隙から発生します。 本記事では、FastAPI を本番で安心して運用するためのハードニング手順を、実装可能な形でまとめます。対象は「すでにAPIが動いているが、運用強度を上げたい」チームです。 1. 入口防御：TLS、ヘッダ、レート制限 TLS終端とForwardedヘッダ ロードバランサ配下で動かす場合、X-Forwarded-For と X-Forwarded-Proto の扱いを明確にします。誤るとクライアントIPが取れず、監査も制限も機能しません。 1 2 3 4 5 from fastapi import FastAPI from starlette.middleware.trustedhost import TrustedHostMiddleware app = FastAPI() app.add_middleware(TrustedHostMiddleware, allowed_hosts=["api.example.com", "*.example.com"]) allowed_hosts をワイルドにしすぎると Host Header Injection の温床になります。 セキュリティヘッダ 最低限次を返します。 Strict-Transport-Security X-Content-Type-Options: nosniff X-Frame-Options: DENY Referrer-Policy APIでも無関係ではありません。管理画面やドキュメントUIを守る意味があります。 レート制限 ブルートフォースと突発負荷に備え、IPまたはAPIキー単位でレート制限を設定します。 認証系: 5 req/min 通常API: 60 req/min 高負荷検索: 20 req/min Redis バックエンド方式にして、アプリ再起動でカウンタが失われないようにします。 ...

はじめに 皆さん、こんにちは！テクノロジーの進化は本当に早いもので、Googleから最新のAIモデル「Gemini 3.1 Pro」が正式に発表されました。 このニュースは世界中のエンジニアを驚かせており、テック系コミュニティの聖地とも言えるHacker Newsでは、投稿からわずか数時間で882ポイントという異例の高評価を獲得しました。これほどまでに注目されているのは、単なるスペックアップを超えた「実用性の進化」があるからです。 「AIの進化が早すぎて追いつけない……」と感じている初心者エンジニアの方も多いかもしれませんが、安心してください。この記事では、Gemini 3.1 Proの何がすごいのか、そして今日からどうやって使いこなすのかを、どこよりも噛み砕いて解説します！ Gemini 3.1 Proとは？ Gemini 3.1 Proは、Googleが開発した「Gemini」シリーズの最新鋭モデルです。従来のGemini 3の長所を引き継ぎつつ、特に「推論（考える力）」と「文脈の理解（記憶力）」が大幅に強化されています。 エンジニアにとってのGemini 3.1 Proは、例えるなら**「プロジェクトの全コードを記憶し、複雑なバグの修正案を即座に提案してくれる、超優秀な先輩エンジニア」**のような存在です。 なぜ「Pro」なのか？ Googleのモデルには「Ultra」「Pro」「Flash」などのラインナップがありますが、Proモデルは「性能」と「コスト・速度」のバランスが最も優れています。開発者がAPIを使ってアプリケーションに組み込む際、最も選ばれているのがこのProシリーズなのです。 ここがすごい！Gemini 3.1 Proの3つの進化点 従来のモデルと比べて、具体的にどこが変わったのでしょうか？注目すべき3つのポイントを挙げます。 1. 「熟考型」の推論プロセス Gemini 3.1 Proには、人間が難しい問題を解くときにじっくり考えるような「System 2 Thinking」に近い仕組みが導入されました。これにより、これまでは間違えやすかった複雑な数学の問題や、高度な論理パズル、さらには大規模なシステムのデバッグにおいて、圧倒的に正確な回答を返せるようになっています。 2. 200万トークンの超長大コンテキスト 「トークン」とは、AIが一度に扱える情報の単位です。Gemini 3.1 Proは、最大で200万トークンという驚異的な量を一度に読み込むことができます。 これは、「厚辞苑数冊分のテキスト」や「数万行のソースコード全体」を丸ごとAIに読み込ませて、その内容について質問できることを意味します。「あの関数の定義、どこにあったっけ？」と探す手間は、もう過去のものになるかもしれません。 3. ハルシネーション（もっともらしい嘘）の劇的な減少 AIが自信満々に嘘をつく現象「ハルシネーション」が、Gemini 3.1 Proでは大幅に抑えられています。特に関数呼び出し（Function Calling）の正確性が増しており、外部ツールやデータベースと連携させた際の信頼性が向上しました。 【実践】PythonでGemini 3.1 Proを動かしてみよう それでは、実際にAPIを使ってGemini 3.1 Proを操作してみましょう。初心者の方でも、以下の3ステップで簡単に始められます。 1. ライブラリの準備 ターミナルで以下のコマンドを実行し、最新のSDKをインストールします。 1 pip install -U google-generativeai

AWS Lambda SnapStartがPythonに対応！コールドスタート解消へ はじめに AWS Lambdaを本番環境で利用している、あるいは利用を検討しているPythonデベロッパーの皆さん。「サーバーレスは便利だけど、あの"最初の"リクエストだけ遅いのが気になる…」と感じたことはありませんか？ API Gatewayと連携させたLambda関数が、しばらくアクセスがないとタイムアウトギリギリになったり、ユーザーに不快な待ち時間を与えてしまったり。この現象は「コールドスタート」と呼ばれ、多くのサーバーレス開発者を悩ませてきた根深い課題です。 これまで、この問題を解決するにはProvisioned Concurrency（プロビジョニングされた同時実行）という有料オプションを利用するのが一般的でしたが、コストとのトレードオフに頭を悩ませるケースも少なくありませんでした。 しかし、2023年末のre:Invent 2023で、ついにこの状況を打開する待望の機能がPythonランタイムにもたらされました。それがAWS Lambda SnapStartです。 これまでJavaランタイムでのみ利用可能だったこの機能がPythonに対応したことで、私たちのサーバーレスアプリケーション開発は新たなステージに進むことになります。本記事では、プロの技術ブロガーとして、Lambda SnapStart for Pythonの仕組みから具体的な使い方、そして現場で活かすための実践的なTipsまで、徹底的に深掘りしていきます。この記事を読み終える頃には、あなたもSnapStartを使いこなし、コールドスタートの悩みから解放されているはずです。 なぜLambda SnapStartが重要なのか？ - コールドスタート問題の再確認 SnapStartの詳細に入る前に、なぜこの機能がこれほどまでに待望されていたのか、その背景にある「コールドスタート問題」を改めて整理しましょう。 Lambdaの実行モデルとライフサイクル AWS Lambdaは、リクエストに応じてコンテナ（実行環境）を起動し、コードを実行するアーキテクチャです。この実行環境は、常に起動しているわけではありません。一定時間リクエストがないと、AWSはコスト効率化のために実行環境を破棄します。 次にリクエストが来たとき、Lambdaは以下のステップを踏んで応答します。 実行環境の確保: 新しい実行環境（マイクロVM）をプロビジョニングします。 コードのダウンロード: S3などから関数のコード（デプロイパッケージ）をダウンロードし、展開します。 ランタイムの初期化: Pythonのランタイム（インタプリタ）を起動します。 関数の初期化 (Initフェーズ): ハンドラ関数の外で定義されたグローバルなコードを実行します。ライブラリのインポート、DBコネクションプールの作成、機械学習モデルのロードなど、比較的時間のかかる処理がここに含まれます。 関数の実行 (Invokeフェーズ): 実際にハンドラ関数を実行し、リクエストを処理します。 このうち、ステップ1から4までを含む最初の呼び出しをコールドスタートと呼びます。一度起動した実行環境は再利用されるため、2回目以降の呼び出し（ウォームスタート）ではステップ5のInvokeフェーズのみが実行され、非常に高速に応答できます。 graph TD subgraph "Cold Start" A[Request] --> B(Allocate MicroVM) B --> C(Download Code) C --> D(Initialize Runtime) D --> E(Run Init Code) E --> F(Run Handler) end F --> G[Response] subgraph "Warm Start" H[Subsequent Request] --> I{Environment Ready?} I -- Yes --> J(Run Handler) I -- No --> B J --> K[Response] end コールドスタートが引き起こす問題 コールドスタートによる遅延は、数十ミリ秒から、場合によっては10秒以上にも及ぶことがあります。特に以下のようなケースで顕著になります。 ...

Python 3.15の新機能：JITコンパイラ標準搭載へ - 待ち望んだパフォーマンス革命がついに始まる はじめに 「Pythonは書きやすいけど、遅い」。これは、多くのエンジニアが一度は耳にしたことがある、あるいは実感したことがある言葉ではないでしょうか。Webアプリケーション開発からデータサイエンス、機械学習まで、Pythonはその圧倒的な生産性と豊富なエコシステムで世界中の開発者を魅了してきました。しかし、その一方で、パフォーマンスが要求される場面では、C/C++による拡張モジュールの作成や、Cython/Numbaといった特殊なツールの導入、あるいはGoやRustといった他の言語の採用を検討せざるを得ない状況がしばしばありました。 もし、あなたがこれまでに、 計算量の多い処理がボトルネックとなり、ユーザー体験を損なっている パフォーマンス向上のためにPython以外の言語知識を要求され、開発の複雑性が増している 高速化ライブラリを導入したものの、環境構築や互換性の問題に悩まされている といった課題に直面したことがあるなら、この記事はまさにあなたのためにあります。 長年の課題であったパフォーマンス問題に終止符を打つべく、Python開発チームは「Faster CPython」プロジェクトを推進してきました。そして、その集大成とも言える機能が、ついに Python 3.15 に標準搭載される見込みです。それが、JIT (Just-In-Time) コンパイラです。 この記事では、Python 3.15で導入されるJITコンパイラが、なぜPythonの歴史における「革命」とまで言えるのか、その仕組みから具体的な効果、そして我々開発者が享受できるメリットと注意点まで、詳細に解説していきます。Pythonの未来を大きく変えるこの新機能の全貌を、一緒に見ていきましょう。 なぜJITコンパイラが今、重要なのか？ - Python高速化の歩み CPython（標準のPython実装）にJITコンパイラが搭載されることの重要性を理解するためには、まずPythonがどのようにコードを実行しているのか、そしてこれまでどのような高速化の試みが行われてきたのかを知る必要があります。 CPythonの実行モデル：インタプリタの長所と短所 私たちが普段書いているPythonコード (.pyファイル) は、そのままではコンピュータが理解できません。CPythonは、以下のステップでコードを実行します。 コンパイル: Pythonのソースコードを、プラットフォームに依存しない中間表現である「バイトコード」に変換します。この結果は .pyc ファイルとしてキャッシュされることがあります。 実行: Python仮想マシン (PVM) と呼ばれるプログラムが、このバイトコードを一行ずつ解釈し、対応するC言語の関数を実行していきます。 +------------------+ (1) コンパイル +-----------------+ (2) 実行 +----------------+ | ソースコード | -------------------> | バイトコード | -------------> | Python仮想マシン | | (hello.py) | | (hello.pyc) | | (PVM) | +------------------+ +-----------------+ +----------------+ | | 実行 V [ 結果 ] この「インタプリタ方式」は、動的型付け（変数の型を実行時に決定する）といったPythonの柔軟性を支える重要な仕組みです。しかし、これがパフォーマンスのボトルネックにもなっています。PVMはバイトコードを実行するたびに、変数の型をチェックし、どの処理を呼び出すかを判断する必要があります。この間接的な処理が、C++やRustのような事前に全てのコードを機械語にコンパイル（AOT: Ahead-Of-Timeコンパイル）する言語に比べて、大きなオーバーヘッドとなるのです。 ...