Class StreamingBodyIO

src/load_item_metadata_function/main.py:30–40

Wrap a boto StreamingBody in the IOBase API.

table_name_prefix = os.environ.get('ItemsTableNamePrefix', 'PersonalizationApiItemMetadata_')

class StreamingBodyIO(RawIOBase):
    """Wrap a boto StreamingBody in the IOBase API."""
    def __init__(self, body):
        self.body = body

    def readable(self):
        return True

    def read(self, n: int = -1):
        n = None if n < 0 else n
        return self.body.read(n)

@tracer.capture_method
def bulk_load_datastore(bucket: str, key: str, namespace: str):