Thoughts on allowing structured classes as a querystring parameter definition

[DefiDebauchery]

This may be a little too over-engineered, but felt it was worth asking. Instead of baking the list of querysting params into the method definition, would it be possible and prudent to accept a structured dataclass (including a Pydantic class) as a defined object, passing that as query_params (or some other argument)?

class MySearchFilter(BaseModel):
    executed: Optional[str] = None
    trusted: Optional[str] = None
    transaction_hash: Optional[str] = None
    ordering: Optional[OrderEnum]
    limit: Optional[int] = 10
    address: str
    ...
    model_config = ConfigDict(use_enum_values=True)

class MyClient(Client):
    def __init__(self, session: ClientSession, txs_url: str) -> None:
        self.base_url = txs_url  # Related to #68 

    @endpoint('/v1/search')
    async def get_search(self, query_params: MySearchFilter) -> MySearchResults
        ...

client = MyClient(txs_url=...)

filter = MySearchFilter(executed=True, limit=5, address='0x1234')
results = await client.get_search(query_params=filter)

This way, the definition for the arguments remains defined, but doesn't overcomplicate the function signature. In turn, this could make building the querystrings a little easier on end-user code, as they can modify a "Filter" class or similar naming convention.

Granted, I imagine you could **unpack it yourself, but you'd still have to define them in the function signature, and if those options ever change, I feel I'd rather update a Filter spec than the signature, especially when you get over a few options (one API endpoint I work with has 27!).

[pmateusz]

Thank you for sharing this feature request. It is indeed something I would like to support.

I shared below the example behaviour I am considering.

@pytest.mark.asyncio()
async def test_get_with_marshaller(mock_tools) -> None:
    # GIVEN
    session = mock_tools.session_with_json_response(json=PRODUCTS)

    @dataclass
    class TimeRange:
        start: datetime.datetime
        end: datetime.datetime
    layout = "%Y-%m-%d"

    def format_time_range(time_range: TimeRange) -> dict[str, str]:
        return {
            "since": time_range.start.strftime(layout),
            "until": time_range.end.strftime(layout),
        }

    class Store(Client):
        def __init__(self) -> None:
            super().__init__(session)
        @endpoint("/api/v1/transactions")
        async def get_transactions(
            self, time_range: Annotated[TimeRange, api_ref(unwrap=format_time_range)]
        ) -> list[Any]:
            ...

    # WHEN
    async with Store() as api:
        result = await api.get_transactions(
            time_range=TimeRange(
                start=datetime.datetime(2006, 1, 2), end=datetime.datetime(2006, 1, 3)
            )
        )

    # THEN
    assert PRODUCTS == result
    session.request.assert_awaited_once_with(
        "GET", "/api/v1/transactions", params={"since": "2006-01-02", "until": "2006-01-03"}
    )

[pmateusz]

The improvement is available in v0.1.12

[DefiDebauchery]

Thank you for adding based on my suggestion! I am having some trouble resolving this for my example, and I just want to make sure I'm on the right track. The parameters I'll be working with will have many different datatypes instead of the clean, specific ones you posted, so I'm hoping something generic could be used here.

Taking my original example (modified to add a datetime field to match yours)

class MySearchFilter(BaseModel):
    ...
    execution_date__lte: Optional[datetime] = None
    limit: Optional[int] = 10
    address: str

    @field_serializer('execution_date__lte')
    def format_date(self, dt: datetime, _info):
        return dt.isoformat(timespec='seconds')

class MyClient(Client):
    def __init__(self, session: ClientSession, txs_url: str) -> None:
        self.base_url = txs_url

    def parse_querystring(self, model: BaseModel) -> dict[str, str]:
        return model.model_dump(mode='json', exclude_unset=True)

    @endpoint('/v1/search')
    async def get_search(self, query_params: Annotated[MySearchFilter, api_ref(unwrap=parse_querystring)]) -> MySearchResults
        ...

client = MyClient(txs_url=...)

filter = MySearchFilter(execution_date__lte: datetime.now(tz=UTC), limit=5, address='0x1234')
results = await client.get_search(query_params=filter)