flync.core

flync.core.annotations

external

External field annotations and output strategies.

class NamingStrategy(*values)

Bases: IntEnum

The strategy on how the external field file/folder will be named.

class OutputStrategy(*values)

Bases: IntFlag

The strategy on how an external field will be generated.

class External(path: str | None = None, root: str | None = None, output_structure: OutputStrategy = <OutputStrategy.AUTO: 1>, naming_strategy: NamingStrategy = NamingStrategy.AUTO)

Bases: object

Indicates this field is loaded from a separate location.

implied

Provides annotations for fields that are automatically derived rather than manually set.

class ImpliedStrategy(*values)

Bases: IntEnum

The strategy on how an implied field will be calculated.

class Implied(strategy: ImpliedStrategy = ImpliedStrategy.AUTO)

Bases: object

Indicates this field is implied instead of loaded/generated.

flync.core.base_models

base_models

Base Model that is used by FLYNC Model classes.

class FLYNCBaseModel

Bases: BaseModel

Base Model that is used by FLYNC Model classes.

model_config = {'extra': 'forbid'}

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

model_post_init(_FLYNCBaseModel__context)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

model_dump(**kwargs)

Override pydantics model_dump to dump with defaults.

unique_names

Base class that ensures instance names are unique.

class UniqueName(*, name: str)

Bases: FLYNCBaseModel

Base class that ensures instance names are unique.

model_config = {'extra': 'forbid'}

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

model_post_init(_FLYNCBaseModel__context)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

registry

The registry is a context-local store that tracks every validated model instance during a single load or build operation. It is the mechanism that lets validators look up objects that were created earlier in the same workspace – for example, resolving a port name string into the actual ECUPort object that was already validated.

Two storage structures are maintained:

class Registry

Bases: object

Base for FLYNC model registries requiring a reset mechanism.

Attributes:

names: Set of registered names for this registry instance. dict_by_class: Mapping from type to a dict of id-keyed instances. list_by_class: Mapping from type to an ordered list of instances.

register_dict_item(instance, instance_id)

Register an instance ID in the dict store for its type.

Args:

instance: The object to register. instance_id: The key for the instance within its type bucket.

register_list_item(instance)

Append an instance to the list store for its type.

Args:

instance: The object to register.

get_dict(cls: type) → dict

Return the registered dict for a given class.

Args:

cls: The type whose dict store should be returned.

Returns:

A dict of instances, or an empty dict if none have been registered for the given class.

get_list(cls: type) → list

Return the registered list for a given class.

Args:

cls: The type whose list store should be returned.

Returns:

A list of registered instances, or an empty list if none have been registered for the given class.

registry_context(registry: Registry)

Context manager that activates a registry for the current context.

Args:

registry: The Registry instance to set as active.

Yields:

The provided registry while the context is active.

get_registry() → Registry

Return the currently active registry.

Returns:

The active Registry instance.

Raises:

RuntimeError: If no registry is active in the current context.

ensure_registry()

Yield the active registry, creating a temporary one if none exists.

Yields:

The existing active Registry if one is set, otherwise a newly created Registry that is cleaned up on exit.

Using the registry with FLYNCWorkspace

Loading a workspace from files

load_workspace() and safe_load_workspace() automatically wrap all YAML loading inside a fresh registry_context(). Every model validated during that call shares the same registry, so cross-references (switch-port names, controller-interface names, service IDs, etc.) resolve correctly. You do not need to manage the registry yourself when using these methods:

from flync.sdk.workspace.flync_workspace import FLYNCWorkspace

workspace = FLYNCWorkspace.load_workspace("my_workspace", "/path/to/workspace")
# All cross-references inside the workspace are already resolved.

Warning

Each call to load_workspace / safe_load_workspace creates an isolated registry. Objects from one workspace cannot be referenced by validators of another workspace loaded in a separate call. If you need to validate connections that span two workspaces you must build a combined model manually (see below).

Building a model programmatically

When you create FLYNC model objects in Python code without going through load_workspace, Pydantic validators run immediately at construction time. Any validator that calls get_registry() will raise a RuntimeError if no registry is active.

Wrap the entire construction block in registry_context() so that all objects – and all validators – share the same store:

from flync.core.base_models.instances_registery import (
    Registry,
    registry_context,
)
from flync.model.flync_4_ecu.switch import Switch, SwitchPort

with registry_context(Registry()):
    port = SwitchPort(name="p0", silicon_port_no=0, ...)
    switch = Switch(name="sw0", ports=[port], ...)
    # port is now registered; switch validators can look it up.

Use ensure_registry() instead when the code may be called either inside or outside an existing registry context (for example, from a helper that is also used during a load_workspace call):

from flync.core.base_models.instances_registery import ensure_registry

with ensure_registry():
    # Uses the caller's active registry if one exists,
    # or creates a new one for this block.
    model = MyModel(...)

dict_instances

Base classes that automatically store created model instances.

class DictInstances

Bases: FLYNCBaseModel, Generic[T]

Base class that registers validated instances in the active Registry under a caller-supplied key.

After Pydantic validation completes, ensure_unique_instances calls get_dict_key() and stores self in registry.dict_by_class[type(self)][key]. Any later validator that holds a name reference to an instance of this class can look it up without building a separate index.

Subclasses must implement get_dict_key().

Looking up a registered instance in a validator

Use get_registry() to obtain the active registry, then call get_dict with the concrete class:

from flync.core.base_models import Registry, get_registry
from pydantic import model_validator

class MyConnection(FLYNCBaseModel):
    port_name: str
    _port: Optional[MyPort] = PrivateAttr(default=None)

    @model_validator(mode="after")
    def resolve_port(self):
        registry: Registry = get_registry()
        instances = registry.get_dict(MyPort)
        self._port = instances.get(self.port_name)
        if self._port is None:
            raise ValueError(f"Port '{self.port_name}' not found")
        return self

This is the pattern used throughout flync.model.flync_4_ecu.internal_topology, for example in ECUPortToXConnection.validate_ecu_port_exists:

registry: Registry = get_registry()
ecu_ports_instances = registry.get_dict(ECUPort)
self._ecu_port = ecu_ports_instances.get(self.ecu_port_name, None)

Injecting a parent back-reference into children

When a child model needs to navigate back to its parent at runtime, override model_post_init on the parent to set a private attribute on each child. Always call super().model_post_init(__context) last:

class MyParent(DictInstances):
    children: List[MyChild] = Field()

    def model_post_init(self, __context):
        for child in self.children:
            child._parent = self
        return super().model_post_init(__context)

See Switch.model_post_init in flync.model.flync_4_ecu.switch for the canonical example:

def model_post_init(self, __context):
    for port in self.ports:
        port._switch = self
    return super().model_post_init(__context)
model_config = {'extra': 'forbid'}

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

model_post_init(_FLYNCBaseModel__context)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

class NamedDictInstances(*, name: str)

Bases: UniqueName, Generic[T]

Base class for named models registered in the active Registry keyed by their name field.

Identical to DictInstances but the registry key is always self.name (via get_instance_key()), so subclasses do not need to implement get_dict_key().

Looking up a registered instance in a validator

The lookup pattern is the same as for DictInstances:

from flync.core.base_models import Registry, get_registry
from pydantic import model_validator

class MyConnection(FLYNCBaseModel):
    iface_name: str
    _iface: Optional[MyIface] = PrivateAttr(default=None)

    @model_validator(mode="after")
    def resolve_iface(self):
        registry: Registry = get_registry()
        instances = registry.get_dict(MyIface)
        self._iface = instances.get(self.iface_name)
        if self._iface is None:
            raise ValueError(
                f"Interface '{self.iface_name}' not found"
            )
        return self

See SwitchPortToControllerInterface.validate_connection_compatibility in flync.model.flync_4_ecu.internal_topology for a real example:

registry: Registry = get_registry()
controller_interface_instances = registry.get_dict(ControllerInterface)
self._iface = controller_interface_instances.get(self.iface_name, None)

Injecting a parent back-reference into children

Same as for DictInstances – override model_post_init on the parent to assign a private back-reference on each child:

class MyParent(NamedDictInstances):
    children: List[MyChild] = Field()

    def model_post_init(self, __context):
        for child in self.children:
            child._parent = self
        return super().model_post_init(__context)
model_config = {'extra': 'forbid'}

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

model_post_init(_FLYNCBaseModel__context)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

list_instances

Base classes that automatically collect model instances in a list.

class ListInstances

Bases: FLYNCBaseModel, Generic[T]

Base class that appends every validated instance to the active Registry list for its concrete type.

After Pydantic validation completes, ensure_unique_instances calls registry.register_list_item(self), which appends self to registry.list_by_class[type(self)]. Unlike DictInstances, there is no key – the list is meant for validators that need to iterate over all instances of a type and filter by some condition.

Looking up instances in a validator

Use get_registry() to obtain the active registry, then call get_list with the concrete class and iterate:

from flync.core.base_models import Registry, get_registry
from pydantic import model_validator

class MyChild(FLYNCBaseModel):
    name: str

    def get_parent(self):
        registry: Registry = get_registry()
        for parent in registry.get_list(MyParent):
            if self in parent.children:
                return parent
        raise ValueError("No parent found for this child")

This is the pattern used in flync.model.flync_4_ecu.controller, for example in ControllerInterface.get_controller:

registry: Registry = get_registry()
controller_instances = registry.get_list(Controller)
for ctrl in controller_instances:
    for interface in ctrl.interfaces:
        if interface.name == self.name:
            return ctrl

Injecting a parent back-reference into children

When a child model needs to navigate back to its parent at runtime (e.g. to call get_switch() on a port), override model_post_init on the parent to set a private attribute on each child. Always call super().model_post_init(__context) last so that the base-class registration runs:

class MyParent(ListInstances):
    children: List[MyChild] = Field()

    def model_post_init(self, __context):
        for child in self.children:
            child._parent = self
        return super().model_post_init(__context)

See Switch.model_post_init in flync.model.flync_4_ecu.switch for the canonical example:

def model_post_init(self, __context):
    for port in self.ports:
        port._switch = self
    return super().model_post_init(__context)
model_config = {'extra': 'forbid'}

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

model_post_init(_FLYNCBaseModel__context)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

class NamedListInstances(*, name: str)

Bases: UniqueName, Generic[T]

Base class that appends every validated named instance to the active Registry list, keyed by type.

Identical to ListInstances but extends UniqueName, so each instance carries a name field that is unique within a registry context. Use this class when instances need a name and are typically looked up by iterating all registered instances of a type.

Looking up instances in a validator

The lookup pattern is the same as for ListInstances – iterate registry.get_list(ClassName) and filter by name or any other attribute:

from flync.core.base_models import Registry, get_registry
from pydantic import model_validator

class MyConnection(FLYNCBaseModel):
    target_name: str
    _target: Optional[MyNode] = PrivateAttr(default=None)

    @model_validator(mode="after")
    def resolve_target(self):
        registry: Registry = get_registry()
        for node in registry.get_list(MyNode):
            if node.name == self.target_name:
                self._target = node
                return self
        raise ValueError(f"Node '{self.target_name}' not found")

Injecting a parent back-reference into children

Same as for ListInstances – override model_post_init on the parent to assign a private back-reference on each child:

class MyParent(NamedListInstances):
    children: List[MyChild] = Field()

    def model_post_init(self, __context):
        for child in self.children:
            child._parent = self
        return super().model_post_init(__context)
model_config = {'extra': 'forbid'}

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

model_post_init(_FLYNCBaseModel__context)

Override this method to perform additional initialization after __init__ and model_construct. This is useful if you want to do some validation that requires the entire model to be initialized.

flync.core.datatypes

class Datatype

Bases: FLYNCBaseModel

Base class of every datatype.

Parameters

namestr

Unique name of the datatype.

descriptionstr, optional

Human-readable description of the datatype.

typestr

Discriminator identifying the concrete datatype kind.

endiannessLiteral[“BE”, “LE”], optional

Byte order used for encoding multibyte values. Defaults to big-endian (“BE”).

member_namestr, optional

When this datatype is stored as a struct member, this field holds the member’s name within the struct (which may differ from the type’s own name). None when the datatype is not a struct member or when member name equals the type name.

class IPv4AddressEntry

Bases: FLYNCBaseModel

Represents an IPv4 address entry for a network interface.

Parameters

addressIPv4Address

The IPv4 address. “0.0.0.0” means dynamic.

ipv4netmaskIPv4Address

The subnet mask in IPv4 format.

class IPv6AddressEntry

Bases: FLYNCBaseModel

Represents an IPv6 address entry for a network interface.

Parameters

addressIPv6Address

The IPv6 address. “::” means dynamic.

ipv6prefixint

The prefix length (0-128).

class MACAddressEntry

Bases: FLYNCBaseModel

Represents an MAC address entry for a network interface.

Parameters:
  • address (str):

    Source MAC address to filter by. Format: “xx:xx:xx:xx:xx:xx”

  • macmask (str):

    The mask in MAC format. Format: “xx:xx:xx:xx:xx:xx”

class MACAddressUnicast

Bases: MACAddressEntry

Represents a Unicast MAC address entry for a network interface.

class MACAddressMulticast

Bases: MACAddressEntry

Represents a Multicast MAC address entry for a network interface.

class BitRange

Bases: Datatype

Represents a range of bits in a signal or data array.

Parameters

namestr

Unique name of the datatype.

descriptionstr, optional

Human-readable description of the datatype.

typestr

Datatype discriminator inherited from Datatype.

endiannessLiteral[“BE”, “LE”], optional

Byte order used for encoding multibyte values. Defaults to big-endian (“BE”).

startint

Starting bit position (inclusive).

endint

Ending bit position (inclusive).

class ValueRange

Bases: FLYNCBaseModel

Defines an inclusive range of integer values.

This datatype is typically used to express valid numeric intervals for parameters, identifiers, or signal values.

Parameters

from_valueint

Lower bound of the range (inclusive).

to_valueint

Upper bound of the range (inclusive).

class ValueTable

Bases: FLYNCBaseModel

Represents a table of values with an associated description.

Parameters

num_valueint

The numeric value associated with this table entry.

descriptionstr

Human-readable description of the table entry.

flync.core.utils

base_utils

Base Utils that can be useful throughout the whole FLYNC Library and toolchain.

read_yaml(path: str | PathLike | Path)

Read a YAML file.

Args:

path (str | os.PathLike | Path): Path to the YAML file

Raises:

err_fatal: If path is not for YAML file.

Returns:

Any: Retrieved data from YAML.

write_to_file(obj, file_to_update)

Read a YAML file.

Args:

obj : FLYNC object to write file_to_update : Path of the file for output

get_yaml_paths(base_path: str | PathLike) → list

Collect absolute paths to yaml files from a base_path.

Args:

base_path (str | os.PathLike): Base Path to FLYNC Config.

Returns:

list: List of absolute file paths to yaml files.

is_mac_address(input: str) → Tuple[bool, str]

Helper to check if an input is a valid MAC address based on pydantic validator.

Args:

input (str): input string that should be checked.

Returns:

bool: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is a MAC address. If the boolean is false, it is not.

is_mac_unicast(input: str) → Tuple[bool, str]

Helper to check if a MAC address is unicast. Unicast if first byte’s least significant bit is 0.

Args:

input (str): input string that should be checked.

Returns:

Union[bool, str]: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is a unicast MAC address. If the boolean is false, it is not.

is_mac_multicast(input: str) → Tuple[bool, str]

Method to check if a MAC address is multicast. Multicast if first byte’s least significant bit is 1.

Args:

input (str): input string that should be checked.

Returns:

Union[bool, str]: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is a MAC multicast address. If the boolean is false, it is not.

is_ip_address(input: IPv4Address | IPv6Address | str) → Tuple[bool, str]

Helper to check if an input is a valid IP address.

Args:

input (IPv4Address | IPv6Address | str): input string that should be checked.

Returns:

bool: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is an IP address. If the boolean is false, it is not.

is_ip_multicast(input: IPv4Address | IPv6Address | str) → Tuple[bool, str]

Method to check if a string is an IP multicast address.

Args:

input (IPv4Address | IPv6Address): input string that should be checked.

Returns:

Union[bool, str]: Returns the result of check as a boolean as well as a message that could be used in logging or exception handling. If the result boolean is true, the provided input is an IP multicast address. If the boolean is false, it is not.

get_duplicates_in_list(input: list) → list

Find duplicates in a list.

Args:

input (list): a list where duplicates are suspected.

Returns:

list: returns a list of the duplicates that were found.

check_obj_in_list(obj, list)

Helper function: To check if the object is in the list or not

deep_iter(obj, stop_class)

Helper function: To deeply iterate through an object and its children. The iteration will stop if an object of type stop_class is reached.

Args:

obj: The object to iterate through. stop_class: The class type at which the iteration should stop.

Returns:

Generator: A generator that yields objects of the specified type.

find_all(base, target_class: Type[T]) → list[T]

Helper to get all fields of certain type in nested data.

Args:

base: The base object to start searching from. target_class: The class to search for.

Returns:

list: A list of all instances of the target class found within the base object.

common_validators

Common Validators are validation methods that are used throughout the whole FLYNC model. The Validators either raise minor, major or fatal errors as pydantic usage proposes.

validate_vlan_id(value)

Validate a VLAN identifier.

None is treated as untagged and returned unchanged. Values in the range 0-4094 are accepted as-is. The reserved value 4095 is accepted but emits a warning via warn(). Anything outside 0-4095 raises a minor validation error.

validate_or_remove(label: str, field_type: Any, severity: str = 'minor')

Factory that returns a BeforeValidator for sub-model fields.

Use inside Annotated to pre-validate a field before Pydantic processes it. If the raw data fails validation all sub-errors are packed into a single error.

  • "minor" severity: the field is removed and the parent model still loads without it.

    The message says “Removing {label}…”.

  • "major" severity: the parent model will fail regardless (the field is required).

    The message reports the validation failure without implying graceful removal.

The parent object’s name field is included in the error message when available via info.data.

Parameters

labelstr

Human-readable field label used in the error message.

field_typeAny

Pydantic-compatible type to validate the data against.

severitystr, optional

Error severity — "minor" (default) or "major".

Returns

Callable

A two-argument validator (data, info) ready for use with BeforeValidator.

validate_list_items_and_remove(label: str, item_type: Any, severity: str = 'minor')

Validate each item in a list individually, removing only invalid entries.

Unlike validate_or_remove(), which discards the entire list when any item fails, this validator keeps valid items and removes only those that fail validation. Per-item errors are forwarded to the _validation_warnings channel so they appear in the final error report even though the model continues building with the remaining valid items.

Use inside Annotated as a BeforeValidator.

Parameters

labelstr

Human-readable field label used in error messages.

item_typeAny

Pydantic-compatible type for each individual list item.

severitystr, optional

Error severity for removed items — "minor" (default) or "major".

Returns

Callable

A two-argument validator (data, info) ready for use with BeforeValidator.

validate_mac_unicast(input: str) → str

Custom Validator for Unicast MAC addresses.

Args:

input (str): MAC address to validate.

Raises:

err_minor: Input is not a Unicast address based on the expected format.

Returns:

Any: Input is handed over.

validate_mac_multicast(input: str) → Any

Custom Validator for Multicast MAC addresses.

Args:

input (str): MAC address to validate.

Raises:

err_minor: Input is not a Multicast address based on the expected format.

Returns:

Any: Input is handed over.

validate_ip_multicast(input: IPv4Address | IPv6Address | str) → Any

Custom Validator for Multicast IP addresses.

Args:

input (IPv4Address | IPv6Address): IP address to validate.

Raises:

err_minor: Input is not a Multicast address based on the expected format.

Returns:

Any: Input is handed over.

validate_any_multicast_address(input: IPv4Address | IPv6Address | str) → Any

Custom Validator for Multicast MAC or IP addresses.

Args:

input (IPv4Address | IPv6Address | str): IP address or MAC Address to validate.

Raises:

err_minor: The address is not a multicast address.

Returns:

Any: Input is handed over.

validate_multicast_list_only_ip(input_list: list)

Custom Validator for a list of Multicast IP addresses.

Args:

input_list (list): List of only Multicast IPs.

Raises:

err_minor: Any of the addresses in the list is not an IP multicast address.

validate_multicast_list(input_list: list)

Custom Validator for a list of Multicast MAC or IP addresses.

Args:

input_list (list): List of Multicast IPs and MACs.

Raises:

err_minor: Any of the addresses in the list is not a multicast address.

validate_ingress_streams_fields(streams, location: str)

Raise err_minor if any stream carries an ipv or ats value.

location is a human-readable label such as "compute node" or "controller interface" used in the error message.

validate_vlan_ids_unique(virtual_interfaces, name: str)

Raise err_major if any VLAN ID appears more than once.

validate_list_items_unique(input_list: list, list_label: str | None = None) → list

Custom Validator for a list of items where every item should be unique.

Args:

input_list (list): List of items.

list_label(str): Add an optional label to the error message.

Raises:

err_major: List contains duplicates.

Returns:

list: Input is handed over.

validate_cbs_idleslopes_fit_portspeed(traffic_classes: list, port_speed: int)

Custom Validator for a list of Traffic Classes to check conformity to MII/MDI speed.

Args:

traffic_classes (list): List of element type TrafficClass.

port_speed (int): MII or MDI speed of the port.

Raises:

err_major: The sum of idleslopes of all shapers on one port must be equal or lower than the port speed.

Returns:

list: Return list of traffic classes as received.

validate_optional_mii_config_compatibility(comp1, comp2, id)

Custom validator for optional MII configuration compatibility between two components.

Args:

comp1 (object): First component that may contain a mii_config attribute.

comp2 (object): Second component that may contain a mii_config attribute.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: One component has an MII config while the other does not.

err_major: Both components have an MII config but the mode values are identical. The modes must differ.

err_major: Both components have an MII config but the speed values are different.

err_major: Both components have an MII config but the type values are different.

validate_compulsory_mii_config_compatibility(comp1, comp2, id)

Validator that enforces a mandatory MII configuration on both components and then checks optional compatibility.

Args:

comp1 (object): First component. Must have mii_config.

comp2 (object): Second component. Must have mii_config.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: Either component is missing a required MII configuration.

err_major: Propagated from validate_optional_mii_config_compatibility() when the optional checks fail.

validate_htb(comp, speed)

Validator that checks an HTB (Hierarchical Token Bucket) configuration against the physical link speed.

Args:

comp (object): Component that owns an htb attribute with child_classes.

speed (int): Link speed of the interface (same unit as the HTB rates).

Raises:

err_major: The sum of the rate values of all child classes exceeds the provided speed.

validate_macsec(comp1, comp2, id)

Validator for MACsec configuration compatibility between two components.

Args:

comp1 (object): First component: May contain a macsec_config.

comp2 (object): Second component: May contain a macsec_config.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: One component has a MACsec config while the other does not.

err_major: MKA (Key Agreement) enabled state differs between the two components.

err_major: macsec_mode differs between the two components.

validate_gptp(comp1, comp2, id)

Validator that checks gPTP (generic Precision Time Protocol) configuration compatibility between two components.

Args:

comp1 (object): First component. May contain a ptp_config.

comp2 (object): Second component. May contain a ptp_config.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: PTP configuration present on one side only.

err_major: Mismatch of the cmlds_linkport_enabled flag between the two components.

err_major: Propagated from validate_gptp_domains() when domain level checks fail.

validate_gptp_domains(comp1, comp2, ptp1, ptp2, id)

Helper that validates matching PTP domains and sync-config types between two components.

Args:

comp1 (object): First component (source of ptp1).

comp2 (object): Second component (source of ptp2).

ptp1 (object): ptp_config of comp1.

ptp2 (object): ptp_config of comp2.

id (Any): Identifier of the connection (used only in error messages).

Raises:

err_major: A domain present in ptp1 is missing in ptp2.

err_major: The sync_config.type of a matching domain is identical on both sides (they must differ for a valid configuration).

validate_elements_in(subset: Iterable[Any], superset: Iterable[Any], msg: str | None = None)

Custom Validator that checks if every element in subset appears at least once in superset. E.g. Validate if port_name is in switch_port_names.

Args:

subset (Iterable[Any]): Subset where elements are expected to be in superset.

superset (Iterable[Any]): Reference set.

Returns:

Iterable[Any]: Return subset as received.

check_prio_unique(traffic_classes)

Check if the traffic class prios are unique across various traffic classes in a controller interface or switch.

check_pcps_different(traffic_classes)

Check if the PCPs are different across traffic classes.

check_ipvs_unique(traffic_classes)

Check if ipvs across traffic classes are unique.

validate_traffic_classes(traffic_classes)

Validate the traffic classes in a controller interface and switch to find out if a pcp, ipv or traffic class prio is reused or not.

none_to_empty_list(v)

Make the field defined as optional [] if accidentally declared by the user as None.

exceptions

Defines custom pydantic errors.

warn(msg: str) → None

Record a validation warning without raising a validation error.

The message is appended to the active warning list (set up by validate_with_policy) and will be returned alongside load_errors so that it appears in the warnings table. If called outside a validate_with_policy context the call is silently ignored.

Parameters

msgstr

Human-readable warning message.

err_minor(msg: str, **ctx) → PydanticCustomError

Factory that returns PydanticCustomError with type minor.

Parameters

msgstr

Error message that may contain placeholders

ctxdict

Context arguments that define key-value pairs to fill the placeholders in msg

Returns

PydanticCustomError

err_major(msg: str, **ctx) → PydanticCustomError

Factory that returns PydanticCustomError with type major.

Parameters

msgstr

Error message that may contain placeholders

ctxdict

Context arguments that define key-value pairs to fill the placeholders in msg

Returns

PydanticCustomError

err_fatal(msg: str, **ctx) → PydanticCustomError

Factory that returns PydanticCustomError with type fatal.

Parameters

msgstr

Error message that may contain placeholders

ctxdict

Context arguments that define key-value pairs to fill the placeholders in msg

Returns

PydanticCustomError

exceptions_handling

Provides utilities for validating models and handling errors.

resolve_alias(model: type[BaseModel], field_name: str) → str

Return the YAML key used for a Pydantic field, considering alias.

get_name_by_alias(model: type[BaseModel], alias: str)

Return the Python field name that corresponds to the given alias.

Parameters

modeltype[BaseModel]

The Pydantic model class to search.

aliasstr

The alias to look up.

Returns

str

The Python attribute name whose alias matches alias.

Raises

KeyError

If no field with the given alias is found.

safe_yaml_position(node: Any, loc: tuple, model: type[BaseModel] | None = None) → Tuple[int | None, int | None]

Given a ruamel.yaml node and a Pydantic loc tuple, return (line, column). Falls back gracefully if key/item is missing.

errors_to_init_errors(errors: List[ErrorDetails], model: type[BaseModel] | None = None, yaml_data: object | None = None, yaml_path: str | None = None) → List[InitErrorDetails]

Convert Pydantic validation errors into InitErrorDetails for re-raising.

Optionally enriches each error with YAML source location information when model and yaml_data are provided, and with the file path when yaml_path is provided.

Parameters

errorsList[ErrorDetails]

The list of errors to convert.

modeltype[BaseModel], optional

The Pydantic model class used to resolve field aliases for YAML position look-ups.

yaml_dataobject, optional

The parsed ruamel.yaml AST of the document, used together with model to locate the error position within the file.

yaml_pathstr, optional

The workspace-relative file path to embed in each error’s context as yaml_path.

Returns

List[InitErrorDetails]

The converted errors, ready to be passed to ValidationError.from_exception_data.

delete_at_loc(data: Any, loc: Tuple)

Helper function to remove the key/item from original object by loc(path to an element within the object).

Parameters

dataAny

Data to remove the item from. Will be mutated.

locTuple

Path to the location of item to remove.

get_unique_errors(errors: List[ErrorDetails]) → List[ErrorDetails]

A function to get the list of unique errors.

Parameters

errors: List[ErrorDetails]

The list of pydantic’s error details

Returns

List[ErrorDetails]

validate_with_policy(model: Type[FLYNCBaseModel], data: Any, path) → Tuple[FLYNCBaseModel | None, List[ErrorDetails]]

Helper function to perform model validation from the given data, collect errors with different severity and perform action based on severity.

For minor/major errors the offending field is removed from the working data via delete_at_loc() and validation is retried, so that the model can still be constructed without the invalid field. The loop continues until either validation succeeds, a fatal error is encountered, or no further progress can be made (all error locations already removed).

Parameters

modelType[FLYNCBaseModel]

Flync model class.

dataAny

Data to validate and instantiate the model with.

Returns

Tuple[Optional[FLYNCBaseModel], List]

Tuple with optional model instance and list of errors.

Raises

ValidationError