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)[source]#

Bases: BaseModel

IPAM Schema.

config: list[IPAMConfig]#
model_config = {}#

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

class openfactory.schemas.infra.IPAMConfig(**data)[source]#

Bases: BaseModel

IPAM Configuration Schema.

aux_addresses: Dict[str, str] | None#
gateway: str | None#
ip_range: str | None#
model_config = {}#

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

subnet: str#
class openfactory.schemas.infra.InfrastructureSchema(**data)[source]#

Bases: BaseModel

Infrastructure Schema.

model_config = {}#

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)[source]#

Bases: RootModel[Dict[str, Node]]

Docker Swarm Managers Schema.

model_config = {}#

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

class openfactory.schemas.infra.Network(**data)[source]#

Bases: BaseModel

Network Schema.

ipam: IPAM#
model_config = {}#

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

name: str | None#
class openfactory.schemas.infra.Networks(**data)[source]#

Bases: BaseModel

Networks Schema.

docker_ingress_network: Network#
model_config = {}#

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

openfactory_network: Network#
class openfactory.schemas.infra.Node(**data)[source]#

Bases: BaseModel

Docker Swarm Node Schema.

ip: IPv4Address#
labels: Dict[str, str] | None#
model_config = {}#

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

class openfactory.schemas.infra.Nodes(**data)[source]#

Bases: BaseModel

Docker Swarm Nodes Schema.

check_unique_ips()[source]#

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 = {}#

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)[source]#

Bases: BaseModel

Volume Schema.

driver_opts: Dict[str, str] | None#
model_config = {}#

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

class openfactory.schemas.infra.Volumes(root=PydanticUndefined, **data)[source]#

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

Volumes Schema.

model_config = {}#

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

class openfactory.schemas.infra.Workers(root=PydanticUndefined, **data)[source]#

Bases: RootModel[Dict[str, Node]]

Docker Swarm Workers Schema.

model_config = {}#

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)[source]#

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.