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, đã đượ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ợ. FastAPI 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 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. SQLAlchemy 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 và Mapped 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 , đâ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à giả kim đị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 . 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: tại đây 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 và 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ả chứa các mô hình sẽ được sử dụng trong API: các đối tượng thực thể alchemist/api/models.py 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 và . Đặt thành 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 sẽ được sử dụng để tạo các đối tượng mới. Ingredient Potion orm_mode True Payload 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 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 lưu trữ trong cơ sở dữ liệu và các kiểu giữa chúng. đại diện cấu trúc dữ liệu quan hệ Tệp chứa các mô hình cho các đối tượng thành phần và thuốc: alchemist/database/models.py 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 . 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. DeclarativeBase Đó cũng là một cách thực hành tốt để tạo một mô hình —Lớp 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 sẽ được giữ đơn giản. trừu tượng Base Base Chuyển sang mô hình , thuộc tính chỉ định tên của bảng cơ sở dữ liệu, trong khi trường 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 dưới dạng một chuỗi. Ingredient __tablename__ name name Mọi thứ trở nên phức tạp hơn trong mô hình . Nó cũng bao gồm các thuộc tính và , 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 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. Potion __tablename__ name Mapped[list["Ingredient"]] 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 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. potion_ingredient_association Chức năng cấu hình mối quan hệ giữa thuốc và các thành phần của nó. Đối số 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 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ã. relationship lazy selectin 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à . Theo tài liệu SQLAlchemy, đối tượng động cơ chịu trách nhiệm kết nối và để 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 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. động cơ Pool Dialect kết nối 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. Phiên 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. Dialect Dấu hỏi cuối cùng là . 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. nhóm kết nối Với kiến thức đó, bây giờ bạn có thể xem tệp , 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: alchemist/database/session.py 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 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. get_db_session Hai dòng đầu tiên của hàm 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. get_db_session 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 , 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ệ. get_db_session Đ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 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. áp dụ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 đọ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 . migrate_tables python scripts/migrate.py Các mô hình hiện có cả trong mã và trong cơ sở dữ liệu và 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. get_db_session 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 , 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 khởi tạo phiên và cho phép tương tác với cơ sở dữ liệu. /ingredients get_db_session Trong thân hàm thực tế, có năm bước diễn ra: Một đối tượng thành phần được tạo từ tải trọng đến. Phương thức 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. add Phiên đã được cam kết. Đố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. 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 bên trong hàm . Trong khi truy vấn cơ sở dữ liệu bằng SQLAlchemy, phương thức thường được sử dụng với truy vấn làm đối số. Trong khi phương thức trả về các bộ dữ liệu dạng hàng, trả về trực tiếp các thực thể ORM, làm cho điểm cuối sạch hơn. scalars get_ingredients execute execute scalars 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 đượ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. scalars Để 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ố được chỉ định trong mối quan hệ. lazy="selectin" 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ư 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ý. Model.objects 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 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. kho lưu trữ 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 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 , 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. DatabaseRepository __init__ Hơn nữa, sử dụng các kiểu chung của Python, với kiểu chung của đượ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. DatabaseRepository Model 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 . Xem xét phần phụ thuộc mới trong tệp : get_db_session 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 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 . 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: get_repository alchemist/api/v2/routes.py 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 , 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à và khai báo mức sử dụng của nó với bạn có thể sử dụng các chú thích loại đơn giản trong điểm cuối: . Annotated DatabaseRepository[db_models.Ingredient] Depends(get_repository(db_models.Ingredient)) 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 của tệp . volumes 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 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: db_session Tạo một công cụ có URL cơ sở dữ liệu đã sửa đổi. 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. 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). Tạo và tạo ra một đối tượng phiên. Đó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: là phiên bản FastAPI từ tệp , với phần phụ thuộc được thay thế bằng phần cố định test_app alchemist/app.py get_db_session db_session là httpx sẽ thực hiện các yêu cầu API đối với client AsyncClient 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 chỉ hiển thị một thử nghiệm để tạo một thành phần: tests/test_api.py 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 Cũng được xuất bản . ở đây