Coverage for src/secchi/api/golang.py: 89%

1"""Go Modules adapter using the public Go module proxy.""" 

2 

3from __future__ import annotations 

4 

5from datetime import datetime 

6from urllib.parse import quote 

7 

8import httpx 

9 

10from secchi.api.sparse import SparseAdapter 

11from secchi.models import PackageInfo, Registry, SearchResult, Version 

12 

13GO_PROXY = "https://proxy.golang.org" 

14 

15 

16def _module_path(name: str) -> str: 

17 return quote(name, safe="/@") 

18 

19 

20class GoModuleAdapter(SparseAdapter): 

21 @property 

22 def registry(self) -> Registry: 

23 return Registry.GO 

24 

25 async def fetch_package(self, name: str) -> PackageInfo: 

26 module = _module_path(name) 

27 async with self._client_scope() as client: 

28 response = await client.get(f"{GO_PROXY}/{module}/@latest") 

29 response.raise_for_status() 

30 data = response.json() 

31 version = data.get("Version", "") 

32 repository = _repository_url(name) 

33 return PackageInfo( 

34 name=name, 

35 registry=Registry.GO, 

36 description=f"Go module {name}", 

37 homepage=f"https://pkg.go.dev/{name}", 

38 repository_url=repository, 

39 documentation_url=f"https://pkg.go.dev/{name}", 

40 latest_version=version, 

41 latest_release_date=_parse_time(data.get("Time")), 

42 versions=[ 

43 Version(version=version, release_date=_parse_time(data.get("Time"))) 

44 ] 

45 if version 

46 else [], 

47 ) 

48 

49 async def fetch_versions(self, name: str) -> list[Version]: 

50 module = _module_path(name) 

51 async with self._client_scope() as client: 

52 response = await client.get(f"{GO_PROXY}/{module}/@v/list") 

53 response.raise_for_status() 

54 versions = [ 

55 Version(version=line) for line in response.text.splitlines() if line 

56 ] 

57 return versions[-100:] 

58 

59 async def search(self, query: str, limit: int = 10) -> list[SearchResult]: 

60 # Go has no central keyword-search endpoint. A module path is resolved 

61 # exactly through the proxy; broader discovery belongs to pkg.go.dev. 

62 try: 

63 info = await self.fetch_package(query) 

64 except httpx.HTTPError: 

65 return [] 

66 return [ 

67 SearchResult( 

68 name=query, 

69 registry=Registry.GO, 

70 version=info.latest_version, 

71 description=info.description, 

72 url=info.documentation_url, 

73 score=1.0, 

74 exact=True, 

75 ) 

76 ] 

77 

78 

79def _repository_url(name: str) -> str: 

80 parts = name.split("/") 

81 if len(parts) >= 3 and parts[0] == "github.com": 

82 return "/".join(parts[:3]) 

83 return "" 

84 

85 

86def _parse_time(raw: str | None) -> datetime | None: 

87 if not raw: 

88 return None 

89 try: 

90 return datetime.fromisoformat(raw.replace("Z", "+00:00")) 

91 except ValueError: 

92 return None