Skip to content

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
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
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)

        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
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
@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
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
@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)

    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
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
@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
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
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
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
"""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.