Pythonを使用した高性能アプリ–FastAPIチュートリアル

公開: 2022-03-11

優れたプログラミング言語フレームワークにより、高品質の製品をより迅速に作成することが容易になります。優れたフレームワークにより、開発エクスペリエンス全体が楽しくなります。 FastAPIは、強力で楽しく使用できる新しいPythonWebフレームワークです。次の機能により、FastAPIを試す価値があります。

速度：FastAPIは、最速のPythonWebフレームワークの1つです。実際、その速度はNode.jsやGoと同等です。これらのパフォーマンステストを確認してください。
詳細で使いやすい開発者向けドキュメント
コードにヒントを入力して、無料のデータ検証と変換を取得します。
依存性注入を使用してプラグインを簡単に作成します。

TODOアプリの構築

FastAPIの背後にある大きなアイデアを探求するために、ユーザーのToDoリストを設定するTODOアプリを作成しましょう。私たちの小さなアプリは、次の機能を提供します。

サインアップとログイン
新しいTODOアイテムを追加する
すべてのTODOのリストを取得する
TODOアイテムの削除/更新

データモデルのSQLAlchemy

私たちのアプリには、ユーザーとTODOの2つのモデルしかありません。 Python用のデータベースツールキットであるSQLAlchemyを使用すると、次のようにモデルを表現できます。

 class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) lname = Column(String) fname = Column(String) email = Column(String, unique=True, index=True) todos = relationship("TODO", back_populates="owner", cascade="all, delete-orphan") class TODO(Base): __tablename__ = "todos" id = Column(Integer, primary_key=True, index=True) text = Column(String, index=True) completed = Column(Boolean, default=False) owner_id = Column(Integer, ForeignKey("users.id")) owner = relationship("User", back_populates="todos")

モデルの準備ができたら、SQLAlchemyの構成ファイルを作成して、データベースとの接続を確立する方法を認識できるようにします。

 import os from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = os.environ['SQLALCHEMY_DATABASE_URL'] engine = create_engine( SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

タイプヒントの力を解き放つ

APIプロジェクトのかなりの部分は、データの検証や変換などの日常的な作業に関係しています。リクエストハンドラーの作成に進む前に、前もってそれに取り組みましょう。 FastAPIでは、pydanticモデルを使用して着信/発信データのスキーマを表現し、これらのpydanticモデルを使用してヒントを入力し、無料のデータ検証と変換を楽しんでいます。これらのモデルはデータベースワークフローとは関係がなく、RESTインターフェースに出入りするデータの形状のみを指定することに注意してください。 pydanticモデルを作成するには、ユーザーとTODOの情報が出入りするすべての方法を考えてください。

従来、新しいユーザーがTODOサービスにサインアップし、既存のユーザーがログインします。これらの対話は両方ともユーザー情報を処理しますが、データの形状は異なります。サインアップ時にはユーザーからのより多くの情報が必要であり、ログイン時には最小限（電子メールとパスワードのみ）が必要です。これは、これら2つの異なる形状のユーザー情報を表現するために2つのpydanticモデルが必要であることを意味します。

ただし、TODOアプリでは、FastAPIに組み込まれているOAuth2サポートを利用して、JSON Web Token（JWT）ベースのログインフローを実現します。ここでUserCreateスキーマを定義して、サインアップエンドポイントに流れるデータと、サインアッププロセスが成功した場合に応答として返すUserBaseスキーマを指定する必要があります。

 from pydantic import BaseModel from pydantic import EmailStr class UserBase(BaseModel): email: EmailStr class UserCreate(UserBase): lname: str fname: str password: str

ここでは、姓、名、パスワードを文字列としてマークしましたが、最小長、最大長、正規表現などのチェックを可能にするpydantic制約付き文字列を使用することで、さらに厳密にすることができます。

TODOアイテムの作成とリストをサポートするために、次のスキーマを定義します。

 class TODOCreate(BaseModel): text: str completed: bool

既存のTODOアイテムの更新をサポートするために、別のスキーマを定義します。

 class TODOUpdate(TODOCreate): id: int

これで、すべてのデータ交換のスキーマの定義が完了しました。ここで、これらのスキーマを使用してデータ変換と検証のすべての面倒な作業を無料で行うリクエストハンドラーに注目します。

ユーザーにサインアップさせる

まず、認証されたユーザーがすべてのサービスにアクセスする必要があるため、ユーザーがサインアップできるようにします。上記で定義したUserCreateおよびUserBaseスキーマを使用して、最初のリクエストハンドラーを記述します。

 @app.post("/api/users", response_model=schemas.User) def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)): """add new user""" user = crud.get_user_by_email(db, user_data.email) if user: raise HTTPException(status_code=409, detail="Email already registered.") signedup_user = crud.create_user(db, user_data) return signedup_user

この短いコードには多くのことが起こっています。デコレータを使用して、HTTP動詞、URI、および成功した応答のスキーマを指定しました。ユーザーが正しいデータを送信したことを確認するために、以前に定義されたUserCreateスキーマを使用してリクエスト本文にヒントを入力しました。このメソッドは、データベースのハンドルを取得するための別のパラメーターを定義します。これは、実際の依存性注入であり、このチュートリアルの後半で説明します。

APIの保護

アプリには次のセキュリティ機能が必要です。

パスワードハッシュ
JWTベースの認証

パスワードハッシュには、Passlibを使用できます。パスワードのハッシュとパスワードが正しいかどうかのチェックを処理する関数を定義しましょう。

 from passlib.context import CryptContext pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def authenticate_user(db, email: str, password: str): user = crud.get_user_by_email(db, email) if not user: return False if not verify_password(password, user.hashed_password): return False return user

JWTベースの認証を有効にするには、JWTを生成し、デコードしてユーザーの資格情報を取得する必要があります。この機能を提供するために、次の関数を定義します。

 # install PyJWT import jwt from fastapi.security import OAuth2PasswordBearer SECRET_KEY = os.environ['SECRET_KEY'] ALGORITHM = os.environ['ALGORITHM'] def create_access_token(*, data: dict, expires_delta: timedelta = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def decode_access_token(db, token): credentials_exception = HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) email: str = payload.get("sub") if email is None: raise credentials_exception token_data = schemas.TokenData(email=email) except PyJWTError: raise credentials_exception user = crud.get_user_by_email(db, email=token_data.email) if user is None: raise credentials_exception return user

ログイン成功時にトークンを発行する

次に、ログインエンドポイントを定義し、OAuth2パスワードフローを実装します。このエンドポイントは、電子メールとパスワードを受け取ります。データベースに対してクレデンシャルをチェックし、成功すると、ユーザーにJSONWebトークンを発行します。

資格情報を受け取るために、FastAPIのセキュリティユーティリティの一部であるOAuth2PasswordRequestFormを利用します。

 @app.post("/api/token", response_model=schemas.Token) def login_for_access_token(db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()): """generate access token for valid credentials""" user = authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException( status_code=HTTP_401_UNAUTHORIZED, detail="Incorrect email or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token(data={"sub": user.email}, expires_delta=access_token_expires) return {"access_token": access_token, "token_type": "bearer"}

依存性注入を使用してDBにアクセスし、エンドポイントを保護する

ログインに成功すると、ユーザーにJWTを提供するログインエンドポイントを設定しました。ユーザーはこのトークンをローカルストレージに保存し、Authorizationヘッダーとしてバックエンドに表示できます。ログインしたユーザーからのアクセスのみを期待するエンドポイントは、トークンをデコードして、リクエスターが誰であるかを確認できます。この種の作業は特定のエンドポイントに関連付けられているのではなく、保護されているすべてのエンドポイントで使用される共有ロジックです。トークンデコードロジックを、任意のリクエストハンドラーで使用できる依存関係として設定することをお勧めします。

FastAPIと言えば、パス操作関数（リクエストハンドラー）はget_current_userに依存します。 get_current_user依存関係は、データベースに接続し、FastAPIのOAuth2PasswordBearerロジックにフックしてトークンを取得する必要があります。 get_current_userを他の関数に依存させることで、この問題を解決します。このようにして、非常に強力な概念である依存関係チェーンを定義できます。

 def get_db(): """provide db session to path operation functions""" try: db = SessionLocal() yield db finally: db.close() def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)): return decode_access_token(db, token) @app.get("/api/me", response_model=schemas.User) def read_logged_in_user(current_user: models.User = Depends(get_current_user)): """return user settings for current user""" return current_user

ログインしたユーザーはTODOをCRUDできます

TODO Create、Read、Update、Delete（CRUD）のパス操作関数を作成する前に、データベースで実際のCRUDを実行するための次のヘルパー関数を定義します。

 def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate): todo = models.TODO(text=todo_data.text, completed=todo_data.completed) todo.owner = current_user db.add(todo) db.commit() db.refresh(todo) return todo def update_todo(db: Session, todo_data: schemas.TODOUpdate): todo = db.query(models.TODO).filter(models.TODO.id == id).first() todo.text = todo_data.text todo.completed = todo.completed db.commit() db.refresh(todo) return todo def delete_todo(db: Session, id: int): todo = db.query(models.TODO).filter(models.TODO.id == id).first() db.delete(todo) db.commit() def get_user_todos(db: Session, userid: int): return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()

これらのdbレベルの関数は、次のRESTエンドポイントで使用されます。

 @app.get("/api/mytodos", response_model=List[schemas.TODO]) def get_own_todos(current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """return a list of TODOs owned by current user""" todos = crud.get_user_todos(db, current_user.id) return todos @app.post("/api/todos", response_model=schemas.TODO) def add_a_todo(todo_data: schemas.TODOCreate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """add a TODO""" todo = crud.create_meal(db, current_user, meal_data) return todo @app.put("/api/todos/{todo_id}", response_model=schemas.TODO) def update_a_todo(todo_id: int, todo_data: schemas.TODOUpdate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """update and return TODO for given id""" todo = crud.get_todo(db, todo_id) updated_todo = crud.update_todo(db, todo_id, todo_data) return updated_todo @app.delete("/api/todos/{todo_id}") def delete_a_meal(todo_id: int, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)): """delete TODO of given id""" crud.delete_meal(db, todo_id) return {"detail": "TODO Deleted"}

テストを書く

TODOAPIのテストをいくつか書いてみましょう。 FastAPIは、人気のあるRequestsライブラリに基づくTestClientクラスを提供し、Pytestを使用してテストを実行できます。

ログインしたユーザーだけがTODOを作成できるようにするために、次のように記述できます。

 from starlette.testclient import TestClient from .main import app client = TestClient(app) def test_unauthenticated_user_cant_create_todos(): todo=dict(text="run a mile", completed=False) response = client.post("/api/todos", data=todo) assert response.status_code == 401

次のテストでは、ログインエンドポイントをチェックし、有効なログインクレデンシャルが提示された場合はJWTを生成します。

 def test_user_can_obtain_auth_token(): response = client.post("/api/token", data=good_credentials) assert response.status_code == 200 assert 'access_token' in response.json() assert 'token_type' in response.json()

まとめる

FastAPIを使用して非常に単純なTODOアプリの実装が完了しました。これまでに、RESTインターフェースを介して着信データと発信データの形状を定義する際に型ヒントの力がうまく利用されていることを確認しました。スキーマを1つの場所で定義し、それをFastAPIに任せて、データの検証と変換を適用します。他の注目すべき機能は、依存性注入です。この概念を使用して、データベース接続を取得し、JWTをデコードして現在ログインしているユーザーを取得し、パスワードとベアラーを使用して単純なOAuth2を実装する共有ロジックをパッケージ化しました。また、依存関係をどのように連鎖させることができるかを見ました。

この概念を簡単に適用して、役割ベースのアクセスなどの機能を追加できます。その上、私たちはフレームワークの特性を学ぶことなく、簡潔で強力なコードを書いています。簡単に言うと、FastAPIは、最新のPythonであるため、学習する必要のない強力なツールのコレクションです。楽しむ。