-
-
Notifications
You must be signed in to change notification settings - Fork 607
/
subsystem.py
127 lines (99 loc) · 5 KB
/
subsystem.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
# Copyright 2015 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
from __future__ import annotations
import functools
import inspect
import re
from abc import ABCMeta
from typing import Any, ClassVar, TypeVar
from pants.engine.internals.selectors import AwaitableConstraints, Get
from pants.option.errors import OptionsError
from pants.option.option_value_container import OptionValueContainer
from pants.option.scope import Scope, ScopedOptions, ScopeInfo, normalize_scope
class Subsystem(metaclass=ABCMeta):
"""A separable piece of functionality that may be reused across multiple tasks or other code.
Subsystems encapsulate the configuration and initialization of things like JVMs,
Python interpreters, SCMs and so on.
Set the `help` class property with a description, which will be used in `./pants help`. For the
best rendering, use soft wrapping (e.g. implicit string concatenation) within paragraphs, but
hard wrapping (`\n`) to separate distinct paragraphs and/or lists.
"""
options_scope: str
help: ClassVar[str]
# Subclasses may override these to specify a deprecated former name for this Subsystem's scope.
# Option values can be read from the deprecated scope, but a deprecation warning will be issued.
# The deprecation warning becomes an error at the given Pants version (which must therefore be
# a valid semver).
deprecated_options_scope: str | None = None
deprecated_options_scope_removal_version: str | None = None
_scope_name_re = re.compile(r"^(?:[a-z0-9_])+(?:-(?:[a-z0-9_])+)*$")
@classmethod
def signature(cls):
"""Returns kwargs to construct a `TaskRule` that will construct the target Subsystem.
TODO: This indirection avoids a cycle between this module and the `rules` module.
"""
partial_construct_subsystem = functools.partial(_construct_subsytem, cls)
# NB: We must populate several dunder methods on the partial function because partial
# functions do not have these defined by default and the engine uses these values to
# visualize functions in error messages and the rule graph.
snake_scope = normalize_scope(cls.options_scope)
name = f"construct_scope_{snake_scope}"
partial_construct_subsystem.__name__ = name
partial_construct_subsystem.__module__ = cls.__module__
partial_construct_subsystem.__doc__ = cls.help
_, class_definition_lineno = inspect.getsourcelines(cls)
partial_construct_subsystem.__line_number__ = class_definition_lineno
return dict(
output_type=cls,
input_selectors=(),
func=partial_construct_subsystem,
input_gets=(
AwaitableConstraints(output_type=ScopedOptions, input_type=Scope, is_effect=False),
),
canonical_name=name,
)
@classmethod
def is_valid_scope_name(cls, s: str) -> bool:
return s == "" or cls._scope_name_re.match(s) is not None
@classmethod
def validate_scope(cls) -> None:
options_scope = getattr(cls, "options_scope", None)
if options_scope is None:
raise OptionsError(f"{cls.__name__} must set options_scope.")
if not cls.is_valid_scope_name(options_scope):
raise OptionsError(
f'Options scope "{options_scope}" is not valid:\nReplace in code with a new '
"scope name consisting of only lower-case letters, digits, underscores, "
"and non-consecutive dashes."
)
@classmethod
def create_scope_info(cls, **scope_info_kwargs) -> ScopeInfo:
"""One place to create scope info, to allow subclasses to inject custom scope args."""
return ScopeInfo(**scope_info_kwargs)
@classmethod
def get_scope_info(cls) -> ScopeInfo:
"""Returns a ScopeInfo instance representing this Subsystem's options scope."""
cls.validate_scope()
return cls.create_scope_info(scope=cls.options_scope, subsystem_cls=cls)
@classmethod
def register_options(cls, register):
"""Register options for this Subsystem.
Subclasses may override and call register(*args, **kwargs).
"""
@classmethod
def register_options_on_scope(cls, options):
"""Trigger registration of this Subsystem's options.
Subclasses should not generally need to override this method.
"""
cls.register_options(options.registration_function_for_subsystem(cls))
def __init__(self, options: OptionValueContainer) -> None:
self.validate_scope()
self.options = options
def __eq__(self, other: Any) -> bool:
if type(self) != type(other):
return False
return bool(self.options == other.options)
_T = TypeVar("_T", bound=Subsystem)
async def _construct_subsytem(subsystem_typ: type[_T]) -> _T:
scoped_options = await Get(ScopedOptions, Scope(str(subsystem_typ.options_scope)))
return subsystem_typ(scoped_options.options)