FastAPI Users

認証ライブラリPythonFastAPIJWTOAuthユーザー管理セキュリティ

認証ライブラリ

FastAPI Users

概要

FastAPI Usersは、FastAPI専用の包括的な認証・ユーザー管理ライブラリです。2025年のFastAPI急速普及に伴い、非同期アプリケーション開発のスタンダードとして大幅に成長しています。JWT、Cookie、OAuth認証を統合サポートし、完全非同期対応による高性能とプラガブルな設計により、モダンAPI開発における認証標準として確立されています。SQLAlchemy、Beanieなどの主要ORMとの統合により、エンタープライズレベルのユーザー管理システムを迅速に構築できます。

詳細

FastAPI Usersは、FastAPIアプリケーション向けの完全なユーザー管理エコシステムです。完全非同期対応により高性能を実現し、プラガブルなアーキテクチャによって柔軟なカスタマイズが可能です。

技術的特徴

完全非同期対応: FastAPIのAsync/Await機構を最大活用した高性能処理
複数認証バックエンド: JWT、Cookie、Database、Redis戦略の組み合わせ対応
プラガブル設計: Transport（トークン伝送方法）とStrategy（トークン生成方式）の分離
OAuth2統合: Google、GitHub等の主要プロバイダーとの簡単連携
データベース非依存: SQLAlchemy、Beanie（MongoDB）等複数ORM対応
OpenAPIスキーマ: 自動ドキュメント生成とSwagger UI統合

2025年における位置づけ

FastAPIの普及加速により、FastAPI Usersは非同期Web開発のデファクトスタンダードとなっています。現代的なマイクロサービスアーキテクチャや、リアルタイム通信を要求するアプリケーションに最適化された設計です。

メリット・デメリット

メリット

開発効率: 即使用可能なルートとコンポーネントで開発加速
高性能: 完全非同期処理による優秀なスループット
柔軟性: モジュラー設計による高いカスタマイズ性
セキュリティ: 業界標準のセキュリティプラクティス内蔵
スケーラビリティ: 複数認証戦略による拡張性
TypeScript連携: FastAPIの型ヒント活用によるタイプセーフティ
エコシステム: 豊富なサンプルコードと活発なコミュニティ

デメリット

学習コスト: FastAPI固有の設計パターン習得が必要
依存関係: FastAPIエコシステムへの強い依存
設定複雑度: 高度な機能活用時の設定複雑性
JWT制限: JWTトークンのサーバーサイド無効化不可
機能過多: 小規模プロジェクトには複雑すぎる場合がある

参考ページ

書き方の例

基本的なインストールとセットアップ

# 基本インストール（SQLAlchemy バックエンド付き）
pip install fastapi-users[sqlalchemy]

# OAuth機能付きでインストール
pip install fastapi-users[sqlalchemy,oauth]

# Beanie（MongoDB）バックエンド付きでインストール
pip install fastapi-users[beanie]

# その他の依存関係
pip install uvicorn[standard] aiosqlite

基本的なユーザーモデルとデータベース設定

# db.py - SQLAlchemyによるユーザーモデル定義
import uuid
from typing import AsyncGenerator

from fastapi import Depends
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(SQLAlchemyBaseUserTableUUID, Base):
    """ユーザーテーブルの定義"""
    # 基本フィールドは SQLAlchemyBaseUserTableUUID から継承
    # id: UUID (Primary Key)
    # email: str (Unique)
    # hashed_password: str
    # is_active: bool
    # is_superuser: bool
    # is_verified: bool
    
    # カスタムフィールドの追加
    first_name: Mapped[str] = mapped_column(String(50), nullable=True)
    last_name: Mapped[str] = mapped_column(String(50), nullable=True)
    phone_number: Mapped[str] = mapped_column(String(20), nullable=True)


# データベース設定
DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_async_engine(DATABASE_URL)
async_session_maker = async_sessionmaker(engine, expire_on_commit=False)


async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        yield session


async def get_user_db(session: AsyncSession = Depends(get_async_session)):
    yield SQLAlchemyUserDatabase(session, User)

Pydanticスキーマの定義

# schemas.py - リクエスト/レスポンススキーマ定義
import uuid
from typing import Optional

from fastapi_users import schemas


class UserRead(schemas.BaseUser[uuid.UUID]):
    """ユーザー情報読み取り用スキーマ"""
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    phone_number: Optional[str] = None


class UserCreate(schemas.BaseUserCreate):
    """ユーザー作成用スキーマ"""
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    phone_number: Optional[str] = None


class UserUpdate(schemas.BaseUserUpdate):
    """ユーザー更新用スキーマ"""
    first_name: Optional[str] = None
    last_name: Optional[str] = None
    phone_number: Optional[str] = None

ユーザーマネージャーと認証バックエンドの設定

# users.py - ユーザーマネージャーと認証設定
import uuid
from typing import Optional

from fastapi import Depends, Request
from fastapi_users import BaseUserManager, UUIDIDMixin
from fastapi_users.authentication import (
    AuthenticationBackend,
    BearerTransport,
    CookieTransport,
    JWTStrategy,
)

from .db import User, get_user_db


SECRET = "your-secret-key-here"  # 本番では環境変数を使用


class UserManager(UUIDIDMixin, BaseUserManager[User, uuid.UUID]):
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

    async def on_after_register(self, user: User, request: Optional[Request] = None):
        print(f"ユーザー {user.id} が登録されました。")

    async def on_after_forgot_password(
        self, user: User, token: str, request: Optional[Request] = None
    ):
        print(f"ユーザー {user.id} がパスワードリセットを要求しました。トークン: {token}")

    async def on_after_request_verify(
        self, user: User, token: str, request: Optional[Request] = None
    ):
        print(f"ユーザー {user.id} の認証トークン: {token}")

    async def validate_password(
        self,
        password: str,
        user: User | UserCreate,
    ) -> None:
        if len(password) < 8:
            raise ValueError("パスワードは8文字以上である必要があります")
        if not any(char.isdigit() for char in password):
            raise ValueError("パスワードには少なくとも1つの数字を含める必要があります")


async def get_user_manager(user_db=Depends(get_user_db)):
    yield UserManager(user_db)


# 認証バックエンド設定

# JWT Bearer認証
bearer_transport = BearerTransport(tokenUrl="auth/jwt/login")

def get_jwt_strategy() -> JWTStrategy:
    return JWTStrategy(secret=SECRET, lifetime_seconds=3600)

jwt_backend = AuthenticationBackend(
    name="jwt",
    transport=bearer_transport,
    get_strategy=get_jwt_strategy,
)

# Cookie認証
cookie_transport = CookieTransport(cookie_max_age=3600)

cookie_backend = AuthenticationBackend(
    name="cookie",
    transport=cookie_transport,
    get_strategy=get_jwt_strategy,
)

# 複数バックエンドを使用
auth_backends = [jwt_backend, cookie_backend]

FastAPIアプリケーションのセットアップ

# app.py - メインアプリケーション
import uuid
from fastapi import FastAPI, Depends
from fastapi_users import FastAPIUsers

from .db import User, create_db_and_tables
from .schemas import UserCreate, UserRead, UserUpdate
from .users import auth_backends, get_user_manager

# FastAPIUsers インスタンス
fastapi_users = FastAPIUsers[User, uuid.UUID](
    get_user_manager,
    auth_backends,
)

# 現在のユーザー取得関数
current_active_user = fastapi_users.current_user(active=True)
current_superuser = fastapi_users.current_user(active=True, superuser=True)

app = FastAPI(title="FastAPI Users Example")

# 認証ルーター
app.include_router(
    fastapi_users.get_auth_router(jwt_backend),
    prefix="/auth/jwt",
    tags=["auth"],
)

app.include_router(
    fastapi_users.get_auth_router(cookie_backend),
    prefix="/auth/cookie",
    tags=["auth"],
)

# ユーザー登録ルーター
app.include_router(
    fastapi_users.get_register_router(UserRead, UserCreate),
    prefix="/auth",
    tags=["auth"],
)

# パスワードリセットルーター
app.include_router(
    fastapi_users.get_reset_password_router(),
    prefix="/auth",
    tags=["auth"],
)

# メール認証ルーター
app.include_router(
    fastapi_users.get_verify_router(UserRead),
    prefix="/auth",
    tags=["auth"],
)

# ユーザー管理ルーター
app.include_router(
    fastapi_users.get_users_router(UserRead, UserUpdate),
    prefix="/users",
    tags=["users"],
)


@app.on_event("startup")
async def on_startup():
    # データベースとテーブルを作成
    await create_db_and_tables()


# 保護されたルートの例
@app.get("/protected")
async def protected_route(user: User = Depends(current_active_user)):
    return f"こんにちは、{user.email}さん！"


@app.get("/admin")
async def admin_route(user: User = Depends(current_superuser)):
    return f"管理者 {user.email} としてログインしています。"

OAuth プロバイダーとの統合

# oauth.py - OAuth設定
from httpx_oauth.clients.google import GoogleOAuth2

google_oauth_client = GoogleOAuth2(
    client_id="YOUR_GOOGLE_CLIENT_ID",
    client_secret="YOUR_GOOGLE_CLIENT_SECRET",
)

# OAuth ルーターの追加 (app.py内)
app.include_router(
    fastapi_users.get_oauth_router(
        google_oauth_client,
        jwt_backend,
        SECRET,
        is_verified_by_default=True,  # OAuth ユーザーは自動的に認証済み
        associate_by_email=True,      # 同じメールアドレスのアカウントを関連付け
    ),
    prefix="/auth/google",
    tags=["auth"],
)

# 既存ユーザーがOAuthアカウントを関連付けるためのルーター
app.include_router(
    fastapi_users.get_oauth_associate_router(google_oauth_client, UserRead, SECRET),
    prefix="/auth/associate/google",
    tags=["auth"],
)

高度なユーザー管理とカスタマイズ

# advanced_users.py - 高度なユーザー管理
import uuid
from typing import Optional
from fastapi import Request, HTTPException, status
from fastapi_users import BaseUserManager
from fastapi_users.password import PasswordHelper

from .db import User
from .schemas import UserCreate


class AdvancedUserManager(BaseUserManager[User, uuid.UUID]):
    reset_password_token_secret = SECRET
    verification_token_secret = SECRET

    def __init__(self, user_db):
        super().__init__(user_db)
        # カスタムパスワードヘルパー
        self.password_helper = PasswordHelper()

    async def create(
        self,
        user_create: UserCreate,
        safe: bool = False,
        request: Optional[Request] = None,
    ) -> User:
        """カスタムユーザー作成ロジック"""
        # メールドメインの制限
        allowed_domains = ["company.com", "trusted-partner.com"]
        email_domain = user_create.email.split("@")[1]
        
        if email_domain not in allowed_domains:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="許可されていないメールドメインです"
            )

        # ユーザー作成前の処理
        user = await super().create(user_create, safe, request)
        
        # ユーザー作成後の処理（例：デフォルトロールの割り当て）
        # await self.assign_default_role(user)
        
        return user

    async def on_after_register(self, user: User, request: Optional[Request] = None):
        """ユーザー登録後の処理"""
        print(f"新規ユーザー登録: {user.email}")
        # メール送信、ログ記録、外部システム連携など
        # await self.send_welcome_email(user)
        # await self.log_user_registration(user)

    async def on_after_login(
        self,
        user: User,
        request: Optional[Request] = None,
        response: Optional[Response] = None,
    ):
        """ログイン後の処理"""
        print(f"ユーザーログイン: {user.email}")
        # ログイン履歴の記録、セッション管理など
        # await self.update_last_login(user)

    async def on_after_update(
        self,
        user: User,
        update_dict: dict,
        request: Optional[Request] = None,
    ):
        """ユーザー情報更新後の処理"""
        print(f"ユーザー情報更新: {user.email}")
        # 変更通知、監査ログなど

    async def validate_password(
        self,
        password: str,
        user: User | UserCreate,
    ) -> None:
        """カスタムパスワード検証"""
        if len(password) < 12:
            raise ValueError("パスワードは12文字以上である必要があります")
        
        if not any(char.isupper() for char in password):
            raise ValueError("パスワードには大文字を含める必要があります")
        
        if not any(char.islower() for char in password):
            raise ValueError("パスワードには小文字を含める必要があります")
        
        if not any(char.isdigit() for char in password):
            raise ValueError("パスワードには数字を含める必要があります")
        
        if not any(char in "!@#$%^&*()_+-=[]{}|;:,.<>?" for char in password):
            raise ValueError("パスワードには特殊文字を含める必要があります")

    async def authenticate(
        self, credentials: OAuth2PasswordRequestForm
    ) -> Optional[User]:
        """カスタム認証ロジック"""
        try:
            user = await self.get_by_email(credentials.username)
        except UserNotExists:
            # タイミング攻撃対策のためのダミー検証
            self.password_helper.hash("dummy")
            return None

        # アカウントロック機能の例
        if hasattr(user, 'failed_login_attempts') and user.failed_login_attempts >= 5:
            raise HTTPException(
                status_code=status.HTTP_423_LOCKED,
                detail="アカウントがロックされています"
            )

        verified, updated_password_hash = self.password_helper.verify_and_update(
            credentials.password, user.hashed_password
        )
        
        if not verified:
            # ログイン失敗回数を増加
            # await self.increment_failed_login_attempts(user)
            return None
        
        # ログイン成功時はカウンターをリセット
        # await self.reset_failed_login_attempts(user)
        
        if updated_password_hash is not None:
            await self.user_db.update(user, {"hashed_password": updated_password_hash})

        return user

動的認証バックエンドとセキュリティ設定

# security.py - セキュリティとアクセス制御
from fastapi import Request, Depends, HTTPException, status
from fastapi_users.authentication import AuthenticationBackend

async def get_enabled_backends(request: Request):
    """リクエストに基づいて有効な認証バックエンドを動的に決定"""
    
    # 管理者向けエンドポイントはJWTのみ
    if request.url.path.startswith("/admin"):
        return [jwt_backend]
    
    # API エンドポイントはJWTのみ
    if request.url.path.startswith("/api"):
        return [jwt_backend]
    
    # その他はCookieとJWTの両方を許可
    return [cookie_backend, jwt_backend]


# 動的認証の使用例
current_user_dynamic = fastapi_users.current_user(
    active=True, 
    get_enabled_backends=get_enabled_backends
)

@app.get("/admin/dashboard")
async def admin_dashboard(user: User = Depends(current_user_dynamic)):
    if not user.is_superuser:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="管理者権限が必要です"
        )
    return {"message": f"管理者ダッシュボード - {user.email}"}


# ロールベースアクセス制御の実装
def require_roles(*required_roles: str):
    """指定されたロールを要求するデコレーター"""
    def role_checker(user: User = Depends(current_active_user)):
        # ユーザーのロールを取得（カスタム実装が必要）
        user_roles = getattr(user, 'roles', [])
        
        if not any(role in user_roles for role in required_roles):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"必要なロール: {', '.join(required_roles)}"
            )
        return user
    
    return role_checker


@app.get("/editor/content")
async def editor_content(user: User = Depends(require_roles("editor", "admin"))):
    return {"message": "編集者専用コンテンツ"}