-
Notifications
You must be signed in to change notification settings - Fork 5
/
services.py
158 lines (129 loc) · 4.88 KB
/
services.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
from contextlib import contextmanager
from logging import Logger
from . import SHORT_COMMIT_ID, __version__
from .diffing import Differ
from .environments import ExecutionEnvironment, ExecutionEnvironmentBuilder
from .extracting import Extractor
from .models import (ApiDescription, ApiDifference, Distribution, Product,
Report)
from .preprocessing import Preprocessor
from .producers import ProduceContext, produce
from .reporting import Reporter
class ServiceProvider:
def __init__(self, name: str | None = None) -> None:
self.name = name or f"aexpy@{__version__}-{SHORT_COMMIT_ID}"
def environmentBuilder(
self, /, logger: Logger | None = None
) -> ExecutionEnvironmentBuilder:
from .extracting.environment import getExtractorEnvironmentBuilder
return getExtractorEnvironmentBuilder(logger=logger)
def preprocessor(self, /, logger: Logger | None = None) -> Preprocessor:
from .preprocessing.counter import FileCounterPreprocessor
return FileCounterPreprocessor(logger=logger)
def extractor(
self, /, logger: Logger | None = None, env: ExecutionEnvironment | None = None
) -> Extractor:
from .extracting.default import DefaultExtractor
return DefaultExtractor(logger=logger, env=env)
def differ(self, /, logger: Logger | None = None) -> Differ:
from .diffing.default import DefaultDiffer
return DefaultDiffer(logger=logger)
def reporter(self, /, logger: Logger | None = None) -> Reporter:
from .reporting.text import TextReporter
return TextReporter(logger=logger)
@contextmanager
def produce[
T: Product
](
self,
product: T,
logger: Logger | None = None,
context: ProduceContext[T] | None = None,
):
if context:
yield context
else:
with produce(product, logger=logger, service=self.name) as context:
yield context
def preprocess(
self,
/,
product: Distribution,
*,
logger: Logger | None = None,
context: ProduceContext[Distribution] | None = None,
):
with self.produce(product, logger=logger, context=context) as context:
with context.using(self.preprocessor(context.logger)) as producer:
producer.preprocess(product)
return context
def extract[
E: ExecutionEnvironment
](
self,
/,
dist: Distribution,
*,
logger: Logger | None = None,
context: ProduceContext[ApiDescription] | None = None,
envBuilder: ExecutionEnvironmentBuilder[E] | None = None,
):
with self.produce(
ApiDescription(distribution=dist), logger=logger, context=context
) as context:
envBuilder = envBuilder or self.environmentBuilder(context.logger)
with envBuilder.use(pyversion=dist.pyversion, logger=context.logger) as env:
with context.using(self.extractor(context.logger, env=env)) as producer:
producer.extract(dist, context.product)
return context
def diff(
self,
/,
old: ApiDescription,
new: ApiDescription,
*,
logger: Logger | None = None,
context: ProduceContext[ApiDifference] | None = None,
):
with self.produce(
ApiDifference(old=old.distribution, new=new.distribution),
logger=logger,
context=context,
) as context:
with context.using(self.differ(context.logger)) as producer:
producer.diff(old, new, context.product)
return context
def report(
self,
/,
diff: ApiDifference,
*,
logger: Logger | None = None,
context: ProduceContext[Report] | None = None,
):
with self.produce(
Report(old=diff.old, new=diff.new),
logger=logger,
context=context,
) as context:
with context.using(self.reporter(context.logger)) as producer:
producer.report(diff, context.product)
return context
def getService():
return ServiceProvider()
def loadServiceFromCode(src: str):
from hashlib import sha256
from importlib.util import module_from_spec, spec_from_loader
spec = spec_from_loader(sha256(src.encode()).hexdigest(), loader=None)
assert spec is not None, "Failed to create module spec."
mod = module_from_spec(spec)
srcCode = compile(src, "<memory>", "exec")
exec("from aexpy.services import *", mod.__dict__)
exec(srcCode, mod.__dict__)
getter = getattr(mod, "getService", None)
assert callable(getter), "Failed to get valid `getService` getter."
service = getter()
assert isinstance(
service, ServiceProvider
), f"Failed to get valid service instance: got {type(service)}."
return service