Xử lý phiên hiệu quả bằng cách sử dụng mẫu kho lưu trữ trong FastAPI

Tôi đã viết một bài báo khoảng năm tháng trước về việc điều chỉnh mẫu kho lưu trữ với FastAPI và tôi đã nhận được rất nhiều lượt đọc (cảm ơn bạn). Tôi đã viết về một cách hiệu quả để xử lý việc xử lý phiên, vẫn sử dụng mẫu kho lưu trữ.

Trước khi bắt đầu ngay, tôi nhận thấy rằng trong quá trình sản xuất, tôi gặp một trong các lỗi sau bất cứ khi nào các API của tôi đang cố thực hiện một giao dịch liên quan đến việc đọc hoặc ghi vào cơ sở dữ liệu:

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

sqlalchemy.exc.PendingRollbackError

Lỗi này chỉ ra rằng có một giao dịch không được cam kết đang diễn ra cần được khôi phục trước khi tiếp tục với bất kỳ hoạt động cơ sở dữ liệu nào khác.

Nguyên nhân phổ biến nhất của lỗi này là một ngoại lệ chưa được xử lý xảy ra trong quá trình giao dịch cơ sở dữ liệu, khiến giao dịch không được cam kết hoặc khôi phục đúng cách.

sqlalchemy.exc.InvalidRequestError

Lỗi này cho biết rằng thao tác hoặc yêu cầu mà bạn đã thực hiện đối với cơ sở dữ liệu không hợp lệ hoặc không được hỗ trợ. Có thể có nhiều nguyên nhân gây ra lỗi này, bao gồm:

• Việc sử dụng API SQLAlchemy không đúng cách, chẳng hạn như gọi một phương thức hoặc truy cập một thuộc tính không tồn tại hoặc không áp dụng được trong ngữ cảnh nhất định.

• Cú pháp SQL hoặc cấu trúc truy vấn không chính xác.

• Ánh xạ thiếu hoặc không chính xác giữa các đối tượng Python và bảng/cột cơ sở dữ liệu.

Tôi chắc chắn rằng bạn có suy nghĩ về cách giải quyết những lỗi này, tuy nhiên, tôi muốn nói rằng mặc dù tôi đã xác định được vấn đề là gì và đã khắc phục - sự cố vẫn tiếp diễn.

Nếu bạn tò mò về cách tôi khắc phục sự cố và giải quyết chúng, bạn có thể cân nhắc thực hiện theo các bước sau:

sqlalchemy.exc.PendingRollbackError:

• Xem lại mã của bạn để biết bất kỳ trường hợp ngoại lệ nào chưa được xử lý có thể ngăn giao dịch được thực hiện hoặc khôi phục đúng cách. Đảm bảo rằng bạn đang xử lý đúng cách các trường hợp ngoại lệ và cam kết hoặc khôi phục giao dịch khi cần.

• Kiểm tra bất kỳ giao dịch dài hạn hoặc giao dịch lồng nhau nào có thể dẫn đến việc quay lui đang chờ xử lý. Đảm bảo rằng tất cả các giao dịch được cam kết hoặc khôi phục kịp thời.

• Xem lại mã của bạn để biết bất kỳ trường hợp nào mà bạn có thể đang bắt đầu một giao dịch mới trước khi xử lý đúng giao dịch trước đó.

sqlalchemy.exc.InvalidRequestError:

• Xem lại thông báo lỗi cụ thể được cung cấp cùng với ngoại lệ để xác định nguyên nhân của yêu cầu không hợp lệ. Nó có thể cung cấp cho bạn các gợi ý về phần nào trong mã hoặc câu lệnh SQL của bạn đang gây ra sự cố.

• Kiểm tra mã SQLAlchemy của bạn xem có bất kỳ cuộc gọi phương thức không chính xác, truy cập thuộc tính hoặc sử dụng API không chính xác nào không.

• Xem lại các câu lệnh SQL của bạn để đảm bảo chúng có cú pháp và cấu trúc chính xác.

• Xác minh rằng lược đồ cơ sở dữ liệu của bạn và ánh xạ SQLAlchemy được đồng bộ hóa. Đảm bảo rằng tất cả các bảng và cột bắt buộc đều tồn tại và các đối tượng Python của bạn được ánh xạ chính xác tới các thực thể cơ sở dữ liệu tương ứng.

Hãy tìm hiểu ngay về cách tôi làm việc với một giải pháp lâu dài đã được chứng minh là có hiệu quả với tôi. Tôi sẽ tiếp tục sử dụng một dự án mà tôi đã thực hiện khi trình bày cách sử dụng các mẫu kho lưu trữ.

Chúng tôi đã có một mô-đun nơi chúng tôi lưu trữ mixin phiên cơ sở orm của mình với các mã sau:

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

Vấn đề với giải pháp này là nếu một ngoại lệ xảy ra trong quá trình giao dịch (đây có thể là bất cứ điều gì: tạo người dùng, cấp tiền cho ví của bạn, v.v.) - thì các ngoại lệ đó không được xử lý đúng cách và phiên cơ sở dữ liệu đang chuyển tiếp không có được rollback.

Sau ba tháng gỡ lỗi và vá lỗi cũng như rất nhiều nghiên cứu, cuối cùng tôi đã có thể xây dựng một cách hiệu quả để xử lý các phiên.

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

Trong mã này:

• DatabaseSession là lớp trình quản lý ngữ cảnh xử lý phiên và đảm bảo rằng phiên được đóng và khôi phục đúng cách trong trường hợp có lỗi.

• Phương thức __enter__ khởi tạo phiên và trả về nó.

• Phương thức __exit__ kiểm tra ngoại lệ và khôi phục phiên nếu xảy ra ngoại lệ. Sau đó, nó đóng phiên và xóa phiên đó khỏi phiên có phạm vi.

• use_database_session là một chức năng tiện ích có thể được sử dụng làm công cụ trang trí hoặc trình quản lý bối cảnh để đơn giản hóa việc sử dụng phiên.

Đây là một ví dụ về cách bạn có thể sử dụng hàm tiện ích 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.

Cách tiếp cận trên cung cấp một cách sạch hơn và hiệu quả hơn để xử lý các phiên, đồng thời đảm bảo rằng chúng được khôi phục hoặc đóng đúng cách trong trường hợp có lỗi. Hãy tiếp tục với cách bạn triển khai mẫu kho lưu trữ trong khi sử dụng phiên cơ sở dữ liệu trong 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

Tiếp theo sẽ là tích hợp kho lưu trữ trên vào lớp dịch vụ của ứng dụng của bạn. Giả sử bạn có chức năng dịch vụ tạo tài khoản người dùng; đây là cách bạn thực hiện bằng phương pháp mới của chúng tôi:

 # 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

Mẫu trên sẽ cho phép bạn đóng gói các hoạt động cơ sở dữ liệu trong các lớp kho lưu trữ trong khi tận dụng phiên cơ sở dữ liệu kế thừa. Nó cũng cung cấp sự tách biệt rõ ràng giữa các mô hình ORM của bạn và logic kho lưu trữ.

Phần kết luận

Tóm lại, xử lý hiệu quả các phiên là rất quan trọng khi xây dựng hệ thống phụ trợ.

Các lỗi như sqlalchemy.exc.PendingRollbackError và sqlalchemy.exc.InvalidRequestError xảy ra trong quá trình giao dịch cơ sở dữ liệu có thể dẫn đến sự không nhất quán của dữ liệu và lỗi ứng dụng nếu không được xử lý đúng cách.

Việc xác định và giải quyết các lỗi này rất quan trọng để duy trì tính toàn vẹn và độ tin cậy của hệ thống.

Để giải quyết các vấn đề liên quan đến xử lý phiên, điều cần thiết là triển khai các chiến lược mạnh mẽ. Một cách tiếp cận là sử dụng trình quản lý ngữ cảnh, chẳng hạn như DatabaseSessionMixin mà chúng tôi đã trình bày trong bài viết.

Trình quản lý ngữ cảnh này đảm bảo rằng các phiên được mở, đóng và khôi phục đúng cách trong trường hợp ngoại lệ. Bằng cách đóng gói logic phiên trong trình quản lý ngữ cảnh, bạn có thể hợp lý hóa việc quản lý phiên và cải thiện khả năng xử lý lỗi.

Ngoài ra, việc tích hợp mẫu kho lưu trữ vào lớp dịch vụ của ứng dụng có thể nâng cao hơn nữa hiệu quả xử lý phiên.

Bằng cách tách các hoạt động cơ sở dữ liệu thành các lớp kho lưu trữ và tận dụng phiên kế thừa từ trình quản lý ngữ cảnh, bạn có thể đạt được tổ chức mã sạch hơn và duy trì sự tách biệt rõ ràng giữa các mô hình ORM và logic kho lưu trữ.

Nhìn chung, xử lý phiên hiệu quả là rất quan trọng để duy trì tính nhất quán của dữ liệu, ngăn ngừa lỗi và đảm bảo tính ổn định của hệ thống phụ trợ.

Bằng cách tuân theo các phương pháp hay nhất, chẳng hạn như sử dụng trình quản lý bối cảnh và áp dụng mẫu kho lưu trữ, nhà phát triển có thể xây dựng các hệ thống mạnh mẽ và đáng tin cậy để quản lý hiệu quả các phiên và xử lý lỗi trong quá trình giao dịch cơ sở dữ liệu.

Tôi sẵn sàng viết các hợp đồng biểu diễn và tích cực tìm kiếm các vai trò hợp đồng liên quan đến việc xây dựng bằng Python (Django, FastAPI, v.v.).