FastAPI'de Depo Desenini Kullanarak Verimli Oturum İşleme

Yaklaşık beş ay önce veri deposu modelinin FastAPI ile uyarlanması hakkında bir makale yazdım ve çok sayıda okuma aldım (teşekkür ederim). Hala depo modelini kullanarak, oturum yönetimini yönetmenin etkili bir yolu hakkında yazmaya geldim.

Hemen konuya geçmeden önce, üretim sırasında API'lerim veritabanına okuma veya yazmayı içeren bir işlem yapmaya çalıştığında aşağıdaki hatalardan birini aldığımı fark ettim:

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

sqlalchemy.exc.PendingRollbackError

Bu hata, başka herhangi bir veritabanı işlemine geçmeden önce geri alınması gereken, devam eden kaydedilmemiş bir işlemin olduğunu gösterir.

Bu hatanın en yaygın nedeni, bir veritabanı işlemi sırasında meydana gelen ve işlemin düzgün bir şekilde gerçekleştirilmesini veya geri alınmasını engelleyen işlenmeyen bir istisnadır.

sqlalchemy.exc.InvalidRequestError

Bu hata, veritabanına yaptığınız bir işlemin veya isteğin geçersiz olduğunu veya desteklenmediğini gösterir. Bu hatanın çeşitli nedenleri olabilir:

• Bir yöntemin çağrılması veya mevcut olmayan veya verilen bağlamda geçerli olmayan bir özniteliğe erişim gibi SQLAlchemy API'nin yanlış kullanımı.

• Yanlış SQL sözdizimi veya sorgu yapısı.

• Python nesneleri ve veritabanı tabloları/sütunları arasında eksik veya yanlış eşleme.

Eminim ki bu hataların nasıl çözülebileceği konusunda düşünceleriniz vardır ancak sorunun ne olduğunu tespit edip düzeltme yapmama rağmen sorunun devam ettiğini belirtmek isterim.

Sorunları nasıl giderdiğimi ve çözdüğümü merak ediyorsanız aşağıdaki adımları izlemeyi düşünebilirsiniz:

sqlalchemy.exc.PendingRollbackError:

• İşlemlerin düzgün bir şekilde gerçekleştirilmesini veya geri alınmasını engelleyebilecek işlenmeyen özel durumlar için kodunuzu inceleyin. İstisnaları doğru şekilde ele aldığınızdan ve gerektiğinde işlemi gerçekleştirdiğinizden veya geri aldığınızdan emin olun.

• Bekleyen geri alma işlemlerine yol açabilecek uzun süredir devam eden veya iç içe geçmiş işlemleri kontrol edin. Tüm işlemlerin zamanında yapıldığından veya geri alındığından emin olun.

• Önceki işlemi doğru şekilde gerçekleştirmeden önce yeni bir işleme başlıyor olabileceğiniz durumlar için kodunuzu gözden geçirin.

sqlalchemy.exc.InvalidRequestError:

• Geçersiz isteğin nedenini belirlemek için istisnayla sağlanan özel hata mesajını inceleyin. Kodunuzun veya SQL ifadenizin hangi bölümünün soruna neden olduğu konusunda size ipuçları verebilir.

• Yanlış yöntem çağrıları, öznitelik erişimleri veya API'nin yanlış kullanımı açısından SQLAlchemy kodunuzu kontrol edin.

• Doğru söz dizimine ve yapıya sahip olduklarından emin olmak için SQL ifadelerinizi gözden geçirin.

• Veritabanı şemanızın ve SQLAlchemy eşlemelerinizin senkronize olduğunu doğrulayın. Gerekli tüm tablo ve sütunların mevcut olduğundan ve Python nesnelerinizin karşılık gelen veritabanı varlıklarıyla düzgün şekilde eşlendiğinden emin olun.

İşime yaradığı kanıtlanmış kalıcı bir çözüm üzerinde nasıl çalıştığıma hemen bakalım. Depo kalıplarının nasıl kullanılacağını gösterirken üzerinde çalıştığım bir projeyi kullanmaya devam edeceğim.

Orm temel oturumumuzu aşağıdaki kodlarla karıştırdığımız bir modülümüz vardı:

 # 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()

Bu çözümdeki sorun, bir işlem sürecinde bir istisna meydana gelirse (bu herhangi bir şey olabilir: bir kullanıcı oluşturmak, cüzdanınıza para yatırmak vb.) istisnaların düzgün şekilde ele alınmaması ve aktarım halindeki veritabanı oturumunun gerçekleşmemesiydi. geri alma alın.

Üç ay süren hata ayıklama, yamalama ve çok sayıda araştırmadan sonra nihayet oturumları yönetmenin etkili bir yolunu oluşturmayı başardım.

 # 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()

Bu kodda:

• DatabaseSession , oturumu yöneten ve bir hata durumunda düzgün şekilde kapatılmasını ve geri alınmasını sağlayan bir bağlam yöneticisi sınıfıdır.

• __enter__ yöntemi oturumu başlatır ve döndürür.

• __exit__ yöntemi istisnaları kontrol eder ve bir istisna oluştuğunda oturumu geri alır. Daha sonra oturumu kapatır ve kapsamı belirlenen oturumdan kaldırır.

• use_database_session , oturum kullanımını basitleştirmek için dekoratör veya içerik yöneticisi olarak kullanılabilen bir yardımcı program işlevidir.

use_database_session yardımcı programını nasıl kullanabileceğinizi gösteren bir örnek:

 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.

Yukarıdaki yaklaşım, oturumları yönetmek için daha temiz ve daha verimli bir yol sağlar ve bir hata durumunda oturumların düzgün bir şekilde geri alınmasını veya kapatılmasını sağlar. ORM'de veritabanı oturumunu kullanırken depo modelini nasıl uygulayacağınıza geçelim.

 # 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

Daha sonra yukarıdaki depoyu uygulamanızın hizmet katmanına entegre etmek olacaktır. Kullanıcı hesabı oluşturan bir hizmet fonksiyonunuz olduğunu varsayalım; yeni yöntemimizi kullanarak bunu şu şekilde yapabilirsiniz:

 # 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

Yukarıdaki model, devralınan veritabanı oturumundan yararlanırken veritabanı işlemlerini depo sınıfları içinde kapsüllemenize olanak tanır. Ayrıca ORM modelleriniz ile depo mantığı arasında temiz bir ayrım sağlar.

Çözüm

Sonuç olarak, arka uç sistemleri oluştururken oturumların verimli bir şekilde yönetilmesi önemlidir.

Veritabanı işlemleri sırasında ortaya çıkan sqlalchemy.exc.PendingRollbackError ve sqlalchemy.exc.InvalidRequestError gibi hatalar, doğru şekilde ele alınmadığı takdirde veri tutarsızlıklarına ve uygulama hatalarına yol açabilmektedir.

Bu hataların belirlenmesi ve çözülmesi, sistemin bütünlüğünü ve güvenilirliğini korumak açısından önemlidir.

Oturum yönetimiyle ilgili sorunları çözmek için sağlam stratejiler uygulamak önemlidir. Bir yaklaşım, makalede gösterdiğimiz DatabaseSessionMixin gibi bağlam yöneticilerini kullanmaktır.

Bu içerik yöneticisi, oturumların düzgün bir şekilde açılmasını, kapatılmasını ve istisnai durumlarda geri alınmasını sağlar. Oturum mantığını bağlam yöneticisi içinde kapsayarak oturum yönetimini kolaylaştırabilir ve hata yönetimini geliştirebilirsiniz.

Ek olarak, depo modelinin uygulamanın hizmet katmanına entegre edilmesi, oturum yönetiminin verimliliğini daha da artırabilir.

Veritabanı işlemlerini depo sınıflarına ayırarak ve bağlam yöneticisinden devralınan oturumu kullanarak, daha temiz kod organizasyonu elde edebilir ve ORM modelleri ile depo mantığı arasında net bir ayrım sağlayabilirsiniz.

Genel olarak verimli oturum yönetimi, veri tutarlılığını korumak, hataları önlemek ve arka uç sistemlerinin kararlılığını sağlamak için çok önemlidir.

Geliştiriciler, bağlam yöneticilerini kullanmak ve veri havuzu modelini benimsemek gibi en iyi uygulamaları izleyerek, oturumları etkili bir şekilde yöneten ve veritabanı işlemleri sırasında hataları ele alan sağlam ve güvenilir sistemler oluşturabilirler.

Konserler yazmaya ve Python (Django, FastAPI, vb.) ile geliştirmeyi içeren sözleşmeli roller aramaya açığım .