def py_to_aws()

in hasher-matcher-actioner/hmalib/common/aws_dataclass.py [0:0]

• 45 lines of code
• 25 McCabe index (conditional complexity)

def py_to_aws(py_field: t.Any, in_type: t.Optional[t.Type[T]] = None) -> T:
    """
    Convert a py item into its AWS equivalent.

    Should exactly inverse aws_to_py
    """
    if in_type is None:
        in_type = type(py_field)
    origin = t.get_origin(in_type)
    args = t.get_args(in_type)

    check_type = origin or in_type

    if isinstance(check_type, t.ForwardRef):
        raise AWSSerializationFailure(
            "Serialization error: "
            f"Expected no forward refs, but detected {check_type}. "
            "Rework your dataclasses to avoid forward references."
        )

    if not isinstance(py_field, check_type):
        raise AWSSerializationFailure(
            "Serialization error: "
            f"Expected {check_type} got {type(py_field)} ({py_field!r})"
        )

    if in_type == int:  # N
        # Technically, this also needs to be converted to decimal,
        # but the boto3 translater seems to handle it fine
        return py_field  # type: ignore # mypy/issues/10003
    if in_type == float:  # N
        # WARNING WARNING
        # floating point is not truly supported in dynamodb
        # We can fake it for numbers without too much precision
        # but Decimal("3.4") != float(3.4)
        return Decimal(str(py_field))  # type: ignore # mypy/issues/10003
    if in_type == Decimal:  # N
        return py_field  # type: ignore # mypy/issues/10003
    if in_type == str:  # S
        return py_field  # type: ignore # mypy/issues/10003
    if in_type == bool:  # BOOL
        return py_field  # type: ignore # mypy/issues/10003
    if in_type == t.Set[str]:  # SS
        return py_field  # type: ignore # mypy/issues/10003
    if in_type == t.Set[int]:  # SN
        return {i for i in py_field}  # type: ignore # mypy/issues/10003
    if in_type == t.Set[float]:  # SN
        # WARNING WARNING
        # floating point is not truly supported in dynamodb
        # See note above
        return {Decimal(str(s)) for s in py_field}  # type: ignore # mypy/issues/10003

    if origin is list:  # L
        return [py_to_aws(v, args[0]) for v in py_field]  # type: ignore # mypy/issues/10003
    # various simple collections that don't fit into a
    # special cases above can likely be coerced into list.
    if origin is set:  # L - Special case
        return [py_to_aws(v, args[0]) for v in py_field]  # type: ignore # mypy/issues/10003

    if origin is dict and args[0] is str:  # M
        return {k: py_to_aws(v, args[1]) for k, v in py_field.items()}  # type: ignore # mypy/issues/10003
    if is_dataclass(in_type):
        return {
            field.name: py_to_aws(getattr(py_field, field.name), field.type)
            for field in fields(in_type)
        }  # type: ignore # mypy/issues/10003

    raise AWSSerializationFailure(f"Missing Serialization logic for {in_type!r}")