def backport_records()

in cronjobs/src/commands/backport_records.py [0:0]
60 lines of code
15 McCabe index (conditional complexity)

def backport_records(event, context, **kwargs):
    """Backport records creations, updates and deletions from one collection to another."""
    server_url = event["server"]
    source_auth = (
        event.get("backport_records_source_auth")
        or os.environ["BACKPORT_RECORDS_SOURCE_AUTH"]
    )
    dest_auth = event.get(
        "backport_records_dest_auth",
        os.getenv("BACKPORT_RECORDS_DEST_AUTH", source_auth),
    )

    mappings = []

    if mappings_env := (
        event.get("backport_records_mappings")
        or os.getenv("BACKPORT_RECORDS_MAPPINGS", "")
    ):
        regexp = re.compile(
            r"^(?P<sbid>[^/]+)/(?P<scid>[^/\?]+)(?P<qs>\?.*)? -> (?P<dbid>[^/]+)/(?P<dcid>[^/]+)$"
        )
        for entry in mappings_env.strip().splitlines():
            if not entry:  # empty lines
                continue
            # bid/cid -> bid/cid
            # bid/cid?field=value -> bid/cid
            if match := regexp.match(entry):
                sbid = match.group("sbid")
                scid = match.group("scid")
                querystring = match.group("qs")
                dbid = match.group("dbid")
                dcid = match.group("dcid")
                filters_dict = parse_querystring(querystring) if querystring else {}
                mappings.append((sbid, scid, filters_dict, dbid, dcid))
            else:
                raise ValueError(f"Invalid syntax in line {entry}")
    else:
        sbid = (
            event.get("backport_records_source_bucket")
            or os.environ["BACKPORT_RECORDS_SOURCE_BUCKET"]
        )
        scid = (
            event.get("backport_records_source_collection")
            or os.environ["BACKPORT_RECORDS_SOURCE_COLLECTION"]
        )
        filters_json = event.get("backport_records_source_filters") or os.getenv(
            "BACKPORT_RECORDS_SOURCE_FILTERS", ""
        )
        filters_dict = json.loads(filters_json or "{}")

        dbid = event.get(
            "backport_records_dest_bucket",
            os.getenv("BACKPORT_RECORDS_DEST_BUCKET", sbid),
        )
        dcid = event.get(
            "backport_records_dest_collection",
            os.getenv("BACKPORT_RECORDS_DEST_COLLECTION", scid),
        )

        if sbid == dbid and scid == dcid:
            raise ValueError("Cannot copy records: destination is identical to source")

        mappings.append((sbid, scid, filters_dict, dbid, dcid))

    safe_headers = event.get(
        "safe_headers", config("SAFE_HEADERS", default=False, cast=bool)
    )

    for mapping in mappings:
        execute_backport(server_url, source_auth, dest_auth, safe_headers, *mapping)