input

class InputScope(*values)[source]

The scope that input must be provided in

tube = 'tube'
process = 'process'
pydantic model InputSpecification[source]

Specification of inputs to a noob tube.

Inputs can be supplied at different times and frequencies, as specified by scope:

tube-scoped inputs may be used in a node’s param specification, and process-scoped inputs may be used as one of a node’s depends.

Inputs can be supplied at a “higher” scope and be accessed by lower scopes: e.g. input requested with a process scope can use input provided when instantiating the tube, if not provided to process but provided to the tube.

Show JSON schema
{
   "title": "InputSpecification",
   "description": "Specification of inputs to a noob tube.\n\nInputs can be supplied at different times and frequencies,\nas specified by `scope`:\n\n- `tube`: When instantiating the tube\n- `process`: Per call to :meth:`.TubeRunner.process`\n\n`tube`-scoped inputs may be used in a node's `param` specification,\nand `process`-scoped inputs may be used as one of a node's `depends`.\n\nInputs can be supplied at a \"higher\" scope and be accessed by lower scopes:\ne.g. input requested with a process scope can use input provided when instantiating the tube,\nif not provided to process but provided to the tube.",
   "type": "object",
   "properties": {
      "id": {
         "title": "Id",
         "type": "string"
      },
      "type": {
         "title": "Type",
         "type": "string"
      },
      "scope": {
         "$ref": "#/$defs/InputScope",
         "default": "tube"
      },
      "description": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "title": "Description"
      }
   },
   "$defs": {
      "InputScope": {
         "description": "The scope that input must be provided in",
         "enum": [
            "tube",
            "process"
         ],
         "title": "InputScope",
         "type": "string"
      }
   },
   "required": [
      "id",
      "type"
   ]
}

Fields:
field description: str | None = None

An optional description of the input value

field id: Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)] [Required]
Constraints:

  • func = <function _not_reserved at 0x77de2d4389a0>

field scope: InputScope = InputScope.tube
field type_: Annotated[str, AfterValidator(func=_is_absolute_identifier)] [Required] (alias 'type')
Constraints:

  • func = <function _is_absolute_identifier at 0x77de2d4385e0>

pydantic model InputCollection[source]

A collection of input specifications used during runtime, split by scope, to validate presence of and to combine inputs.

Show JSON schema
{
   "title": "InputCollection",
   "description": "A collection of input specifications used during runtime, split by scope,\nto validate presence of and to combine inputs.",
   "type": "object",
   "properties": {
      "specs": {
         "additionalProperties": {
            "additionalProperties": {
               "$ref": "#/$defs/InputSpecification"
            },
            "type": "object"
         },
         "propertyNames": {
            "$ref": "#/$defs/InputScope"
         },
         "title": "Specs",
         "type": "object"
      }
   },
   "$defs": {
      "InputScope": {
         "description": "The scope that input must be provided in",
         "enum": [
            "tube",
            "process"
         ],
         "title": "InputScope",
         "type": "string"
      },
      "InputSpecification": {
         "description": "Specification of inputs to a noob tube.\n\nInputs can be supplied at different times and frequencies,\nas specified by `scope`:\n\n- `tube`: When instantiating the tube\n- `process`: Per call to :meth:`.TubeRunner.process`\n\n`tube`-scoped inputs may be used in a node's `param` specification,\nand `process`-scoped inputs may be used as one of a node's `depends`.\n\nInputs can be supplied at a \"higher\" scope and be accessed by lower scopes:\ne.g. input requested with a process scope can use input provided when instantiating the tube,\nif not provided to process but provided to the tube.",
         "properties": {
            "id": {
               "title": "Id",
               "type": "string"
            },
            "type": {
               "title": "Type",
               "type": "string"
            },
            "scope": {
               "$ref": "#/$defs/InputScope",
               "default": "tube"
            },
            "description": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "title": "Description"
            }
         },
         "required": [
            "id",
            "type"
         ],
         "title": "InputSpecification",
         "type": "object"
      }
   }
}

Fields:
field specs: dict[InputScope, dict[Annotated[str, AfterValidator(func=_is_identifier), AfterValidator(func=_not_reserved)], InputSpecification]] [Optional]
add_input(scope: InputScope, input: dict) → None[source]

Add some scope’s input to the input collection.

collect(edges: list[Edge], input: dict) → dict[source]
filter_input(scope: InputScope, input: dict) → dict[source]

filter input to only specified keys, emitting an ExtraInput warning if found.

get(key: str, input: dict | None = None) → Any[source]

Get a value from the inputs at any scope, if present

get_node_params(params: dict) → dict[source]

Get tube-scoped params specified as inputs needed when instantiating a node

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

  • self – The BaseModel instance.

  • context – The context.

validate_input(scope: InputScope, input: dict) → dict[source]

Check that the required inputs are present in one of several input dicts, and then filter to only specified input

INPUT_PATTERN: ClassVar[Pattern] = re.compile('input\\.(?P<key>.*)')
property chain: ChainMap

Make a chainmap of inputs at different scopes

(for possible expansion of number of scopes, to e.g. a runner scope)