feat: add reflection modules

This commit is contained in:
Henry Li
2026-01-14 07:16:07 +08:00
parent 88ed3841c7
commit 86524a65f6
2 changed files with 84 additions and 0 deletions

View File

@@ -0,0 +1,3 @@
from .resolvers import resolve_class, resolve_variable
__all__ = ["resolve_class", "resolve_variable"]

View File

@@ -0,0 +1,81 @@
from importlib import import_module
from typing import Type, TypeVar
T = TypeVar("T")
def resolve_variable[T](
variable_path: str,
expected_type: Type[T] | tuple[Type, ...] | None = None,
) -> T:
"""Resolve a variable from a path.
Args:
variable_path: The path to the variable (e.g. "parent_package_name.sub_package_name.module_name:variable_name").
expected_type: Optional type or tuple of types to validate the resolved variable against.
If provided, uses isinstance() to check if the variable is an instance of the expected type(s).
Returns:
The resolved variable.
Raises:
ImportError: If the module path is invalid or the attribute doesn't exist.
ValueError: If the resolved variable doesn't pass the validation checks.
"""
try:
module_path, variable_name = variable_path.rsplit(":", 1)
except ValueError as err:
raise ImportError(
f"{variable_path} doesn't look like a variable path. Example: parent_package_name.sub_package_name.module_name:variable_name"
) from err
try:
module = import_module(module_path)
except ImportError as err:
raise ImportError(f"Could not import module {module_path}") from err
try:
variable = getattr(module, variable_name)
except AttributeError as err:
raise ImportError(
f"Module {module_path} does not define a {variable_name} attribute/class"
) from err
# Type validation
if expected_type is not None:
if not isinstance(variable, expected_type):
type_name = (
expected_type.__name__
if isinstance(expected_type, type)
else " or ".join(t.__name__ for t in expected_type)
)
raise ValueError(
f"{variable_path} is not an instance of {type_name}, got {type(variable).__name__}"
)
return variable
def resolve_class(class_path: str, base_class: Type[T] | None = None) -> Type[T]:
"""Resolve a class from a module path and class name.
Args:
class_path: The path to the class (e.g. "langchain_openai:ChatOpenAI").
base_class: The base class to check if the resolved class is a subclass of.
Returns:
The resolved class.
Raises:
ImportError: If the module path is invalid or the attribute doesn't exist.
ValueError: If the resolved object is not a class or not a subclass of base_class.
"""
model_class = resolve_variable(class_path, expected_type=type)
if not isinstance(model_class, type):
raise ValueError(f"{class_path} is not a valid class")
if base_class is not None and not issubclass(model_class, base_class):
raise ValueError(f"{class_path} is not a subclass of {base_class.__name__}")
return model_class