Merge pull request #48 from Carreau/_info

Add the _info submodule

.github/workflows/tests.yml

@@ -25,7 +25,7 @@ jobs:
       - run: |
           pip install pip --upgrade
           pip install -ve '.[tests]'
-          pip install mypy sphinx
+          pip install mypy sphinx sphobjinv aiohttp
       - run:
           mypy
 
@@ -47,7 +47,7 @@ jobs:
       - run: |
           pip install pip --upgrade
           pip install -ve '.[tests]'
-          pip install mypy sphinx
+          pip install mypy sphinx 
           pip install pytest-reportlog
 
       - name: Test

intersphinx_registry/_info.py

@@ -0,0 +1,80 @@
+"""
+Experimental, print info about all (or a subset of), all the known inventories.
+
+As this is likely to make _many_ concurrent connections; 
+we use aoihttp and sphobjinv which are not listed as dependencies.
+"""
+
+import asyncio
+import sys
+from urllib.parse import urljoin
+
+import aiohttp
+from sphobjinv import Inventory  # type: ignore
+
+from intersphinx_registry import get_intersphinx_mapping
+
+if len(sys.argv) not in [2, 3]:
+    sys.exit(
+        """Usage: python -m intersphinx_registry._info <package>[,package|:all]
+
+        This prints debug info about registered package.
+
+        WARNING: :all: will query _all_ the domains; this can be considered
+
+        """
+    )
+
+packages = set(sys.argv[1].split(","))
+if packages == {":all:"}:
+    from intersphinx_registry import _get_all_mappings
+
+    packages = set(list(_get_all_mappings().keys()))
+
+print("Looking up metadata for", len(packages), "package")
+
+# there will be only one url
+urls = [
+    (k, u[0], (u[1] if u[1] else "objects.inv"))
+    for k, u in get_intersphinx_mapping(packages=packages).items()
+]
+
+
+async def fetch(session, key, base_url, obj):
+    final_url = urljoin(base_url, obj)
+    if obj != "objects.inv":
+        print("Obj for ", base_url, obj)
+    async with session.get(final_url) as response:
+        return key, base_url, await response.read()
+
+
+async def fetch_all(urls):
+    async with aiohttp.ClientSession() as session:
+        tasks = [fetch(session, *url) for url in urls]
+        return await asyncio.gather(*tasks)
+
+
+results = asyncio.run(fetch_all(urls))
+
+flattened = []
+for key, base_url, data in results:
+    try:
+        inv = Inventory(zlib=data)
+    except Exception:
+        print("ERROR with", key, base_url, file=sys.stderr)
+        continue
+    flattened.append((key, base_url, inv.project, inv.version, inv.count))
+
+if not flattened:
+    sys.exit("Could not reach any pacakges")
+width = [len(x) for x in flattened[0][:-1]]
+
+for item in flattened:
+    width = [max(w, len(x)) for w, x in zip(width, item[:-1])]
+
+
+for key, url, proj, version, count in flattened:
+    w_key, w_url, w_proj, w_version = width
+    print(
+        f"{key:<{w_key}}  {proj!r:<{w_proj+2}} {version:<{w_version}} {count:<5} {url:<{w_url}}"
+    )

intersphinx_registry/lookup.py

@@ -1,14 +1,13 @@
 import sys
 
-from intersphinx_registry import get_intersphinx_mapping
+from . import get_intersphinx_mapping
 from urllib.parse import urljoin
 
 from typing import Optional
 
 from sphinx.util.inventory import InventoryFile
 from io import BytesIO
 
-# filename = 'https://ipython.readthedocs.io/en/latest/'
 import requests
 
 if len(sys.argv) not in [2, 3]: