Skip to content

Registry API

sweet_tea.registry.Registry

Global registry for class definitions that can be instantiated via factories.

This registry automatically discovers and registers classes from packages, supporting optional dependencies that may not be installed. Classes with missing dependencies are skipped with a warning rather than failing registration.

The registry supports typed lookups for abstract factories, allowing filtering by inheritance hierarchy.

Thread-safe: All registry operations are synchronized to prevent race conditions in multi-threaded environments.

Source code in sweet_tea/registry.py 
 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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
class Registry:
    """
    Global registry for class definitions that can be instantiated via factories.

    This registry automatically discovers and registers classes from packages,
    supporting optional dependencies that may not be installed. Classes with
    missing dependencies are skipped with a warning rather than failing registration.

    The registry supports typed lookups for abstract factories, allowing filtering
    by inheritance hierarchy.

    Thread-safe: All registry operations are synchronized to prevent race conditions
    in multi-threaded environments.
    """

    # Threading lock for synchronizing registry operations
    __lock = threading.RLock()

    # This is the registry of packages
    __registry: list[Entry] = []

    __lookup: dict[Any, list[Entry]] = {}

    __lookup_keys: list[Any] = []

    # Logger instance using the global settings
    __logger = logging.getLogger()

    @classmethod
    def entries(cls) -> list[Entry]:
        """Get all registered entries."""
        with cls.__lock:
            return cls.__registry.copy()

    @classmethod
    def typed_entries(cls, lookup_type: Any = Any) -> list[Entry]:
        """
        Get entries that are subclasses of the specified type.

        Args:
            lookup_type: The base class to filter by. Defaults to Any.

        Returns:
            List of entries where the class_def is a subclass of lookup_type.
        """
        with cls.__lock:
            if lookup_type not in cls.__lookup_keys:
                cls.__lookup_keys.append(lookup_type)
                if lookup_type is Any:
                    # Any matches everything - return all entries without filtering
                    cls.__lookup[lookup_type] = cls.__registry.copy()
                else:
                    cls.__lookup[lookup_type] = [
                        filtered_type
                        for filtered_type in cls.__registry
                        if issubclass(filtered_type.class_def, lookup_type)
                    ]
            return cls.__lookup[lookup_type].copy()

    @classmethod
    def register(
        cls, key: str, class_def: type, library: str = "", label: str = ""
    ) -> None:
        """
        Register a class with the registry.

        Args:
            key: Name used to reference the class for instantiation.
            class_def: The class type to register.
            library: Name of the library the class belongs to.
            label: Optional label for categorizing classes (e.g., for different environments).
        """
        new_entry = Entry(
            key=key.lower(),
            class_def=class_def,
            library=library.lower(),
            label=label.lower(),
        )

        with cls.__lock:
            # Add entry if it is not currently present. Prevents duplicate entry.
            if new_entry not in cls.__registry:
                cls.__registry.append(new_entry)
                # Refresh every previously-queried lookup slot whose type matches
                # this class. Without this, ancestor-type slots cached before the
                # registration go stale (see GH #6).
                for lookup_type in cls.__lookup_keys:
                    if lookup_type is Any:
                        cls.__lookup[lookup_type].append(new_entry)
                    elif isinstance(lookup_type, type) and issubclass(
                        class_def, lookup_type
                    ):
                        cls.__lookup[lookup_type].append(new_entry)

    @classmethod
    def fill_registry(
        cls,
        path: str | None = None,
        module: str | None = None,
        library: str = "",
        label: str = "",
    ) -> None:
        """
        Recursively scan and register classes from packages starting from the given path.

        This method automatically discovers all classes in the package hierarchy,
        supporting optional dependencies by gracefully skipping modules that fail
        to import due to missing packages.

        Args:
            path: Package path where modules are located. If None, uses the caller's module path.
            module: Name of the root module. If None, inferred from path.
            library: Name of the library for categorization.
            label: Optional label for categorizing classes.
        """
        with cls.__lock:
            # Determine the path to scan
            if path is None:
                _module = inspect.getmodule(inspect.stack()[1][0])
                if _module is None or _module.__file__ is None:
                    raise SweetTeaError("Cannot determine module path automatically")
                path = str(Path(_module.__file__).parent)

            # Ensure path is a string
            path_str = str(path)

            # Make sure the root module is correctly specified
            if module is None:
                module = os.path.basename(path_str)

            if not library:
                library = module

            # Location of package
            pkg_dir = path_str

            # Loop over the modules. If it is a package, the make recursive call, otherwise for each non-package
            # module imports the module and registers it.
            for _, name, is_a_package in pkgutil.iter_modules([pkg_dir]):
                pkg_name = f"{module}.{name}"
                if not is_a_package:
                    cls.__add_entry_to_registry(
                        label=label, library=library, name_of_package=pkg_name
                    )
                else:
                    # Make recursive call to the
                    cls.fill_registry(
                        path=os.path.join(pkg_dir, name),
                        module=pkg_name,
                        library=library,
                        label=label,
                    )

    @classmethod
    def __add_entry_to_registry(
        cls, label: str, library: str, name_of_package: str
    ) -> None:
        """
        Import a module and register all classes defined in it.

        Handles optional dependencies by issuing warnings for ImportError/ModuleNotFoundError
        and continuing, while raising SweetTeaError for other exceptions.

        Args:
            label: Optional label for categorizing classes.
            library: Name of the library the classes belong to.
            name_of_package: Full module name to import and scan.

        Raises:
            SweetTeaError: For non-import related errors during module processing.
        """
        try:
            # exec("import " + name_of_package)
            module = importlib.import_module(name_of_package)

            classes = []
            for name, obj in inspect.getmembers(module, inspect.isclass):
                if obj.__module__ == name_of_package:
                    classes.append((name, obj))

            for class_name, class_def in classes:
                Registry.register(
                    key=class_name.lower(),
                    class_def=class_def,
                    library=library,
                    label=label,
                )

        except (ImportError, ModuleNotFoundError):
            # Optional dependency not installed - issue warning and skip this module
            warnings.warn(
                f"Skipping module {name_of_package} due to missing optional dependency",
                SweetTeaWarning,
                stacklevel=2,
            )
            # Continue without registering this module
        except Exception:
            # Other errors (e.g., syntax errors, runtime errors) should still fail
            error_message = traceback.format_exc()
            cls.__logger.error(
                f"Error processing module {name_of_package}: {error_message}"
            )
            raise SweetTeaError(error_message)

__add_entry_to_registry(label, library, name_of_package) classmethod

Import a module and register all classes defined in it.

Handles optional dependencies by issuing warnings for ImportError/ModuleNotFoundError and continuing, while raising SweetTeaError for other exceptions.

Parameters:

Name Type Description Default
label str

Optional label for categorizing classes.

 required
library str

Name of the library the classes belong to.

 required
name_of_package str

Full module name to import and scan.

 required

Raises:

Type Description
SweetTeaError

For non-import related errors during module processing.
Source code in sweet_tea/registry.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
@classmethod
def __add_entry_to_registry(
    cls, label: str, library: str, name_of_package: str
) -> None:
    """
    Import a module and register all classes defined in it.

    Handles optional dependencies by issuing warnings for ImportError/ModuleNotFoundError
    and continuing, while raising SweetTeaError for other exceptions.

    Args:
        label: Optional label for categorizing classes.
        library: Name of the library the classes belong to.
        name_of_package: Full module name to import and scan.

    Raises:
        SweetTeaError: For non-import related errors during module processing.
    """
    try:
        # exec("import " + name_of_package)
        module = importlib.import_module(name_of_package)

        classes = []
        for name, obj in inspect.getmembers(module, inspect.isclass):
            if obj.__module__ == name_of_package:
                classes.append((name, obj))

        for class_name, class_def in classes:
            Registry.register(
                key=class_name.lower(),
                class_def=class_def,
                library=library,
                label=label,
            )

    except (ImportError, ModuleNotFoundError):
        # Optional dependency not installed - issue warning and skip this module
        warnings.warn(
            f"Skipping module {name_of_package} due to missing optional dependency",
            SweetTeaWarning,
            stacklevel=2,
        )
        # Continue without registering this module
    except Exception:
        # Other errors (e.g., syntax errors, runtime errors) should still fail
        error_message = traceback.format_exc()
        cls.__logger.error(
            f"Error processing module {name_of_package}: {error_message}"
        )
        raise SweetTeaError(error_message)

entries() classmethod

Get all registered entries.

Source code in sweet_tea/registry.py 
58
59
60
61
62
@classmethod
def entries(cls) -> list[Entry]:
    """Get all registered entries."""
    with cls.__lock:
        return cls.__registry.copy()

fill_registry(path=None, module=None, library='', label='') classmethod

Recursively scan and register classes from packages starting from the given path.

This method automatically discovers all classes in the package hierarchy, supporting optional dependencies by gracefully skipping modules that fail to import due to missing packages.

Parameters:

Name Type Description Default
path str | None

Package path where modules are located. If None, uses the caller's module path.

 None
module str | None

Name of the root module. If None, inferred from path.

 None
library str

Name of the library for categorization.

 ''
label str

Optional label for categorizing classes.

 ''
Source code in sweet_tea/registry.py 
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
@classmethod
def fill_registry(
    cls,
    path: str | None = None,
    module: str | None = None,
    library: str = "",
    label: str = "",
) -> None:
    """
    Recursively scan and register classes from packages starting from the given path.

    This method automatically discovers all classes in the package hierarchy,
    supporting optional dependencies by gracefully skipping modules that fail
    to import due to missing packages.

    Args:
        path: Package path where modules are located. If None, uses the caller's module path.
        module: Name of the root module. If None, inferred from path.
        library: Name of the library for categorization.
        label: Optional label for categorizing classes.
    """
    with cls.__lock:
        # Determine the path to scan
        if path is None:
            _module = inspect.getmodule(inspect.stack()[1][0])
            if _module is None or _module.__file__ is None:
                raise SweetTeaError("Cannot determine module path automatically")
            path = str(Path(_module.__file__).parent)

        # Ensure path is a string
        path_str = str(path)

        # Make sure the root module is correctly specified
        if module is None:
            module = os.path.basename(path_str)

        if not library:
            library = module

        # Location of package
        pkg_dir = path_str

        # Loop over the modules. If it is a package, the make recursive call, otherwise for each non-package
        # module imports the module and registers it.
        for _, name, is_a_package in pkgutil.iter_modules([pkg_dir]):
            pkg_name = f"{module}.{name}"
            if not is_a_package:
                cls.__add_entry_to_registry(
                    label=label, library=library, name_of_package=pkg_name
                )
            else:
                # Make recursive call to the
                cls.fill_registry(
                    path=os.path.join(pkg_dir, name),
                    module=pkg_name,
                    library=library,
                    label=label,
                )

register(key, class_def, library='', label='') classmethod

Register a class with the registry.

Parameters:

Name Type Description Default
key str

Name used to reference the class for instantiation.

 required
class_def type

The class type to register.

 required
library str

Name of the library the class belongs to.

 ''
label str

Optional label for categorizing classes (e.g., for different environments).

 ''
Source code in sweet_tea/registry.py 
 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
@classmethod
def register(
    cls, key: str, class_def: type, library: str = "", label: str = ""
) -> None:
    """
    Register a class with the registry.

    Args:
        key: Name used to reference the class for instantiation.
        class_def: The class type to register.
        library: Name of the library the class belongs to.
        label: Optional label for categorizing classes (e.g., for different environments).
    """
    new_entry = Entry(
        key=key.lower(),
        class_def=class_def,
        library=library.lower(),
        label=label.lower(),
    )

    with cls.__lock:
        # Add entry if it is not currently present. Prevents duplicate entry.
        if new_entry not in cls.__registry:
            cls.__registry.append(new_entry)
            # Refresh every previously-queried lookup slot whose type matches
            # this class. Without this, ancestor-type slots cached before the
            # registration go stale (see GH #6).
            for lookup_type in cls.__lookup_keys:
                if lookup_type is Any:
                    cls.__lookup[lookup_type].append(new_entry)
                elif isinstance(lookup_type, type) and issubclass(
                    class_def, lookup_type
                ):
                    cls.__lookup[lookup_type].append(new_entry)

typed_entries(lookup_type=Any) classmethod

Get entries that are subclasses of the specified type.

Parameters:

Name Type Description Default
lookup_type Any

The base class to filter by. Defaults to Any.

 Any

Returns:

Type Description
list[Entry]

List of entries where the class_def is a subclass of lookup_type.
Source code in sweet_tea/registry.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@classmethod
def typed_entries(cls, lookup_type: Any = Any) -> list[Entry]:
    """
    Get entries that are subclasses of the specified type.

    Args:
        lookup_type: The base class to filter by. Defaults to Any.

    Returns:
        List of entries where the class_def is a subclass of lookup_type.
    """
    with cls.__lock:
        if lookup_type not in cls.__lookup_keys:
            cls.__lookup_keys.append(lookup_type)
            if lookup_type is Any:
                # Any matches everything - return all entries without filtering
                cls.__lookup[lookup_type] = cls.__registry.copy()
            else:
                cls.__lookup[lookup_type] = [
                    filtered_type
                    for filtered_type in cls.__registry
                    if issubclass(filtered_type.class_def, lookup_type)
                ]
        return cls.__lookup[lookup_type].copy()

Methods

register()

sweet_tea.registry.Registry.register(key, class_def, library='', label='') classmethod

Register a class with the registry.

Parameters:

Name Type Description Default
key str

Name used to reference the class for instantiation.

 required
class_def type

The class type to register.

 required
library str

Name of the library the class belongs to.

 ''
label str

Optional label for categorizing classes (e.g., for different environments).

 ''
Source code in sweet_tea/registry.py 
 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
@classmethod
def register(
    cls, key: str, class_def: type, library: str = "", label: str = ""
) -> None:
    """
    Register a class with the registry.

    Args:
        key: Name used to reference the class for instantiation.
        class_def: The class type to register.
        library: Name of the library the class belongs to.
        label: Optional label for categorizing classes (e.g., for different environments).
    """
    new_entry = Entry(
        key=key.lower(),
        class_def=class_def,
        library=library.lower(),
        label=label.lower(),
    )

    with cls.__lock:
        # Add entry if it is not currently present. Prevents duplicate entry.
        if new_entry not in cls.__registry:
            cls.__registry.append(new_entry)
            # Refresh every previously-queried lookup slot whose type matches
            # this class. Without this, ancestor-type slots cached before the
            # registration go stale (see GH #6).
            for lookup_type in cls.__lookup_keys:
                if lookup_type is Any:
                    cls.__lookup[lookup_type].append(new_entry)
                elif isinstance(lookup_type, type) and issubclass(
                    class_def, lookup_type
                ):
                    cls.__lookup[lookup_type].append(new_entry)

entries()

sweet_tea.registry.Registry.entries() classmethod

Get all registered entries.

Source code in sweet_tea/registry.py 
58
59
60
61
62
@classmethod
def entries(cls) -> list[Entry]:
    """Get all registered entries."""
    with cls.__lock:
        return cls.__registry.copy()

typed_entries()

sweet_tea.registry.Registry.typed_entries(lookup_type=Any) classmethod

Get entries that are subclasses of the specified type.

Parameters:

Name Type Description Default
lookup_type Any

The base class to filter by. Defaults to Any.

 Any

Returns:

Type Description
list[Entry]

List of entries where the class_def is a subclass of lookup_type.
Source code in sweet_tea/registry.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@classmethod
def typed_entries(cls, lookup_type: Any = Any) -> list[Entry]:
    """
    Get entries that are subclasses of the specified type.

    Args:
        lookup_type: The base class to filter by. Defaults to Any.

    Returns:
        List of entries where the class_def is a subclass of lookup_type.
    """
    with cls.__lock:
        if lookup_type not in cls.__lookup_keys:
            cls.__lookup_keys.append(lookup_type)
            if lookup_type is Any:
                # Any matches everything - return all entries without filtering
                cls.__lookup[lookup_type] = cls.__registry.copy()
            else:
                cls.__lookup[lookup_type] = [
                    filtered_type
                    for filtered_type in cls.__registry
                    if issubclass(filtered_type.class_def, lookup_type)
                ]
        return cls.__lookup[lookup_type].copy()

fill_registry()

sweet_tea.registry.Registry.fill_registry(path=None, module=None, library='', label='') classmethod

Recursively scan and register classes from packages starting from the given path.

This method automatically discovers all classes in the package hierarchy, supporting optional dependencies by gracefully skipping modules that fail to import due to missing packages.

Parameters:

Name Type Description Default
path str | None

Package path where modules are located. If None, uses the caller's module path.

 None
module str | None

Name of the root module. If None, inferred from path.

 None
library str

Name of the library for categorization.

 ''
label str

Optional label for categorizing classes.

 ''
Source code in sweet_tea/registry.py 
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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
@classmethod
def fill_registry(
    cls,
    path: str | None = None,
    module: str | None = None,
    library: str = "",
    label: str = "",
) -> None:
    """
    Recursively scan and register classes from packages starting from the given path.

    This method automatically discovers all classes in the package hierarchy,
    supporting optional dependencies by gracefully skipping modules that fail
    to import due to missing packages.

    Args:
        path: Package path where modules are located. If None, uses the caller's module path.
        module: Name of the root module. If None, inferred from path.
        library: Name of the library for categorization.
        label: Optional label for categorizing classes.
    """
    with cls.__lock:
        # Determine the path to scan
        if path is None:
            _module = inspect.getmodule(inspect.stack()[1][0])
            if _module is None or _module.__file__ is None:
                raise SweetTeaError("Cannot determine module path automatically")
            path = str(Path(_module.__file__).parent)

        # Ensure path is a string
        path_str = str(path)

        # Make sure the root module is correctly specified
        if module is None:
            module = os.path.basename(path_str)

        if not library:
            library = module

        # Location of package
        pkg_dir = path_str

        # Loop over the modules. If it is a package, the make recursive call, otherwise for each non-package
        # module imports the module and registers it.
        for _, name, is_a_package in pkgutil.iter_modules([pkg_dir]):
            pkg_name = f"{module}.{name}"
            if not is_a_package:
                cls.__add_entry_to_registry(
                    label=label, library=library, name_of_package=pkg_name
                )
            else:
                # Make recursive call to the
                cls.fill_registry(
                    path=os.path.join(pkg_dir, name),
                    module=pkg_name,
                    library=library,
                    label=label,
                )