FastAPI와 함께 SQLAlchemy 2.0을 사용하기 위한 패턴 및 사례

Django와 Flask는 많은 Python 엔지니어에게 첫 번째 선택으로 남아 있지만 FastAPI는 이미 부인할 수 없을 만큼 신뢰할 수 있는 선택으로 인식되었습니다. 이는 개발자에게 백엔드 애플리케이션 구축을 위한 무한한 가능성을 제공하는 매우 유연하고 잘 최적화된 구조화된 프레임워크입니다.

데이터베이스 작업은 대부분의 백엔드 애플리케이션에서 필수적인 측면입니다. 결과적으로 ORM은 백엔드 코드에서 중요한 역할을 합니다. 그러나 Django와 달리 FastAPI에는 ORM이 내장되어 있지 않습니다. 적합한 라이브러리를 선택하고 이를 코드베이스에 통합하는 것은 전적으로 개발자의 책임입니다.

Python 엔지니어는 SQLAlchemy를 가장 널리 사용되는 ORM으로 널리 간주합니다. 이는 2006년부터 사용되어 수천 개의 프로젝트에 채택된 전설적인 라이브러리입니다. 2023년에는 버전 2.0으로 대규모 업데이트가 이루어졌습니다. FastAPI와 마찬가지로 SQLAlchemy는 개발자에게 특정 방식으로 사용하도록 강요하지 않고도 강력한 기능과 유틸리티를 제공합니다. 기본적으로 이는 개발자가 적합하다고 생각하는 대로 사용할 수 있도록 지원하는 다용도 툴킷입니다.

FastAPI와 SQLAlchemy는 천생연분입니다. 둘 다 강력하고 고유한 응용 프로그램을 만들 수 있는 안정적이고 성능이 뛰어난 최신 기술입니다. 이 문서에서는 SQLAlchemy 2.0을 ORM으로 활용하는 FastAPI 백엔드 애플리케이션을 만드는 방법을 살펴봅니다. 내용은 다음과 같습니다.

• Mapped 및 mapped_column 사용하여 모델 구축
• 추상 모델 정의
• 데이터베이스 세션 처리
• ORM을 사용하여
• 모든 모델에 대한 공통 저장소 클래스 생성
• 테스트 설정 준비 및 테스트 추가

이후에는 FastAPI 애플리케이션을 SQLAlchemy ORM과 쉽게 결합할 수 있습니다. 또한 잘 구조화되고 강력하며 성능이 뛰어난 애플리케이션을 만들기 위한 모범 사례와 패턴을 익히게 됩니다.

전제 조건

기사에 포함된 코드 예제는 재료 및 물약 개체를 생성하고 읽기 위한 기본 API인 alchemist 프로젝트에서 가져온 것입니다. 이 기사의 주요 초점은 FastAPI와 SQLAlchemy의 조합을 탐색하는 것입니다. 다음과 같은 다른 주제는 다루지 않습니다.

• Docker 설정 구성
• uvicorn 서버 시작
• 린트 설정

이러한 주제에 관심이 있다면 코드베이스를 검토하여 직접 탐색해 볼 수 있습니다. 연금술사 프로젝트의 코드 저장소에 액세스하려면 여기 링크를 따르십시오. 또한 아래에서 프로젝트의 파일 구조를 찾을 수 있습니다.

 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

트리가 커 보일 수도 있지만 일부 내용은 이 기사의 요점과 관련이 없습니다. 또한 특정 영역에서는 코드가 필요한 것보다 간단하게 나타날 수 있습니다. 예를 들어, 프로젝트에는 다음이 부족합니다.

• Dockerfile의 프로덕션 단계
• 마이그레이션을 위한 Alembic 설정
• 테스트용 하위 디렉터리

이는 복잡성을 줄이고 불필요한 오버헤드를 피하기 위해 의도적으로 수행되었습니다. 그러나 보다 생산 준비가 완료된 프로젝트를 처리하는 경우 이러한 요소를 염두에 두는 것이 중요합니다.

API 요구사항

앱 개발을 시작할 때 앱에서 사용할 모델을 고려하는 것이 중요합니다. 이러한 모델은 앱이 작동하고 API에 노출될 객체 와 엔터티를 나타냅니다. 연금술사 앱의 경우 재료와 물약이라는 두 가지 항목이 있습니다. API는 이러한 엔터티를 생성하고 검색할 수 있어야 합니다. alchemist/api/models.py 파일에는 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는 Ingredient 및 Potion 모델을 반환합니다. 구성에서 orm_mode True 로 설정하면 나중에 SQLAlchemy 개체를 사용하는 것이 더 쉬워집니다. Payload 모델은 새로운 객체를 생성하는 데 활용됩니다.

pydantic을 사용하면 클래스의 역할과 기능이 더욱 상세하고 명확해집니다. 이제 데이터베이스 모델을 생성할 차례입니다.

모델 선언

모델은 본질적으로 무언가를 표현한 것 입니다. API의 맥락에서 모델은 백엔드가 요청 본문에서 기대하는 것과 응답 데이터에서 반환할 것을 나타냅니다. 반면에 데이터베이스 모델은 더 복잡하며 데이터베이스에 저장된 데이터 구조 와 이들 간의 관계 유형을 나타냅니다.

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", )

SQLAlchemy의 모든 모델은 DeclarativeBase 클래스로 시작됩니다. 이를 상속하면 Python 유형 검사기와 호환되는 데이터베이스 모델을 구축할 수 있습니다.

모든 모델에 필요한 필드를 포함하는 추상 모델(이 경우 Base 클래스)을 만드는 것도 좋은 방법입니다. 이러한 필드에는 모든 개체의 고유 식별자인 기본 키가 포함됩니다. 추상 모델은 객체가 생성되거나 업데이트될 때 자동으로 설정되는 객체의 생성 및 업데이트 날짜도 저장하는 경우가 많습니다. 그러나 Base 모델은 단순하게 유지됩니다.

Ingredient 모델로 이동하면 __tablename__ 속성은 데이터베이스 테이블의 이름을 지정하고, name 필드는 새로운 SQLAlchemy 구문을 사용하여 유형 주석으로 모델 필드를 선언할 수 있습니다. 이 간결하고 현대적인 접근 방식은 name 필드를 문자열로 인식하므로 유형 검사기와 IDE에 강력하면서도 유리합니다.

Potion 모델에서는 상황이 더욱 복잡해집니다. __tablename__ 및 name 속성도 포함되어 있지만 그 외에 재료와의 관계도 저장됩니다. Mapped[list["Ingredient"]] 사용하면 물약에 여러 성분이 포함될 수 있으며 이 경우 관계는 다대다(M2M)입니다. 이는 단일 성분이 여러 물약에 할당될 수 있음을 의미합니다.

M2M에는 일반적으로 두 엔터티 간의 연결을 저장하는 연관 테이블 생성과 관련된 추가 구성이 필요합니다. 이 경우, potion_ingredient_association 개체는 재료와 물약의 식별자만 저장하지만 물약에 필요한 특정 재료의 양과 같은 추가 속성도 포함할 수 있습니다.

relationship 기능은 물약과 그 성분 간의 관계를 구성합니다. lazy 인수는 관련 항목을 로드하는 방법을 지정합니다. 즉, 물약을 가져올 때 SQLAlchemy는 관련 재료로 무엇을 해야 할까요? selectin 으로 설정하면 재료가 물약과 함께 로드되므로 코드에서 추가 쿼리가 필요하지 않습니다.

ORM을 사용할 때는 잘 디자인된 모델을 구축하는 것이 중요합니다. 이 작업이 완료되면 다음 단계는 데이터베이스와의 연결을 설정하는 것입니다.

세션 핸들러

데이터베이스 작업 시, 특히 SQLAlchemy를 사용할 때 다음 개념을 이해하는 것이 중요합니다.

• 방언
• 엔진
• 연결
• 연결 풀
• 세션

이 모든 용어 중에서 가장 중요한 것은 엔진 입니다. SQLAlchemy 문서에 따르면 엔진 개체는 데이터베이스 연결 및 동작을 용이하게 하기 위해 Pool 과 Dialect 연결하는 역할을 담당합니다. 간단히 말해서, 엔진 개체는 데이터베이스 연결의 소스이고, 연결은 SQL 문 실행, 트랜잭션 관리, 데이터베이스에서 결과 검색과 같은 높은 수준의 기능을 제공합니다.

세션은 단일 트랜잭션 내에서 관련 작업을 그룹화하는 작업 단위입니다. 이는 기본 데이터베이스 연결에 대한 추상화이며 연결 및 트랜잭션 동작을 효율적으로 관리합니다.

Dialect 는 특정 데이터베이스 백엔드에 대한 지원을 제공하는 구성 요소입니다. SQLAlchemy와 데이터베이스 사이의 중개자 역할을 하며 통신 세부 사항을 처리합니다. 연금술사 프로젝트는 Postgres를 데이터베이스로 사용하므로 방언은 이 특정 데이터베이스 유형과 호환되어야 합니다.

마지막 물음표는 연결 풀 입니다. SQLAlchemy의 맥락에서 연결 풀은 데이터베이스 연결 모음을 관리하는 메커니즘입니다. 각 요청에 대해 새 연결을 생성하는 대신 기존 연결을 재사용하여 데이터베이스 작업의 성능과 효율성을 향상시키도록 설계되었습니다. 연결 풀은 연결을 재사용함으로써 새 연결을 설정하고 해제하는 오버헤드를 줄여 성능을 향상시킵니다.

해당 지식을 다루었으므로 이제 데이터베이스 연결을 위한 종속성으로 사용될 함수가 포함된 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

주목해야 할 첫 번째 중요한 세부 사항은 get_db_session 함수가 생성기 함수라는 것입니다. 이는 FastAPI 종속성 시스템이 생성기를 지원하기 때문입니다. 결과적으로 이 함수는 성공 및 실패 시나리오를 모두 처리할 수 있습니다.

get_db_session 함수의 처음 두 줄은 데이터베이스 엔진과 세션을 생성합니다. 그러나 세션 개체를 컨텍스트 관리자로 사용할 수도 있습니다. 이를 통해 잠재적인 예외와 성공적인 결과를 더 효과적으로 제어할 수 있습니다.

SQLAlchemy가 연결 닫기를 처리하지만 연결이 완료된 후 연결을 처리하는 방법을 명시적으로 선언하는 것이 좋습니다. get_db_session 함수에서는 모든 것이 잘 되면 세션이 커밋되고 예외가 발생하면 롤백됩니다.

이 코드는 asyncio 확장을 중심으로 구축되었다는 점에 유의하는 것이 중요합니다. SQLAlchemy의 이 기능을 사용하면 앱이 데이터베이스와 비동기적으로 상호 작용할 수 있습니다. 이는 데이터베이스에 대한 요청이 다른 API 요청을 차단하지 않아 앱을 더욱 효율적으로 만든다는 의미입니다.

모델과 연결이 설정되면 다음 단계는 모델이 데이터베이스에 추가되었는지 확인하는 것입니다.

빠른 마이그레이션

SQLAlchemy 모델은 데이터베이스의 구조를 나타냅니다. 그러나 단순히 생성한다고 해서 데이터베이스가 즉시 변경되는 것은 아닙니다. 변경하려면 먼저 적용 해야 합니다. 이는 일반적으로 모든 모델을 추적하고 그에 따라 데이터베이스를 업데이트하는 alembic과 같은 마이그레이션 라이브러리를 사용하여 수행됩니다.

이 시나리오에서는 모델에 대한 추가 변경이 계획되어 있지 않으므로 기본 마이그레이션 스크립트로 충분합니다. 다음은 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())

간단히 말하면 migrate_tables 함수는 모델의 구조를 읽고 SQLAlchemy 엔진을 사용하여 데이터베이스에 이를 다시 생성합니다. 이 스크립트를 실행하려면 python scripts/migrate.py 명령을 사용하십시오.

모델은 이제 코드와 데이터베이스 모두에 존재하며 get_db_session 데이터베이스와의 상호 작용을 용이하게 할 수 있습니다. 이제 API 로직 작업을 시작할 수 있습니다.

ORM을 사용한 API

앞서 언급했듯이 재료와 물약에 대한 API는 다음 세 가지 작업을 지원하도록 되어 있습니다.

• 객체 생성
• 객체 나열
• ID로 객체 검색

사전 준비 덕분에 SQLAlchemy를 ORM으로, FastAPI를 웹 프레임워크로 사용하여 이러한 모든 기능을 이미 구현할 수 있습니다. 시작하려면 alchemist/api/v1/routes.py 파일에 있는 성분 API를 검토하세요.

 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)

/ingredients API에는 세 가지 경로를 사용할 수 있습니다. POST 엔드포인트는 이전에 생성된 모델 및 데이터베이스 세션의 객체로 재료 페이로드를 사용합니다. get_db_session 생성기 함수는 세션을 초기화하고 데이터베이스 상호 작용을 활성화합니다.

실제 함수 본문에는 5가지 단계가 수행됩니다.

1. 들어오는 페이로드에서 성분 객체가 생성됩니다.
2. 세션 개체의 add 메소드는 세션 추적 시스템에 구성 요소 개체를 추가하고 데이터베이스에 삽입하기 위해 보류 중인 것으로 표시합니다.
3. 세션이 커밋되었습니다.
4. 해당 속성이 데이터베이스 상태와 일치하는지 확인하기 위해 성분 개체가 새로 고쳐집니다.
5. 데이터베이스 구성요소 인스턴스는 from_orm 메소드를 사용하여 API 모델 인스턴스로 변환됩니다.

빠른 테스트를 위해 실행 중인 앱에 대해 간단한 컬을 실행할 수 있습니다.

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

응답에는 데이터베이스에서 오는 ID를 가진 성분 개체가 있어야 합니다.

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

SQLAlchemy의 여러 추상화 계층은 간단한 API에 불필요해 보일 수 있지만 ORM 세부 정보를 분리하여 유지하고 SQLAlchemy의 효율성과 확장성에 기여합니다. asyncio와 결합하면 ORM 기능이 API에서 매우 잘 작동합니다.

나머지 두 끝점은 덜 복잡하고 유사점을 공유합니다. 더 깊이 설명할 가치가 있는 부분 중 하나는 get_ingredients 함수 내에서 scalars 메서드를 사용하는 것입니다. SQLAlchemy를 사용하여 데이터베이스를 쿼리하는 동안 execute 메서드는 쿼리와 함께 인수로 사용되는 경우가 많습니다. execute 메소드는 행과 같은 튜플을 반환하지만 scalars ORM 엔터티를 직접 반환하여 엔드포인트를 더 깔끔하게 만듭니다.

이제 동일한 파일에서 물약 API를 고려하십시오.

 @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)

물약의 GET 끝점은 재료의 끝점과 동일합니다. 그러나 POST 함수에는 추가 코드가 필요합니다. 이는 물약을 만들려면 최소한 하나의 재료 ID를 포함해야 하기 때문입니다. 즉, 재료를 가져와서 새로 만든 물약에 연결해야 한다는 뜻입니다. 이를 달성하기 위해 scalars 메서드가 다시 사용되지만 이번에는 가져온 재료의 ID를 지정하는 쿼리가 사용됩니다. 물약을 만드는 나머지 과정은 재료와 동일합니다.

엔드포인트를 테스트하기 위해 다시 컬 명령을 실행할 수 있습니다.

 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"}'

결과적으로 다음과 같은 응답이 발생합니다.

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

관계에 지정된 lazy="selectin" 인수 덕분에 각 성분이 물약 내에서 완전한 개체로 표시된다는 점에 유의하는 것이 중요합니다.

API는 기능적이지만 코드에 심각한 문제가 있습니다. SQLAlchemy는 원하는 대로 데이터베이스와 상호 작용할 수 있는 자유를 제공하지만 Django의 Model.objects 와 유사한 높은 수준의 "관리자" 유틸리티를 제공하지 않습니다. 결과적으로, 이를 직접 생성해야 하며 이는 본질적으로 재료 및 물약 API에 사용되는 논리입니다. 하지만 이 로직을 별도의 공간에 추출하지 않고 엔드포인트에 직접 보관하게 되면 중복된 코드가 많이 발생하게 됩니다. 또한 쿼리나 모델을 변경하면 관리하기가 점점 더 어려워집니다.

다음 장에서는 ORM 코드 추출을 위한 우아한 솔루션인 저장소 패턴을 소개합니다.

저장소

저장소 패턴을 사용하면 데이터베이스 작업의 세부 사항을 추상화할 수 있습니다. 연금술사의 예와 같이 SQLAlchemy를 사용하는 경우 저장소 클래스는 여러 모델을 관리하고 데이터베이스 세션과 상호 작용하는 일을 담당합니다.

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

DatabaseRepository 클래스는 이전에 엔드포인트에 포함된 모든 논리를 보유합니다. 차이점은 특정 모델 클래스를 __init__ 메서드에 전달할 수 있어 각 엔드포인트에서 코드를 복제하는 대신 모든 모델에 대해 코드를 재사용할 수 있다는 것입니다.

또한 DatabaseRepository 는 추상 데이터베이스 모델에 바인딩된 Model 일반 유형과 함께 Python 제네릭을 사용합니다. 이를 통해 저장소 클래스는 정적 유형 검사를 통해 더 많은 이점을 얻을 수 있습니다. 특정 모델과 함께 사용하는 경우 저장소 메서드의 반환 유형은 이 특정 모델을 반영합니다.

저장소는 데이터베이스 세션을 사용해야 하기 때문에 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

간단히 말해서, get_repository 함수는 종속성 팩토리입니다. 먼저 리포지토리를 사용할 데이터베이스 모델을 선택합니다. 그런 다음 데이터베이스 세션을 수신하고 리포지토리 개체를 초기화하는 데 사용되는 종속성을 반환합니다. 더 잘 이해하려면 alchemist/api/v2/routes.py 파일에서 새 API를 확인하세요. POST 끝점만 표시되지만 코드가 어떻게 개선되는지에 대한 명확한 아이디어를 제공하는 데 충분합니다.

 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)

주목해야 할 첫 번째 중요한 기능은 FastAPI 종속성을 사용하는 새로운 방법인 Annotated 사용하는 것입니다. 종속성의 반환 유형을 DatabaseRepository[db_models.Ingredient] 로 지정하고 Depends(get_repository(db_models.Ingredient)) 사용하여 사용법을 선언하면 엔드포인트에서 간단한 유형 주석을 사용할 수 있습니다: repository: IngredientRepository .

저장소 덕분에 엔드포인트는 모든 ORM 관련 부담을 저장할 필요가 없습니다. 더 복잡한 물약의 경우에도 두 개의 저장소를 동시에 사용하기만 하면 됩니다.

두 개의 저장소를 초기화하면 세션이 두 번 초기화되는지 궁금할 수 있습니다. 내 대답은 아니오 야. FastAPI 종속성 시스템은 단일 요청에서 동일한 종속성 호출을 캐시합니다. 이는 세션 초기화가 캐시되고 두 저장소 모두 정확히 동일한 세션 개체를 사용함을 의미합니다. SQLAlchemy와 FastAPI 조합의 또 다른 뛰어난 기능입니다.

API는 완벽하게 작동하며 재사용 가능한 고성능 데이터 액세스 계층을 갖추고 있습니다. 다음 단계는 몇 가지 엔드투엔드 테스트를 작성하여 요구 사항이 충족되는지 확인하는 것입니다.

테스트

테스트는 소프트웨어 개발에서 중요한 역할을 합니다. 프로젝트에는 단위, 통합 및 E2E(엔드 투 엔드) 테스트가 포함될 수 있습니다. 일반적으로 의미 있는 단위 테스트를 많이 수행하는 것이 가장 좋지만, 전체 워크플로가 올바르게 작동하는지 확인하기 위해 최소한 몇 개의 E2E 테스트를 작성하는 것도 좋습니다.

연금술사 앱에 대한 일부 E2E 테스트를 생성하려면 두 개의 추가 라이브러리가 필요합니다.

• 실제로 테스트를 생성하고 실행하는 pytest
• 테스트 내에서 비동기 요청을 수행하는 httpx

이것들이 설치되면 다음 단계는 별도의 테스트 데이터베이스를 마련하는 것입니다. 기본 데이터베이스가 오염되거나 삭제되는 것을 원하지 않습니다. Alchemist에는 Docker 설정이 포함되어 있으므로 두 번째 데이터베이스를 생성하려면 간단한 스크립트만 필요합니다. scripts/create_test_db.sh 파일의 코드를 살펴보세요.

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

스크립트를 실행하려면 Postgres 컨테이너에 볼륨으로 추가해야 합니다. docker-compose.yaml 파일의 volumes 섹션에 이를 포함하면 됩니다.

준비의 마지막 단계는 tests/conftest.py 파일 내에 pytest 픽스처를 만드는 것입니다.

 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

테스트에서 변경해야 할 필수 사항 중 하나는 앱이 데이터베이스와 상호 작용하는 방식입니다. 여기에는 데이터베이스 URL을 변경하는 것뿐만 아니라 빈 데이터베이스로 시작하여 각 테스트를 격리시키는 것도 포함됩니다.

db_session 설비는 이 두 가지 목표를 모두 달성합니다. 본체는 다음 단계를 수행합니다.

1. 수정된 데이터베이스 URL로 엔진을 생성합니다.
2. 테스트에 깨끗한 데이터베이스가 있는지 확인하려면 기존 테이블을 모두 삭제하세요.
3. 데이터베이스 내부에 모든 테이블을 생성합니다(마이그레이션 스크립트와 동일한 코드).
4. 세션 객체를 생성하고 생성합니다.
5. 테스트가 완료되면 세션을 수동으로 닫습니다.

마지막 단계를 컨텍스트 관리자로 구현할 수도 있지만 이 경우에는 수동으로 닫기가 제대로 작동합니다.

나머지 두 개의 고정 장치는 설명이 필요하지 않습니다.

• test_app 은 alchemist/app.py 파일의 FastAPI 인스턴스이며, get_db_session 종속성은 db_session 고정 장치로 대체되었습니다.
• client 는 test_app 에 대해 API 요청을 만드는 httpx AsyncClient 입니다.

이 모든 설정이 완료되면 최종적으로 실제 테스트를 작성할 수 있습니다. 간결성을 위해 아래 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

테스트에서는 재정의된 종속성을 사용하여 FastAPI 인스턴스에 요청하는 픽스처에서 생성된 클라이언트 개체를 사용합니다. 결과적으로 테스트는 테스트가 완료된 후 지워지는 별도의 데이터베이스와 상호 작용할 수 있습니다. 두 API의 나머지 테스트 스위트 구조는 거의 동일합니다.

요약

FastAPI와 SQLAlchemy는 현대적이고 강력한 백엔드 애플리케이션을 만들기 위한 탁월한 기술입니다. 그들이 제공하는 자유, 단순성 및 유연성은 Python 기반 프로젝트를 위한 최고의 옵션 중 하나입니다. 개발자가 모범 사례와 패턴을 따르면 데이터베이스 작업과 API 논리를 쉽게 처리하는 성능이 뛰어나고 강력하며 잘 구조화된 애플리케이션을 만들 수 있습니다. 이 기사는 이 놀라운 조합을 설정하고 유지하는 방법에 대한 좋은 이해를 제공하는 것을 목표로 했습니다.

출처

연금술사 프로젝트의 소스 코드는 여기에서 찾을 수 있습니다: link .