目的
コードを書く機会はそれなりに多いが、日ごろから持ち前の鳥頭が存分に活かされすぎているため書き方を一瞬で忘れる。
そのため、仕事用端末からもプライベートの端末からも閲覧できる形でスニペットとして残しておきたい
今のところ、1から書くと多すぎる上に他の記事などでうまくまとまってる可能性の高いもの（Pandasとか）はそちらを参照することにする。
一覧
FastAPI（v0.103.1)、Flask（v2.3.3）
基本的なAPIエンドポイント
FastAPI
from fastapi import FastAPI
app = FastAPI()
@app.get("/data")
async def get_data():
return {"message": "データを取得しました"}
Flask
from flask import Flask, jsonify
app = Flask(__name__)
@app.route("/data", methods=["GET"])
def get_data():
return jsonify({"message": "データを取得しました"})
パスパラメータの使用
FastAPI
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id, "description": "Itemの詳細"}
Flask
@app.route("/items/<int:item_id>", methods=["GET"])
def get_item(item_id):
return jsonify({"item_id": item_id, "description": "Itemの詳細"})
クエリパラメータの使用
FastAPI
@app.get("/search/")
async def search_items(query: str = None, limit: int = 10):
return {"query": query, "limit": limit}
Flask
from flask import request
@app.route("/search", methods=["GET"])
def search_items():
query = request.args.get("query", "")
limit = request.args.get("limit", 10)
return jsonify({"query": query, "limit": limit})
POSTリクエストとデータバリデーション
FastAPI
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str
price: float
@app.post("/items/")
async def create_item(item: Item):
return {"message": "アイテムが作成されました", "item": item}
Flask
from flask import request, abort
@app.route("/items", methods=["POST"])
def create_item():
data = request.get_json()
if not data or "name" not in data or "price" not in data:
abort(400, description="Invalid data")
return jsonify({"message": "アイテムが作成されました", "item": data})
エラーハンドリング
FastAPI
from fastapi import HTTPException
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id not in range(1, 101):
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}
Flask
@app.errorhandler(404)
def not_found(error):
return jsonify({"error": "Not Found"}), 404
@app.route("/items/<int:item_id>", methods=["GET"])
def get_item(item_id):
if item_id > 100:
abort(404)
return jsonify({"item_id": item_id})
SQLAlchemy（v2.0.21）Tortoise-ORM（v0.19.3）
データベース接続と依存性の注入
SQLAlchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base
from fastapi import Depends, FastAPI, HTTPException
# データベース接続設定
DATABASE_URL = "postgresql+psycopg2://user:password@localhost/dbname"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()
Tortoise-ORM
from tortoise import Tortoise, fields
from tortoise.models import Model
from fastapi import FastAPI
app = FastAPI()
@app.on_event("startup")
async def init_db():
await Tortoise.init(
db_url="sqlite://db.sqlite3",
modules={"models": ["__main__"]}
)
await Tortoise.generate_schemas()
@app.on_event("shutdown")
async def close_db():
await Tortoise.close_connections()
モデルの定義
SQLAlchemy
from sqlalchemy import Column, Integer, String, Float
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
price = Column(Float)
Tortoise-ORM
class Item(Model):
id = fields.IntField(pk=True)
name = fields.CharField(max_length=255)
price = fields.FloatField()
class Meta:
table = "items"
データベースセッションを取得する依存関数の定義
SQLAlchemy
from contextlib import contextmanager
def get_db() -> Session:
db = SessionLocal()
try:
yield db
finally:
db.close()
Tortoise-ORM
from fastapi import Depends, HTTPException
async def get_item(item_id: int):
item = await Item.get_or_none(id=item_id)
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
CRUD操作とルーティング
SQLAlchemy
app = FastAPI()
@app.post("/items/", response_model=Item)
def create_item(name: str, price: float, db: Session = Depends(get_db)):
db_item = Item(name=name, price=price)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
@app.get("/items/{item_id}")
def read_item(item_id: int, db: Session = Depends(get_db)):
item = db.query(Item).filter(Item.id == item_id).first()
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
Tortoise-ORM
@app.post("/items/")
async def create_item(name: str, price: float):
item = await Item.create(name=name, price=price)
return {"message": "Item created", "item": item}
@app.get("/items/{item_id}")
async def read_item(item_id: int):
item = await get_item(item_id)
return item
Pydantic（v2.4.2）Marshmallow（v3.19.0）
基本的なデータモデルの定義
Pydantic
from pydantic import BaseModel, Field
class Item(BaseModel):
id: int
name: str = Field(..., max_length=50)
price: float = Field(..., gt=0, description="商品の価格")
description: str | None = None
Marshmallow
from marshmallow import Schema, fields, validate
class ItemSchema(Schema):
id = fields.Int(required=True)
name = fields.Str(required=True, validate=validate.Length(max=50))
price = fields.Float(required=True, validate=validate.Range(min=0.01))
description = fields.Str(missing=None)
リクエストボディのバリデーション
Pydantic
from fastapi import FastAPI
app = FastAPI()
@app.post("/items/")
async def create_item(item: Item):
return {"message": "アイテムが作成されました", "item": item}
データのシリアライズとデシリアライズ
Marshmallow
item_schema = ItemSchema()
# デシリアライズ（バリデーション付き）
item_data = {"id": 1, "name": "Example Item", "price": 10.5}
item = item_schema.load(item_data) # バリデーション成功でPythonオブジェクトを取得
# シリアライズ
item_json = item_schema.dump(item)
ネストされたデータモデル
Pydantic
class User(BaseModel):
username: str
email: str
class Order(BaseModel):
order_id: int
user: User
items: list[Item]
Marshmallow
class UserSchema(Schema):
username = fields.Str(required=True)
email = fields.Email(required=True)
class OrderSchema(Schema):
order_id = fields.Int(required=True)
user = fields.Nested(UserSchema, required=True)
items = fields.List(fields.Nested(ItemSchema), required=True)
カスタムバリデーション
Pydantic
from pydantic import field_validator, ValidationError
class Item(BaseModel):
name: str
price: float
quantity: int
@field_validator("price")
def validate_price(cls, v):
if v <= 0:
raise ValueError("価格は0より大きくなければなりません")
return v
Marhmallow
from marshmallow import validates, ValidationError
class ItemSchema(Schema):
name = fields.Str(required=True)
price = fields.Float(required=True)
quantity = fields.Int(required=True)
@validates("price")
def validate_price(self, value):
if value <= 0:
raise ValidationError("価格は0より大きくなければなりません")
PyJWT（v2.8.0）
トークンの生成
import jwt
from datetime import datetime, timedelta
SECRET_KEY = "your_secret_key"
ALGORITHM = "HS256"
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
トークンのデコードと検証
from jwt import PyJWTError
def decode_access_token(token: str):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except PyJWTError:
return None
FastAPIと組み合わせたトークンベースの認証
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
async def get_current_user(token: str = Depends(oauth2_scheme)):
payload = decode_access_token(token)
if payload is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="トークンが無効です",
headers={"WWW-Authenticate": "Bearer"},
)
return payload
aiohttp（v3.8.5）
基本的な非同期GETリクエスト
import aiohttp
import asyncio
async def fetch_data(url: str):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
# 使用例
url = "https://api.example.com/data"
data = asyncio.run(fetch_data(url))
print(data)
非同期POSTリクエスト
async def post_data(url: str, payload: dict):
async with aiohttp.ClientSession() as session:
async with session.post(url, json=payload) as response:
return await response.json()
# 使用例
payload = {"key": "value"}
data = asyncio.run(post_data("https://api.example.com/post", payload))
print(data)
エラーハンドリング
async def fetch_with_error_handling(url: str):
async with aiohttp.ClientSession() as session:
try:
async with session.get(url) as response:
response.raise_for_status()
return await response.json()
except aiohttp.ClientError as e:
print(f"HTTPエラーが発生しました: {e}")
return None
python-dotenv（v1.0.0）
.envファイルの作成
# .envファイル
DATABASE_URL="postgresql://user:password@localhost/dbname"
SECRET_KEY="your_secret_key"
DEBUG=True
その後、pip install python-dotenv==1.0.0でライブラリインストール
環境変数の読み込み
import os
from dotenv import load_dotenv
# .envファイルの読み込み
load_dotenv()
# 環境変数の取得
database_url = os.getenv("DATABASE_URL")
secret_key = os.getenv("SECRET_KEY")
debug_mode = os.getenv("DEBUG") == "True" # Booleanに変換
print("Database URL:", database_url)
print("Secret Key:", secret_key)
print("Debug Mode:", debug_mode)
環境変数が存在しない場合のデフォルト値
# デフォルト値の設定
database_url = os.getenv("DATABASE_URL", "postgresql://user:password@localhost/defaultdb")
print("Database URL:", database_url)
FastAPIやFlaskとの組み合わせ
from fastapi import FastAPI
import os
from dotenv import load_dotenv
# .envファイルの読み込み
load_dotenv()
# 環境変数の取得
secret_key = os.getenv("SECRET_KEY")
app = FastAPI()
@app.get("/")
async def read_root():
return {"Secret Key": secret_key}
エラーハンドリング
# 必須の環境変数が設定されていない場合にエラーを表示
def get_env_var(var_name):
value = os.getenv(var_name)
if value is None:
raise EnvironmentError(f"{var_name} is not set in the environment.")
return value
# 使用例
database_url = get_env_var("DATABASE_URL")