Coverage for src / lexigram / admin / contributors / resource_collector.py: 65%

1"""Collects, namespaces, and validates Resource classes from contributors.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Sequence 

6from typing import TYPE_CHECKING 

7 

8from lexigram.admin.dashboard.naming_policy import NamingPolicy 

9from lexigram.admin.resources.namespace import apply_namespace 

10from lexigram.logging import get_logger 

11 

12if TYPE_CHECKING: 

13 from lexigram.contracts.admin.protocols import AdminContributorProtocol 

14 

15logger = get_logger(__name__) 

16 

17 

18class ResourceCollector: 

19 """Collects Resource classes from contributors, applies namespacing and collision detection. 

20 

21 Usage: 

22 collector = ResourceCollector(naming_policy) 

23 resources = collector.collect(contributors) 

24 """ 

25 

26 def __init__(self, naming_policy: NamingPolicy) -> None: 

27 self._naming = naming_policy 

28 

29 def collect(self, contributors: Sequence[AdminContributorProtocol]) -> list[type]: 

30 """Iterate contributors, collect and namespace their Resource classes. 

31 

32 Returns a list of Resource classes with namespaced names. When 

33 collision_mode is ``"warn"`` the first contributor's resource wins; 

34 in ``"error"`` mode a duplicate raises ``NameCollisionError``. 

35 """ 

36 out: list[type] = [] 

37 seen: dict[str, type] = {} 

38 for contributor in contributors: 

39 for resource_cls in contributor.get_resources(): 

40 ns = self._naming.namespaced( 

41 contributor.package_source, 

42 getattr(resource_cls, "name", None) 

43 or resource_cls.__name__.replace("Resource", "").lower(), 

44 ) 

45 self._naming.register("resource", ns) 

46 if ns in seen: 

47 continue 

48 wrapped = apply_namespace(resource_cls, ns) 

49 seen[ns] = wrapped 

50 out.append(wrapped) 

51 return out