add AttributeParameters and link to AttributeSchema (#6274)

* add AttributeParameters and link to AttributeSchema

* few small updates

* support content validation for regex, min/max_length

* update constraint validators for parameters

* fix some unit tests

* integrate parameters into schema diff, deprecate regex/min/max

* regenerate schemas

* update tests

* add changelog

* add unit tests

* undo deprecation b/c it raises 10000 warnings

* update docs

* remove unnecessary typing ignores

* handle double falsy values

* add new ConstraintIdentifier enum

* update comment

* maybe fix flaky test

* allow parameters fields to specify validator behavior

Author: ajtmccarty

backend/infrahub/core/attribute.py

@@ -232,30 +232,28 @@ def validate_content(cls, value: Any, name: str, schema: AttributeSchema) -> Non
         Raises:
             ValidationError: Content of the attribute value is not valid
         """
-        if schema.regex:
+        if regex := schema.get_regex():
             if schema.kind == "List":
                 validation_values = [str(entry) for entry in value]
             else:
                 validation_values = [str(value)]
 
             for validation_value in validation_values:
                 try:
-                    is_valid = re.match(pattern=schema.regex, string=str(validation_value))
+                    is_valid = re.match(pattern=regex, string=str(validation_value))
                 except re.error as exc:
-                    raise ValidationError(
-                        {name: f"The regex defined in the schema is not valid ({schema.regex!r})"}
-                    ) from exc
+                    raise ValidationError({name: f"The regex defined in the schema is not valid ({regex!r})"}) from exc
 
                 if not is_valid:
-                    raise ValidationError({name: f"{validation_value} must conform with the regex: {schema.regex!r}"})
+                    raise ValidationError({name: f"{validation_value} must conform with the regex: {regex!r}"})
 
-        if schema.min_length:
-            if len(value) < schema.min_length:
-                raise ValidationError({name: f"{value} must have a minimum length of {schema.min_length!r}"})
+        if min_length := schema.get_min_length():
+            if len(value) < min_length:
+                raise ValidationError({name: f"{value} must have a minimum length of {min_length!r}"})
 
-        if schema.max_length:
-            if len(value) > schema.max_length:
-                raise ValidationError({name: f"{value} must have a maximum length of {schema.max_length!r}"})
+        if max_length := schema.get_max_length():
+            if len(value) > max_length:
+                raise ValidationError({name: f"{value} must have a maximum length of {max_length!r}"})
 
         if schema.enum:
             try:

backend/infrahub/core/models.py

@@ -20,6 +20,7 @@
     from infrahub.core.schema.schema_branch import SchemaBranch
 
 GENERIC_ATTRIBUTES_TO_IGNORE = ["namespace", "name", "branch"]
+PROPERTY_NAMES_TO_IGNORE = ["regex", "min_length", "max_length"]
 
 
 class NodeKind(BaseModel):
@@ -252,11 +253,37 @@ def _process_attrs_rels(
             if not sub_field_diff:
                 raise ValueError("sub_field_diff must be defined, unexpected situation")
 
-            for prop_name in sub_field_diff.changed:
+            for prop_name, prop_diff in sub_field_diff.changed.items():
+                if prop_name in PROPERTY_NAMES_TO_IGNORE:
+                    continue
+
                 field_info = field.model_fields[prop_name]
                 field_update = str(field_info.json_schema_extra.get("update"))  # type: ignore[union-attr]
 
-                schema_path = SchemaPath(  # type: ignore[call-arg]
+                if isinstance(prop_diff, HashableModelDiff):
+                    for param_field_name in prop_diff.changed:
+                        # override field_update if this field has its own json_schema_extra.update
+                        try:
+                            prop_field = getattr(field, prop_name)
+                            param_field_info = prop_field.model_fields[param_field_name]
+                            param_field_update = str(param_field_info.json_schema_extra.get("update"))
+                        except (AttributeError, KeyError):
+                            param_field_update = None
+
+                        schema_path = SchemaPath(
+                            schema_kind=schema.kind,
+                            path_type=path_type,
+                            field_name=field_name,
+                            property_name=f"{prop_name}.{param_field_name}",
+                        )
+
+                        self._process_field(
+                            schema_path=schema_path,
+                            field_update=param_field_update or field_update,
+                        )
+                    continue
+
+                schema_path = SchemaPath(
                     schema_kind=schema.kind,
                     path_type=path_type,
                     field_name=field_name,

backend/infrahub/core/schema/attribute_parameters.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from pydantic import Field
+
+from infrahub.core.constants.schema import UpdateSupport
+from infrahub.core.models import HashableModel
+
+
+def get_attribute_parameters_class_for_kind(kind: str) -> type[AttributeParameters]:
+    return {
+        "Text": TextAttributeParameters,
+        "TextArea": TextAttributeParameters,
+    }.get(kind, AttributeParameters)
+
+
+class AttributeParameters(HashableModel):
+    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},
+    )

backend/infrahub/core/schema/attribute_schema.py

@@ -2,15 +2,17 @@
 
 import enum
 from enum import Enum
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Self
 
-from pydantic import field_validator, model_validator
+from pydantic import Field, ValidationInfo, field_validator, model_validator
 
 from infrahub import config
+from infrahub.core.constants.schema import UpdateSupport
 from infrahub.core.enums import generate_python_enum
 from infrahub.core.query.attribute import default_attribute_query_filter
 from infrahub.types import ATTRIBUTE_KIND_LABELS, ATTRIBUTE_TYPES
 
+from .attribute_parameters import AttributeParameters, TextAttributeParameters, get_attribute_parameters_class_for_kind
 from .generated.attribute_schema import GeneratedAttributeSchema
 
 if TYPE_CHECKING:
@@ -21,6 +23,14 @@
     from infrahub.database import InfrahubDatabase
 
 
+def get_attribute_schema_class_for_kind(kind: str) -> type[AttributeSchema]:
+    attribute_schema_class_by_kind: dict[str, type[AttributeSchema]] = {
+        "Text": TextAttributeSchema,
+        "TextArea": TextAttributeSchema,
+    }
+    return attribute_schema_class_by_kind.get(kind, AttributeSchema)
+
+
 class AttributeSchema(GeneratedAttributeSchema):
     _sort_by: list[str] = ["name"]
     _enum_class: type[enum.Enum] | None = None
@@ -53,16 +63,36 @@ def kind_options(cls, v: str) -> str:
 
     @model_validator(mode="before")
     @classmethod
-    def validate_dropdown_choices(cls, values: dict[str, Any]) -> dict[str, Any]:
+    def validate_dropdown_choices(cls, values: Any) -> Any:
         """Validate that choices are defined for a dropdown but not for other kinds."""
-        if values.get("kind") != "Dropdown" and values.get("choices"):
-            raise ValueError(f"Can only specify 'choices' for kind=Dropdown: {values['kind']}")
-
-        if values.get("kind") == "Dropdown" and not values.get("choices"):
+        if isinstance(values, dict):
+            kind = values.get("kind")
+            choices = values.get("choices")
+        elif isinstance(values, AttributeSchema):
+            kind = values.kind
+            choices = values.choices
+        else:
+            return values
+        if kind != "Dropdown" and choices:
+            raise ValueError(f"Can only specify 'choices' for kind=Dropdown: {kind}")
+
+        if kind == "Dropdown" and not choices:
             raise ValueError("The property 'choices' is required for kind=Dropdown")
 
         return values
 
+    @field_validator("parameters", mode="before")
+    @classmethod
+    def set_parameters_type(cls, value: Any, info: ValidationInfo) -> Any:
+        """Override parameters class if using base AttributeParameters class and should be using a subclass"""
+        kind = info.data["kind"]
+        expected_parameters_class = get_attribute_parameters_class_for_kind(kind=kind)
+        if value is None:
+            return expected_parameters_class()
+        if not isinstance(value, expected_parameters_class) and isinstance(value, AttributeParameters):
+            return expected_parameters_class(**value.model_dump())
+        return value
+
     def get_class(self) -> type[BaseAttribute]:
         return ATTRIBUTE_TYPES[self.kind].get_infrahub_class()
 
@@ -106,7 +136,7 @@ def update_from_generic(self, other: AttributeSchema) -> None:
 
     def to_node(self) -> dict[str, Any]:
         fields_to_exclude = {"id", "state", "filters"}
-        fields_to_json = {"computed_attribute"}
+        fields_to_json = {"computed_attribute", "parameters"}
         data = self.model_dump(exclude=fields_to_exclude | fields_to_json)
 
         for field_name in fields_to_json:
@@ -117,6 +147,15 @@ def to_node(self) -> dict[str, Any]:
 
         return data
 
+    def get_regex(self) -> str | None:
+        return self.regex
+
+    def get_min_length(self) -> int | None:
+        return self.min_length
+
+    def get_max_length(self) -> int | None:
+        return self.max_length
+
     async def get_query_filter(
         self,
         name: str,
@@ -144,3 +183,39 @@ async def get_query_filter(
             partial_match=partial_match,
             support_profiles=support_profiles,
         )
+
+
+class TextAttributeSchema(AttributeSchema):
+    parameters: TextAttributeParameters = Field(
+        default_factory=TextAttributeParameters,
+        description="Extra parameters specific to text attributes",
+        json_schema_extra={"update": UpdateSupport.VALIDATE_CONSTRAINT.value},
+    )
+
+    @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
+            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
+            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
+            self.max_length = self.parameters.max_length = final_max_length
+        return self
+
+    def get_regex(self) -> str | None:
+        return self.parameters.regex
+
+    def get_min_length(self) -> int | None:
+        return self.parameters.min_length
+
+    def get_max_length(self) -> int | None:
+        return self.parameters.max_length

backend/infrahub/core/schema/basenode_schema.py

@@ -13,7 +13,7 @@
 from infrahub.core.constants import RelationshipCardinality, RelationshipKind
 from infrahub.core.models import HashableModel, HashableModelDiff
 
-from .attribute_schema import AttributeSchema
+from .attribute_schema import AttributeSchema, get_attribute_schema_class_for_kind
 from .generated.base_node_schema import GeneratedBaseNodeSchema
 from .relationship_schema import RelationshipSchema
 
@@ -74,6 +74,30 @@ def __hash__(self) -> int:
         Be careful hash generated from hash() have a salt by default and they will not be the same across run"""
         return hash(self.get_hash())
 
+    @field_validator("attributes", mode="before")
+    @classmethod
+    def set_attribute_type(cls, raw_attributes: Any) -> Any:
+        if not isinstance(raw_attributes, list):
+            return raw_attributes
+        attribute_schemas_with_types: list[Any] = []
+        for raw_attr in raw_attributes:
+            if not isinstance(raw_attr, (dict, AttributeSchema)):
+                attribute_schemas_with_types.append(raw_attr)
+                continue
+            if isinstance(raw_attr, dict):
+                kind = raw_attr.get("kind")
+                attribute_type_class = get_attribute_schema_class_for_kind(kind=kind)
+                attribute_schemas_with_types.append(attribute_type_class(**raw_attr))
+                continue
+
+            expected_attr_schema_class = get_attribute_schema_class_for_kind(kind=raw_attr.kind)
+            if not isinstance(raw_attr, expected_attr_schema_class):
+                final_attr = expected_attr_schema_class(**raw_attr.model_dump())
+            else:
+                final_attr = raw_attr
+            attribute_schemas_with_types.append(final_attr)
+        return attribute_schemas_with_types
+
     def to_dict(self) -> dict:
         data = self.model_dump(
             exclude_unset=True, exclude_none=True, exclude_defaults=True, exclude={"attributes", "relationships"}

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

@@ -31,6 +31,7 @@
     RelationshipKind,
     UpdateSupport,
 )
+from infrahub.core.schema.attribute_parameters import AttributeParameters
 from infrahub.core.schema.attribute_schema import AttributeSchema
 from infrahub.core.schema.computed_attribute import ComputedAttribute
 from infrahub.core.schema.dropdown import DropdownChoice
@@ -506,21 +507,21 @@ def to_dict(self) -> dict[str, Any]:
         SchemaAttribute(
             name="regex",
             kind="Text",
-            description="Regex uses to limit the characters allowed in for the attributes.",
+            description="Regex uses to limit the characters allowed in for the attributes. (deprecated: please use parameters.regex instead)",
             optional=True,
             extra={"update": UpdateSupport.VALIDATE_CONSTRAINT},
         ),
         SchemaAttribute(
             name="max_length",
             kind="Number",
-            description="Set a maximum number of characters allowed for a given attribute.",
+            description="Set a maximum number of characters allowed for a given attribute. (deprecated: please use parameters.max_length instead)",
             optional=True,
             extra={"update": UpdateSupport.VALIDATE_CONSTRAINT},
         ),
         SchemaAttribute(
             name="min_length",
             kind="Number",
-            description="Set a minimum number of characters allowed for a given attribute.",
+            description="Set a minimum number of characters allowed for a given attribute. (deprecated: please use parameters.min_length instead)",
             optional=True,
             extra={"update": UpdateSupport.VALIDATE_CONSTRAINT},
         ),
@@ -617,6 +618,15 @@ def to_dict(self) -> dict[str, Any]:
             optional=True,
             extra={"update": UpdateSupport.ALLOWED},
         ),
+        SchemaAttribute(
+            name="parameters",
+            kind="JSON",
+            internal_kind=AttributeParameters,
+            optional=True,
+            description="Extra parameters specific to this kind of attribute",
+            extra={"update": UpdateSupport.VALIDATE_CONSTRAINT},
+            default_factory="AttributeParameters",
+        ),
         SchemaAttribute(
             name="deprecation",
             kind="Text",

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

@@ -8,6 +8,7 @@
 
 from infrahub.core.constants import AllowOverrideType, BranchSupportType, HashableModelState
 from infrahub.core.models import HashableModel
+from infrahub.core.schema.attribute_parameters import AttributeParameters  # noqa: TC001
 from infrahub.core.schema.computed_attribute import ComputedAttribute  # noqa: TC001
 from infrahub.core.schema.dropdown import DropdownChoice  # noqa: TC001
 
@@ -44,17 +45,17 @@ class GeneratedAttributeSchema(HashableModel):
     )
     regex: str | None = Field(
         default=None,
-        description="Regex uses to limit the characters allowed in for the attributes.",
+        description="Regex uses to limit the characters allowed in for the attributes. (deprecated: please use parameters.regex instead)",
         json_schema_extra={"update": "validate_constraint"},
     )
     max_length: int | None = Field(
         default=None,
-        description="Set a maximum number of characters allowed for a given attribute.",
+        description="Set a maximum number of characters allowed for a given attribute. (deprecated: please use parameters.max_length instead)",
         json_schema_extra={"update": "validate_constraint"},
     )
     min_length: int | None = Field(
         default=None,
-        description="Set a minimum number of characters allowed for a given attribute.",
+        description="Set a minimum number of characters allowed for a given attribute. (deprecated: please use parameters.min_length instead)",
         json_schema_extra={"update": "validate_constraint"},
     )
     label: str | None = Field(
@@ -112,6 +113,11 @@ class GeneratedAttributeSchema(HashableModel):
         description="Type of allowed override for the attribute.",
         json_schema_extra={"update": "allowed"},
     )
+    parameters: AttributeParameters = 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,
         description="Mark attribute as deprecated and provide a user-friendly message to display",