Coverage for /home/runner/work/coxbuild/coxbuild/src/coxbuild/managers.py: 0%

1import asyncio 

2import logging 

3from dataclasses import dataclass, field 

4 

5from coxbuild.configurations.builders import (ConfigurationBuilderCollection, 

6 getDefaultBuilder) 

7 

8from .configurations import Configuration 

9from .extensions import Extension 

10from .pipelines import Pipeline, PipelineResult 

11from .runtime import ExecutionState 

12from .services import Service 

13from .tasks import Task 

14 

15logger = logging.getLogger("managers") 

16 

17 

18@dataclass 

19class Manager: 

20 """Manage extensions and execution.""" 

21 extensions: dict[str, Extension] = field(default_factory=dict) 

22 configBuilders: ConfigurationBuilderCollection = field( 

23 default_factory=getDefaultBuilder) 

24 

25 def copy(self) -> "Manager": 

26 """Copy manager.""" 

27 return Manager(extensions=self.extensions) 

28 

29 def register(self, ext: Extension): 

30 """Register extension.""" 

31 if ext.uri not in self.extensions: 

32 self.extensions[ext.uri] = ext 

33 else: 

34 logger.warning(f"Register existed extensions {ext.uri}.") 

35 

36 def unregister(self, ext: Extension | str): 

37 """Unregister extension.""" 

38 if isinstance(ext, Extension): 

39 ext = ext.uri 

40 if ext in self.extensions: 

41 del self.extensions[ext] 

42 

43 def _load(self, *exts: Extension, pipeline: Pipeline, service: Service) -> None: 

44 def regTask(t: Task): 

45 t.extension = ext 

46 pipeline.register(t) 

47 

48 for ext in exts: 

49 logger.debug(f"Import extension: {ext.name}({ext.uri})") 

50 for t in ext.tasks: 

51 regTask(t) 

52 for eh in ext.events: 

53 logger.debug( 

54 f"Registering event handler: {eh.name} in {ext.name}({ext.uri}).") 

55 eh.extension = ext 

56 service.register(eh) 

57 regTask(eh.handler) 

58 for ph in ext.pipelineHooks: 

59 logger.debug( 

60 f"Registering pipeline hook: in {ext.name}.") 

61 ph.extension = ext 

62 pipeline.hook(ph) 

63 logger.debug(f"Imported extension: {ext.name}({ext.uri})") 

64 

65 async def executeAsync(self, *tasks: str) -> PipelineResult: 

66 """Execute tasks asynchronously.""" 

67 from coxbuild.extensions import builtin 

68 from coxbuild.extensions.loader import fromModule 

69 

70 pipeline = Pipeline() 

71 config = Configuration() 

72 service = Service() 

73 

74 executionState = ExecutionState(config) 

75 executionState.manager = self 

76 executionState.configuration = config 

77 executionState.service = service 

78 executionState.pipeline = pipeline 

79 

80 self.configBuilders.build(config) 

81 

82 self._load(*self.extensions.values(), fromModule(builtin), 

83 pipeline=pipeline, service=service) 

84 runner = pipeline(*(tasks or ["default"])) 

85 runner.context.config = config 

86 return await runner 

87 

88 def execute(self, *tasks: str) -> PipelineResult: 

89 """Execute tasks.""" 

90 return asyncio.run(self.executeAsync(*tasks))