Coverage for src/lumen/admin/registry.py: 100%

1""" 

2This module handles downloading FHIR packages from the official FHIR package registry 

3(packages.fhir.org). 

4""" 

5 

6from __future__ import annotations 

7 

8import httpx 

9import structlog 

10 

11log = structlog.get_logger(__name__) 

12 

13 

14class RegistryDownloadError(Exception): 

15 """Raised when a package cannot be downloaded from the registry.""" 

16 

17 

18async def download_package(package_id: str, version: str) -> bytes: 

19 """ 

20 Query packages.fhir.org for the given package_id, find the tarball URL 

21 for the specified version, and download it. 

22 

23 Returns the raw bytes of the .tgz file. 

24 """ 

25 registry_url = f"https://packages.fhir.org/{package_id}" 

26 

27 async with httpx.AsyncClient() as client: 

28 try: 

29 # 1. Fetch package metadata 

30 resp = await client.get(registry_url, headers={"Accept": "application/json"}) 

31 resp.raise_for_status() 

32 data = resp.json() 

33 except Exception as e: 

34 raise RegistryDownloadError(f"Failed to fetch metadata from {registry_url}: {e}") from e 

35 

36 # 2. Extract tarball URL for the requested version 

37 versions = data.get("versions", {}) 

38 if version not in versions: 

39 raise RegistryDownloadError( 

40 f"Version {version} not found for package {package_id}. " 

41 f"Available versions: {', '.join(versions.keys())}" 

42 ) 

43 

44 tarball_url = versions[version].get("dist", {}).get("tarball") 

45 if not tarball_url: 

46 raise RegistryDownloadError(f"No tarball URL found for {package_id}@{version}.") 

47 

48 log.info( 

49 "package_registry.download_started", package=f"{package_id}@{version}", url=tarball_url 

50 ) 

51 

52 # 3. Download the tarball 

53 try: 

54 # Some Simplifier tarballs might redirect or take a bit longer 

55 tarball_resp = await client.get(tarball_url, follow_redirects=True, timeout=60.0) 

56 tarball_resp.raise_for_status() 

57 

58 content = tarball_resp.content 

59 log.info( 

60 "package_registry.download_finished", 

61 package=f"{package_id}@{version}", 

62 size_bytes=len(content), 

63 ) 

64 return content 

65 except Exception as e: 

66 msg = f"Failed to download tarball from {tarball_url}: {e}" 

67 raise RegistryDownloadError(msg) from e