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. 環境設定
   - 必要なパッケージのインストール
        ```bash
        pip install fastapi uvicorn authlib sqlalchemy
        ```
    - FastAPIアプリケーションの初期設定
    まず、プロジェクトのディレクトリを作成。
    次に、FastAPIアプリケーションの基本的な構造を作成します。
    `main.py`という名前のファイルを作成し、以下のコードを記述します。
        ```python
        # 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アプリケーションを起動するためのコマンドです。
         ```bash
        uvicorn main:app --reload
        ```
        このコマンドは、`main.py` ファイルの `app` インスタンスを指定しており、`--reload` オプションは開発中にファイルが変更されたときに自動的にサーバーを再起動します。
        
        以上でFastAPIアプリケーションの初期設定は完了です。この状態で、ブラウザを開き `http://127.0.0.1:8000` にアクセスすると、`{"Hello": "World"}` というレスポンスが表示されるはずです。これでFastAPIを使ったアプリケーション開発の基盤が整いました。


2. Authlibの統合
   - グローバル設定に必要な値を定義
       ```python
       CONFIG_ENDPOINT= "endpoint-url"
       CLIENT_ID = "your-client-id"
       CLIENT_SECRET = "your-client-secret"
       REDIRECT_URI = "your-redirect-uri"
       ```
   - Authlibクライアントの設定
       ```python
       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エンドポイントの実装
       ```python
       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の準備
   - データベース接続とテーブルのセットアップ
       ```python
       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`を定義しています。
        ```python
        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を作成します。
       ```python
       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への接続
        ```python
        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クライアント実装](https://sogo.dev/posts/2022/12/openid-connect-fastapi)
- [FastAPIの公式ドキュメント](https://fastapi.tiangolo.com/)
- [Authlibドキュメント](https://docs.authlib.org/)
- [SQLAlchemy公式ドキュメント](https://www.sqlalchemy.org/)
- [MySQLドキュメント](https://dev.mysql.com/doc/)