def transform_data()

in aws/lambda/github-webhook-rds-sync/utils.py [0:0]

• 38 lines of code
• 16 McCabe index (conditional complexity)

def transform_data(obj: Dict[str, Any]) -> Dict[str, Any]:
    """
    Run in-place transformations on obj to translate fields into the appropriate
    type for storage:
        * lists -> JSON encoded data
        * dates -> Python datetimes
    """
    for key, value in obj.items():
        if value is None:
            # Don't bother writing nulls, they can mess with object fields
            continue
        if isinstance(value, list):
            obj[key] = json.dumps(value)
        elif is_date(key, value) and value is not None:
            if isinstance(value, int):
                # convert from timestamp
                obj[key] = datetime.datetime.fromtimestamp(value)
            elif isinstance(value, datetime.datetime):
                obj[key] = value
            elif isinstance(value, str):
                formats = [
                    "%Y-%m-%dT%H:%M:%SZ",
                    "%Y-%m-%dT%H:%M:%S%z",
                    "%Y-%m-%dT%H:%M:%S.%fZ",
                    "%Y-%m-%dT%H:%M:%S.%f%z",
                ]
                date = None

                for format in formats:
                    try:
                        date = datetime.datetime.strptime(value, format)
                    except ValueError:
                        pass

                if date is None:
                    raise RuntimeError(value)
                obj[key] = date
            else:
                raise RuntimeError(f"Unknown date type {key}: {value}")
        elif isinstance(value, str):
            # TODO: Use utf8mb4 on the DB instead of this which deletes all
            # unicode chars
            obj[key] = value.encode("ascii", "ignore").decode()

            if len(obj[key]) >= 65530:
                obj[key] = obj[key][:65530]

    for key, item in NAME_MAP.items():
        caster, _ = item
        if key in obj:
            obj[key] = caster(obj[key])

    return obj