FastAPI Schema를 제대로 다루는 방법

FastAPI Schema를 제대로 다루는 방법

FastAPI에는 Schema라는 개념이 존재한다. 만약에 스프링이나 nestJS로 개발을 해봤던 사람이라면 DTO라는 이름이 더 익숙할 것이다. 간단하게 말하자면 DTO란 Data Transfer Object의 약자로서 어떤 메소드나 클래스간 객체정보를 주고 받을 때 특정 모양으로 주고 받겠다는 일종의 약속이다.

 

FastAPI의 스키마는 pydantic model에 종속돼있다. 말이 종속이지 그냥 pydantic 패키지를 그대로 갖다 쓴다고 해도 무리가 아닐 정도다. FastAPI의 스키마는 아래처럼 생겼다.

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    model: str
    manufacturer: str
    price: float
    tax: float

볼 수 있다시피 pydantic의 BaseModel을 상속받아 만들어지는 걸 알 수 있다. 이제 이걸 이용해 api endpoint를 구성해보자.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    model: str
    manufacturer: str
    price: float
    tax: float


@app.get("/")
async def root(
        item: Item
):
    return {"message": "Hello World"}

이렇게 기본 엔드포인트에서 Item이라는 자료형을 받겠다고 선언하면

이렇게 해당 자료형을 받겠다는 example이 자동으로 생성된다. 이것은 다른 방법으로도 활용할 수 있는데

from fastapi import FastAPI, Depends
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    model: str
    manufacturer: str
    price: float
    tax: float


@app.get("/")
async def root(
        item: Item = Depends()
):
    return {"message": "Hello World"}

이렇게 Depends라는 값을 schema에 등록하면

Body로 받겠다는 부분이 Parameter로 변경된 걸 볼 수 있다. 어디서 많이 본 느낌 아닌가? 맞다. search query를 구성할 때 흔히 쓰이는 방법이다. 하지만 위에서 볼 수 있다시피 required라는 제한이 걸려있는 걸 볼 수 있다. search query인데 모든 멤버에 required가 걸려있는 건 말이 안 된다. 그래서 이것을 모두 optional로 바꾸는 작업이 필요하다.

정말 쉽게는

from pydantic import BaseModel
from typing import Optional

class Item(BaseModel):
    name: Optional[str]
    model: Optional[str]
    manufacturer: Optional[str]
    price: Optional[float]
    tax: Optional[float]

이렇게 구성할 수 있지만 Item이라는 스키마의 기본값이 전부 옵셔널로 들어가면 나중에 PUT method 엔드포인트를 구성할 때 Optional을 전부 걷어내야하는 번거로움이 존재한다. 아, 여기서 PUT method와 PATCH method에 대한 명확한 구분이 안가는 사람이라면 이 글을 참조해보는 걸 추천한다. 간단히 말하자면 PUT은 어떤 한 요소의 전체 값을 바꾼다고 약속된 메소드이고 PATCH는 일부를 바꾸겠다는 메소드이다.

 

그래서 나는 이 방법을 쓴다.

from pydantic.main import ModelMetaclass

class AllOptional(ModelMetaclass):
    def __new__(self, name, bases, namespaces, **kwargs):
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            annotations.update(base.__annotations__)
        for field in annotations:
            if not field.startswith('__'):
                annotations[field] = Optional[annotations[field]]
        namespaces['__annotations__'] = annotations
        return super().__new__(self, name, bases, namespaces, **kwargs)

이 클래스는 메타클래스로 지정됐을 때 그 클래스의 멤버변수를 전부 Optional로 바꿔주는 기능을 갖고 있다.

 

from fastapi import FastAPI, Depends
from pydantic import BaseModel
from pydantic.main import ModelMetaclass
from typing import Optional

app = FastAPI()


class AllOptional(ModelMetaclass):
    def __new__(self, name, bases, namespaces, **kwargs):
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            annotations.update(base.__annotations__)
        for field in annotations:
            if not field.startswith('__'):
                annotations[field] = Optional[annotations[field]]
        namespaces['__annotations__'] = annotations
        return super().__new__(self, name, bases, namespaces, **kwargs)


class Item(BaseModel):
    name: str
    model: str
    manufacturer: str
    price: float
    tax: float


class OptionalItem(Item, metaclass=AllOptional):
    pass


@app.get("/")
async def root(
        item: OptionalItem = Depends()
):
    return {"message": "Hello World"}

pydantic BaseModel로 만들어지는 클래스를 상속받고 거기에 metaclass=AllOptional을 지정해주면 이렇게

Required가 벗겨진 parameters를 만날 수 있다.

 

 

하지만 실무에서 어떤 schema를 다룰 땐 넣을 땐 필요 없고 나중에 가져올 땐 필요한 것들이 있다. 흔하게 id, created_datetime, updated_datetime등이 그런 것이다. 그리고 때때로는 몇몇 멤버들을 제거하여 보여줘야할 때도 있다. spring이나 nest에서는 setFieldToIgnore함수나 Omit클래스를 사용함으로써 이런 상황에 유연하게 대처할 수 있지만 FastAPI를 다뤄본 사람은 알 수 있다시피 그런 게 없다.

 

그래서 정말 많은 시행착오 끝에 FastAPI를 기존에 쓰던 DTO처럼 사용하는 방법을 정립했다. 지금은 이게 베스트 프랙티스라고 해도 무관할 것이다. 왜냐. 수많은 예제를 찾아봤지만 완벽한 대안이 없어서 내가 직접 정의하고 만들었으니 말이다.

 

 

위에서 사용한 Item이라는 schema를 사용해서 예제를 작성해보자면 이렇다.

from pydantic import BaseModel

class BaseItem(BaseModel):
    name: str
    model: str
    manufacturer: str
    price: float
    tax: float

가장 기본이 되는 요소들을 BaseItem으로 정의한다. Upsert에 사용되는 요소들을 정의한다고 보면 된다. pydantic model은 기본적으로 Optional이나 =None으로 처리해주지 않으면 Required로 처리되기 때문에 PUT 메소드에 사용하는 스키마가 될 것이다.

class Item(BaseItem, metaclass=AllOptional):
    id: int
    created_datetime: datetime
    updated_datetime: datetime

BaseItem을 상속받는 Item이다. BaseModel을 상속받은 클래스를 또 상속받아도 상속받은 클래스 또한 pydantic model이 되므로 이렇게 간단하게 선언할 수 있다. 이것은 Upsert에 쓰이지 않는 Item의 기본 요소들을 포함해서 던져주는 데 사용된다. 이것은 주로 아이템을 조회할 때 쓰이는 스키마이므로 Optional을 달지 않아도 괜찮지 않을까...? 했지만 Optional이 아닌 필드가 있으면 response로 해당 모델을 불러올 때 Required로 선언된 멤버의 값이 null이나 None이면 에러를 내뱉는다. 사담이지만 Data를 보내는 게 아니라 가져올 때도 해당 모델에 값이 없으면 못불러온다? 이건 말이 안 된다고 본다. 나중에 Contribute할 수 있는 부분이 될 것 같다.

class FindBase(BaseModel):
    count: int
    page: int
    order: str


class FindItem(FindBase, BaseItem, metaclass=AllOptional):
    pass

Item을 찾는데 쓰이는 Parameter는 이렇게 구성할 수 있다. 위에서 선언한 BaseItem과 FindBase를 상속받고 metaclass=AllOptional로 모두 옵셔널로 처리해 패러미터의 required option을 벗겨내준다. 그리고 마지막으로

class Omit(ModelMetaclass):
    def __new__(self, name, bases, namespaces, **kwargs):
        omit_fields = getattr(namespaces.get("Config", {}), "omit_fields", {})
        fields = namespaces.get('__fields__', {})
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            fields.update(base.__fields__)
            annotations.update(base.__annotations__)
        merged_keys = fields.keys() & annotations.keys()
        [merged_keys.add(field) for field in fields]
        new_fields = {}
        new_annotations = {}
        for field in merged_keys:
            if not field.startswith('__') and field not in omit_fields:
                new_annotations[field] = annotations.get(field, fields[field].type_)
                new_fields[field] = fields[field]
        namespaces['__annotations__'] = new_annotations
        namespaces['__fields__'] = new_fields
        return super().__new__(self, name, bases, namespaces, **kwargs)

이건 Omit기능을 지원하는 metaclass를 내가 제작한 것이다. 분명히 필요함직 한데 없더라.

class BaseItem(BaseModel):
    name: str
    model: str
    manufacturer: str
    price: float
    tax: float


class OmittedTaxPrice(BaseItem, metaclass=Omit):
    class Config:
        omit_fields = {'tax', 'price'}

이렇게 class Config에 omit_fields를 선언해놓으면 해당 필드를 제거한 새로운 schema를 얻을 수 있다.

 

※그리고 정말주의할 점이 있다면 metaclass라는 것은 클래스가 생성될 때마다 실행되는 것이므로 상속관계에 엮인 class들끼리는 단 하나만 선언할 수 있다. 예를 들어

class BaseItem(BaseModel):
    name: str
    model: str
    manufacturer: str
    price: float
    tax: float


class Item(BaseItem, metaclass=AllOptional):
    id: int
    created_datetime: datetime
    updated_datetime: datetime



class TestItem(Item):
    additional_number: int

이렇게 metaclass=AllOptional이 선언된 Item을 TestItem에서 상속받으면 AllOptional은 두번 실행된다. Item Class를 만들 때 한 번, 그 Item을 상속받는 TestItem을 만들 때 또 한 번. 그래서 Item을 상속받는 TestItem의 멤버변수인 additional_number도 Optional이 된다. 그것이 metaclass이다. 그래서 만약에 여기서 다른 metaclass인 Omit을 지정하면 프로그램이 깨진다.

TypeError: metaclass conflict: the metaclass of a derived class must be a (non-strict) subclass of the metaclasses of all its bases

"하위클래스의 메타클래스는 상속받는 모든 베이스(부모) 메타클래스의 하위클래스여야 합니다"

예전에 이 에러를 만났을 땐 그냥 안되는 구나 하고 생각했었는데 지금 다시 보니까 metaclass들끼리도 상속관계가 명확하면 다중선언이 가능한건가 싶다. 나중에 실험해봐야지..

 

그러니까 metaclass는 무조건 schema로 사용되기 전 마지막 상속단계의 class에서 선언해서 사용하는게 가독성도 좋고 기능을 직관적으로 파악하기에도 쉽다.

 

뭐 아무튼 위에서 쭉 얘기한 내용을 자주 쓰이는 4개의 http method인 GET, POST, PUT, PATCH와 붙여서 써보면 이렇게 구성이 된다. 내 git repo에서도 확인할 수 있다. https://github.com/jujumilk3/fastapi-best-practice/tree/schema

from datetime import datetime
from fastapi import FastAPI, Depends
from pydantic import BaseModel
from pydantic.main import ModelMetaclass
from typing import Optional, List


app = FastAPI()


class AllOptional(ModelMetaclass):
    def __new__(self, name, bases, namespaces, **kwargs):
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            annotations.update(base.__annotations__)
        for field in annotations:
            if not field.startswith('__'):
                annotations[field] = Optional[annotations[field]]
        namespaces['__annotations__'] = annotations
        return super().__new__(self, name, bases, namespaces, **kwargs)


class Omit(ModelMetaclass):
    def __new__(self, name, bases, namespaces, **kwargs):
        omit_fields = getattr(namespaces.get("Config", {}), "omit_fields", {})
        fields = namespaces.get('__fields__', {})
        annotations = namespaces.get('__annotations__', {})
        for base in bases:
            fields.update(base.__fields__)
            annotations.update(base.__annotations__)
        merged_keys = fields.keys() & annotations.keys()
        [merged_keys.add(field) for field in fields]
        new_fields = {}
        new_annotations = {}
        for field in merged_keys:
            if not field.startswith('__') and field not in omit_fields:
                new_annotations[field] = annotations.get(field, fields[field].type_)
                new_fields[field] = fields[field]
        namespaces['__annotations__'] = new_annotations
        namespaces['__fields__'] = new_fields
        return super().__new__(self, name, bases, namespaces, **kwargs)


class BaseItem(BaseModel):
    name: str
    model: str
    manufacturer: str
    price: float
    tax: float


class UpsertItem(BaseItem, metaclass=AllOptional):
    pass


class Item(BaseItem, metaclass=AllOptional):
    id: int
    created_datetime: datetime
    updated_datetime: datetime


class OmittedTaxPrice(BaseItem, metaclass=Omit):
    class Config:
        omit_fields = {'tax', 'price'}


class FindBase(BaseModel):
    count: int
    page: int
    order: str


class FindItem(FindBase, BaseItem, metaclass=AllOptional):
    pass


@app.get("/")
async def root(
):
    return {"message": "Hello World"}


@app.get("/items/", response_model=List[Item])
async def find_items(
        find_query: FindItem = Depends()
):
    return {"hello": "world"}


@app.post("/items/", response_model=Item)
async def create_item(
        schema: UpsertItem
):
    return {"hello": "world"}


@app.patch("/items/{id}", response_model=Item)
async def update_item(
        id: int,
        schema: UpsertItem
):
    return {"hello": "world"}


@app.put("/items/{id}", response_model=Item)
async def put_item(
        id: int,
        schema: BaseItem
):
    return {"hello": "world"}


@app.get("/items/omitted", response_model=OmittedTaxPrice)
async def omitted_item(
        schema: OmittedTaxPrice
):
    return {"hello": "world"}

 

이렇게 구성하면 아래와 같은 swagger문서가 생성된다.

1. parameter로 구성된 find items endpoint

 

2. create item

 

3. PUT method

PUT METHOD의 사용법에 맞게 name이라는 항목을 누락시켜서 보내면 name좀 담아서 보내달라고 422에러가 반환된다. BaseItem을 Body로 받겠다는 선언으로 PUT을 구성한 것이다.

 

4. PATCH method

 

5. Omitted

 

이렇게 FastAPI schema를 다루는 방법을 알아보았다. 뭔가 사용하다보면 더 필요한 기능이 보일지 모르겠지만 일단 이정도로 필요한 엔드포인트와 docs구성은 무리 없이 진행할 수 있다. FastAPI의 schema를 적극적으로 사용하여 docs를 작성하려는 사람이라면 꼭 이를 참조하여 시행착오를 줄였으면 하는 바람이다.