From 598565d5603dcc73a9a322a78268de49809152d0 Mon Sep 17 00:00:00 2001 From: shamilbi Date: Wed, 8 Jul 2026 21:37:38 +0200 Subject: [PATCH] Accept-Encoding: gzip --- morgan/__init__.py | 47 ++++++++++++++++++++++++++++------------------ 1 file changed, 29 insertions(+), 18 deletions(-) diff --git a/morgan/__init__.py b/morgan/__init__.py index 391517e..30aac4a 100644 --- a/morgan/__init__.py +++ b/morgan/__init__.py @@ -2,6 +2,7 @@ import argparse import configparser +import gzip import hashlib import json import os @@ -161,24 +162,7 @@ def _mirror( # noqa: C901, PLR0912 self._processed_pkgs.add(requirement) # Mark as processed return None - data: dict | None = None - - # get information about this package from the Simple API in JSON - # format as per PEP 691 - request = urllib.request.Request( # noqa: S310 - f"{self.index_url}{requirement.name}/", - headers={ - "Accept": "application/vnd.pypi.simple.v1+json", - }, - ) - - response_url = "" - with urllib.request.urlopen(request) as response: # noqa: S310 - data = json.load(response) - response_url = str(response.url) - if not data: - msg = f"Failed loading metadata: {response}" - raise RuntimeError(msg) + data, response_url = download_req(self.index_url, requirement.name) # check metadata version ~1.0 v_str = data["meta"]["api-version"] @@ -851,5 +835,32 @@ def my_url(arg): Mirrorer(args).copy_server() +def download_req(index_url: str, req_name: str) -> tuple[dict, str]: + # get information about this package from the Simple API in JSON + # format as per PEP 691 + url = index_url.rstrip("/") + request = urllib.request.Request( # noqa: S310 + f"{url}/{req_name}/", + headers={ + "Accept": "application/vnd.pypi.simple.v1+json", + "Accept-Encoding": "gzip", + }, + ) + + response_url = "" + with urllib.request.urlopen(request) as response: # noqa: S310 + # Check if response is gzip-encoded + if response.headers.get("Content-Encoding") == "gzip": + with gzip.GzipFile(fileobj=response) as gzip_response: + data = json.load(gzip_response) + else: + data = json.load(response) + response_url = str(response.url) + if data: + return data, response_url + msg = f"Failed loading metadata: {response}" + raise RuntimeError(msg) + + if __name__ == "__main__": main()