Module `aws_lambda_powertools.utilities.batch.types`

Expand source code

#
# type specifics
#
import sys
from typing import List, Optional, Type, Union

from typing_extensions import TypedDict

has_pydantic = "pydantic" in sys.modules

# For IntelliSense and Mypy to work, we need to account for possible SQS subclasses
# We need them as subclasses as we must access their message ID or sequence number metadata via dot notation
if has_pydantic:
    from aws_lambda_powertools.utilities.parser.models import DynamoDBStreamRecordModel
    from aws_lambda_powertools.utilities.parser.models import (
        KinesisDataStreamRecord as KinesisDataStreamRecordModel,
    )
    from aws_lambda_powertools.utilities.parser.models import SqsRecordModel

    BatchTypeModels = Optional[
        Union[Type[SqsRecordModel], Type[DynamoDBStreamRecordModel], Type[KinesisDataStreamRecordModel]]
    ]
    BatchSqsTypeModel = Optional[Type[SqsRecordModel]]
else:
    BatchTypeModels = "BatchTypeModels"  # type: ignore
    BatchSqsTypeModel = "BatchSqsTypeModel"  # type: ignore


class PartialItemFailures(TypedDict):
    itemIdentifier: str


class PartialItemFailureResponse(TypedDict):
    batchItemFailures: List[PartialItemFailures]

Classes

class PartialItemFailureResponse (*args, **kwargs)

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Expand source code

class PartialItemFailureResponse(TypedDict):
    batchItemFailures: List[PartialItemFailures]

Ancestors

builtins.dict

Class variables

var batchItemFailures : List[PartialItemFailures]

class PartialItemFailures (*args, **kwargs)

Expand source code

class PartialItemFailures(TypedDict):
    itemIdentifier: str

Ancestors

builtins.dict

Class variables

var itemIdentifier : str