Expand Up @@ -257,11 +257,11 @@ async def push_data(self, data: JsonSerializable, **kwargs: Unpack[PushDataKwarg """ # Handle singular items if not isinstance(data, list): items = await self._check_and_serialize(data) items = await self.check_and_serialize(data) return await self._resource_client.push_items(items, **kwargs)
# Handle lists payloads_generator = (await self._check_and_serialize(item, index) for index, item in enumerate(data)) payloads_generator = (await self.check_and_serialize(item, index) for index, item in enumerate(data))
# Invoke client in series to preserve the order of data async for items in self._chunk_by_size(payloads_generator): Expand Down Expand Up @@ -417,7 +417,8 @@ async def iterate_items( ): yield item
async def _check_and_serialize(self, item: JsonSerializable, index: int | None = None) -> str: @classmethod async def check_and_serialize(cls, item: JsonSerializable, index: int | None = None) -> str: """Serializes a given item to JSON, checks its serializability and size against a limit.
Args: Expand All @@ -438,8 +439,8 @@ async def _check_and_serialize(self, item: JsonSerializable, index: int | None = raise ValueError(f'Data item{s}is not serializable to JSON.') from exc
payload_size = ByteSize(len(payload.encode('utf-8'))) if payload_size > self._EFFECTIVE_LIMIT_SIZE: raise ValueError(f'Data item{s}is too large (size: {payload_size}, limit: {self._EFFECTIVE_LIMIT_SIZE})') if payload_size > cls._EFFECTIVE_LIMIT_SIZE: raise ValueError(f'Data item{s}is too large (size: {payload_size}, limit: {cls._EFFECTIVE_LIMIT_SIZE})')
return payload
Expand Down