Coverage for src/metador_core/plugin/entrypoints.py: 86%

1"""Processing of entry points for Metador plugins.""" 

2from __future__ import annotations 

3 

4import re 

5from dataclasses import dataclass 

6from typing import Any, Dict, List, Optional, Tuple 

7 

8from importlib_metadata import Distribution, entry_points 

9 

10from ..schema.plugins import PluginPkgMeta, SemVerTuple 

11from .types import ( 

12 EPGroupName, 

13 from_ep_group_name, 

14 is_metador_ep_group, 

15 to_ep_group_name, 

16) 

17 

18_eps = entry_points() 

19"""All entry points.""" 

20 

21pkg_meta = {} 

22"""Collected infos about packages that provide plugins (filled by get_group).""" 

23 

24 

25def get_group(group_name: str) -> Dict[str, Any]: 

26 """Get a dict of all available entrypoints for a Metador plugin group.""" 

27 ep_grp = to_ep_group_name(group_name) 

28 plugins: Dict[str, Any] = {} 

29 

30 for ep in _eps.select(group=ep_grp): 

31 if ep.name in plugins: 

32 # TODO: will importlib_metadata even return colliding packages? 

33 # should be figured out (quite important to know) 

34 msg = f"{group_name}: a plugin named '{ep.name}' is already registered!" 

35 raise TypeError(msg) 

36 

37 plugins[ep.name] = ep 

38 if ep.dist.name not in pkg_meta: 

39 pkg_meta[ep.dist.name] = PluginPkgMeta.for_package(ep.dist.name) 

40 

41 return plugins 

42 

43 

44# ---- 

45 

46 

47@dataclass 

48class DistMeta: 

49 name: str 

50 version: Tuple[int, int, int] 

51 plugins: Dict[str, List[str]] 

52 repository_url: Optional[str] 

53 

54 

55def distmeta_for(dist: Distribution) -> DistMeta: 

56 """Extract required metadata from importlib_metadata distribution object.""" 

57 ver = dist.version 

58 if not re.fullmatch("[0-9]+\\.[0-9]+\\.[0-9]+", ver): 

59 msg = f"Invalid version string of {dist.name}: {ver}" 

60 raise TypeError(msg) 

61 parsed_ver: SemVerTuple = tuple(map(int, ver.split("."))) # type: ignore 

62 

63 # parse entry point groups 

64 epgs = filter( 

65 is_metador_ep_group, 

66 dist.entry_points.groups, 

67 ) 

68 eps = { 

69 from_ep_group_name(EPGroupName(epg)): list( 

70 map(lambda x: x.name, dist.entry_points.select(group=epg)) 

71 ) 

72 for epg in epgs 

73 } 

74 

75 repo_url: Optional[str] = None 

76 try: 

77 urls = dist.metadata.get_all("Project-URL") 

78 url = next(filter(lambda u: u.startswith("Repository,"), urls or [])) 

79 repo_url = url.split()[1] # from "Repository, http://..." 

80 except StopIteration: 

81 pass 

82 return DistMeta( 

83 name=dist.name, version=parsed_ver, plugins=eps, repository_url=repo_url 

84 )