【備忘】よくつかうコードスニペット: Python外部ライブラリ系

## 目的

コードを書く機会はそれなりに多いが、日ごろから持ち前の鳥頭が存分に活かされすぎているため書き方を一瞬で忘れる。
そのため、仕事用端末からもプライベートの端末からも閲覧できる形でスニペットとして残しておきたい
今のところ、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")

```