OpenFactory Infrastructure Schemas

Infrastructure Configuration Schemas for OpenFactory

This module defines Pydantic models used to validate and process infrastructure configuration files in OpenFactory. These configurations typically describe networking, nodes, volumes, and IP address management (IPAM) settings for Docker Swarm-based deployments.

Key Components:

• Volumes: Defines named Docker volumes and associated driver options.

• Nodes: Represents Docker Swarm managers and workers with IP and labels.

• Networks: Includes support for IPAM configuration and multiple network definitions.

• InfrastructureSchema: Top-level model combining nodes, volumes, and networks.

• get_infrastructure_from_config_file: Utility function to load and validate YAML-based infrastructure config.

Validation Features:

• Ensures that manager and worker node IP addresses are unique across the cluster.

• Uses Pydantic aliases for compatibility with external config file formats.

• Provides rich error messages via user notification hooks.

Example YAML Structure:

nodes:
  managers:
    manager-1:
      ip: 192.168.1.10
  workers:
    worker-1:
      ip: 192.168.1.11

volumes:
  my-volume:
    driver_opts:
      type: nfs
      o: addr=192.168.1.20,nolock,soft,rw
      device: ":/data"

networks:
  openfactory-network:
    name: ofa-net
    ipam:
      config:
        - subnet: 192.168.0.0/16
          gateway: 192.168.0.1

Usage:

Call get_infrastructure_from_config_file(path) to load and validate the infrastructure config.

This module is typically used by OpenFactory’s bootstrap and deployment tooling.

class openfactory.schemas.infra.IPAM(**data)

Bases: BaseModel

IPAM Schema.

config: list[IPAMConfig]

model_config: ClassVar[ConfigDict] = {}

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

class openfactory.schemas.infra.IPAMConfig(**data)

Bases: BaseModel

IPAM Configuration Schema.

aux_addresses: Dict[str, str] | None

gateway: str | None

ip_range: str | None

model_config: ClassVar[ConfigDict] = {}

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

subnet: str

class openfactory.schemas.infra.InfrastructureSchema(**data)

Bases: BaseModel

Infrastructure Schema.

model_config: ClassVar[ConfigDict] = {}

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

networks: Dict[str, Network] | None

nodes: Nodes

volumes: Volumes | None

class openfactory.schemas.infra.Managers(root=PydanticUndefined, **data)

Bases: RootModel[Dict[str, Node]]

Docker Swarm Managers Schema.

model_config: ClassVar[ConfigDict] = {}

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

class openfactory.schemas.infra.Network(**data)

Bases: BaseModel

Network Schema.

ipam: IPAM

model_config: ClassVar[ConfigDict] = {}

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

name: str | None

class openfactory.schemas.infra.Networks(**data)

Bases: BaseModel

Networks Schema.

docker_ingress_network: Network

model_config: ClassVar[ConfigDict] = {}

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

openfactory_network: Network

class openfactory.schemas.infra.Node(**data)

Bases: BaseModel

Docker Swarm Node Schema.

ip: IPv4Address

labels: Dict[str, str] | None

model_config: ClassVar[ConfigDict] = {}

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

class openfactory.schemas.infra.Nodes(**data)

Bases: BaseModel

Docker Swarm Nodes Schema.

check_unique_ips()

Validates that IP addresses are unique across managers and workers.

Return type:

Dict

Parameters:

values (Dict) – Dictionary of values to validate.

Returns:

Dict – Validated values.

Raises:

ValueError – If IP addresses are not unique.

managers: Dict[str, Node] | None

model_config: ClassVar[ConfigDict] = {}

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

workers: Dict[str, Node] | None

class openfactory.schemas.infra.Volume(**data)

Bases: BaseModel

Volume Schema.

driver_opts: Dict[str, str] | None

model_config: ClassVar[ConfigDict] = {}

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

class openfactory.schemas.infra.Volumes(root=PydanticUndefined, **data)

Bases: RootModel[Dict[str, Union[Volume, NoneType]]]

Volumes Schema.

model_config: ClassVar[ConfigDict] = {}

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

class openfactory.schemas.infra.Workers(root=PydanticUndefined, **data)

Bases: RootModel[Dict[str, Node]]

Docker Swarm Workers Schema.

model_config: ClassVar[ConfigDict] = {}

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

openfactory.schemas.infra.get_infrastructure_from_config_file(infra_yaml_config_file)

Load and validate infrastructure configuration from a YAML file.

This function reads a YAML file containing infrastructure configuration, validates its content using the InfrastructureSchema Pydantic model, and returns the parsed data as a dictionary.

Return type:

Optional[Dict[str, InfrastructureSchema]]

Parameters:

infra_yaml_config_file (str) – Path to the YAML file defining the infrastructure configuration.

Returns:

Optional[Dict[str, InfrastructureSchema]] –

A dictionary of validated infrastructure configuration data,

or None if validation fails.

Note

In case of validation errors, user notifications will be triggered and None will be returned.