✨

supabaseとFastAPI(バックエンド)

2025/03/30に公開

FastAPI

Supabase

tech

概要

api=FastAPI DB=supabase
これでアプリ作成をしようと思ったので備忘録作成
以前も FastAPI で認証機能を作成したが、今回は API として運用する
ORM やマイグレーションで使っていた技術の仕様をはっきり頭に入れておきたかったので、もう一度基本をアウトプットする

ORM とは

Object-Relational Mapping
データ設計を python で行うことができるオブジェクト指向言語でリレーションを記述できる

sqlalchemy

Schema: DB 構造

from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey

# メタデータオブジェクトの作成
metadata = MetaData()

# テーブル定義の例
users = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50)),
    Column('email', String(120), unique=True)
)

Types

from sqlalchemy import (
    Integer,    # 整数
    String,     # 文字列
    Text,       # 長いテキスト
    Boolean,    # 真偽値
    DateTime,   # 日時
    Float,      # 浮動小数点
    Numeric,    # 精度の高い数値
    Date,       # 日付
    Time        # 時間
)

SQL Expression Language
SQLAlchemy が提供する、Python コードを使って SQL クエリを構築するための言語/API

from sqlalchemy import select, and_, or_, desc
from sqlalchemy.sql import text

# テーブル定義の例
users = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    Column('age', Integer)
)

alembic
マイグレーション機能
Django からマイグレーションだけを切り出したような機能をしている

アプリ例

ディレクトリ構成

keikakun_api/
├── alembic/
│   ├── versions/
│   │   └── 2c90a84a1b40_create_staff.py
│   ├── env.py
│   └── script.py.mako
├── app/
│   ├── models/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   └── staff.py
│   └── __init__.py
└── alembic.ini

環境構築

開発環境 DB = "postgresql://ユーザー:パスワード@接続 URL/DB 名"

supabase

初期設定

supabase init

ログイン

supabase login

リモートの DB と接続

supabase link --project-ref <urlの一部>

docker + supabase

supabase start

supabase stop

migration + ORM

ORM
sqlalchemy
migration
alembic
初期化

alembic init alembic

alembic.ini

sqlalchemy.url = 'DB_URL'

モデル定義
models/OO.py(ex.staff)

from sqlalchemy import Column, String, Boolean, DateTime, Enum as SQLEnum
from sqlalchemy.dialects.postgresql import UUID
from app.models.base import Base

class StaffRole(enum.Enum):
    EMPLOYEE = "employee"
    MANAGER = "manager"
    ADMIN = "admin"

class Staff(Base):
    __tablename__ = "staff"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    # その他のカラム定義

models/base.py

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

SQLALCHEMY_DATABASE_URL = "postgresql://postgres:postgres@localhost:54322/postgres"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

Alembic の環境設定(alembic/env.py)

from app.models.base import Base
from app.models.staff import Staff

target_metadata = Base.metadata

マイグレーションの作成と実行

# マイグレーションファイルの生成
alembic revision --autogenerate -m "Create staff table"

# マイグレーションの実行
alembic upgrade head

マイグレーションファイル

def upgrade() -> None:
    staff_role = ENUM('employee', 'manager', 'admin', name='staff_role')
    staff_role.create(op.get_bind())

    op.create_table(
        'staff',
        sa.Column('id', UUID(), nullable=False),
        # その他のカラム定義
    )

def downgrade() -> None:
    op.drop_table('staff')
    sa.Enum(name='staff_role').drop(op.get_bind())

alembic について

sqlalchemy で作成した ORM をマイグレーションファイルに反映させる
env.py にモデルを記載
マイグレーションの手順を実行
ロールバック方法

FAILED: Target database is not up to date.

まず、現在のデータベースの状態を確認：

alembic current

履歴を確認：

alembic history

データベースを完全に巻き戻す（最初の状態に戻す）：

alembic downgrade base

または、特定のリビジョンまで巻き戻す場合：

alembic downgrade <revision_id>

その後、最新の状態まで更新：

alembic upgrade head

クラスメソッド　バリデーション　 sqlalchemy

ビジネスロジックの分離

現在作成中のアプリ

endpoints
crud 処理をまとめる
auth
管理者権限をまとめる(admin を持っているか)
models
DB 二定義するテーブル
schemas
crud 処理などで用いるフィールド　バリデーションを定義

keikakun_api/
├── app/
│   ├── api/
│   │   └── v1/
│   │       ├── api.py
│   │       └── endpoints/
│   │           ├── service_offices.py  # 事業所CRUD操作のエンドポイント
│   │           └── staff.py            # スタッフ管理のエンドポイント
│   ├── auth/
│   │   ├── dependencies.py             # 認証関連の依存関数
│   │   └── permissions.py              # 権限チェック関数
│   ├── core/
│   │   └── config.py                   # アプリケーション設定
│   ├── db/
│   │   ├── database.py                 # データベース接続設定
│   │   └── session.py                  # セッション管理
│   ├── exceptions.py                   # カスタム例外クラス
│   ├── models/                         # SQLAlchemyモデル
│   │   ├── base.py                     # 基本モデルクラス
│   │   ├── enums.py                    # 列挙型定義
│   │   ├── permission.py               # 権限モデル
│   │   ├── role_change_request.py      # 役割変更リクエストモデル
│   │   ├── service_office.py           # 事業所モデル
│   │   └── staff.py                    # スタッフモデル
│   └── schemas/                        # Pydanticスキーマ
│       ├── service_office.py           # 事業所スキーマ
│       └── staff.py                    # スタッフスキーマ
├── migration/                          # データベースマイグレーション
│   └── env.py                          # Alembic環境設定
├── tests/                              # テストディレクトリ
│   ├── api/
│   │   └── v1/
│   │       └── test_service_offices.py # API エンドポイントテスト
│   ├── test_db_session.py              # テスト用DBセッション設定
│   └── test_service_office.py          # モデルレベルのテスト
└── README.md                           # プロジェクト説明

認証

Nextjs supabase
middleware.ts -> auth/callback

開発環境

fornt
localhost:3000
backend
localhost:8000
db
supabase
localhost:54322

エラーハンドリング
原因の切り分け

GitHubで編集を提案

概要