FastAPI と SQLalchemy を使って、Authlib による OIDC 認証と MySQL にユーザーを登録する

はじめに

こんにちは！
今回はFastAPIでWebアプリケーションを構築する過程で、Authlibを利用してOpenID Connect（OIDC）認証を実装する方法と、SQLAlchemyを使用してMySQLデータベースへのユーザー登録を行うプロセスを共有します。

使用する技術

• FastAPI: 非同期Python Webフレームワーク
• Authlib: OIDCを始めとするOAuth系のライブラリ
• SQLAlchemy: PythonのORMツール
• MySQL: リレーショナルデータベース管理システム

OIDC認証の流れ

OpenID Connectはユーザーが認証を希望するサービスへ安全にアクセスするためのプロトコルです。ユーザーがアクセスすると、認証プロバイダにリダイレクトされ、認証後にはアプリケーションにトークンが返されます。このトークンにはユーザーの確認情報が含まれています。

FastAPIでのOIDC認証実装手順

1. 環境設定

   • 必要なパッケージのインストール

     pip install fastapi uvicorn authlib sqlalchemy
     

   • FastAPIアプリケーションの初期設定
     まず、プロジェクトのディレクトリを作成。
     次に、FastAPIアプリケーションの基本的な構造を作成します。
     main.pyという名前のファイルを作成し、以下のコードを記述します。

     # main.py
     
     from fastapi import FastAPI
     
     # FastAPIアプリケーションインスタンスを作成します。
     app = FastAPI()
     
     # ルートパスへのGETリクエストに対するエンドポイントを定義します。
     @app.get("/")
     async def read_root():
         return {"Hello": "World"}
     

     ここではFastAPIアプリケーションをインスタンス化し、ルートパス / に対する
     単純なGETリクエストを処理する最も基本的なAPIエンドポイントを定義しています。
     レスポンスとしてJSON形式の挨拶文を返しています。
     最後に、UVicornを使用してアプリケーションを実行します。以下はローカル開発環境でFastAPIアプリケーションを起動するためのコマンドです。

     uvicorn main:app --reload
     

     このコマンドは、main.py ファイルの app インスタンスを指定しており、--reload オプションは開発中にファイルが変更されたときに自動的にサーバーを再起動します。

     以上でFastAPIアプリケーションの初期設定は完了です。この状態で、ブラウザを開き http://127.0.0.1:8000 にアクセスすると、{"Hello": "World"} というレスポンスが表示されるはずです。これでFastAPIを使ったアプリケーション開発の基盤が整いました。

2. Authlibの統合

   • グローバル設定に必要な値を定義

     CONFIG_ENDPOINT= "endpoint-url"
     CLIENT_ID = "your-client-id"
     CLIENT_SECRET = "your-client-secret"
     REDIRECT_URI = "your-redirect-uri"
     

   • Authlibクライアントの設定

     from authlib.integrations.starlette_client import OAuth
     
     oauth = OAuth()
     oauth.register(
      name="auth0",
      server_metadata_url=config("CONFIG_ENDPOINT"),
      client_id=config("OIDC_CLIENT_ID"),
      client_secret=config("CLIENT_SECRET"),
      client_kwargs={
          "scope": "openid profile email",
          },
      )
     

   client_kwargsのscopeではユーザー情報のどの部分にアクセスする許可を要求するかを指定するためのものです。例えば、openid email profileがあります。

   • Callbackエンドポイントの実装

     from fastapi import FastAPI, Depends, HTTPException
     from starlette.responses import RedirectResponse
     
     app = FastAPI()
     
     @router.get("/api/login")
      async def login(request: Request):
          redirect_uri = config("REDIRECT_URL")
          return await oauth.auth0.authorize_redirect(request, redirect_uri)
      
      
      @router.get("/api/auth")
      async def auth(request: Request, db: AsyncSession = Depends(get_db)):
          try:
              token = await oauth.auth0.authorize_access_token(request)
          except OAuthError as e:
              print("An error occurred while verifying authorization response:", e)
              raise HTTPException(status_code=401)
      
          userinfo = token.get("userinfo")
          if not userinfo:
              raise ValueError()
      
          user_dict = dict(userinfo)
          sub = user_dict["sub"]
          user_name = user_dict["user_name"]
          email = user_dict["email"]
          user = await get_user_from_db(db, sub)
      
          if user is None:
              user = User(sub=sub, user_name=user_name, email=email)
              await user_signup(db, user)
      
          request.session["id_token"] = token.get("id_token")
          return RedirectResponse(url='/some-protected-page')
     

     今回は後ほど定義するget_user_from_db関数でユーザの登録有無を確認し、
     条件分岐し未登録であればuser_signup関数でDBへ登録するという実装にしました。

3. SQLAlchemyとMySQLの準備

   • データベース接続とテーブルのセットアップ

     from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
     from sqlalchemy.orm import sessionmaker, declarative_base
     
      ASYNC_DB_URL = "mysql+aiomysql://root:root@db:3306/vd"
      async_engine = create_async_engine(ASYNC_DB_URL, echo=True)
      async_session = sessionmaker(
          autocommit=False, autoflush=False, bind=async_engine, class_=AsyncSession
      )
     
      Base = declarative_base()
      async def get_db():
          async with async_session() as session:
              yield session
     

4. ユーザー登録処理の追加

   • ユーザーモデルとテーブルの定義
     schemas.pyでは下記のようにUserを定義しています。

     from sqlalchemy import Column, Integer, String
     from .database import Base
     
     
     class User(BaseModel):
         employee_Number: str = Field(max_length=12)
         user_name: str
         email: str
     
         class Config:
             from_attributes = True
     

   • データベースのテーブルとそれらのテーブルに対する操作を定義するためのmodel.pyを作成します。

     from sqlalchemy import Column, Integer, String
     from .database import Base
     
     
     class User(Base):
         __tablename__ = "users"
         
         sub = Column(Integer, primary_key=True, index=True)
         user_name = Column(String(100), nullable=False)
         email = Column(String(100), nullable=False)
     

   • ユーザーアカウントの作成と検証: MySQLへの接続

     from sqlalchemy.ext.asyncio import AsyncSession
     from sqlalchemy.future import select
     from . import model
     from .schemas import User
     
     async def user_signup(db: AsyncSession, user: User):
         db_user = model.User(
             sub=user.employee_Number, user_name=user.user_name, email=user.email
         )
         db.add(db_user)
         await db.commit()
         await db.refresh(db_user)
         return db_user
         
     async def get_user_from_db(db: AsyncSession, employee_Number: str):
         result = await db.execute(
             select(model.User).where(model.User.employee_Number == employee_Number)
             )
         user = result.scalars().first()
         return user
     

まとめ

Authlibを活用したOIDC認証と、SQLAlchemyを利用したMySQLデータベースへのユーザー情報の登録プロセスを作成しました。
あまり初学者向けの日本語記事が見受けられず、実装を頑張ったような気がしています。
気がしているだけで、実際はそのようなことはなかったのもしれません。

記憶というのはそういうものですよね。

スペシャルサンクス

GitHub Copilot様

参考リンク

• よくあるSPA+API構成でのOpenID Connectクライアント実装
• FastAPIの公式ドキュメント
• Authlibドキュメント
• SQLAlchemy公式ドキュメント
• MySQLドキュメント