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

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 _escape_case(component: str) -> str: 

17 """Go module proxy case-encoding: each uppercase letter becomes ``!`` + 

18 its lowercase form (e.g. ``BurntSushi`` -> ``!burnt!sushi``). 

19 

20 Required so the proxy can serve module paths as files on case-insensitive 

21 filesystems; see https://go.dev/ref/mod#module-proxy. Without it, any 

22 module path containing an uppercase letter 404s against the real proxy. 

23 """ 

24 return "".join(f"!{char.lower()}" if char.isupper() else char for char in component) 

25 

26 

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

28 return quote(_escape_case(name), safe="/@!") 

29 

30 

31class GoModuleAdapter(SparseAdapter): 

32 @property 

33 def registry(self) -> Registry: 

34 return Registry.GO 

35 

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

37 module = _module_path(name) 

38 async with self._client_scope() as client: 

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

40 response.raise_for_status() 

41 data = response.json() 

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

43 repository = _repository_url(name) 

44 return PackageInfo( 

45 name=name, 

46 registry=Registry.GO, 

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

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

49 repository_url=repository, 

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

51 latest_version=version, 

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

53 versions=[ 

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

55 ] 

56 if version 

57 else [], 

58 ) 

59 

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

61 module = _module_path(name) 

62 async with self._client_scope() as client: 

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

64 response.raise_for_status() 

65 versions = [ 

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

67 ] 

68 return versions[-100:] 

69 

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

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

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

73 try: 

74 info = await self.fetch_package(query) 

75 except httpx.HTTPError: 

76 return [] 

77 return [ 

78 SearchResult( 

79 name=query, 

80 registry=Registry.GO, 

81 version=info.latest_version, 

82 description=info.description, 

83 url=info.documentation_url, 

84 score=1.0, 

85 exact=True, 

86 ) 

87 ] 

88 

89 

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

91 parts = name.split("/") 

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

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

94 return "" 

95 

96 

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

98 if not raw: 

99 return None 

100 try: 

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

102 except ValueError: 

103 return None