はじめに
こんにちは！
今回はFastAPIでWebアプリケーションを構築する過程で、Authlibを利用してOpenID Connect（OIDC）認証を実装する方法と、SQLAlchemyを使用してMySQLデータベースへのユーザー登録を行うプロセスを共有します。
使用する技術
- FastAPI: 非同期Python Webフレームワーク
- Authlib: OIDCを始めとするOAuth系のライブラリ
- SQLAlchemy: PythonのORMツール
- MySQL: リレーショナルデータベース管理システム
OIDC認証の流れ
OpenID Connectはユーザーが認証を希望するサービスへ安全にアクセスするためのプロトコルです。ユーザーがアクセスすると、認証プロバイダにリダイレクトされ、認証後にはアプリケーションにトークンが返されます。このトークンにはユーザーの確認情報が含まれています。
FastAPIでのOIDC認証実装手順
-
環境設定
-
必要なパッケージのインストール
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を使ったアプリケーション開発の基盤が整いました。
-
-
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へ登録するという実装にしました。
- グローバル設定に必要な値を定義
-
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
- データベース接続とテーブルのセットアップ
-
ユーザー登録処理の追加
- ユーザーモデルとテーブルの定義
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データベースへのユーザー情報の登録プロセスを作成しました。
あまり初学者向けの日本語記事が見受けられず、実装を頑張ったような気がしています。
気がしているだけで、実際はそのようなことはなかったのもしれません。
記憶というのはそういうものですよね。