Skip to content

taps.plugins

taps/plugins.py

register() 

register(
    kind: PluginType,
) -> Callable[[type[ConfigType]], type[ConfigType]]

Decorator for registering plugin configurations.

Example

An app config can be defined and registered using a name. 

from concurrent.futures import Executor

from pydantic import Field

from taps.plugins import register
from taps.executor.config import ExecutorConfig

@register('executor')
class FooConfig(ExecutorConfig):
    name: Literal['foo'] = 'foo'
    n: int = Field(1, description='count')

    def get_executor(self) -> Executor:
        ...

Registration will make the executor named "foo" available within the CLI and configuration files. ```

Parameters:

  • kind (PluginType) –

    Kind of plugin that is being registered.

Source code in taps/plugins.py 
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
def register(
    kind: PluginType,
) -> Callable[[type[ConfigType]], type[ConfigType]]:
    """Decorator for registering plugin configurations.

    Example:
        An app config can be defined and registered using a name.
        ```python
        from concurrent.futures import Executor

        from pydantic import Field

        from taps.plugins import register
        from taps.executor.config import ExecutorConfig

        @register('executor')
        class FooConfig(ExecutorConfig):
            name: Literal['foo'] = 'foo'
            n: int = Field(1, description='count')

            def get_executor(self) -> Executor:
                ...
        ```

        Registration will make the executor named "foo" available within
        the CLI and configuration files.
        ```

    Args:
        kind: Kind of plugin that is being registered.
    """

    def _decorator(cls: type[ConfigType]) -> type[ConfigType]:
        try:
            registry = _REGISTERED_CONFIGS[kind]
        except KeyError:
            raise ValueError(f'Unknown plugin type "{kind}".') from None

        try:
            name = cls.model_fields['name'].default
            registry[name] = cls  # type: ignore[index]
        except Exception as e:
            raise RuntimeError(
                f'Failed to register {cls.__name__} as a {kind} plugin.',
            ) from e

        return cls

    return _decorator

get_app_configs() 

get_app_configs() -> dict[str, type[AppConfig]]

Get all registered application configs.

Returns:

Source code in taps/plugins.py 
90
91
92
93
94
95
96
def get_app_configs() -> dict[str, type[AppConfig]]:
    """Get all registered application configs.

    Returns:
        Mapping of application name to the config type.
    """
    return _REGISTERED_APP_CONFIGS.copy()

get_executor_configs() 

get_executor_configs() -> dict[str, type[ExecutorConfig]]

Get all registered executor configs.

Returns:

Source code in taps/plugins.py 
 99
100
101
102
103
104
105
def get_executor_configs() -> dict[str, type[ExecutorConfig]]:
    """Get all registered executor configs.

    Returns:
        Mapping of executor name to the config type.
    """
    return _REGISTERED_EXECUTOR_CONFIGS.copy()

get_filter_configs() 

get_filter_configs() -> dict[str, type[FilterConfig]]

Get all registered filter configs.

Returns:

Source code in taps/plugins.py 
108
109
110
111
112
113
114
def get_filter_configs() -> dict[str, type[FilterConfig]]:
    """Get all registered filter configs.

    Returns:
        Mapping of filter name to the config type.
    """
    return _REGISTERED_FILTER_CONFIGS.copy()

get_transformer_configs() 

get_transformer_configs() -> (
    dict[str, type[TransformerConfig]]
)

Get all registered transformer configs.

Returns:

Source code in taps/plugins.py 
117
118
119
120
121
122
123
def get_transformer_configs() -> dict[str, type[TransformerConfig]]:
    """Get all registered transformer configs.

    Returns:
        Mapping of transformer name to the config type.
    """
    return _REGISTERED_TRANSFORMER_CONFIGS.copy()