plumpy.ports module

Module for process ports

class plumpy.ports.InputPort(name: str, valid_type: Type[Any] | None = None, help: str | None = None, default: Any = (), required: bool = True, validator: Callable[[Any, Port], str | None] | None = None)

Bases: Port

A simple input port for a value being received by a process

property default: Any

get_description() → Dict[str, str]

Return a description of the InputPort, which will be a dictionary of its attributes

Returns:

a dictionary of the stringified InputPort attributes

has_default() → bool

static required_override(required: bool, default: Any) → bool

If a default is specified an input should no longer be marked as required. Otherwise the input should always be marked explicitly to be not required even if a default is specified.

class plumpy.ports.OutputPort(name: str, valid_type: Type[Any] | None = None, help: str | None = None, required: bool = True, validator: Callable[[Any, Port], str | None] | None = None)

Bases: Port

class plumpy.ports.Port(name: str, valid_type: Type[Any] | None = None, help: str | None = None, required: bool = True, validator: Callable[[Any, Port], str | None] | None = None)

Bases: object

Specifications relating to a general input/output value including properties like whether it is required, valid types, the help string, etc.

get_description() → Dict[str, Any]

Return a description of the Port, which will be a dictionary of its attributes

Returns:

a dictionary of the stringified Port attributes

property help: str | None

Get the help string for this port

Returns:

the help string

property name: str

property required: bool

Is this port required?

Returns:

True if required, False otherwise

property valid_type: Type[Any] | None

Get the valid value type for this port if one is specified

Returns:

the value value type

validate(value: Any, breadcrumbs: Sequence[str] = ()) → PortValidationError | None

Validate a value to see if it is valid for this port

Parameters:

• value – the value to check

• breadcrumbs – a tuple of the path to having reached this point in validation

property validator: Callable[[Any, Port], str | None] | None

Get the validator for this port

Returns:

the validator

Return type:

Callable[[Any], Tuple[bool, Optional[str]]]

class plumpy.ports.PortNamespace(name: str = '', help: str | None = None, required: bool = True, validator: Callable[[Any, Port], str | None] | None = None, valid_type: Type[Any] | None = None, default: Any = (), dynamic: bool = False, populate_defaults: bool = True)

Bases: MutableMapping, Port

A container for Ports. Effectively it maintains a dictionary whose members are either a Port or yet another PortNamespace. This allows for the nesting of ports

NAMESPACE_SEPARATOR = '.'

_abc_impl = <_abc._abc_data object>

_ports: Dict[str, Port | PortNamespace]

absorb(port_namespace: PortNamespace, exclude: Sequence[str] | None = None, include: Sequence[str] | None = None, namespace_options: Dict[str, Any] | None = None) → List[str]

Absorb another PortNamespace instance into oneself, including all its mutable properties and ports.

Mutable properties of self will be overwritten with those of the port namespace that is to be absorbed. The same goes for the ports, meaning that any ports with a key that already exists in self will be overwritten. The namespace_options dictionary can be used to yet override the mutable properties of the port namespace that is to be absorbed. The exclude and include tuples can be used to exclude or include certain ports and both are mutually exclusive.

Parameters:

• port_namespace – instance of PortNamespace that is to be absorbed into self

• exclude – input keys to exclude from being exposed

• include – input keys to include as exposed inputs

• namespace_options – a dictionary with mutable PortNamespace property values to override

Returns:

list of the names of the ports that were absorbed

create_port_namespace(name: str, **kwargs: Any) → PortNamespace

Create and return a new PortNamespace in this PortNamespace. If the name is namespaced, the sub PortNamespaces will be created recursively, except if one of the namespaces is already occupied at any level by a Port in which case a ValueError will be thrown

Parameters:

• name – name (potentially namespaced) of the port to create and return

• kwargs – constructor arguments that will be used only for the construction of the terminal PortNamespace

Returns:

PortNamespace

Raises:

ValueError if any sub namespace is occupied by a non-PortNamespace port

property default: Any

property dynamic: bool

get_description() → Dict[str, Dict[str, Any]]

Return a dictionary with a description of the ports this namespace contains Nested PortNamespaces will be properly recursed and Ports will print their properties in a list

Returns:

a dictionary of descriptions of the Ports contained within this PortNamespace

get_port(name: str, create_dynamically: bool = False) → Port | PortNamespace

Retrieve a (namespaced) port from this PortNamespace. If any of the sub namespaces of the terminal port itself cannot be found, a ValueError will be raised

Parameters:

• name – name (potentially namespaced) of the port to retrieve.

• create_dynamically – If set to True, dynamically create the requested port if it doesn’t exist and the namespace is dynamic, instead of raising a ValueError.

Returns:

Port

Raises:

ValueError if port or namespace does not exist

has_default() → bool

property populate_defaults: bool

property ports: Dict[str, Port | PortNamespace]

pre_process(port_values: MutableMapping[str, Any]) → AttributesFrozendict

Map port values onto the port namespace, filling in values for ports with a default.

Parameters:

port_values – the dictionary with supplied port values

Returns:

an AttributesFrozenDict with pre-processed port value mapping, complemented with port default values

project(port_values: MutableMapping[str, Any]) → MutableMapping[str, Any]

Project a (nested) dictionary of port values onto the port dictionary of this PortNamespace. That is to say, return those keys of the dictionary that are shared by this PortNamespace. If a matching key corresponds to another PortNamespace, this method will be called recursively, passing the sub dictionary belonging to that port name.

Parameters:

port_values – a dictionary where keys are port names and values are actual input values

static strip_namespace(namespace: str, separator: str, rules: Sequence[str] | None = None) → List[str] | None

Filter given exclude/include rules staring with namespace and strip the first level.

For example if the namespace is base and the rules are:

('base.a', 'base.sub.b','relax.base.c', 'd')

the function will return:

('a', 'sub.c')

If the rules are None, that is what is returned as well.

Parameters:

• namespace – the string name of the namespace

• separator – the namespace separator string

• rules – the list or tuple of exclude or include rules to strip

Returns:

None if rules=None or the list of stripped rules

property valid_type: Type[Any] | None

Get the valid value type for this port if one is specified

Returns:

the value value type

validate(port_values: Mapping[str, Any] | None = None, breadcrumbs: Sequence[str] = ()) → PortValidationError | None

Validate the namespace port itself and subsequently all the port_values it contains

Parameters:

• port_values – an arbitrarily nested dictionary of parsed port values

• breadcrumbs – a tuple of the path to having reached this point in validation

Returns:

None or tuple containing 0: error string 1: tuple of breadcrumb strings to where the validation failed

validate_dynamic_ports(port_values: MutableMapping[str, Any], breadcrumbs: Sequence[str] = ()) → PortValidationError | None

Validate port values with respect to the dynamic properties of the port namespace. It will check if the namespace is actually dynamic and if all values adhere to the valid types of the namespace if those are specified

Parameters:

• port_values (dict) – an arbitrarily nested dictionary of parsed port values

• breadcrumbs (Tuple[str]) – a tuple of the path to having reached this point in validation

Returns:

if invalid returns a string with the reason for the validation failure, otherwise None

Return type:

Optional[str]

validate_ports(port_values: MutableMapping[str, Any], breadcrumbs: Sequence[str]) → PortValidationError | None

Validate port values with respect to the explicitly defined ports of the port namespace. Ports values that are matched to an actual Port will be popped from the dictionary

Parameters:

• port_values – an arbitrarily nested dictionary of parsed port values

• breadcrumbs – a tuple of breadcrumbs showing the path to to the value being validated

Returns:

None or tuple containing 0: error string 1: tuple of breadcrumb strings to where the validation failed

exception plumpy.ports.PortValidationError(message: str, port: str)

Bases: Exception

Error when validation fails on a port

property message: str

Get the validation error message

Returns:

the error message

property port: str

Get the port breadcrumbs

Returns:

the port where the error occurred