FastAPI 유효성 검사

Request Body Validate

API 요청을 받으면 최소한으로 수행해야 하는 유효성 검사가 있습니다. 보안의 가장 기본인 SQL Injection 공격 이외에도 전달받은 데이터가 필요로 하는 최소한의 규칙 등이 있습니다. password를 예로 들자면 대소문자, 숫자, 특수문자를 포함해 최소 8에서 최대 20자리까지의 텍스트를 만족한다던가 혹은 전화번호는 숫자와 +- 등으로만 구성이 되어있는지 판단할 경우가 생기기 마련이죠. FastAPI 에서는 pydantic 이라는 Opensource 를 통해서 검증할 수가 있습니다.

BaseModel

pydantic 에서 제공하는 BaseModel과 Field를 이용하면 기본적인 유효성 검증을 손쉽게 처리할 수 있으며, 직접 유효성 검증을 하는 코드를 삽입하는 것 보다 훨씬 직관적이고 간편하게 사용할 수 있는 장점이 있습니다.

from pydantic import BaseModel

class ModelIn(BaseModel):

    password: str = Field(title='비밀번호', regex='^(?=.*[a-zA-Z])(?=.*[!@#$%^*+=-])(?=.*[0-9]).{8,20}$')
    id: Optional[int] = Field(title='ID', default=None, ge=0)

    class Config:
        orm_mode = True

위 예시는 password의 경우 대소문자와 숫자 그리고 특수문자를 포함한 8~20자리의 비밀번호를 허용, id는 null 혹은 0보다 큰 숫자만 허용한다는 유효성 검증의 예시입니다.

@validator

Field의 매개변수로 사용하는 gt, lt, min_length… 등만으로도 상당수의 유효성 검사를 만족할 수는 있지만, 사용자 정의 유효성 검사 및 개체 간의 복잡한 관계는 @ decorator를 통해 수행할 수 있습니다.

class UserModel(BaseModel):
    name: str = Field()
    email: str = Field()
    password: str = Field()

    @validator('name')
    def name_must_contain_space(cls, v):
        if ' ' not in v:
            raise ValueError('must contain a space')
        return v

위 코드는 변수 name에 대한 customized validator로 @validator({필드명}) 으로 처리 할 수 있습니다. 이때 필드명은 복수로 설정할 수 있으며 예시는 다음과 같습니다.

@validator('email', 'password')
def decrypt_fields(cls, raw):
    ...
    return

@root_validator

필드별 유효성 검증 외에도 요청받은 전체 모델의 유효성 검증을 해야 하는 경우도 필요합니다.
예를 들면 복수의 리스트를 전달받았을 때 전체 항목에서 중복된 값이 존재하는지 여부를 판별하는데 쓰일 수도 있죠.

class OrganizationModelIn(BaseModel):
    industries: List[IndustryModelIn] = Field()
    ...

    @root_validator
    def duplicate_not_allowed(cls, values):
        industries = values.get('industries')
        ids = [industry.id for industry in industries]
        id_set = set(ids)
        if len(id_set) != len(ids):
            raise ValueError('industry id`s duplicate not allowed')

        return values

E2EE (End to End Encryption)

패스워드나 기타 식별데이터는 HTTPS 프로토콜을 사용한다고 하더라도 암호화를 거쳐 종단 간에 통신이 이루어져야 합니다.
위에서 password 필드에 정규식 ^(?=.*[a-zA-Z])(?=.*[!@#$%^*+=-])(?=.*[0-9]).{8,20}$을 넣었지만, 암호화된 필드 값으로는 애초에 정규식을 판별할 수 없다는 문제가 있으며, 이에 pydantic에서는 validator decorator에 pre=키워드를 이용해 validate 함수를 먼저 호출한 후 필드에 적용된 정규식을 검사할 수 있도록 제공하고 있습니다.

@validator('email', 'password', pre=True)
def decrypt_fields(cls, raw):
    from app.core.utils.crypto import rsa
    try:
        dec = rsa.dec(raw)
    except Exception as e:
        from app.core.errors.exceptions import CommonExceptions
        raise CommonExceptions.InvalidEncryptionField(e)
    return dec

이렇게 validate 함수를 구현해 놓으면 1차로 decrypt_fields 함수를 먼저 실행, 복호화된 값을 토대로 다시 한번 Field에 적용된 validate 검사를 진행하게 됩니다.

Validator Sequence

validator decorator를 통해 유효성 검증을 하게 되면 @validator가 먼저 실행되고, @root_validator가 나중에 실행 되게 됩니다. 각 validate 함수에 print를 찍어보면 아래와 같은 결과가 나옵니다

@validator('email', 'password', pre=True)
def decrypt_fields(cls, raw):
    print(f'decrypt_fields')
    ...
    return dec

@root_validator
def root_validator(cls, values):
    print(f'root_validator')
    return values

decrypt_fields
decrypt_fields
root_validator

2%부족한 BaseModel

요청을 받은 Body 데이터 중에 industries의 id가 데이터베이스에 있는지를 확인해야 새로 생성된 모델과의 제대로 된 관계 설정을 할 수가 있습니다. Django의 drf를 통해 유효성 검사를 경험해 보았던 제게는 Serializer 안에서 접근할 수 있는 ORM 객체와 관련 함수에 대한 아쉬움이 상당히 많이 남아

async def some(session: Session = Depends(get_db)):
    ...
    return

등과 같이 Depends함수를 통해 database session 을 가져오는 것처럼 BaseModel에서도 가져올 수 있지 않을까라는 생각을 하며 여러 가지 시도를 해보았습니다.
하지만 pydantic에서 sqlalchemy를 하나의 커넥션으로 가져오는 방법에 드라마틱한 방식은 없다는 쪽으로 결론을 내고 Request Body의 유효성 검증이 모두 끝난 후 Depends를 통해 전체적으로 유효성 검증을 해 보았습니다.

def model_in(model: OrganizationModelIn, database: Session = Depends(get_db)):
    industries = model.industries
    errors = []
    from pydantic.error_wrappers import ErrorWrapper
    for idx, industry in enumerate(industries):
    if industry.id and 0 < industry.id and not is_exists_industry(database, industry.id):
        e = OrganizationExceptions.Join.UnDefinedIndustryId
        errors.append(ErrorWrapper(e(industry.id), loc=('body', 'industries', idx, 'id')), )

    if len(errors) != 0:
        raise ValidationException(errors=errors, model=model)
    return model, database

@router.post()
async def post_join(model_n_session=Depends(model_in)):
    ...
    return

crud에서 raise 처리를 하는 걸 고민해 보았으나, is_exists_industry함수는 industry.id에 1:1로 대응하는 게 낫겠다는 판단에 model_in이라고 하는 validate 함수를 만드는 방식으로 진행했습니다. 위와 같이 실제 router에 제공되는 메서드에서 model_in이라는 함수를 통해 session을 하나의 커넥션으로 묶어 실제 데이터베이스와의 유효성 검증을 수행하도록 해보았습니다.

CRUD 단에서 데이터를 Insert 할 때에 실제 유효한 ID를 전달받은 것인지 확인해 볼 수도 있었지만, CRUD 각각의 함수에 대한 테스트 문제, 그리고, pydantic이 기본적으로 매개변수 유효성 검사를 통과하지 못할 때 내려주는 422 에러와 동일한 형태인

{
  "detail": [
    {
      "loc": ["body", "industries", 0],
      "msg": "정의 되지 않은 Industry 객체의 ID({id})가 입력되었습니다",
      "type": "type_error"
    }
  ]
}

처럼 내려 주고 싶다는 고집때문에 위와 같은 형태를 고안해 보았네요.

이전 API와의 동질화

아이메디신의 API는 FastAPI 외에도 SpringBoot를 사용하고 있습니다. API Host와 Response는 어쩔 수 없이 다른 형태로 사용한다고 하더라도 FrontEnd 쪽에서 처리하는 변수명 명명 규칙을 최대한 변화를 주지 않도록 하기 위해서 CamelCase 문법을 따르도록 수정을 해주어야 하는 상황이 발생하였습니다. pydantic Field에는 alias=를 통해 내부에서 사용되는 변수명과 request / response에서 사용되는 변수명을 다르게 할 수 있습니다. 예를 들면

field_name = str = Field(alias='fieldName')

위와 같이 FastAPI 내부에서는 python의 변수 형태인 field_name 을 사용할 수 있으면서 외부에서 사용되는 변수명은 fieldName으로 사용할 수 있는 건데요. 모든 필드명에 이런 식으로 alias를 붙이는 건 효율성이 떨어진다는 문제점이 있어 pyhumps를 이용해 BaseModel을 상속받은 CamelModel을 사용해 보는 것으로 방향을 잡아 보았습니다.

from humps import camelize

class CamelModel(BaseModel):
    class Config:
        alias_generator = camelize
        allow_population_by_field_name = True

이제 BaseModel 대신 CamelModel을 impliment 하는 것으로 Feild(alias=)를 생략할 수가 있습니다.

class IndustryModelIn(CamelModel):
    id: Optional[PositiveInt] = Field(title='ID', default=None)
    industry_title: Optional[str] = Field(title='산업 명', default=None)

위 모델에 대응하는 Request Body는 다음과 같습니다

{
  "id": null,
  "industryTitle": "title"
}

마치며

밸리데이션을 처리하는 데 있어 어느 정도까지 깊게 구현하느냐는 API의 퍼포먼스와도 크게 영향이 있다 보니, 무턱대고 아무렇게나 살을 붙이기는 어려운 것 같습니다. 더욱이 표준을 중요시하는 제게 문서에서 추천하지 않는 디자인 패턴을 적용한다는 건 거부감이 상당히 많이 들기도 했지만, 어떻게든 응답의 형태를 맞추고 싶다는 욕망과의 싸움이 삽질을 한참이나 하게 된 이유가 된 것 같기도 하네요. 실제로 Django/DRF를 사용하다 API의 퍼포먼스 개선을 위해 FastAPI로 환경을 구성하였으나, 그 장점을 살리지 못한다면 FastAPI로 갈아탄 이유가 없다는 생각을 많이 하게 된 시간을 가졌습니다.