Các mẫu và phương pháp sử dụng SQLAlchemy 2.0 với FastAPI

Trong khi Django và Flask vẫn là lựa chọn đầu tiên của nhiều kỹ sư Python, FastAPI đã được công nhận là một lựa chọn đáng tin cậy không thể phủ nhận. Nó là một khung cấu trúc có tính linh hoạt cao, được tối ưu hóa tốt, mang đến cho nhà phát triển khả năng vô tận để xây dựng các ứng dụng phụ trợ.

Làm việc với cơ sở dữ liệu là một khía cạnh thiết yếu của hầu hết các ứng dụng phụ trợ. Do đó, ORM đóng một vai trò quan trọng trong mã phụ trợ. Tuy nhiên, không giống như Django, FastAPI không tích hợp sẵn ORM. Nhà phát triển hoàn toàn có trách nhiệm chọn một thư viện phù hợp và tích hợp nó vào cơ sở mã.

Các kỹ sư Python coi SQLAlchemy là ORM phổ biến nhất hiện có. Đó là một thư viện huyền thoại đã được sử dụng từ năm 2006 và đã được hàng nghìn dự án áp dụng. Vào năm 2023, nó đã nhận được một bản cập nhật lớn cho phiên bản 2.0. Tương tự như FastAPI, SQLAlchemy cung cấp cho các nhà phát triển các tính năng và tiện ích mạnh mẽ mà không bắt buộc họ phải sử dụng chúng theo một cách cụ thể. Về cơ bản, đó là một bộ công cụ đa năng cho phép các nhà phát triển sử dụng nó theo bất kỳ cách nào họ thấy phù hợp.

FastAPI và SQLAlchemy là một cặp trời sinh. Chúng đều là những công nghệ đáng tin cậy, hiệu quả và hiện đại, cho phép tạo ra các ứng dụng mạnh mẽ và độc đáo. Bài viết này khám phá việc tạo ứng dụng phụ trợ FastAPI sử dụng SQLAlchemy 2.0 làm ORM. Nội dung bao gồm:

• xây dựng mô hình bằng Mapped và mapped_column
• xác định một mô hình trừu tượng
• xử lý phiên cơ sở dữ liệu
• sử dụng ORM
• tạo một lớp kho lưu trữ chung cho tất cả các mô hình
• chuẩn bị thiết lập thử nghiệm và thêm thử nghiệm

Sau đó, bạn sẽ có thể kết hợp ứng dụng FastAPI với SQLAlchemy ORM một cách dễ dàng. Ngoài ra, bạn sẽ làm quen với các phương pháp và mẫu tốt nhất để tạo các ứng dụng có cấu trúc tốt, mạnh mẽ và hoạt động hiệu quả.

điều kiện tiên quyết

Các mã ví dụ trong bài viết đến từ dự án nhà giả kim , đây là một API cơ bản để tạo và đọc các đối tượng thành phần và thuốc. Trọng tâm chính của bài viết là khám phá sự kết hợp giữa FastAPI và SQLAlchemy. Nó không bao gồm các chủ đề khác, chẳng hạn như:

• định cấu hình thiết lập Docker
• khởi động máy chủ uvicorn
• thiết lập linting

Nếu bạn quan tâm đến những chủ đề này, bạn có thể tự khám phá chúng bằng cách kiểm tra cơ sở mã. Để truy cập vào kho lưu trữ mã của dự án nhà giả kim, vui lòng theo liên kết này tại đây . Ngoài ra, bạn có thể tìm thấy cấu trúc tệp của dự án bên dưới:

 alchemist ├─ alchemist │ ├─ api │ │ ├─ v1 │ │ │ ├─ __init__.py │ │ │ └─ routes.py │ │ ├─ v2 │ │ │ ├─ __init__.py │ │ │ ├─ dependencies.py │ │ │ └─ routes.py │ │ ├─ __init__.py │ │ └─ models.py │ ├─ database │ │ ├─ __init__.py │ │ ├─ models.py │ │ ├─ repository.py │ │ └─ session.py │ ├─ __init__.py │ ├─ app.py │ └─ config.py ├─ requirements │ ├─ base.txt │ └─ dev.txt ├─ scripts │ ├─ create_test_db.sh │ ├─ migrate.py │ └─ run.sh ├─ tests │ ├─ conftest.py │ └─ test_api.py ├─ .env ├─ .gitignore ├─ .pre-commit-config.yaml ├─ Dockerfile ├─ Makefile ├─ README.md ├─ docker-compose.yaml ├─ example.env └─ pyproject.toml

Mặc dù cây có vẻ lớn, nhưng một số nội dung không liên quan đến điểm chính của bài viết này. Ngoài ra, mã có thể xuất hiện đơn giản hơn những gì cần thiết ở một số khu vực nhất định. Chẳng hạn, dự án thiếu:

• giai đoạn sản xuất trong Dockerfile
• thiết lập alembic cho di chuyển
• thư mục con cho các bài kiểm tra

Điều này được thực hiện có chủ ý để giảm độ phức tạp và tránh chi phí không cần thiết. Tuy nhiên, điều quan trọng là phải ghi nhớ những yếu tố này nếu xử lý một dự án sẵn sàng sản xuất hơn.

Yêu cầu API

Khi bắt đầu phát triển một ứng dụng, điều quan trọng là phải xem xét các mô hình mà ứng dụng của bạn sẽ sử dụng. Các mô hình này sẽ đại diện cho các đối tượng và thực thể mà ứng dụng của bạn sẽ làm việc cùng và sẽ được hiển thị trong API. Trong trường hợp của ứng dụng giả kim, có hai thực thể: nguyên liệu và thuốc. API sẽ cho phép tạo và truy xuất các thực thể này. Tệp giả alchemist/api/models.py chứa các mô hình sẽ được sử dụng trong API:

 import uuid from pydantic import BaseModel, Field class Ingredient(BaseModel): """Ingredient model.""" pk: uuid.UUID name: str class Config: orm_mode = True class IngredientPayload(BaseModel): """Ingredient payload model.""" name: str = Field(min_length=1, max_length=127) class Potion(BaseModel): """Potion model.""" pk: uuid.UUID name: str ingredients: list[Ingredient] class Config: orm_mode = True class PotionPayload(BaseModel): """Potion payload model.""" name: str = Field(min_length=1, max_length=127) ingredients: list[uuid.UUID] = Field(min_items=1)

API sẽ trả về các mô hình Ingredient và Potion . Đặt orm_mode thành True trong cấu hình sẽ giúp làm việc với các đối tượng SQLAlchemy dễ dàng hơn trong tương lai. Các mô hình Payload sẽ được sử dụng để tạo các đối tượng mới.

Việc sử dụng pydantic làm cho các lớp chi tiết và rõ ràng hơn về vai trò và chức năng của chúng. Bây giờ là lúc tạo các mô hình cơ sở dữ liệu.

khai báo mô hình

Một mô hình về cơ bản là một đại diện của một cái gì đó. Trong ngữ cảnh của API, các mô hình biểu thị những gì chương trình phụ trợ mong đợi trong phần thân yêu cầu và những gì nó sẽ trả về trong dữ liệu phản hồi. Mặt khác, các mô hình cơ sở dữ liệu phức tạp hơn và biểu thị các cấu trúc dữ liệu được lưu trữ trong cơ sở dữ liệu và các kiểu quan hệ giữa chúng.

Tệp alchemist/database/models.py chứa các mô hình cho các đối tượng thành phần và thuốc:

 import uuid from sqlalchemy import Column, ForeignKey, Table, orm from sqlalchemy.dialects.postgresql import UUID class Base(orm.DeclarativeBase): """Base database model.""" pk: orm.Mapped[uuid.UUID] = orm.mapped_column( primary_key=True, default=uuid.uuid4, ) potion_ingredient_association = Table( "potion_ingredient", Base.metadata, Column("potion_id", UUID(as_uuid=True), ForeignKey("potion.pk")), Column("ingredient_id", UUID(as_uuid=True), ForeignKey("ingredient.pk")), ) class Ingredient(Base): """Ingredient database model.""" __tablename__ = "ingredient" name: orm.Mapped[str] class Potion(Base): """Potion database model.""" __tablename__ = "potion" name: orm.Mapped[str] ingredients: orm.Mapped[list["Ingredient"]] = orm.relationship( secondary=potion_ingredient_association, backref="potions", lazy="selectin", )

Mọi mô hình trong SQLAlchemy đều bắt đầu với lớp DeclarativeBase . Kế thừa từ nó cho phép xây dựng các mô hình cơ sở dữ liệu tương thích với trình kiểm tra kiểu Python.

Đó cũng là một cách thực hành tốt để tạo một mô hình trừu tượng —Lớp Base trong trường hợp này—bao gồm các trường bắt buộc trong tất cả các mô hình. Các trường này bao gồm khóa chính, là mã định danh duy nhất của mọi đối tượng. Mô hình trừu tượng cũng thường lưu trữ ngày tạo và cập nhật của một đối tượng, được đặt tự động khi một đối tượng được tạo hoặc cập nhật. Tuy nhiên, mô hình Base sẽ được giữ đơn giản.

Chuyển sang mô hình Ingredient , thuộc tính __tablename__ chỉ định tên của bảng cơ sở dữ liệu, trong khi trường name sử dụng cú pháp SQLAlchemy mới, cho phép khai báo các trường mô hình bằng chú thích loại. Cách tiếp cận ngắn gọn và hiện đại này vừa mạnh mẽ vừa thuận lợi cho các trình kiểm tra kiểu và IDE, vì nó nhận dạng trường name dưới dạng một chuỗi.

Mọi thứ trở nên phức tạp hơn trong mô hình Potion . Nó cũng bao gồm các thuộc tính __tablename__ và name , nhưng trên hết, nó lưu trữ mối quan hệ với các thành phần. Việc sử dụng Mapped[list["Ingredient"]] chỉ ra rằng thuốc có thể chứa nhiều thành phần và trong trường hợp này, mối quan hệ là nhiều-nhiều (M2M). Điều này có nghĩa là một thành phần duy nhất có thể được chỉ định cho nhiều loại thuốc.

M2M yêu cầu cấu hình bổ sung, thường liên quan đến việc tạo bảng liên kết lưu trữ các kết nối giữa hai thực thể. Trong trường hợp này, đối tượng potion_ingredient_association chỉ lưu trữ các mã định danh của thành phần và thuốc, nhưng nó cũng có thể bao gồm các thuộc tính bổ sung, chẳng hạn như lượng của một thành phần cụ thể cần thiết cho thuốc.

Chức năng relationship cấu hình mối quan hệ giữa thuốc và các thành phần của nó. Đối số lazy chỉ định cách tải các mục liên quan. Nói cách khác: SQLAlchemy nên làm gì với các thành phần liên quan khi bạn đang tìm nạp một lọ thuốc. Đặt nó thành selectin có nghĩa là các thành phần sẽ được nạp cùng với thuốc, loại bỏ nhu cầu truy vấn bổ sung trong mã.

Xây dựng các mô hình được thiết kế tốt là rất quan trọng khi làm việc với ORM. Khi điều này được thực hiện, bước tiếp theo là thiết lập kết nối với cơ sở dữ liệu.

Trình xử lý phiên

Khi làm việc với cơ sở dữ liệu, đặc biệt là khi sử dụng SQLAlchemy, điều cần thiết là phải hiểu các khái niệm sau:

• thổ ngữ
• động cơ
• sự liên quan
• nhóm kết nối
• phiên họp

Trong số tất cả các điều khoản này, điều quan trọng nhất là động cơ . Theo tài liệu SQLAlchemy, đối tượng động cơ chịu trách nhiệm kết nối Pool và Dialect để tạo điều kiện thuận lợi cho hành vi và kết nối cơ sở dữ liệu. Nói một cách đơn giản hơn, đối tượng engine là nguồn của kết nối cơ sở dữ liệu, trong khi kết nối cung cấp các chức năng cấp cao như thực thi câu lệnh SQL, quản lý giao dịch và truy xuất kết quả từ cơ sở dữ liệu.

Phiên là một đơn vị công việc nhóm các hoạt động liên quan trong một giao dịch. Nó là một sự trừu tượng hóa các kết nối cơ sở dữ liệu cơ bản và quản lý hiệu quả các kết nối và hành vi giao dịch.

Dialect là một thành phần cung cấp hỗ trợ cho một phụ trợ cơ sở dữ liệu cụ thể. Nó hoạt động như một trung gian giữa SQLAlchemy và cơ sở dữ liệu, xử lý các chi tiết của giao tiếp. Dự án nhà giả kim sử dụng Postgres làm cơ sở dữ liệu, vì vậy phương ngữ phải tương thích với loại cơ sở dữ liệu cụ thể này.

Dấu hỏi cuối cùng là nhóm kết nối . Trong ngữ cảnh của SQLAlchemy, nhóm kết nối là một cơ chế quản lý tập hợp các kết nối cơ sở dữ liệu. Nó được thiết kế để cải thiện hiệu suất và hiệu quả của các hoạt động cơ sở dữ liệu bằng cách sử dụng lại các kết nối hiện có thay vì tạo các kết nối mới cho mỗi yêu cầu. Bằng cách sử dụng lại các kết nối, nhóm kết nối giảm chi phí thiết lập và chia nhỏ các kết nối mới, dẫn đến hiệu suất được cải thiện.

Với kiến thức đó, bây giờ bạn có thể xem tệp alchemist/database/session.py , chứa một hàm sẽ được sử dụng làm phụ thuộc để kết nối với cơ sở dữ liệu:

 from collections.abc import AsyncGenerator from sqlalchemy import exc from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.config import settings async def get_db_session() -> AsyncGenerator[AsyncSession, None]: engine = create_async_engine(settings.DATABASE_URL) factory = async_sessionmaker(engine) async with factory() as session: try: yield session await session.commit() except exc.SQLAlchemyError as error: await session.rollback() raise

Chi tiết quan trọng đầu tiên cần lưu ý là hàm get_db_session là một hàm tạo. Điều này là do hệ thống phụ thuộc FastAPI hỗ trợ trình tạo. Do đó, chức năng này có thể xử lý cả trường hợp thành công và không thành công.

Hai dòng đầu tiên của hàm get_db_session tạo một công cụ cơ sở dữ liệu và một phiên. Tuy nhiên, đối tượng phiên cũng có thể được sử dụng làm trình quản lý ngữ cảnh. Điều này cho phép bạn kiểm soát nhiều hơn đối với các trường hợp ngoại lệ tiềm năng và kết quả thành công.

Mặc dù SQLAlchemy xử lý việc đóng kết nối, nhưng thực tế tốt là khai báo rõ ràng cách xử lý kết nối sau khi thực hiện xong. Trong hàm get_db_session , phiên được cam kết nếu mọi thứ diễn ra suôn sẻ và được khôi phục nếu có ngoại lệ.

Điều quan trọng cần lưu ý là mã này được xây dựng xung quanh tiện ích mở rộng asyncio. Tính năng này của SQLAlchemy cho phép ứng dụng tương tác với cơ sở dữ liệu một cách không đồng bộ. Điều đó có nghĩa là các yêu cầu tới cơ sở dữ liệu sẽ không chặn các yêu cầu API khác, giúp ứng dụng hoạt động hiệu quả hơn.

Khi các mô hình và kết nối được thiết lập, bước tiếp theo là đảm bảo rằng các mô hình được thêm vào cơ sở dữ liệu.

Di chuyển nhanh

Các mô hình SQLAlchemy đại diện cho các cấu trúc của cơ sở dữ liệu. Tuy nhiên, chỉ cần tạo chúng không dẫn đến những thay đổi ngay lập tức đối với cơ sở dữ liệu. Để thực hiện các thay đổi, trước tiên bạn phải áp dụng chúng. Điều này thường được thực hiện bằng cách sử dụng thư viện di chuyển, chẳng hạn như alembic, theo dõi mọi mô hình và cập nhật cơ sở dữ liệu tương ứng.

Vì không có thay đổi nào nữa đối với các mô hình được lên kế hoạch trong kịch bản này, nên một tập lệnh di chuyển cơ bản là đủ. Dưới đây là mã ví dụ từ tệp scripts/migrate.py .

 import asyncio import logging from sqlalchemy.ext.asyncio import create_async_engine from alchemist.config import settings from alchemist.database.models import Base logger = logging.getLogger() async def migrate_tables() -> None: logger.info("Starting to migrate") engine = create_async_engine(settings.DATABASE_URL) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("Done migrating") if __name__ == "__main__": asyncio.run(migrate_tables())

Nói một cách đơn giản, hàm migrate_tables đọc cấu trúc của các mô hình và tạo lại nó trong cơ sở dữ liệu bằng cách sử dụng công cụ SQLAlchemy. Để chạy tập lệnh này, hãy sử dụng lệnh python scripts/migrate.py .

Các mô hình hiện có cả trong mã và trong cơ sở dữ liệu và get_db_session có thể hỗ trợ tương tác với cơ sở dữ liệu. Bây giờ bạn có thể bắt đầu làm việc với logic API.

API với ORM

Như đã đề cập trước đây, API cho nguyên liệu và thuốc nhằm hỗ trợ ba hoạt động:

• tạo đối tượng
• liệt kê các đối tượng
• truy xuất các đối tượng bằng ID

Nhờ có sự chuẩn bị trước, tất cả các tính năng này đã có thể được triển khai với SQLAlchemy làm ORM và FastAPI làm khung web. Để bắt đầu, hãy xem lại API thành phần nằm trong tệp giả alchemist/api/v1/routes.py :

 import uuid from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.api import models from alchemist.database import models as db_models from alchemist.database.session import get_db_session router = APIRouter(prefix="/v1", tags=["v1"]) @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = db_models.Ingredient(**data.dict()) session.add(ingredient) await session.commit() await session.refresh(ingredient) return models.Ingredient.from_orm(ingredient) @router.get("/ingredients", status_code=status.HTTP_200_OK) async def get_ingredients( session: AsyncSession = Depends(get_db_session), ) -> list[models.Ingredient]: ingredients = await session.scalars(select(db_models.Ingredient)) return [models.Ingredient.from_orm(ingredient) for ingredient in ingredients] @router.get("/ingredients/{pk}", status_code=status.HTTP_200_OK) async def get_ingredient( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = await session.get(db_models.Ingredient, pk) if ingredient is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ingredient does not exist", ) return models.Ingredient.from_orm(ingredient)

Trong API /ingredients , có ba tuyến có sẵn. Điểm cuối POST lấy tải trọng thành phần làm đối tượng từ mô hình đã tạo trước đó và phiên cơ sở dữ liệu. Hàm tạo get_db_session khởi tạo phiên và cho phép tương tác với cơ sở dữ liệu.

Trong thân hàm thực tế, có năm bước diễn ra:

1. Một đối tượng thành phần được tạo từ tải trọng đến.
2. Phương thức add của đối tượng phiên bổ sung đối tượng thành phần vào hệ thống theo dõi phiên và đánh dấu nó là đang chờ chèn vào cơ sở dữ liệu.
3. Phiên đã được cam kết.
4. Đối tượng thành phần được làm mới để đảm bảo các thuộc tính của nó khớp với trạng thái cơ sở dữ liệu.
5. Phiên bản thành phần cơ sở dữ liệu được chuyển đổi thành phiên bản mô hình API bằng cách sử dụng phương thức from_orm .

Để kiểm tra nhanh, có thể thực hiện một thao tác cuộn tròn đơn giản đối với ứng dụng đang chạy:

 curl -X 'POST' \ 'http://localhost:8000/api/v1/ingredients' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty water"}'

Trong phản hồi, phải có một đối tượng thành phần có ID đến từ cơ sở dữ liệu:

 { "pk":"2eb255e9-2172-4c75-9b29-615090e3250d", "name":"Salty water" }

Mặc dù nhiều lớp trừu tượng của SQLAlchemy có vẻ không cần thiết đối với một API đơn giản, nhưng chúng giữ cho các chi tiết ORM được tách biệt và góp phần vào hiệu quả cũng như khả năng mở rộng của SQLAlchemy. Khi được kết hợp với asyncio, các tính năng ORM hoạt động rất tốt trong API.

Hai điểm cuối còn lại ít phức tạp hơn và có những điểm tương đồng. Một phần đáng được giải thích sâu hơn là việc sử dụng phương thức scalars bên trong hàm get_ingredients . Trong khi truy vấn cơ sở dữ liệu bằng SQLAlchemy, phương thức execute thường được sử dụng với truy vấn làm đối số. Trong khi phương thức execute trả về các bộ dữ liệu dạng hàng, scalars trả về trực tiếp các thực thể ORM, làm cho điểm cuối sạch hơn.

Bây giờ, hãy xem xét API độc dược, trong cùng một tệp:

 @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: data_dict = data.dict() ingredients = await session.scalars( select(db_models.Ingredient).where( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) ) potion = db_models.Potion(**data_dict, ingredients=list(ingredients)) session.add(potion) await session.commit() await session.refresh(potion) return models.Potion.from_orm(potion) @router.get("/potions", status_code=status.HTTP_200_OK) async def get_potions( session: AsyncSession = Depends(get_db_session), ) -> list[models.Potion]: potions = await session.scalars(select(db_models.Potion)) return [models.Potion.from_orm(potion) for potion in potions] @router.get("/potions/{pk}", status_code=status.HTTP_200_OK) async def get_potion( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: potion = await session.get(db_models.Potion, pk) if potion is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Potion does not exist", ) return models.Potion.from_orm(potion)

Điểm cuối GET cho bình thuốc giống với điểm cuối của thành phần. Tuy nhiên, chức năng POST yêu cầu mã bổ sung. Điều này là do việc tạo thuốc bao gồm ít nhất một ID thành phần, có nghĩa là các thành phần phải được tìm nạp và liên kết với thuốc mới được tạo. Để đạt được điều này, phương pháp scalars được sử dụng lại, nhưng lần này với một truy vấn chỉ định ID của các thành phần được tìm nạp. Phần còn lại của quy trình tạo thuốc giống hệt với phần nguyên liệu.

Để kiểm tra điểm cuối, có thể thực hiện lại lệnh curl.

 curl -X 'POST' \ 'http://localhost:8000/api/v1/potions' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty soup", "ingredients": ["0b4f1de5-e780-418d-a74d-927afe8ac954"}'

Nó dẫn đến phản ứng sau:

 { "pk": "d4929197-3998-4234-a5f7-917dc4bba421", "name": "Salty soup", "ingredients": [ { "pk": "0b4f1de5-e780-418d-a74d-927afe8ac954", "name": "Salty water" } ] }

Điều quan trọng cần lưu ý là mỗi thành phần được thể hiện dưới dạng một đối tượng hoàn chỉnh trong lọ thuốc, nhờ đối số lazy="selectin" được chỉ định trong mối quan hệ.

Các API đang hoạt động, nhưng có một vấn đề lớn với mã. Mặc dù SQLAlchemy cho phép bạn tự do tương tác với cơ sở dữ liệu theo ý muốn, nhưng nó không cung cấp bất kỳ tiện ích "trình quản lý" cấp cao nào tương tự như Model.objects của Django. Do đó, bạn sẽ cần phải tự tạo nó, về cơ bản là logic được sử dụng trong API thành phần và thuốc. Tuy nhiên, nếu bạn giữ logic này trực tiếp trong các điểm cuối mà không giải nén nó vào một không gian riêng biệt, bạn sẽ nhận được rất nhiều mã trùng lặp. Ngoài ra, việc thực hiện các thay đổi đối với truy vấn hoặc mô hình sẽ ngày càng khó quản lý.

Chương sắp tới giới thiệu mẫu kho lưu trữ: một giải pháp tinh tế để trích xuất mã ORM.

Kho

Mẫu kho lưu trữ cho phép trừu tượng hóa các chi tiết làm việc với cơ sở dữ liệu. Trong trường hợp sử dụng SQLAlchemy, chẳng hạn như trong ví dụ về nhà giả kim, lớp kho lưu trữ sẽ chịu trách nhiệm quản lý nhiều mô hình và tương tác với phiên cơ sở dữ liệu.

Hãy xem đoạn mã sau từ tệp alchemist/database/repository.py :

 import uuid from typing import Generic, TypeVar from sqlalchemy import BinaryExpression, select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models Model = TypeVar("Model", bound=models.Base) class DatabaseRepository(Generic[Model]): """Repository for performing database queries.""" def __init__(self, model: type[Model], session: AsyncSession) -> None: self.model = model self.session = session async def create(self, data: dict) -> Model: instance = self.model(**data) self.session.add(instance) await self.session.commit() await self.session.refresh(instance) return instance async def get(self, pk: uuid.UUID) -> Model | None: return await self.session.get(self.model, pk) async def filter( self, *expressions: BinaryExpression, ) -> list[Model]: query = select(self.model) if expressions: query = query.where(*expressions) return list(await self.session.scalars(query))

Lớp DatabaseRepository chứa tất cả logic trước đó đã được đưa vào các điểm cuối. Sự khác biệt là nó cho phép lớp mô hình cụ thể được chuyển vào phương thức __init__ , giúp có thể sử dụng lại mã cho tất cả các mô hình thay vì sao chép nó ở mỗi điểm cuối.

Hơn nữa, DatabaseRepository sử dụng các kiểu chung của Python, với kiểu chung của Model được liên kết với mô hình cơ sở dữ liệu trừu tượng. Điều này cho phép lớp kho lưu trữ được hưởng lợi nhiều hơn từ việc kiểm tra kiểu tĩnh. Khi được sử dụng với một mô hình cụ thể, các kiểu trả về của các phương thức lưu trữ sẽ phản ánh mô hình cụ thể này.

Vì kho lưu trữ cần sử dụng phiên cơ sở dữ liệu nên nó phải được khởi tạo cùng với phụ thuộc get_db_session . Xem xét phần phụ thuộc mới trong tệp alchemist/api/v2/dependencies.py :

 from collections.abc import Callable from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models, repository, session def get_repository(  model: type[models.Base], ) -> Callable[[AsyncSession], repository.DatabaseRepository]:  def func(session: AsyncSession = Depends(session.get_db_session)):    return repository.DatabaseRepository(model, session)  return func

Nói một cách đơn giản, hàm get_repository là một nhà máy phụ thuộc. Đầu tiên, nó lấy mô hình cơ sở dữ liệu mà bạn sẽ sử dụng kho lưu trữ. Sau đó, nó trả về phần phụ thuộc, phần phụ thuộc này sẽ được sử dụng để nhận phiên cơ sở dữ liệu và khởi tạo đối tượng kho lưu trữ. Để hiểu rõ hơn, hãy xem API mới từ tệp alchemist/api/v2/routes.py . Nó chỉ hiển thị các điểm cuối POST, nhưng nó đủ để cung cấp cho bạn ý tưởng rõ ràng hơn về cách mã được cải thiện:

 from typing import Annotated from fastapi import APIRouter, Depends, status from alchemist.api import models from alchemist.api.v2.dependencies import get_repository from alchemist.database import models as db_models from alchemist.database.repository import DatabaseRepository router = APIRouter(prefix="/v2", tags=["v2"]) IngredientRepository = Annotated[  DatabaseRepository[db_models.Ingredient],  Depends(get_repository(db_models.Ingredient)), ] PotionRepository = Annotated[  DatabaseRepository[db_models.Potion],  Depends(get_repository(db_models.Potion)), ] @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, repository: IngredientRepository, ) -> models.Ingredient: ingredient = await repository.create(data.dict()) return models.Ingredient.from_orm(ingredient) @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, ingredient_repository: IngredientRepository, potion_repository: PotionRepository, ) -> models.Potion: data_dict = data.dict() ingredients = await ingredient_repository.filter( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) potion = await potion_repository.create({**data_dict, "ingredients": ingredients}) return models.Potion.from_orm(potion)

Tính năng quan trọng đầu tiên cần lưu ý là việc sử dụng Annotated , một cách mới để làm việc với các phụ thuộc FastAPI. Bằng cách chỉ định loại trả về của phần phụ thuộc là DatabaseRepository[db_models.Ingredient] và khai báo mức sử dụng của nó với Depends(get_repository(db_models.Ingredient)) bạn có thể sử dụng các chú thích loại đơn giản trong điểm cuối: repository: IngredientRepository .

Nhờ có kho lưu trữ, các điểm cuối không phải lưu trữ tất cả gánh nặng liên quan đến ORM. Ngay cả trong trường hợp thuốc phức tạp hơn, tất cả những gì bạn cần làm là sử dụng hai kho cùng một lúc.

Bạn có thể thắc mắc nếu khởi tạo hai kho lưu trữ sẽ khởi tạo phiên hai lần. Câu trả lời là không. Hệ thống phụ thuộc FastAPI lưu trữ các cuộc gọi phụ thuộc giống nhau trong một yêu cầu. Điều này có nghĩa là quá trình khởi tạo phiên được lưu vào bộ nhớ cache và cả hai kho lưu trữ đều sử dụng cùng một đối tượng phiên. Một tính năng tuyệt vời khác của sự kết hợp giữa SQLAlchemy và FastAPI.

API có đầy đủ chức năng và có lớp truy cập dữ liệu hiệu suất cao, có thể tái sử dụng. Bước tiếp theo là đảm bảo các yêu cầu được đáp ứng bằng cách viết một số bài kiểm tra đầu cuối.

thử nghiệm

Kiểm thử đóng một vai trò quan trọng trong phát triển phần mềm. Các dự án có thể chứa các bài kiểm tra đơn vị, tích hợp và đầu cuối (E2E). Mặc dù thông thường tốt nhất là có nhiều bài kiểm tra đơn vị có ý nghĩa, nhưng bạn cũng nên viết ít nhất một vài bài kiểm tra E2E để đảm bảo toàn bộ quy trình làm việc hoạt động chính xác.

Để tạo một số thử nghiệm E2E cho ứng dụng giả kim, cần có thêm hai thư viện:

• pytest để thực sự tạo và chạy thử nghiệm
• httpx để thực hiện các yêu cầu không đồng bộ trong các bài kiểm tra

Sau khi chúng được cài đặt, bước tiếp theo là có một cơ sở dữ liệu thử nghiệm riêng biệt. Bạn không muốn cơ sở dữ liệu mặc định của mình bị ô nhiễm hoặc bị hủy bỏ. Vì nhà giả kim bao gồm thiết lập Docker nên chỉ cần một tập lệnh đơn giản để tạo cơ sở dữ liệu thứ hai. Hãy xem mã từ tệp scripts/create_test_db.sh :

 #!/bin/bash psql -U postgres psql -c "CREATE DATABASE test"

Để tập lệnh được thực thi, tập lệnh phải được thêm dưới dạng một ổ đĩa vào bộ chứa Postgres. Điều này có thể đạt được bằng cách đưa nó vào phần volumes của tệp docker-compose.yaml .

Bước chuẩn bị cuối cùng là tạo các đồ đạc pytest trong tệp tests/conftest.py :

 from collections.abc import AsyncGenerator import pytest import pytest_asyncio from fastapi import FastAPI from httpx import AsyncClient from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.app import app from alchemist.config import settings from alchemist.database.models import Base from alchemist.database.session import get_db_session @pytest_asyncio.fixture() async def db_session() -> AsyncGenerator[AsyncSession, None]: """Start a test database session.""" db_name = settings.DATABASE_URL.split("/")[-1] db_url = settings.DATABASE_URL.replace(f"/{db_name}", "/test") engine = create_async_engine(db_url) async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.create_all) session = async_sessionmaker(engine)() yield session await session.close() @pytest.fixture() def test_app(db_session: AsyncSession) -> FastAPI: """Create a test app with overridden dependencies.""" app.dependency_overrides[get_db_session] = lambda: db_session return app @pytest_asyncio.fixture() async def client(test_app: FastAPI) -> AsyncGenerator[AsyncClient, None]: """Create an http client.""" async with AsyncClient(app=test_app, base_url="http://test") as client: yield client

Một điều cần thiết phải thay đổi trong các thử nghiệm là cách ứng dụng tương tác với cơ sở dữ liệu. Điều này bao gồm không chỉ thay đổi URL cơ sở dữ liệu mà còn đảm bảo rằng mỗi thử nghiệm được tách biệt bằng cách bắt đầu với một cơ sở dữ liệu trống.

Lịch thi đấu db_session hoàn thành cả hai mục tiêu này. Cơ thể của nó thực hiện các bước sau:

1. Tạo một công cụ có URL cơ sở dữ liệu đã sửa đổi.
2. Xóa tất cả các bảng hiện có để đảm bảo rằng bài kiểm tra có cơ sở dữ liệu sạch.
3. Tạo tất cả các bảng bên trong cơ sở dữ liệu (cùng mã như trong tập lệnh di chuyển).
4. Tạo và tạo ra một đối tượng phiên.
5. Đóng phiên theo cách thủ công khi kiểm tra xong.

Mặc dù bước cuối cùng cũng có thể được thực hiện như một trình quản lý ngữ cảnh, nhưng việc đóng thủ công chỉ hoạt động tốt trong trường hợp này.

Hai đồ đạc còn lại sẽ khá dễ hiểu:

• test_app là phiên bản FastAPI từ tệp alchemist/app.py , với phần phụ thuộc get_db_session được thay thế bằng phần cố định db_session
• client là httpx AsyncClient sẽ thực hiện các yêu cầu API đối với test_app

Với tất cả những điều này đã được thiết lập, cuối cùng thì các bài kiểm tra thực tế cũng có thể được viết. Để ngắn gọn, ví dụ bên dưới từ tệp tests/test_api.py chỉ hiển thị một thử nghiệm để tạo một thành phần:

 from fastapi import status class TestIngredientsAPI: """Test cases for the ingredients API.""" async def test_create_ingredient(self, client): response = await client.post("/api/v2/ingredients", json={"name": "Carrot"}) assert response.status_code == status.HTTP_201_CREATED pk = response.json().get("pk") assert pk is not None response = await client.get("/api/v2/ingredients") assert response.status_code == status.HTTP_200_OK assert len(response.json()) == 1 assert response.json()[0]["pk"] == pk

Thử nghiệm sử dụng một đối tượng máy khách được tạo trong một lịch thi đấu, đối tượng này đưa ra các yêu cầu đối với phiên bản FastAPI với phần phụ thuộc bị ghi đè. Do đó, bài kiểm tra có thể tương tác với một cơ sở dữ liệu riêng biệt sẽ bị xóa sau khi kiểm tra xong. Cấu trúc của bộ thử nghiệm còn lại cho cả hai API gần như giống nhau.

Bản tóm tắt

FastAPI và SQLAlchemy là những công nghệ tuyệt vời để tạo các ứng dụng phụ trợ hiện đại và mạnh mẽ. Sự tự do, đơn giản và linh hoạt mà chúng mang lại khiến chúng trở thành một trong những lựa chọn tốt nhất cho các dự án dựa trên Python. Nếu các nhà phát triển tuân theo các mẫu và phương pháp hay nhất, họ có thể tạo các ứng dụng có cấu trúc tốt, mạnh mẽ và hiệu quả để xử lý các hoạt động cơ sở dữ liệu và logic API một cách dễ dàng. Bài viết này nhằm mục đích cung cấp cho bạn hiểu biết tốt về cách thiết lập và duy trì sự kết hợp tuyệt vời này.

nguồn

Bạn có thể tìm thấy mã nguồn của dự án nhà giả kim tại đây: liên kết .