poprox_recommender.lkpipeline.config#

Pydantic models for pipeline configuration and serialization support.

Functions

hash_config(config)

Compute the hash of a configuration model.

Classes

PipelineComponent(*, code[, config, inputs])

PipelineConfig(*, meta[, inputs, defaults, ...])

Root type for serialized pipeline configuration.

PipelineInput(*, name, types)

PipelineLiteral(*, encoding, value)

Literal nodes represented in the pipeline.

PipelineMeta(*[, name, version, hash])

Pipeline metadata.
class poprox_recommender.lkpipeline.config.PipelineConfig(*, meta, inputs=None, defaults=None, components=None, aliases=None, literals=None)#

Bases: BaseModel

Root type for serialized pipeline configuration. A pipeline config contains the full configuration, components, and wiring for the pipeline, but does not contain the

Parameters:
meta: PipelineMeta#

Pipeline metadata.

inputs: list[PipelineInput]#

Pipeline inputs.

defaults: dict[str, str]#

Default pipeline wirings.

components: OrderedDict[str, PipelineComponent]#

Pipeline components, with their configurations and wiring.

aliases: dict[str, str]#

Pipeline node aliases.

literals: dict[str, PipelineLiteral]#

Literals

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'aliases': FieldInfo(annotation=dict[str, str], required=False, default_factory=dict), 'components': FieldInfo(annotation=OrderedDict[str, PipelineComponent], required=False, default_factory=OrderedDict), 'defaults': FieldInfo(annotation=dict[str, str], required=False, default_factory=dict), 'inputs': FieldInfo(annotation=list[PipelineInput], required=False, default_factory=list), 'literals': FieldInfo(annotation=dict[str, PipelineLiteral], required=False, default_factory=dict), 'meta': FieldInfo(annotation=PipelineMeta, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class poprox_recommender.lkpipeline.config.PipelineMeta(*, name=None, version=None, hash=None)#

Bases: BaseModel

Pipeline metadata.

Parameters:

  • name (str | None)

  • version (str | None)

  • hash (str | None)

name: str | None#

The pipeline name.

version: str | None#

The pipeline version.

hash: str | None#

The pipeline configuration hash. This is optional, particularly when hand-crafting pipeline configuration files.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'hash': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'name': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'version': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class poprox_recommender.lkpipeline.config.PipelineInput(*, name, types)#

Bases: BaseModel

Parameters:
name: str#

The name for this input.

types: set[str] | None#

The list of types for this input.

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'name': FieldInfo(annotation=str, required=True), 'types': FieldInfo(annotation=Union[set[str], NoneType], required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

class poprox_recommender.lkpipeline.config.PipelineComponent(*, code, config=None, inputs=None)#

Bases: BaseModel

Parameters:
model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'code': FieldInfo(annotation=str, required=True), 'config': FieldInfo(annotation=Union[dict[str, object], NoneType], required=False, default=None), 'inputs': FieldInfo(annotation=Union[dict[str, str], list[str]], required=False, default_factory=dict)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

code: str#

The path to the component’s implementation, either a class or a function. This is a Python qualified path of the form module:name.

config: dict[str, object] | None#

The component configuration. If not provided, the component will be created with its default constructor parameters.

inputs: dict[str, str] | list[str]#

The component’s input wirings, mapping input names to node names. For certain meta-nodes, it is specified as a list instead of a dict.

class poprox_recommender.lkpipeline.config.PipelineLiteral(*, encoding, value)#

Bases: BaseModel

Literal nodes represented in the pipeline.

Parameters:

  • encoding (Literal['json', 'base85'])

  • value (JsonValue)

model_computed_fields: ClassVar[dict[str, ComputedFieldInfo]] = {}#

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'encoding': FieldInfo(annotation=Literal['json', 'base85'], required=True), 'value': FieldInfo(annotation=JsonValue, required=True)}#

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

decode()#

Decode the represented literal.

Return type:

Any

poprox_recommender.lkpipeline.config.hash_config(config)#

Compute the hash of a configuration model.

Parameters:

config (BaseModel)

Return type:

str

On this page