Product Blocks

Defining a Product Block

Warning

You should only have one root Product Block on your domain model!

A Product Block is a reusable collection of product attributes that lives under a Product Type (which we covered in the Product Types section of these docs). A Product Block allows you to define the bulk of the attributes that you want to assign to your Product definition. At it's most basic, you will have a root product block that is stored on your Product Type, as shown in the Product Types documentation. Building off of that example, let's examine a basic product block by examining the NodeBlock product block from the example workflow orchestrator:

# Copyright 2019-2023 SURF.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle
from pydantic import computed_field


class NodeBlockInactive(ProductBlockModel, product_block_name="Node"):
    role_id: int | None = None
    type_id: int | None = None
    site_id: int | None = None
    node_status: str | None = None  # should be NodeStatus, but strEnum is not supported (yet?)
    node_name: str | None = None
    node_description: str | None = None
    ims_id: int | None = None
    nrm_id: int | None = None
    ipv4_ipam_id: int | None = None
    ipv6_ipam_id: int | None = None


class NodeBlockProvisioning(NodeBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
    role_id: int
    type_id: int
    site_id: int
    node_status: str  # should be NodeStatus, but strEnum is not supported (yet?)
    node_name: str
    node_description: str | None = None
    ims_id: int | None = None
    nrm_id: int | None = None
    ipv4_ipam_id: int | None = None
    ipv6_ipam_id: int | None = None

    @computed_field  # type: ignore[misc]
    @property
    def title(self) -> str:
        return f"node {self.node_name} status {self.node_status}"


class NodeBlock(NodeBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
    role_id: int
    type_id: int
    site_id: int
    node_status: str  # should be NodeStatus, but strEnum is not supported (yet?)
    node_name: str
    node_description: str | None = None
    ims_id: int
    nrm_id: int
    ipv4_ipam_id: int
    ipv6_ipam_id: int

Breaking this product block down a bit more, we see 3 classes, NodeBlockInactive, NodeBlockProvisioning, and finally NodeBlock. These three classes are built off of each-other, with the lowest level class (NodeBlockInactive) based off of the ProductBlockModel base class. These classes have a number of attributes, referred to as "Resource Types", which you can read more about here. Looking at this ProductBlockModel base class tells us a lot about how to use it in our workflows:

orchestrator.domain.base.ProductBlockModel

Bases: orchestrator.domain.base.DomainModel

This is the base class for all product block models.

This class should have been called SubscriptionInstanceModel.

ProductTable Blocks are represented as dataclasses with pydantic runtime validation.

Different stages of a subscription lifecycle could require different product block definition.Mainly to support mandatory fields when a subscription is active. To support this a lifecycle specific product block definition can be created by subclassing the generic product block with keyword argument 'lifecycle' and overriding its fields.

All product blocks are related to a database ProductBlockTable object through the product_block_name that is given as class keyword argument.

Define a product block:

>>> class BlockInactive(ProductBlockModel, product_block_name="Virtual Circuit"):
...    int_field: Optional[int] = None
...    str_field: Optional[str] = None

>>> class Block(BlockInactive, lifecycle=[SubscriptionLifecycle.ACTIVE]):
...    int_field: int
...    str_field: str

This example defines a product_block with two different contraints based on lifecycle. Block is valid only for ACTIVE And BlockInactive for all other states. product_block_name must be defined on the base class and need not to be defined on the others

Create a new empty product block:

>>> example1 = BlockInactive()  # doctest: +SKIP

Create a new instance based on a dict in the state:

>>> example2 = BlockInactive(**state)  # doctest:+SKIP

To retrieve a ProductBlockModel from the database:

>>> BlockInactive.from_db(subscription_instance_id)  # doctest:+SKIP

Source code in orchestrator/domain/base.py

class ProductBlockModel(DomainModel):
    r"""This is the base class for all product block models.

    This class should have been called SubscriptionInstanceModel.

    ProductTable Blocks are represented as dataclasses with pydantic runtime validation.

    Different stages of a subscription lifecycle could require different product block
    definition.Mainly to support mandatory fields when a subscription is active. To support
    this a lifecycle specific product block definition can be created by subclassing the
    generic product block with keyword argument 'lifecycle' and overriding its fields.

    All product blocks are related to a database ProductBlockTable object through the `product_block_name`
    that is given as class keyword argument.

    Define a product block:

        >>> class BlockInactive(ProductBlockModel, product_block_name="Virtual Circuit"):
        ...    int_field: Optional[int] = None
        ...    str_field: Optional[str] = None

        >>> class Block(BlockInactive, lifecycle=[SubscriptionLifecycle.ACTIVE]):
        ...    int_field: int
        ...    str_field: str

    This example defines a product_block with two different contraints based on lifecycle. `Block` is valid only for `ACTIVE`
    And `BlockInactive` for all other states.
    `product_block_name` must be defined on the base class and need not to be defined on the others

    Create a new empty product block:

        >>> example1 = BlockInactive()  # doctest: +SKIP

    Create a new instance based on a dict in the state:

        >>> example2 = BlockInactive(**state)  # doctest:+SKIP

    To retrieve a ProductBlockModel from the database:

        >>> BlockInactive.from_db(subscription_instance_id)  # doctest:+SKIP
    """

    registry: ClassVar[dict[str, type["ProductBlockModel"]]] = {}  # pragma: no mutate
    __names__: ClassVar[set[str]] = set()
    product_block_id: ClassVar[UUID]
    description: ClassVar[str]
    tag: ClassVar[str]
    _db_model: SubscriptionInstanceTable = PrivateAttr()

    # Product block name. This needs to be an instance var because its part of the API (we expose it to the frontend)
    # Is actually optional since abstract classes don't have it.
    # TODO #427 name is used as both a ClassVar and a pydantic Field, for which Pydantic 2.x raises
    #  warnings (which may become errors)
    name: str | None
    subscription_instance_id: UUID
    owner_subscription_id: UUID
    label: str | None = None

    @classmethod
    def _fix_pb_data(cls) -> None:
        if not cls.name:
            raise ValueError(f"Cannot create instance of abstract class. Use one of {cls.__names__}")

        # Would have been nice to do this in __init_subclass__ but that runs outside the app context so we can't
        # access the db. So now we do it just before we instantiate the instance
        if not hasattr(cls, "product_block_id"):
            product_block = db.session.scalars(
                select(ProductBlockTable).filter(ProductBlockTable.name == cls.name)
            ).one()
            cls.product_block_id = product_block.product_block_id
            cls.description = product_block.description
            cls.tag = product_block.tag

    @classmethod
    def __pydantic_init_subclass__(  # type: ignore[override]
        cls,
        *,
        product_block_name: str | None = None,
        lifecycle: list[SubscriptionLifecycle] | None = None,
        **kwargs: Any,
    ) -> None:
        super().__pydantic_init_subclass__(lifecycle=lifecycle, **kwargs)

        if product_block_name is not None:
            # This is a concrete product block base class (so not a abstract super class or a specific lifecycle version)
            cls.name = product_block_name
            cls.__base_type__ = cls
            cls.__names__ = {cls.name}
            cls.registry[cls.name] = cls
        elif lifecycle is None:
            # Abstract class, no product block name
            cls.name = None
            cls.__names__ = set()

        # For everything except abstract classes
        if cls.name is not None:
            register_specialized_type(cls, lifecycle)

            # Add ourselves to any super class. That way we can match a superclass to an instance when loading
            for klass in cls.__mro__:
                if issubclass(klass, ProductBlockModel):
                    klass.__names__.add(cls.name)

        cls.__doc__ = make_product_block_docstring(cls, lifecycle)

    @classmethod
    def diff_product_block_in_database(cls) -> dict[str, set[str]]:
        """Return any differences between the attrs defined on the domain model and those on product blocks in the database.

        This is only needed to check if the domain model and database models match which would be done during testing...
        """
        if not cls.name:
            # This is a superclass we can't check that
            return {}

        product_block_db = db.session.scalars(
            select(ProductBlockTable).where(ProductBlockTable.name == cls.name)
        ).one_or_none()
        product_blocks_in_db = {pb.name for pb in product_block_db.depends_on} if product_block_db else set()

        product_blocks_in_model = cls._get_depends_on_product_block_types()
        product_blocks_types_in_model = get_depends_on_product_block_type_list(product_blocks_in_model)

        product_blocks_in_model = set(flatten(map(attrgetter("__names__"), product_blocks_types_in_model)))  # type: ignore

        missing_product_blocks_in_db = product_blocks_in_model - product_blocks_in_db  # type: ignore
        missing_product_blocks_in_model = product_blocks_in_db - product_blocks_in_model  # type: ignore

        resource_types_model = set(cls._non_product_block_fields_)
        resource_types_db = {rt.resource_type for rt in product_block_db.resource_types} if product_block_db else set()

        missing_resource_types_in_db = resource_types_model - resource_types_db
        missing_resource_types_in_model = resource_types_db - resource_types_model

        logger.debug(
            "ProductBlockTable blocks diff",
            product_block_db=product_block_db.name if product_block_db else None,
            product_blocks_in_db=product_blocks_in_db,
            product_blocks_in_model=product_blocks_in_model,
            resource_types_db=resource_types_db,
            resource_types_model=resource_types_model,
            missing_product_blocks_in_db=missing_product_blocks_in_db,
            missing_product_blocks_in_model=missing_product_blocks_in_model,
            missing_resource_types_in_db=missing_resource_types_in_db,
            missing_resource_types_in_model=missing_resource_types_in_model,
        )

        missing_data: dict[str, Any] = {}
        for product_block_model in product_blocks_types_in_model:
            if product_block_model.name == cls.name or product_block_model.name in missing_data:
                continue
            missing_data.update(product_block_model.diff_product_block_in_database())

        diff: dict[str, set[str]] = {
            k: v
            for k, v in {
                "missing_product_blocks_in_db": missing_product_blocks_in_db,
                "missing_product_blocks_in_model": missing_product_blocks_in_model,
                "missing_resource_types_in_db": missing_resource_types_in_db,
                "missing_resource_types_in_model": missing_resource_types_in_model,
            }.items()
            if v
        }

        if diff:
            missing_data[cls.name] = diff

        return missing_data

    @classmethod
    def new(cls: type[B], subscription_id: UUID, **kwargs: Any) -> B:
        """Create a new empty product block.

        We need to use this instead of the normal constructor because that assumes you pass in
        all required values. That is cumbersome since that means creating a tree of product blocks.

        This is similar to `from_product_id()`
        """
        sub_instances = cls._init_instances(subscription_id, set(kwargs.keys()))

        subscription_instance_id = uuid4()

        # Make sure product block stuff is already set if new is the first usage of this class
        cls._fix_pb_data()

        db_model = SubscriptionInstanceTable(
            product_block_id=cls.product_block_id,
            subscription_instance_id=subscription_instance_id,
            subscription_id=subscription_id,
        )
        db.session.enable_relationship_loading(db_model)

        if kwargs_name := kwargs.pop("name", None):
            # Not allowed to change the product block model name at runtime. This is only possible through
            # the `product_block_name=..` metaclass parameter
            logger.warning("Ignoring `name` keyword to ProductBlockModel.new()", rejected_name=kwargs_name)
        model = cls(
            name=cls.name,
            subscription_instance_id=subscription_instance_id,
            owner_subscription_id=subscription_id,
            label=db_model.label,
            **sub_instances,
            **kwargs,
        )
        model._db_model = db_model
        return model

    @classmethod
    def _load_instances_values(cls, instance_values: list[SubscriptionInstanceValueTable]) -> dict[str, str]:
        """Load non product block fields (instance values).

        Args:
            instance_values: List of instance values from database

        Returns:
            Dict of fields to use for constructor

        """
        instance_values_dict: State = {}
        list_field_names = set()

        # Set default values
        for field_name, field_type in cls._non_product_block_fields_.items():
            # Ensure that empty lists are handled OK
            if is_list_type(field_type):
                instance_values_dict[field_name] = []
                list_field_names.add(field_name)
            elif is_optional_type(field_type):
                # Initialize "optional required" fields
                instance_values_dict[field_name] = None

        for siv in instance_values:
            # check the type of the siv in the instance and act accordingly: only lists and scalar values supported
            resource_type_name = siv.resource_type.resource_type
            if resource_type_name in list_field_names:
                instance_values_dict[resource_type_name].append(siv.value)
            else:
                instance_values_dict[resource_type_name] = siv.value

        return instance_values_dict

    @classmethod
    def _from_other_lifecycle(
        cls: type[B],
        other: "ProductBlockModel",
        status: SubscriptionLifecycle,
        subscription_id: UUID,
    ) -> B:
        """Create new domain model from instance while changing the status.

        This makes sure we always have a specific instance.
        """
        if not cls.__base_type__:
            cls = ProductBlockModel.registry.get(other.name, cls)  # type:ignore
            cls = lookup_specialized_type(cls, status)

        data = cls._data_from_lifecycle(other, status, subscription_id)

        cls._fix_pb_data()
        model = cls(**data)
        model._db_model = other._db_model
        return model

    @classmethod
    def from_db(
        cls: type[B],
        subscription_instance_id: UUID | None = None,
        subscription_instance: SubscriptionInstanceTable | None = None,
        status: SubscriptionLifecycle | None = None,
    ) -> B:
        """Create a product block based on a subscription instance from the database.

        This function is similar to `from_subscription()`

            >>> subscription_instance_id = KNOWN_UUID_IN_DB  # doctest:+SKIP
            >>> si_from_db = db.SubscriptionInstanceTable.query.get(subscription_instance_id)  # doctest:+SKIP
            >>> example3 = ProductBlockModel.from_db(subscription_instance=si_from_db)  # doctest:+SKIP
            >>> example4 = ProductBlockModel.from_db(subscription_instance_id=subscription_instance_id)  # doctest:+SKIP
        """
        # Fill values from actual subscription
        if subscription_instance_id:
            subscription_instance = db.session.get(SubscriptionInstanceTable, subscription_instance_id)
        if subscription_instance:
            subscription_instance_id = subscription_instance.subscription_instance_id
        assert subscription_instance_id  # noqa: S101
        assert subscription_instance  # noqa: S101

        if not status:
            status = SubscriptionLifecycle(subscription_instance.subscription.status)

        if not cls.__base_type__:
            cls = ProductBlockModel.registry.get(subscription_instance.product_block.name, cls)  # type:ignore
            cls = lookup_specialized_type(cls, status)

        elif not issubclass(cls, lookup_specialized_type(cls, status)):
            raise ValueError(f"{cls} is not valid for lifecycle {status}")

        label = subscription_instance.label

        instance_values = cls._load_instances_values(subscription_instance.values)
        sub_instances = cls._load_instances(
            subscription_instance.depends_on,
            status,
            match_domain_attr=True,
            in_use_by_id_boundary=subscription_instance_id,
        )

        cls._fix_pb_data()
        try:
            model = cls(
                name=cls.name,
                subscription_instance_id=subscription_instance_id,
                owner_subscription_id=subscription_instance.subscription_id,
                label=label,
                subscription=subscription_instance.subscription,
                **instance_values,  # type: ignore
                **sub_instances,
            )
            model._db_model = subscription_instance

            return model
        except ValidationError:
            logger.exception(
                "Subscription is not correct in database",
                loaded_instance_values=instance_values,
                loaded_sub_instances=sub_instances,
            )
            raise

    def _save_instance_values(
        self, product_block: ProductBlockTable, current_values: list[SubscriptionInstanceValueTable]
    ) -> list[SubscriptionInstanceValueTable]:
        """Save non product block fields (instance values).

        Returns:
            List of database instances values to save

        """
        resource_types = {rt.resource_type: rt for rt in product_block.resource_types}
        current_values_dict: dict[str, list[SubscriptionInstanceValueTable]] = defaultdict(list)
        for siv in current_values:
            current_values_dict[siv.resource_type.resource_type].append(siv)

        subscription_instance_values = []
        for field_name, field_type in self._non_product_block_fields_.items():
            assert (  # noqa: S101
                field_name in resource_types
            ), f"Domain model {self.__class__} does not match the ProductBlockTable {product_block.name}, missing: {field_name} {resource_types}"

            resource_type = resource_types[field_name]
            value = getattr(self, field_name)
            if value is None:
                continue
            if is_list_type(field_type):
                for val, siv in zip_longest(value, current_values_dict[field_name]):
                    if val is not None:
                        if siv:
                            siv.value = str(val)
                            subscription_instance_values.append(siv)
                        else:
                            subscription_instance_values.append(
                                SubscriptionInstanceValueTable(resource_type=resource_type, value=str(val))
                            )
            else:
                if field_name in current_values_dict:
                    current_value = current_values_dict[field_name][0]
                    current_value.value = str(value)
                    subscription_instance_values.append(current_value)
                else:
                    subscription_instance_values.append(
                        SubscriptionInstanceValueTable(resource_type=resource_type, value=str(value))
                    )
        return subscription_instance_values

    def _set_instance_domain_model_attrs(
        self,
        subscription_instance: SubscriptionInstanceTable,
        subscription_instance_mapping: dict[str, list[SubscriptionInstanceTable]],
    ) -> None:
        """Save the domain model attribute to the database.

        This function iterates through the subscription instances and stores the domain model attribute in the
        hierarchy relationship.

        Args:
            subscription_instance: The subscription instance object.
            subscription_instance_mapping: a mapping of the domain model attribute a underlying instances

        Returns:
            None

        """
        depends_on_block_relations = []
        # Set the domain_model_attrs in the database
        for domain_model_attr, instances in subscription_instance_mapping.items():
            instance: SubscriptionInstanceTable
            for index, instance in enumerate(instances):
                relation = SubscriptionInstanceRelationTable(
                    in_use_by_id=subscription_instance.subscription_instance_id,
                    depends_on_id=instance.subscription_instance_id,
                    order_id=index,
                    domain_model_attr=domain_model_attr,
                )
                depends_on_block_relations.append(relation)
        subscription_instance.depends_on_block_relations = depends_on_block_relations

    def save(
        self, *, subscription_id: UUID, status: SubscriptionLifecycle
    ) -> tuple[list[SubscriptionInstanceTable], SubscriptionInstanceTable]:
        """Save the current model instance to the database.

        This means saving the whole tree of subscription instances and separately saving all instance
        values for this instance. This is called automatically when you return a subscription to the state
        in a workflow step.

        Args:
            status: current SubscriptionLifecycle to check if all constraints match
            subscription_id: Optional subscription id needed if this is a new model

        Returns:
            List of saved instances

        """
        if not self.name:
            raise ValueError(f"Cannot create instance of abstract class. Use one of {self.__names__}")

        # Make sure we have a valid subscription instance database model
        subscription_instance: SubscriptionInstanceTable | None = db.session.get(
            SubscriptionInstanceTable, self.subscription_instance_id
        )
        if subscription_instance:
            # Make sure we do not use a mapped session.
            db.session.refresh(subscription_instance)

            # If this is a "foreign" instance we just stop saving and return it so only its relation is saved
            # We should not touch these themselves
            if self.subscription and subscription_instance.subscription_id != subscription_id:
                return [], subscription_instance

            self._db_model = subscription_instance
        else:
            subscription_instance = self._db_model
            # We only need to add to the session if the subscription_instance does not exist.
            db.session.add(subscription_instance)

        subscription_instance.subscription_id = subscription_id

        db.session.flush()

        # Everything is ok, make sure we are of the right class
        specialized_type = lookup_specialized_type(self.__class__, status)
        if specialized_type and not isinstance(self, specialized_type):
            raise ValueError(
                f"Lifecycle status {status} requires specialized type {specialized_type!r}, was: {type(self)!r}"
            )

        # Actually save stuff
        subscription_instance.label = self.label
        subscription_instance.values = self._save_instance_values(
            subscription_instance.product_block, subscription_instance.values
        )

        sub_instances, depends_on_instances = self._save_instances(subscription_id, status)

        # Save the subscription instances relations.
        self._set_instance_domain_model_attrs(subscription_instance, depends_on_instances)

        return sub_instances + [subscription_instance], subscription_instance

    @property
    def subscription(self) -> SubscriptionTable:
        return self.db_model.subscription

    @property
    def db_model(self) -> SubscriptionInstanceTable:
        return self._db_model

    @property
    def in_use_by(self) -> list[SubscriptionInstanceTable]:
        """This provides a list of product blocks that depend on this product block."""
        return self._db_model.in_use_by

    @property
    def depends_on(self) -> list[SubscriptionInstanceTable]:
        """This provides a list of product blocks that this product block depends on."""
        return self._db_model.depends_on

depends_on property

depends_on: list[orchestrator.db.SubscriptionInstanceTable]

This provides a list of product blocks that this product block depends on.

in_use_by property

in_use_by: list[orchestrator.db.SubscriptionInstanceTable]

This provides a list of product blocks that depend on this product block.

diff_product_block_in_database classmethod

diff_product_block_in_database() -> dict[str, set[str]]

Return any differences between the attrs defined on the domain model and those on product blocks in the database.

This is only needed to check if the domain model and database models match which would be done during testing...

Source code in orchestrator/domain/base.py

@classmethod
def diff_product_block_in_database(cls) -> dict[str, set[str]]:
    """Return any differences between the attrs defined on the domain model and those on product blocks in the database.

    This is only needed to check if the domain model and database models match which would be done during testing...
    """
    if not cls.name:
        # This is a superclass we can't check that
        return {}

    product_block_db = db.session.scalars(
        select(ProductBlockTable).where(ProductBlockTable.name == cls.name)
    ).one_or_none()
    product_blocks_in_db = {pb.name for pb in product_block_db.depends_on} if product_block_db else set()

    product_blocks_in_model = cls._get_depends_on_product_block_types()
    product_blocks_types_in_model = get_depends_on_product_block_type_list(product_blocks_in_model)

    product_blocks_in_model = set(flatten(map(attrgetter("__names__"), product_blocks_types_in_model)))  # type: ignore

    missing_product_blocks_in_db = product_blocks_in_model - product_blocks_in_db  # type: ignore
    missing_product_blocks_in_model = product_blocks_in_db - product_blocks_in_model  # type: ignore

    resource_types_model = set(cls._non_product_block_fields_)
    resource_types_db = {rt.resource_type for rt in product_block_db.resource_types} if product_block_db else set()

    missing_resource_types_in_db = resource_types_model - resource_types_db
    missing_resource_types_in_model = resource_types_db - resource_types_model

    logger.debug(
        "ProductBlockTable blocks diff",
        product_block_db=product_block_db.name if product_block_db else None,
        product_blocks_in_db=product_blocks_in_db,
        product_blocks_in_model=product_blocks_in_model,
        resource_types_db=resource_types_db,
        resource_types_model=resource_types_model,
        missing_product_blocks_in_db=missing_product_blocks_in_db,
        missing_product_blocks_in_model=missing_product_blocks_in_model,
        missing_resource_types_in_db=missing_resource_types_in_db,
        missing_resource_types_in_model=missing_resource_types_in_model,
    )

    missing_data: dict[str, Any] = {}
    for product_block_model in product_blocks_types_in_model:
        if product_block_model.name == cls.name or product_block_model.name in missing_data:
            continue
        missing_data.update(product_block_model.diff_product_block_in_database())

    diff: dict[str, set[str]] = {
        k: v
        for k, v in {
            "missing_product_blocks_in_db": missing_product_blocks_in_db,
            "missing_product_blocks_in_model": missing_product_blocks_in_model,
            "missing_resource_types_in_db": missing_resource_types_in_db,
            "missing_resource_types_in_model": missing_resource_types_in_model,
        }.items()
        if v
    }

    if diff:
        missing_data[cls.name] = diff

    return missing_data

from_db classmethod

from_db(
    subscription_instance_id: UUID | None = None,
    subscription_instance: SubscriptionInstanceTable | None = None,
    status: SubscriptionLifecycle | None = None,
) -> B

Create a product block based on a subscription instance from the database.

This function is similar to from_subscription()

>>> subscription_instance_id = KNOWN_UUID_IN_DB  # doctest:+SKIP
>>> si_from_db = db.SubscriptionInstanceTable.query.get(subscription_instance_id)  # doctest:+SKIP
>>> example3 = ProductBlockModel.from_db(subscription_instance=si_from_db)  # doctest:+SKIP
>>> example4 = ProductBlockModel.from_db(subscription_instance_id=subscription_instance_id)  # doctest:+SKIP

Source code in orchestrator/domain/base.py

@classmethod
def from_db(
    cls: type[B],
    subscription_instance_id: UUID | None = None,
    subscription_instance: SubscriptionInstanceTable | None = None,
    status: SubscriptionLifecycle | None = None,
) -> B:
    """Create a product block based on a subscription instance from the database.

    This function is similar to `from_subscription()`

        >>> subscription_instance_id = KNOWN_UUID_IN_DB  # doctest:+SKIP
        >>> si_from_db = db.SubscriptionInstanceTable.query.get(subscription_instance_id)  # doctest:+SKIP
        >>> example3 = ProductBlockModel.from_db(subscription_instance=si_from_db)  # doctest:+SKIP
        >>> example4 = ProductBlockModel.from_db(subscription_instance_id=subscription_instance_id)  # doctest:+SKIP
    """
    # Fill values from actual subscription
    if subscription_instance_id:
        subscription_instance = db.session.get(SubscriptionInstanceTable, subscription_instance_id)
    if subscription_instance:
        subscription_instance_id = subscription_instance.subscription_instance_id
    assert subscription_instance_id  # noqa: S101
    assert subscription_instance  # noqa: S101

    if not status:
        status = SubscriptionLifecycle(subscription_instance.subscription.status)

    if not cls.__base_type__:
        cls = ProductBlockModel.registry.get(subscription_instance.product_block.name, cls)  # type:ignore
        cls = lookup_specialized_type(cls, status)

    elif not issubclass(cls, lookup_specialized_type(cls, status)):
        raise ValueError(f"{cls} is not valid for lifecycle {status}")

    label = subscription_instance.label

    instance_values = cls._load_instances_values(subscription_instance.values)
    sub_instances = cls._load_instances(
        subscription_instance.depends_on,
        status,
        match_domain_attr=True,
        in_use_by_id_boundary=subscription_instance_id,
    )

    cls._fix_pb_data()
    try:
        model = cls(
            name=cls.name,
            subscription_instance_id=subscription_instance_id,
            owner_subscription_id=subscription_instance.subscription_id,
            label=label,
            subscription=subscription_instance.subscription,
            **instance_values,  # type: ignore
            **sub_instances,
        )
        model._db_model = subscription_instance

        return model
    except ValidationError:
        logger.exception(
            "Subscription is not correct in database",
            loaded_instance_values=instance_values,
            loaded_sub_instances=sub_instances,
        )
        raise

new classmethod

new(subscription_id: UUID, **kwargs: Any) -> B

Create a new empty product block.

We need to use this instead of the normal constructor because that assumes you pass in all required values. That is cumbersome since that means creating a tree of product blocks.

This is similar to from_product_id()

Source code in orchestrator/domain/base.py

@classmethod
def new(cls: type[B], subscription_id: UUID, **kwargs: Any) -> B:
    """Create a new empty product block.

    We need to use this instead of the normal constructor because that assumes you pass in
    all required values. That is cumbersome since that means creating a tree of product blocks.

    This is similar to `from_product_id()`
    """
    sub_instances = cls._init_instances(subscription_id, set(kwargs.keys()))

    subscription_instance_id = uuid4()

    # Make sure product block stuff is already set if new is the first usage of this class
    cls._fix_pb_data()

    db_model = SubscriptionInstanceTable(
        product_block_id=cls.product_block_id,
        subscription_instance_id=subscription_instance_id,
        subscription_id=subscription_id,
    )
    db.session.enable_relationship_loading(db_model)

    if kwargs_name := kwargs.pop("name", None):
        # Not allowed to change the product block model name at runtime. This is only possible through
        # the `product_block_name=..` metaclass parameter
        logger.warning("Ignoring `name` keyword to ProductBlockModel.new()", rejected_name=kwargs_name)
    model = cls(
        name=cls.name,
        subscription_instance_id=subscription_instance_id,
        owner_subscription_id=subscription_id,
        label=db_model.label,
        **sub_instances,
        **kwargs,
    )
    model._db_model = db_model
    return model

save

save(
    *, subscription_id: UUID, status: SubscriptionLifecycle
) -> tuple[list[SubscriptionInstanceTable], SubscriptionInstanceTable]

Save the current model instance to the database.

This means saving the whole tree of subscription instances and separately saving all instance values for this instance. This is called automatically when you return a subscription to the state in a workflow step.

Parameters:

• status (orchestrator.types.SubscriptionLifecycle) –

  current SubscriptionLifecycle to check if all constraints match

• subscription_id (uuid.UUID) –

  Optional subscription id needed if this is a new model

Returns:

• tuple[list[orchestrator.db.SubscriptionInstanceTable], orchestrator.db.SubscriptionInstanceTable] –

  List of saved instances

Source code in orchestrator/domain/base.py

def save(
    self, *, subscription_id: UUID, status: SubscriptionLifecycle
) -> tuple[list[SubscriptionInstanceTable], SubscriptionInstanceTable]:
    """Save the current model instance to the database.

    This means saving the whole tree of subscription instances and separately saving all instance
    values for this instance. This is called automatically when you return a subscription to the state
    in a workflow step.

    Args:
        status: current SubscriptionLifecycle to check if all constraints match
        subscription_id: Optional subscription id needed if this is a new model

    Returns:
        List of saved instances

    """
    if not self.name:
        raise ValueError(f"Cannot create instance of abstract class. Use one of {self.__names__}")

    # Make sure we have a valid subscription instance database model
    subscription_instance: SubscriptionInstanceTable | None = db.session.get(
        SubscriptionInstanceTable, self.subscription_instance_id
    )
    if subscription_instance:
        # Make sure we do not use a mapped session.
        db.session.refresh(subscription_instance)

        # If this is a "foreign" instance we just stop saving and return it so only its relation is saved
        # We should not touch these themselves
        if self.subscription and subscription_instance.subscription_id != subscription_id:
            return [], subscription_instance

        self._db_model = subscription_instance
    else:
        subscription_instance = self._db_model
        # We only need to add to the session if the subscription_instance does not exist.
        db.session.add(subscription_instance)

    subscription_instance.subscription_id = subscription_id

    db.session.flush()

    # Everything is ok, make sure we are of the right class
    specialized_type = lookup_specialized_type(self.__class__, status)
    if specialized_type and not isinstance(self, specialized_type):
        raise ValueError(
            f"Lifecycle status {status} requires specialized type {specialized_type!r}, was: {type(self)!r}"
        )

    # Actually save stuff
    subscription_instance.label = self.label
    subscription_instance.values = self._save_instance_values(
        subscription_instance.product_block, subscription_instance.values
    )

    sub_instances, depends_on_instances = self._save_instances(subscription_id, status)

    # Save the subscription instances relations.
    self._set_instance_domain_model_attrs(subscription_instance, depends_on_instances)

    return sub_instances + [subscription_instance], subscription_instance

Nesting Product Blocks

When building a more complicated product, you will likely want to start nesting layers of product blocks. Some of these might just be used by one product, but some of these will be a reference to a product block in use by multiple product types. For an example of this, let's look at the CorePortBlock product block from the example workflow orchestrator:

Example: example-orchestrator/products/product_blocks/core_port.py

# Copyright 2019-2023 SURF.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from orchestrator.domain.base import ProductBlockModel
from orchestrator.types import SubscriptionLifecycle
from pydantic import computed_field

from products.product_blocks.node import NodeBlock, NodeBlockInactive, NodeBlockProvisioning


class CorePortBlockInactive(ProductBlockModel, product_block_name="CorePort"):
    port_name: str | None = None
    enabled: bool | None = True
    ims_id: int | None = None
    nrm_id: int | None = None
    node: NodeBlockInactive | None = None
    ipv6_ipam_id: int | None = None


class CorePortBlockProvisioning(CorePortBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
    port_name: str | None = None
    enabled: bool
    ims_id: int | None = None
    nrm_id: int | None = None
    node: NodeBlockProvisioning
    ipv6_ipam_id: int | None = None

    @computed_field  # type: ignore[misc]
    @property
    def title(self) -> str:
        return f"core port {self.port_name} on {self.node.node_name}"


class CorePortBlock(CorePortBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
    port_name: str | None = None
    enabled: bool
    ims_id: int
    nrm_id: int
    node: NodeBlock
    ipv6_ipam_id: int

Note in this example how the attribute node is type-hinted with NodeBlockInactive. This tells the WFO to bring in the entire tree of product blocks that are referenced, in this case, NodeBlockInactive when you load the CorePortBlockInactive into your workflow step. To read more on this concept, read through the Product Modelling page in the architecture section of the docs.

Creating Database Migrations

Just like Product types, you'll need to create a database migration to properly wire-up the product block in the orchestrator's database. A migration file for this example NodeBlock model looks like this:

Example: example-orchestrator/migrations/versions/schema/2023-10-27_a84ca2e5e4db_add_node.py

"""Add node product.

Revision ID: a84ca2e5e4db
Revises: a77227fe5455
Create Date: 2023-10-27 11:25:40.994878

"""
from uuid import uuid4

from alembic import op
from orchestrator.migrations.helpers import (
    create,
    create_workflow,
    delete,
    delete_workflow,
    ensure_default_workflows,
)
from orchestrator.targets import Target

# revision identifiers, used by Alembic.
revision = "a84ca2e5e4db"
down_revision = "a77227fe5455"
branch_labels = None
depends_on = None

new_products = {
    "products": {
        "node Cisco": {
            "product_id": uuid4(),
            "product_type": "Node",
            "description": "Network node",
            "tag": "NODE",
            "status": "active",
            "product_blocks": [
                "Node",
            ],
            "fixed_inputs": {
                "node_type": "Cisco",
            },
        },
        "node Nokia": {
            "product_id": uuid4(),
            "product_type": "Node",
            "description": "Network node",
            "tag": "NODE",
            "status": "active",
            "product_blocks": [
                "Node",
            ],
            "fixed_inputs": {
                "node_type": "Nokia",
            },
        },
        "node Cumulus": {
            "product_id": uuid4(),
            "product_type": "Node",
            "description": "Network node",
            "tag": "NODE",
            "status": "active",
            "product_blocks": [
                "Node",
            ],
            "fixed_inputs": {
                "node_type": "Cumulus",
            },
        },
        "node FRR": {
            "product_id": uuid4(),
            "product_type": "Node",
            "description": "Network node",
            "tag": "NODE",
            "status": "active",
            "product_blocks": [
                "Node",
            ],
            "fixed_inputs": {
                "node_type": "FRR",
            },
        },
    },
    "product_blocks": {
        "Node": {
            "product_block_id": uuid4(),
            "description": "node product block",
            "tag": "NODE",
            "status": "active",
            "resources": {
                "role_id": "ID in CMDB of role of the node in the network",
                "type_id": "ID in CMDB of type of the node",
                "site_id": "ID in CMDB of site where the node is located",
                "node_status": "Operational status of the node",
                "node_name": "Unique name of the node",
                "node_description": "Description of the node",
                "ims_id": "ID of the node in the inventory management system",
                "nrm_id": "ID of the node in the network resource manager",
                "ipv4_ipam_id": "ID of the node’s iPv4 loopback address in IPAM",
                "ipv6_ipam_id": "ID of the node’s iPv6 loopback address in IPAM",
            },
            "depends_on_block_relations": [],
        },
    },
    "workflows": {},
}

new_workflows = [
    {
        "name": "create_node",
        "target": Target.CREATE,
        "description": "Create node",
        "product_type": "Node",
    },
    {
        "name": "modify_node",
        "target": Target.MODIFY,
        "description": "Modify node",
        "product_type": "Node",
    },
    {
        "name": "modify_sync_ports",
        "target": Target.MODIFY,
        "description": "Update node interfaces",
        "product_type": "Node",
    },
    {
        "name": "terminate_node",
        "target": Target.TERMINATE,
        "description": "Terminate node",
        "product_type": "Node",
    },
    {
        "name": "validate_node",
        "target": Target.SYSTEM,
        "description": "Validate node",
        "product_type": "Node",
    },
]


def upgrade() -> None:
    conn = op.get_bind()
    create(conn, new_products)
    for workflow in new_workflows:
        create_workflow(conn, workflow)
    ensure_default_workflows(conn)


def downgrade() -> None:
    conn = op.get_bind()
    for workflow in new_workflows:
        delete_workflow(conn, workflow["name"])

    delete(conn, new_products)

Thankfully, you don't have to write these database migrations by hand, you can simply use the main.py db migrate-domain-models command that is part of the orchestrator CLI, documented here.

Automatically Generating Product Types

If all of this seems like too much work, then good news, as all clever engineers before us have done, we've fixed that with YAML! Using the WFO CLI, you can generate your product types directly from a YAML. For more information on how to do that, check out the CLI generate command documentation.