FastAPI のリポジトリ パターンを使用した効率的なセッション処理

FastAPI を使用したリポジトリ パターンの適応に関する記事を約 5 か月前に書きました。そして、たくさんの読者を獲得しました (ありがとうございます)。私は、引き続きリポジトリ パターンを使用して、セッション処理を処理する効率的な方法について書きに来ました。

本題に入る前に、本番環境で API がデータベースの読み取りまたは書き込みを伴うトランザクションを実行しようとすると、次のいずれかのエラーが発生することに気付きました。

• sqlalchemy.exc.PendingRollbackError
• sqlalchemy.exc.InvalidRequestError

sqlalchemy.exc.PendingRollbackError

このエラーは、他のデータベース操作を続行する前にロールバックする必要がある、コミットされていないトランザクションが進行中であることを示します。

このエラーの最も一般的な原因は、データベース トランザクション中に発生する未処理の例外により、トランザクションが適切にコミットまたはロールバックされなくなることです。

sqlalchemy.exc.InvalidRequestError

このエラーは、データベースに対して行った操作またはリクエストが無効であるか、サポートされていないことを示します。このエラーには、次のようなさまざまな原因が考えられます。

• SQLAlchemy API の不適切な使用法 (メソッドの呼び出しや、存在しない、または指定されたコンテキストに適用できない属性へのアクセスなど)。

• SQL 構文またはクエリ構造が正しくありません。

• Python オブジェクトとデータベースのテーブル/列の間のマッピングが欠落しているか、正しくありません。

これらのエラーをどのように解決できるかについて考えがあることは確かですが、問題の内容を特定して修正したにもかかわらず、問題は解決しません。

トラブルシューティングと解決方法に興味がある場合は、次の手順に従うことを検討してください。

sqlalchemy.exc.PendingRollbackエラー:

• コードを確認して、トランザクションの適切なコミットまたはロールバックを妨げる可能性のある未処理の例外がないか確認してください。例外を適切に処理し、必要に応じてトランザクションをコミットまたはロールバックしていることを確認してください。

• 保留中のロールバックにつながる可能性のある、長時間実行されているトランザクションまたはネストされたトランザクションがないか確認してください。すべてのトランザクションが適時にコミットまたはロールバックされるようにしてください。

• 前のトランザクションを適切に処理する前に、新しいトランザクションを開始する可能性があるかどうかコードを確認してください。

sqlalchemy.exc.InvalidRequestError:

• 例外とともに提供される特定のエラー メッセージを確認して、無効なリクエストの原因を特定します。コードまたは SQL ステートメントのどの部分が問題の原因となっているかについてのヒントが得られる場合があります。

• SQLAlchemy コードに不正なメソッド呼び出し、属性アクセス、または API の不正な使用法がないか確認してください。

• SQL ステートメントを見直して、構文と構造が正しいことを確認してください。

• データベース スキーマと SQLAlchemy マッピングが同期していることを確認します。必要なテーブルと列がすべて存在し、Python オブジェクトが対応するデータベース エンティティに適切にマップされていることを確認してください。

私がどのようにして効果があることが証明された永続的な解決策に取り組んだのかを見ていきましょう。リポジトリ パターンの使用方法をデモしていたときに取り組んだプロジェクトの使用に進みます。

次のコードを含む orm ベース セッション ミックスインを保存するモジュールがありました。

 # SQLAlchemy Imports from sqlalchemy.orm import Session # Own Imports from config.database import SessionLocal from core.settings import ledger_settings class ORMSessionMixin: """Base orm session mixin for interacting with the database.""" def __init__(self): """ Get the next database session from the database pool. """ self.orm: Session = self.get_db().__next__() def get_db(self): """ This method creates a database session, yields it, rollback the transaction if there's an exception and then finally closes the session. Yields: db: scoped database session """ db = SessionLocal() try: yield db except Exception: db.rollback() finally: db.close()

このソリューションの問題は、トランザクションのプロセス (ユーザーの作成、ウォレットへの資金提供など、あらゆるプロセスが考えられます) で例外が発生した場合、例外が適切に処理されず、転送中のデータベース セッションが処理されないことでした。ロールバックを取得します。

3 か月にわたるデバッグとパッチ適用、および多くの研究を経て、ついにセッションを処理する効率的な方法を構築することができました。

 # SQLAlchemy Imports import sqlalchemy from sqlalchemy.orm import Session # Own Imports from config.database.connection import SessionLocal class DatabaseSessionMixin: """Database session mixin.""" def __enter__(self) -> Session: self.db = SessionLocal() return self.db def __exit__(self, exc_type, exc_val, exc_tb): try: if exc_type is not None: self.db.rollback() except sqlalchemy.exc.SQLAlchemyError: pass finally: self.db.close() SessionLocal.remove() def use_database_session(): return DatabaseSessionMixin()

このコードでは:

• DatabaseSession 、セッションを処理し、エラーが発生した場合にセッションが適切に閉じられ、ロールバックされるようにするコンテキスト マネージャー クラスです。

• __enter__メソッドはセッションを初期化してそれを返します。

• __exit__メソッドは例外をチェックし、例外が発生した場合はセッションをロールバックします。次に、セッションを閉じて、スコープ指定されたセッションから削除します。

• use_database_session 、セッションの使用を簡素化するデコレーターまたはコンテキスト マネージャーとして使用できるユーティリティ関数です。

use_database_session ユーティリティ関数の使用方法の例を次に示します。

 with use_database_session() as db: # perform logic that uses the session # ... # After exiting the context, the session will be automatically closed and removed from the scoped session.

上記のアプローチは、セッションを処理するためのよりクリーンで効率的な方法を提供し、エラーが発生した場合にセッションが適切にロールバックまたはクローズされることを保証します。 ORM でデータベース セッションを使用しながらリポジトリ パターンを実装する方法に進みましょう。

 # SQLAlchemy Imports import sqlalchemy from sqlalchemy.orm import Session class BaseRepository: def __init__(self, session: Session): self.db = session class UserRepository(BaseRepository): """Operations to interact with the `users` table in the database.""" def get(self, user_id: int) -> User: """This method gets a user from the database.""" user = ( self.db.query(User) .filter(User.id == user_id) .first() ) return user def create(self, name: str, email: str, password: str) -> User: """This method creates a user.""" user = User(name=name, email=email, password=password) self.db.add(user) self.db.commit() self.db.refresh(user) return user def update_user(self, user_id: int, updated_data: dict): """This method updates a user.""" user = self.get(user_id) if user: for key, value in updated_data.items(): setattr(user, key, value) self.db.commit() return user return None def delete_user(self, user_id): """This method deletes a user.""" user = self.get_user(user_id) if user: self.db.delete(user) self.db.commit() return True return False

次に、上記のリポジトリをアプリケーションのサービス層に統合します。ユーザーアカウントを作成するサービス関数があるとします。新しい方法を使用してそれを行う方法は次のとおりです。

 # Apps Imports from apps.users.models import User from apps.users.repo import UserRepository from apps.users.schemas.auth import UserCreate # Config Imports from config.security.hashers import password from config.database.session_mixin import use_database_session async def create_user(user: UserCreate) -> User: """ This function creates a new user in the database. :param user: schemas.UserCreate :type user: schemas.UserCreate :return: The user object """ with use_database_session() as db: users_repo = UserRepository(db) user = users_repo.create( user.name, user.email, password.hash(user.password) ) return user

上記のパターンを使用すると、継承されたデータベース セッションを利用しながら、リポジトリ クラス内でデータベース操作をカプセル化できます。また、ORM モデルとリポジトリ ロジックが明確に分離されます。

結論

結論として、バックエンド システムを構築する場合は、セッションを効率的に処理することが重要です。

データベース トランザクション中に発生するsqlalchemy.exc.PendingRollbackErrorやsqlalchemy.exc.InvalidRequestErrorなどのエラーは、適切に処理されないとデータの不整合やアプリケーションの障害を引き起こす可能性があります。

これらのエラーを特定して解決することは、システムの整合性と信頼性を維持するために重要です。

セッション処理に関連する問題に対処するには、堅牢な戦略を実装することが不可欠です。 1 つのアプローチは、この記事で説明したDatabaseSessionMixinなどのコンテキスト マネージャーを使用することです。

このコンテキスト マネージャーは、例外が発生した場合にセッションが適切に開かれ、閉じられ、ロールバックされることを保証します。コンテキスト マネージャー内でセッション ロジックをカプセル化することで、セッション管理を合理化し、エラー処理を改善できます。

さらに、リポジトリ パターンをアプリケーションのサービス層に統合すると、セッション処理の効率がさらに向上します。

データベース操作をリポジトリ クラスに分離し、コンテキスト マネージャーから継承されたセッションを利用することにより、よりクリーンなコード編成を実現し、ORM モデルとリポジトリ ロジック間の明確な分離を維持できます。

全体として、効率的なセッション処理は、データの一貫性を維持し、エラーを防止し、バックエンド システムの安定性を確保するために非常に重要です。

コンテキスト マネージャーの使用やリポジトリ パターンの採用などのベスト プラクティスに従うことで、開発者は効果的にセッションを管理し、データベース トランザクション中のエラーを処理する堅牢で信頼性の高いシステムを構築できます。

私はギグの執筆に前向きで、Python (Django、FastAPI など) での構築に関わる契約上の役割を積極的に探しています。