Add Number attribute min/max/excluded values (#6498)

Author: LucasG0

backend/infrahub/core/node/__init__.py

@@ -287,7 +287,7 @@ async def handle_pool(self, db: InfrahubDatabase, attribute: BaseAttribute, erro
             and number_pool.node_attribute.value == attribute.name
         ):
             try:
-                next_free = await number_pool.get_resource(db=db, branch=self._branch, node=self)
+                next_free = await number_pool.get_resource(db=db, branch=self._branch, node=self, attribute=attribute)
             except PoolExhaustedError:
                 errors.append(
                     ValidationError({f"{attribute.name}.from_pool": f"The pool {number_pool.node.value} is exhausted."})

backend/infrahub/core/node/resource_manager/number_pool.py

@@ -2,22 +2,44 @@
 
 from typing import TYPE_CHECKING
 
+from infrahub.core import registry
 from infrahub.core.query.resource_manager import NumberPoolGetReserved, NumberPoolGetUsed, NumberPoolSetReserved
+from infrahub.core.schema.attribute_parameters import NumberAttributeParameters
 from infrahub.exceptions import PoolExhaustedError
 
 from .. import Node
 
 if TYPE_CHECKING:
+    from infrahub.core.attribute import BaseAttribute
     from infrahub.core.branch import Branch
     from infrahub.database import InfrahubDatabase
 
 
 class CoreNumberPool(Node):
+    def get_attribute_nb_excluded_values(self) -> int:
+        """
+        Returns the number of excluded values for the attribute of the number pool.
+        """
+
+        pool_node = registry.schema.get(name=self.node.value)  # type: ignore [attr-defined]
+        attribute = [attribute for attribute in pool_node.attributes if attribute.name == self.node_attribute.value][0]  # type: ignore [attr-defined]
+        if not isinstance(attribute.parameters, NumberAttributeParameters):
+            return 0
+
+        sum_excluded_values = 0
+        excluded_ranges = attribute.parameters.get_excluded_ranges()
+        for start_range, end_range in excluded_ranges:
+            sum_excluded_values += end_range - start_range + 1
+
+        res = len(attribute.parameters.get_excluded_single_values()) + sum_excluded_values
+        return res
+
     async def get_resource(
         self,
         db: InfrahubDatabase,
         branch: Branch,
         node: Node,
+        attribute: BaseAttribute,
         identifier: str | None = None,
     ) -> int:
         identifier = identifier or node.get_id()
@@ -31,35 +53,40 @@ async def get_resource(
             return reservation
 
         # If we have not returned a value we need to find one if avaiable
-        number = await self.get_next(db=db, branch=branch)
+        number = await self.get_next(db=db, branch=branch, attribute=attribute)
 
         query_set = await NumberPoolSetReserved.init(
             db=db, pool_id=self.get_id(), identifier=identifier, reserved=number
         )
         await query_set.execute(db=db)
         return number
 
-    async def get_next(self, db: InfrahubDatabase, branch: Branch) -> int:
+    async def get_next(self, db: InfrahubDatabase, branch: Branch, attribute: BaseAttribute) -> int:
         query = await NumberPoolGetUsed.init(db=db, branch=branch, pool=self, branch_agnostic=True)
         await query.execute(db=db)
         taken = [result.get_as_optional_type("av.value", return_type=int) for result in query.results]
+        parameters = attribute.schema.parameters
         next_number = find_next_free(
             start=self.start_range.value,  # type: ignore[attr-defined]
             end=self.end_range.value,  # type: ignore[attr-defined]
             taken=taken,
+            parameters=parameters if isinstance(parameters, NumberAttributeParameters) else None,
         )
         if next_number is None:
             raise PoolExhaustedError("There are no more values available in this pool.")
 
         return next_number
 
 
-def find_next_free(start: int, end: int, taken: list[int | None]) -> int | None:
+def find_next_free(
+    start: int, end: int, taken: list[int | None], parameters: NumberAttributeParameters | None
+) -> int | None:
     used_numbers = [number for number in taken if number is not None]
     used_set = set(used_numbers)
 
     for num in range(start, end + 1):
         if num not in used_set:
-            return num
+            if parameters is None or parameters.is_valid_value(num):
+                return num
 
     return None

backend/infrahub/core/schema/attribute_parameters.py

@@ -14,6 +14,7 @@ def get_attribute_parameters_class_for_kind(kind: str) -> type[AttributeParamete
         "NumberPool": NumberPoolParameters,
         "Text": TextAttributeParameters,
         "TextArea": TextAttributeParameters,
+        "Number": NumberAttributeParameters,
     }
     return param_classes.get(kind, AttributeParameters)
 
@@ -23,6 +24,88 @@ class Config:
         extra = "forbid"
 
 
+class TextAttributeParameters(AttributeParameters):
+    regex: str | None = Field(
+        default=None,
+        description="Regular expression that attribute value must match if defined",
+        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
+    )
+    min_length: int | None = Field(
+        default=None,
+        description="Set a minimum number of characters allowed.",
+        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
+    )
+    max_length: int | None = Field(
+        default=None,
+        description="Set a maximum number of characters allowed.",
+        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
+    )
+
+
+class NumberAttributeParameters(AttributeParameters):
+    min_value: int | None = Field(
+        default=None,
+        description="Set a minimum value allowed.",
+        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
+    )
+    max_value: int | None = Field(
+        default=None,
+        description="Set a maximum value allowed.",
+        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
+    )
+    excluded_values: str | None = Field(
+        default=None,
+        description="List of values or range of values not allowed for the attribute, format is: '100,150-200,280,300-400'",
+        pattern=r"^(\d+(?:-\d+)?)(?:,\d+(?:-\d+)?)*$",
+        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
+    )
+
+    @model_validator(mode="after")
+    def validate_ranges(self) -> Self:
+        ranges = self.get_excluded_ranges()
+        for i, (start_range_1, end_range_1) in enumerate(ranges):
+            if start_range_1 > end_range_1:
+                raise ValueError("`start_range` can't be less than `end_range`")
+
+            # Check for overlapping ranges
+            for start_range_2, end_range_2 in ranges[i + 1 :]:
+                if not (end_range_1 < start_range_2 or start_range_1 > end_range_2):
+                    raise ValueError("Excluded ranges cannot overlap")
+
+        return self
+
+    def get_excluded_single_values(self) -> list[int]:
+        if not self.excluded_values:
+            return []
+
+        results = [int(value) for value in self.excluded_values.split(",") if "-" not in value]
+        return results
+
+    def get_excluded_ranges(self) -> list[tuple[int, int]]:
+        if not self.excluded_values:
+            return []
+
+        ranges = []
+        for value in self.excluded_values.split(","):
+            if "-" in value:
+                start, end = map(int, value.split("-"))
+                ranges.append((start, end))
+
+        return ranges
+
+    def is_valid_value(self, value: int) -> bool:
+        if self.min_value is not None and value < self.min_value:
+            return False
+        if self.max_value is not None and value > self.max_value:
+            return False
+        if value in self.get_excluded_single_values():
+            return False
+        for start, end in self.get_excluded_ranges():
+            if start <= value <= end:
+                return False
+        return True
+
+
 class NumberPoolParameters(AttributeParameters):
     end_range: int = Field(
         default=sys.maxsize,
@@ -43,23 +126,5 @@ class NumberPoolParameters(AttributeParameters):
     @model_validator(mode="after")
     def validate_ranges(self) -> Self:
         if self.start_range > self.end_range:
-            raise ValueError("start_range can't be less than end_range")
+            raise ValueError("`start_range` can't be less than `end_range`")
         return self
-
-
-class TextAttributeParameters(AttributeParameters):
-    regex: str | None = Field(
-        default=None,
-        description="Regular expression that attribute value must match if defined",
-        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
-    )
-    min_length: int | None = Field(
-        default=None,
-        description="Set a minimum number of characters allowed.",
-        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
-    )
-    max_length: int | None = Field(
-        default=None,
-        description="Set a maximum number of characters allowed.",
-        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
-    )

backend/infrahub/core/schema/attribute_schema.py

@@ -14,6 +14,7 @@
 
 from .attribute_parameters import (
     AttributeParameters,
+    NumberAttributeParameters,
     NumberPoolParameters,
     TextAttributeParameters,
     get_attribute_parameters_class_for_kind,
@@ -33,6 +34,7 @@ def get_attribute_schema_class_for_kind(kind: str) -> type[AttributeSchema]:
         "NumberPool": NumberPoolSchema,
         "Text": TextAttributeSchema,
         "TextArea": TextAttributeSchema,
+        "Number": NumberAttributeSchema,
     }
     return attribute_schema_class_by_kind.get(kind, AttributeSchema)
 
@@ -219,19 +221,13 @@ class TextAttributeSchema(AttributeSchema):
     @model_validator(mode="after")
     def reconcile_parameters(self) -> Self:
         if self.regex != self.parameters.regex:
-            final_regex = self.parameters.regex or self.regex
-            if not final_regex:  # falsy parameters.regex override falsy regex
-                final_regex = self.parameters.regex
+            final_regex = self.parameters.regex if self.parameters.regex is not None else self.regex
             self.regex = self.parameters.regex = final_regex
         if self.min_length != self.parameters.min_length:
-            final_min_length = self.parameters.min_length or self.min_length
-            if not final_min_length:  # falsy parameters.min_length override falsy min_length
-                final_min_length = self.parameters.min_length
+            final_min_length = self.parameters.min_length if self.parameters.min_length is not None else self.min_length
             self.min_length = self.parameters.min_length = final_min_length
         if self.max_length != self.parameters.max_length:
-            final_max_length = self.parameters.max_length or self.max_length
-            if not final_max_length:  # falsy parameters.max_length override falsy max_length
-                final_max_length = self.parameters.max_length
+            final_max_length = self.parameters.max_length if self.parameters.max_length is not None else self.max_length
             self.max_length = self.parameters.max_length = final_max_length
         return self
 
@@ -243,3 +239,11 @@ def get_min_length(self) -> int | None:
 
     def get_max_length(self) -> int | None:
         return self.parameters.max_length
+
+
+class NumberAttributeSchema(AttributeSchema):
+    parameters: NumberAttributeParameters = Field(
+        default_factory=NumberAttributeParameters,
+        description="Extra parameters specific to number attributes",
+        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
+    )

backend/infrahub/core/schema/definitions/internal.py

@@ -31,7 +31,12 @@
     RelationshipKind,
     UpdateSupport,
 )
-from infrahub.core.schema.attribute_parameters import AttributeParameters, NumberPoolParameters, TextAttributeParameters
+from infrahub.core.schema.attribute_parameters import (
+    AttributeParameters,
+    NumberAttributeParameters,
+    NumberPoolParameters,
+    TextAttributeParameters,
+)
 from infrahub.core.schema.attribute_schema import AttributeSchema
 from infrahub.core.schema.computed_attribute import ComputedAttribute
 from infrahub.core.schema.dropdown import DropdownChoice
@@ -624,7 +629,12 @@ def to_dict(self) -> dict[str, Any]:
         SchemaAttribute(
             name="parameters",
             kind="JSON",
-            internal_kind=[AttributeParameters, TextAttributeParameters, NumberPoolParameters],
+            internal_kind=[
+                AttributeParameters,
+                TextAttributeParameters,
+                NumberAttributeParameters,
+                NumberPoolParameters,
+            ],
             optional=True,
             description="Extra parameters specific to this kind of attribute",
             extra={"update": UpdateSupport.VALIDATE_CONSTRAINT},

backend/infrahub/core/schema/generated/attribute_schema.py

@@ -10,6 +10,7 @@
 from infrahub.core.models import HashableModel
 from infrahub.core.schema.attribute_parameters import (
     AttributeParameters,  # noqa: TC001
+    NumberAttributeParameters,  # noqa: TC001
     NumberPoolParameters,  # noqa: TC001
     TextAttributeParameters,  # noqa: TC001
 )
@@ -117,10 +118,12 @@ class GeneratedAttributeSchema(HashableModel):
         description="Type of allowed override for the attribute.",
         json_schema_extra={"update": "allowed"},
     )
-    parameters: AttributeParameters | TextAttributeParameters | NumberPoolParameters = Field(
-        default_factory=AttributeParameters,
-        description="Extra parameters specific to this kind of attribute",
-        json_schema_extra={"update": "validate_constraint"},
+    parameters: AttributeParameters | TextAttributeParameters | NumberAttributeParameters | NumberPoolParameters = (
+        Field(
+            default_factory=AttributeParameters,
+            description="Extra parameters specific to this kind of attribute",
+            json_schema_extra={"update": "validate_constraint"},
+        )
     )
     deprecation: str | None = Field(
         default=None,

backend/infrahub/core/validators/__init__.py

@@ -1,3 +1,5 @@
+from infrahub.core.validators.attribute.min_max import AttributeNumberChecker
+
 from .attribute.choices import AttributeChoicesChecker
 from .attribute.enum import AttributeEnumChecker
 from .attribute.kind import AttributeKindChecker
@@ -26,6 +28,9 @@
     "attribute.max_length.update": AttributeLengthChecker,
     ConstraintIdentifier.ATTRIBUTE_PARAMETERS_MIN_LENGTH_UPDATE.value: AttributeLengthChecker,
     ConstraintIdentifier.ATTRIBUTE_PARAMETERS_MAX_LENGTH_UPDATE.value: AttributeLengthChecker,
+    ConstraintIdentifier.ATTRIBUTE_PARAMETERS_MIN_VALUE_UPDATE.value: AttributeNumberChecker,
+    ConstraintIdentifier.ATTRIBUTE_PARAMETERS_MAX_VALUE_UPDATE.value: AttributeNumberChecker,
+    ConstraintIdentifier.ATTRIBUTE_PARAMETERS_EXCLUDED_VALUES_UPDATE.value: AttributeNumberChecker,
     "attribute.unique.update": AttributeUniquenessChecker,
     "attribute.optional.update": AttributeOptionalChecker,
     "attribute.choices.update": AttributeChoicesChecker,

backend/infrahub/core/validators/attribute/choices.py

@@ -42,7 +42,6 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: dict[str, Any]) -> No
             LIMIT 1
         }
         WITH full_path, node, attribute_value, value_relationship
-        WITH full_path, node, attribute_value, value_relationship
         WHERE all(r in relationships(full_path) WHERE r.status = "active")
         AND attribute_value IS NOT NULL
         AND attribute_value <> $null_value

backend/infrahub/core/validators/attribute/enum.py

@@ -41,7 +41,6 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: dict[str, Any]) -> No
             LIMIT 1
         }
         WITH full_path, node, attribute_value, value_relationship
-        WITH full_path, node, attribute_value, value_relationship
         WHERE all(r in relationships(full_path) WHERE r.status = "active")
         AND attribute_value IS NOT NULL
         AND attribute_value <> $null_value

backend/infrahub/core/validators/attribute/kind.py

@@ -48,7 +48,6 @@ async def query_init(self, db: InfrahubDatabase, **kwargs: dict[str, Any]) -> No
             LIMIT 1
         }
         WITH full_path, node, attribute_value, value_relationship
-        WITH full_path, node, attribute_value, value_relationship
         WHERE all(r in relationships(full_path) WHERE r.status = "active")
         AND attribute_value IS NOT NULL
         AND attribute_value <> $null_value