vipr.plugins.adapter package¶

Submodules¶

vipr.plugins.adapter.controller module¶

class vipr.plugins.adapter.controller.AdapterController(*args: Any, **kw: Any)¶

Bases: ArgparseController

CLI commands for adapter helper tooling.

class Meta¶

Bases: object

help = 'Adapter helper commands'¶

label = 'adapter'¶

stacked_on = 'base'¶

stacked_type = 'nested'¶

init_config()¶: Generate a full VIPR inference YAML scaffold from adapter/discovery data.

vipr.plugins.adapter.decorators module¶

vipr.plugins.adapter.decorators.data_loader(name: str, *, passthrough: bool = False, auto_dataset: bool = False) → Callable[[Callable[[...], Any]], Callable[[...], Any]]¶: Register a data_loader function.

vipr.plugins.adapter.decorators.hook(name: str, *, target: str, injects: list[str] | tuple[str, ...] | set[str] | None = None, weight: int = 0, enabled_in_config: bool = True, passthrough: bool = False) → Callable[[Callable[[...], Any]], Callable[[...], Any]]¶: Register a hook function.

vipr.plugins.adapter.decorators.model_loader(name: str, *, passthrough: bool = False) → Callable[[Callable[[...], Any]], Callable[[...], Any]]¶: Register a model_loader function.

vipr.plugins.adapter.decorators.postprocess_filter(*_args, **_kwargs)¶

vipr.plugins.adapter.decorators.predictor(name: str, *, passthrough: bool = False) → Callable[[Callable[[...], Any]], Callable[[...], Any]]¶: Register a predictor function.

vipr.plugins.adapter.decorators.preprocess_filter(*_args, **_kwargs)¶

vipr.plugins.adapter.init_config module¶

exception vipr.plugins.adapter.init_config.InitConfigError¶

Bases: AdapterError

Raised when adapter init-config generation fails.

vipr.plugins.adapter.init_config.build_init_config_yaml(*, plugin_name: str, selected_data_loader: str | None = None, selected_model_loader: str | None = None, selected_predictor: str | None = None) → str¶

vipr.plugins.adapter.init_config.write_init_config(*, content: str, output_path: str | None, force: bool = False) → str | None¶

vipr.plugins.adapter.registry module¶

vipr.plugins.adapter.registry.clear_registry_for_tests() → None¶: Reset adapter state (test helper).

vipr.plugins.adapter.registry.get_filter_short_name_registry(module_name: str | None = None) → dict[str, list[dict[str, object]]]¶

Return filter short-name registry.

Result shape:

{

“my_normalize”: [

{: “module_name”: “…”, “class”: “…”, “method”: “…”, “hook”: “…”, “weight”: 0, “enabled_in_config”: True

}

]

}

vipr.plugins.adapter.registry.get_module_or_raise(module_name: str)¶

vipr.plugins.adapter.registry.get_module_state(module_name: str) → ModuleState¶

vipr.plugins.adapter.registry.normalize_identifier(value: str) → str¶

vipr.plugins.adapter.registry.publish_generated_class(module_name: str, class_name: str, cls: type) → None¶

vipr.plugins.adapter.registry.raise_on_name_collision(state: ModuleState, kind: str, name: str, module_name: str) → None¶

vipr.plugins.adapter.registry.register_filter_short_name(module_name: str, entry: FilterShortNameEntry) → None¶

vipr.plugins.adapter.registry.resolve_filter_short_name(short_name: str, *, module_name: str | None = None) → dict[str, object]¶

Resolve one short name to exactly one filter entry.

Raises AdapterError on missing or ambiguous resolution.

vipr.plugins.adapter.runtime module¶

class vipr.plugins.adapter.runtime.AdapterRuntime(app)¶

Bases: object

Runtime registrar attached to app as app.adapter_runtime.

list_registered_modules() → tuple[str, ...]¶

register(module_name: str) → None¶: Register one module snapshot into handler/discovery systems.

vipr.plugins.adapter.runtime.load(app) → None¶: Internal VIPR plugin entry point (plugin.adapter).

vipr.plugins.adapter.runtime.plugin(module_name: str) → Callable[[Any], None]¶: Freeze module registrations and return an entry-point callable.

vipr.plugins.adapter.signature module¶

vipr.plugins.adapter.signature.build_params_model(*, func: Callable[[...], Any], kind: str, name: str, module_name: str, excluded_positional_count: int, passthrough: bool, excluded_parameter_names: set[str] | None = None) → type[BaseModel]¶

vipr.plugins.adapter.signature.warn_data_loader_return_type(func: Callable[[...], Any], name: str, module_name: str) → None¶

vipr.plugins.adapter.types module¶

exception vipr.plugins.adapter.types.AdapterError¶

Bases: VIPRError

Raised when adapter registration or execution fails.

class vipr.plugins.adapter.types.CallbackRegistration(kind: 'str', name: 'str', target: 'str', module_name: 'str', wrapper_class: 'type', method_name: 'str', params_model: 'type[BaseModel]', passthrough: 'bool', weight: 'int' = 0, enabled_in_config: 'bool' = True)¶

Bases: object

enabled_in_config: bool = True¶

kind: str¶

method_name: str¶

module_name: str¶

name: str¶

params_model: type[BaseModel]¶

passthrough: bool¶

target: str¶

weight: int = 0¶

wrapper_class: type¶

class vipr.plugins.adapter.types.FilterShortNameEntry(module_name: 'str', name: 'str', class_path: 'str', method: 'str', hook: 'str', weight: 'int', enabled_in_config: 'bool')¶

Bases: object

class_path: str¶

enabled_in_config: bool¶

hook: str¶

method: str¶

module_name: str¶

name: str¶

weight: int¶

class vipr.plugins.adapter.types.HandlerRegistration(kind: 'str', name: 'str', module_name: 'str', wrapper_class: 'type', params_model: 'type[BaseModel]', passthrough: 'bool', auto_dataset: 'bool' = False)¶

Bases: object

auto_dataset: bool = False¶

kind: str¶

module_name: str¶

name: str¶

params_model: type[BaseModel]¶

passthrough: bool¶

wrapper_class: type¶

class vipr.plugins.adapter.types.ModuleState(handlers: 'list[HandlerRegistration]' = <factory>, callbacks: 'list[CallbackRegistration]' = <factory>, names: 'dict[str, str]' = <factory>, filter_short_names: 'dict[str, FilterShortNameEntry]' = <factory>, handlers_snapshot: 'tuple[HandlerRegistration, ...] | None' = None, callbacks_snapshot: 'tuple[CallbackRegistration, ...] | None' = None)¶

Bases: object

callbacks: list[CallbackRegistration]¶

callbacks_snapshot: tuple[CallbackRegistration, ...] | None = None¶

filter_short_names: dict[str, FilterShortNameEntry]¶

handlers: list[HandlerRegistration]¶

handlers_snapshot: tuple[HandlerRegistration, ...] | None = None¶

names: dict[str, str]¶

vipr.plugins.adapter.wrappers module¶

vipr.plugins.adapter.wrappers.make_callback_wrapper(*, kind: str, func: Callable[[...], Any], name: str, module_name: str) → tuple[type, str]¶

vipr.plugins.adapter.wrappers.make_data_loader_wrapper(*, func: Callable[[...], Any], name: str, module_name: str, params_model: type[BaseModel], passthrough: bool, auto_dataset: bool) → type¶

vipr.plugins.adapter.wrappers.make_model_loader_wrapper(*, func: Callable[[...], Any], name: str, module_name: str, params_model: type[BaseModel], passthrough: bool) → type¶

vipr.plugins.adapter.wrappers.make_predictor_wrapper(*, func: Callable[[...], Any], name: str, module_name: str, params_model: type[BaseModel], passthrough: bool) → type¶

vipr.plugins.adapter.wrappers.maybe_convert_auto_dataset(*, result: Any, handler_name: str, auto_dataset: bool) → Any¶

vipr.plugins.adapter.wrappers.register_callback(*, kind: str, name: str, target: str, func: Callable[[...], Any], excluded_positional_count: int, passthrough: bool, weight: int = 0, enabled_in_config: bool = True, injects: set[str] | None = None) → Callable[[...], Any]¶

vipr.plugins.adapter.wrappers.register_handler(*, kind: str, name: str, func: Callable[[...], Any], excluded_positional_count: int, passthrough: bool, auto_dataset: bool = False) → Callable[[...], Any]¶

Module contents¶

Function-as-handler adapter public API.

exception vipr.plugins.adapter.AdapterError¶

Bases: VIPRError

Raised when adapter registration or execution fails.

class vipr.plugins.adapter.AdapterRuntime(app)¶

Bases: object

Runtime registrar attached to app as app.adapter_runtime.

list_registered_modules() → tuple[str, ...]¶

register(module_name: str) → None¶: Register one module snapshot into handler/discovery systems.

vipr.plugins.adapter.data_loader(name: str, *, passthrough: bool = False, auto_dataset: bool = False) → Callable[[Callable[[...], Any]], Callable[[...], Any]]¶: Register a data_loader function.

vipr.plugins.adapter.get_filter_short_name_registry(module_name: str | None = None) → dict[str, list[dict[str, object]]]¶

Return filter short-name registry.

Result shape:

{

“my_normalize”: [

{: “module_name”: “…”, “class”: “…”, “method”: “…”, “hook”: “…”, “weight”: 0, “enabled_in_config”: True

}

]

}

vipr.plugins.adapter.hook(name: str, *, target: str, injects: list[str] | tuple[str, ...] | set[str] | None = None, weight: int = 0, enabled_in_config: bool = True, passthrough: bool = False) → Callable[[Callable[[...], Any]], Callable[[...], Any]]¶: Register a hook function.

vipr.plugins.adapter.load(app) → None¶: Internal VIPR plugin entry point (plugin.adapter).

vipr.plugins.adapter.model_loader(name: str, *, passthrough: bool = False) → Callable[[Callable[[...], Any]], Callable[[...], Any]]¶: Register a model_loader function.

vipr.plugins.adapter.plugin(module_name: str) → Callable[[Any], None]¶: Freeze module registrations and return an entry-point callable.

vipr.plugins.adapter.postprocess_filter(*_args, **_kwargs)¶

vipr.plugins.adapter.predictor(name: str, *, passthrough: bool = False) → Callable[[Callable[[...], Any]], Callable[[...], Any]]¶: Register a predictor function.

vipr.plugins.adapter.preprocess_filter(*_args, **_kwargs)¶

vipr.plugins.adapter.resolve_filter_short_name(short_name: str, *, module_name: str | None = None) → dict[str, object]¶

Resolve one short name to exactly one filter entry.

Raises AdapterError on missing or ambiguous resolution.