직렬화 (Serialization)¶

요청 본문이나 응답은 @serialize 와 @deserialize 로 모델 클래스와 서로 변환할 수 있습니다. ahttp_client 는 dataclasses (추가 의존성 없음), pydantic, marshmallow 코덱을 기본으로 갖추고 있습니다.

응답을 Pydantic 모델로 역직렬화하기¶

from typing import Annotated

import aiohttp
from pydantic import BaseModel, ConfigDict
from pydantic.alias_generators import to_camel

from ahttp_client import AsyncSession, Query, get
from ahttp_client.serializer import deserialize


class Station(BaseModel):
    id: str
    name: str
    model_config = ConfigDict(alias_generator=to_camel)


class MetroAPI(AsyncSession):
    def __init__(self) -> None:
        super().__init__("https://api.yhs.kr", aiohttp.ClientSession)

    @get("/metro/station", directly_response=True)
    @deserialize(by_alias=True)
    async def search_stations(
        self, name: Annotated[str, Query]
    ) -> list[Station]:
        raise AssertionError("direct deserialization skips the method body")

@deserialize 는 메소드의 반환 애너테이션(list[Station])을 보고 Station 에 등록된 deserializer를 고른 다음, Response.model 로 응답에서 그 값을 읽어옵니다. directly_response=True 와 같이 쓰면 메소드 본문은 아예 실행되지 않고, 역직렬화된 모델이 곧바로 반환됩니다.

@serialize 는 방향만 반대일 뿐 동작은 똑같습니다. 반환 애너테이션 대신 요청 본문 전체를 나타내는 Body 애너테이션 매개변수에서 serializer를 고릅니다 (HTTP 구성요소(Component) 참고).

모델 타입 선택하기¶

애너테이션만으로 모델 타입을 추론할 수 없다면 @serialize(CreateUser) 나 @deserialize(Station) 처럼 model 을 직접 넘기면 됩니다. model 을 뺀 나머지 keyword arguments는 모델 타입이 정해진 뒤에 내부 serializer/deserializer로 넘어갑니다. 이게 바로 “late-bind” 방식인데, late_bind() 가 일단 옵션을 저장해두면, 애너테이션 타입을 알게 되는 시점에 set_model()/set_model() 이 실제 값을 확정합니다.

기본 제공 코덱¶

dataclasses¶

추가 의존성이 필요하지 않습니다. 중첩된 dataclass, 리스트는 물론 datetime, UUID, Decimal, Enum 같은 흔히 쓰이는 필드 타입까지, @dataclasses.dataclass 로 데코레이팅된 클래스는 모두 자동으로 지원됩니다.

from dataclasses import dataclass
from typing import Annotated

from ahttp_client import AsyncSession, Body, Response, post
from ahttp_client.serializer import serialize, deserialize


@dataclass
class CreateUser:
    name: str
    age: int


class UserAPI(AsyncSession):
    @post("/users")
    @serialize()
    @deserialize()
    async def create_user(
        self, response: Response, user: Annotated[CreateUser, Body]
    ) -> CreateUser:
        return response.model

pydantic¶

참고

pydantic 패키지가 필요합니다.

pip install pydantic

pydantic.BaseModel 서브클래스는 모두 지원됩니다. serializer/deserializer의 keyword arguments(by_alias, exclude_none, strict 등)는 pydantic.BaseModel.model_dump/TypeAdapter.validate_python 인자와 그대로 매핑되며, 자세한 옵션은 PydanticSerializeOptions 와 PydanticDeserializeOptions 를 참고하세요.

marshmallow¶

참고

marshmallow 패키지가 필요합니다.

pip install marshmallow

dataclasses나 pydantic 코덱과 달리, 스키마 자체는 데이터의 파이썬 타입이 아니기 때문에 marshmallow.Schema 인스턴스 는 항상 schema keyword argument로 직접 전달해야 합니다. 같은 이유로, Body/반환 애너테이션으로부터의 자동 추론에 의존하는 대신 스키마 클래스 도 model 에 함께 명시하는 것이 좋습니다 (평범한 dict 는 dataclass나 pydantic 모델과 달리 그 자체만으로는 어떤 스키마를 사용해야 할지 알 수 없습니다):

from typing import Annotated

from marshmallow import Schema, fields

from ahttp_client import AsyncSession, Body, Response, post
from ahttp_client.serializer import serialize, deserialize


class UserSchema(Schema):
    name = fields.Str()
    age = fields.Int()


class UserAPI(AsyncSession):
    @post("/users")
    @serialize(model=UserSchema, schema=UserSchema())
    @deserialize(model=UserSchema, schema=UserSchema())
    async def create_user(
        self, response: Response, user: Annotated[dict, Body]
    ) -> dict:
        return response.model

Reference¶

class BaseCodec¶

모델별 구현을 결정하는 코덱의 기본 클래스입니다.

property is_late_bind: bool¶: 이 코덱이 모델 타입을 기다리고 있는 상태인지 반환합니다.

classmethod is_model_type(model_type: type[Any]) → bool¶: 애너테이션에 설정된 모델 타입이 포함되어 있는지 반환합니다.

static is_sequence(value: Any) → bool¶: value 가 변환할 값들의 컬렉션인지 반환합니다.

classmethod late_bind(**kwargs: Any) → BaseCodec¶: 모델 타입이 확인될 때까지 옵션을 저장해두는 placeholder를 반환합니다.

class BaseSerializer¶

기반 클래스: BaseCodec, ABC, Generic[ModelT]

모델을 전송 가능한 값으로 변환하는 기본 코덱입니다.

classmethod from_model(model: type[ModelT], **kwargs: Any) → BaseSerializer[ModelT] | None¶: 모델 애너테이션에 등록된 serializer가 있으면 반환합니다.

multiple_serialize(model: Sequence[ModelT]) → list[Any]¶: 모델들의 시퀀스를 직렬화합니다.

serialize(model: ModelT | Sequence[ModelT]) → Any | list[Any]¶: 모델 하나 또는 모델들의 시퀀스를 직렬화합니다.

classmethod set_model(model: type[ModelT], origin_cls: BaseCodec, **kwargs: Any) → BaseSerializer[ModelT] | None¶

모델 애너테이션에 대해 late-bind된 serializer를 확정합니다.

keyword arguments는 origin_cls 에 저장된 옵션을 재정의(override)합니다.

예외 발생:: ValueError – origin_cls 가 late-bind 상태가 아닌 경우 발생합니다.

abstractmethod single_serialize(model: ModelT) → Any¶: 모델 하나를 직렬화합니다.

class BaseDeserializer¶

기반 클래스: BaseCodec, ABC, Generic[ModelT]

전송 가능한 값을 모델로 변환하는 기본 코덱입니다.

deserialize(data: Any | Sequence[Any]) → ModelT | list[ModelT]¶: 값 하나 또는 값들의 시퀀스를 역직렬화합니다.

classmethod from_model(model: type[ModelT], **kwargs: Any) → BaseDeserializer[ModelT] | None¶: 모델 애너테이션에 등록된 deserializer가 있으면 반환합니다.

get_data(response: Response) → Any¶

응답으로부터 deserializer의 입력값을 추출합니다.

응답 객체를 직접 사용하는 커스텀 deserializer는 이 항등(identity) 구현을 그대로 활용할 수 있습니다.

multiple_deserialize(data: Sequence[Any]) → list[ModelT]¶: 값들의 시퀀스를 모델들로 역직렬화합니다.

classmethod set_model(model: type[ModelT], origin_cls: BaseCodec, **kwargs: Any) → BaseDeserializer[ModelT] | None¶

모델 애너테이션에 대해 late-bind된 deserializer를 확정합니다.

keyword arguments는 origin_cls 에 저장된 옵션을 재정의(override)합니다.

예외 발생:: ValueError – origin_cls 가 late-bind 상태가 아닌 경우 발생합니다.

abstractmethod single_deserialize(data: Any) → ModelT¶: 값 하나를 모델로 역직렬화합니다.

serialize(model: type[ModelT] | None = None, **serializer_kwargs: Unpack[PydanticSerializeOptions]) → RequestDecorator[Any, Any]¶

serialize(model: type[ModelT] | None = None, **serializer_kwargs: Unpack[DataclassSerializeOptions]) → RequestDecorator[Any, Any]

serialize(model: type[ModelT] | None = None, **serializer_kwargs: Unpack[MarshmallowSerializeOptions]) → RequestDecorator[Any, Any]

본문 전체를 직렬화하도록 요청을 데코레이팅합니다.

model 이 비어 있으면, 본문 전체를 나타내는 매개변수의 애너테이션으로부터 serializer가 선택됩니다. serializer의 keyword arguments는 모델 타입이 결정된 뒤에 적용됩니다.

매개변수:

model (Optional[type[ModelT]]) – 등록된 serializer를 선택하는 데 사용되는 모델 타입입니다.
**serializer_kwargs – serializer 생성자로 전달되는 keyword arguments입니다.

예외 발생:

TypeError – 해당 모델 타입에 등록된 serializer가 없는 경우 발생합니다.

deserialize(model: type[ModelT] | None = None, **deserializer_kwargs: Unpack[PydanticDeserializeOptions]) → RequestDecorator[Any, Any]¶

deserialize(model: type[ModelT] | None = None, **deserializer_kwargs: Unpack[DataclassDeserializeOptions]) → RequestDecorator[Any, Any]

deserialize(model: type[ModelT] | None = None, **deserializer_kwargs: Unpack[MarshmallowDeserializeOptions]) → RequestDecorator[Any, Any]

HTTP 응답을 역직렬화하도록 요청을 데코레이팅합니다.

model 이 비어 있으면, 반환 애너테이션으로부터 deserializer가 선택됩니다. deserializer의 keyword arguments는 모델 타입이 결정된 뒤에 적용됩니다.

매개변수:

model (Optional[type[ModelT]]) – 등록된 deserializer를 선택하는 데 사용되는 모델 타입입니다.
**deserializer_kwargs – deserializer 생성자로 전달되는 keyword arguments입니다.

예외 발생:

TypeError – 해당 모델 타입에 등록된 deserializer가 없는 경우 발생합니다.