def _get_ping_schemas()

in sync/legacy.py [0:0]

• 13 lines of code
• 4 McCabe index (conditional complexity)

def _get_ping_schemas() -> Dict[str, Sequence[str]]:
    """
    Fetches the latest version of the schema tarball from GitHub and returns a dict of ping names and their schema
    versions.

    Example:
    {
        'account-ecosystem': ['account-ecosystem.4.schema.json'],
        'android-anr-report': ['android-anr-report.1.schema.json', 'android-anr-report.2.schema.json'],  # noqa: E501
        'anonymous': ['anonymous.4.schema.json'],
    }
    """
    print("Fetching schemas from GitHub...")
    schemas = requests.get(SCHEMA_URL)
    schema_file = BytesIO(schemas.content)

    with tarfile.open(fileobj=schema_file, mode="r|gz") as tar:
        schema_versions = defaultdict(list)
        for member in tar:
            if member.name.startswith(
                "mozilla-pipeline-schemas-generated-schemas/schemas/telemetry/"
            ) and member.name.endswith(".schema.json"):
                *_, ping_name, schema_name = member.name.split("/")
                schema_versions[ping_name].append(schema_name)
    return schema_versions