Skip to content

Constraint checkers reference

UniqueConstraintChecker

autocrud.resource_manager.unique_handler.UniqueConstraintChecker

Bases: IConstraintChecker

Checks that unique-annotated fields are not duplicated.

The checker owns the list of unique fields. When unique_fields is not given explicitly it is auto-detected from the model's Unique annotations. It also ensures every unique field is present in the :class:ResourceManager's indexed fields (auto-adding if missing).

PARAMETER DESCRIPTION
rm

The owning :class:ResourceManager (needed to query storage).

TYPE: ResourceManager

unique_fields

Explicit list of field names to enforce uniqueness on. When None (the default) the fields are auto-detected from Unique annotations on the model.

TYPE: list[str] | None DEFAULT: None

Source code in autocrud/resource_manager/unique_handler.py 
 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
class UniqueConstraintChecker(IConstraintChecker):
    """Checks that unique-annotated fields are not duplicated.

    The checker **owns** the list of unique fields.  When *unique_fields* is
    not given explicitly it is auto-detected from the model's ``Unique``
    annotations.  It also ensures every unique field is present in the
    :class:`ResourceManager`'s indexed fields (auto-adding if missing).

    Args:
        rm: The owning :class:`ResourceManager` (needed to query storage).
        unique_fields: Explicit list of field names to enforce uniqueness on.
            When ``None`` (the default) the fields are auto-detected from
            ``Unique`` annotations on the model.
    """

    def __init__(
        self,
        rm: ResourceManager,
        unique_fields: list[str] | None = None,
    ) -> None:
        self.rm: ResourceManager = rm
        self._unique_fields: list[str] = (
            list(unique_fields)
            if unique_fields is not None
            else extract_unique_fields(rm.resource_type)
        )
        # Auto-ensure every unique field is indexed.
        self._ensure_indexed()

    @property
    def unique_fields(self) -> list[str]:
        """The list of field names enforced by this checker."""
        return self._unique_fields

    # -- internal helpers ----------------------------------------------------

    def _ensure_indexed(self) -> None:
        """Add unique fields to RM's indexed fields if not already present."""
        for field_name in self._unique_fields:
            raw_type = get_field_raw_type(
                self.rm.resource_type, field_name, default=UNSET
            )
            self.rm.add_indexed_field(
                IndexableField(field_path=field_name, field_type=raw_type)
            )

    # -- IConstraintChecker --------------------------------------------------

    def check(
        self,
        data: Any,
        *,
        exclude_resource_id: str | None = None,
    ) -> None:
        """Raise :class:`UniqueConstraintError` if any unique-annotated field
        value is already used by another (non-deleted) resource.
        """
        if not self._unique_fields:
            return
        for field_path in self._unique_fields:
            value = getattr(data, field_path, None)
            if value is None:
                continue
            query = ResourceMetaSearchQuery(
                conditions=[
                    DataSearchCondition(
                        field_path=field_path,
                        operator=DataSearchOperator.equals,
                        value=value,
                    )
                ],
                is_deleted=False,
                limit=1,
                sorts=[
                    ResourceMetaSearchSort(
                        key=ResourceMetaSortKey.updated_time,
                        direction=ResourceMetaSortDirection.ascending,
                    )
                ],
            )
            matches: list[ResourceMeta] = self.rm.storage.search(query)
            if not matches:
                continue
            first: ResourceMeta = matches[0]
            if (
                exclude_resource_id is not None
                and first.resource_id == exclude_resource_id
            ):
                continue  # this resource is the earliest owner — no conflict
            raise UniqueConstraintError(field_path, value, first.resource_id)

    def data_relevant_changed(self, current_data: Any, new_data: Any) -> bool:
        """Return ``True`` if any unique-constrained field value differs."""
        if not self._unique_fields:
            return False
        return any(
            getattr(new_data, f, None) != getattr(current_data, f, None)
            for f in self._unique_fields
        )

Attributes

rm instance-attribute 

rm: ResourceManager = rm

_unique_fields instance-attribute 

_unique_fields: list[str] = (
    list(unique_fields)
    if unique_fields is not None
    else extract_unique_fields(resource_type)
)

unique_fields property 

unique_fields: list[str]

The list of field names enforced by this checker.

Functions

__init__ 

__init__(
    rm: ResourceManager,
    unique_fields: list[str] | None = None,
) -> None
Source code in autocrud/resource_manager/unique_handler.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(
    self,
    rm: ResourceManager,
    unique_fields: list[str] | None = None,
) -> None:
    self.rm: ResourceManager = rm
    self._unique_fields: list[str] = (
        list(unique_fields)
        if unique_fields is not None
        else extract_unique_fields(rm.resource_type)
    )
    # Auto-ensure every unique field is indexed.
    self._ensure_indexed()

_ensure_indexed 

_ensure_indexed() -> None

Add unique fields to RM's indexed fields if not already present.

Source code in autocrud/resource_manager/unique_handler.py 
72
73
74
75
76
77
78
79
80
def _ensure_indexed(self) -> None:
    """Add unique fields to RM's indexed fields if not already present."""
    for field_name in self._unique_fields:
        raw_type = get_field_raw_type(
            self.rm.resource_type, field_name, default=UNSET
        )
        self.rm.add_indexed_field(
            IndexableField(field_path=field_name, field_type=raw_type)
        )

check 

check(
    data: Any, *, exclude_resource_id: str | None = None
) -> None

Raise :class:UniqueConstraintError if any unique-annotated field value is already used by another (non-deleted) resource.

Source code in autocrud/resource_manager/unique_handler.py 
 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
def check(
    self,
    data: Any,
    *,
    exclude_resource_id: str | None = None,
) -> None:
    """Raise :class:`UniqueConstraintError` if any unique-annotated field
    value is already used by another (non-deleted) resource.
    """
    if not self._unique_fields:
        return
    for field_path in self._unique_fields:
        value = getattr(data, field_path, None)
        if value is None:
            continue
        query = ResourceMetaSearchQuery(
            conditions=[
                DataSearchCondition(
                    field_path=field_path,
                    operator=DataSearchOperator.equals,
                    value=value,
                )
            ],
            is_deleted=False,
            limit=1,
            sorts=[
                ResourceMetaSearchSort(
                    key=ResourceMetaSortKey.updated_time,
                    direction=ResourceMetaSortDirection.ascending,
                )
            ],
        )
        matches: list[ResourceMeta] = self.rm.storage.search(query)
        if not matches:
            continue
        first: ResourceMeta = matches[0]
        if (
            exclude_resource_id is not None
            and first.resource_id == exclude_resource_id
        ):
            continue  # this resource is the earliest owner — no conflict
        raise UniqueConstraintError(field_path, value, first.resource_id)

data_relevant_changed 

data_relevant_changed(
    current_data: Any, new_data: Any
) -> bool

Return True if any unique-constrained field value differs.

Source code in autocrud/resource_manager/unique_handler.py 
127
128
129
130
131
132
133
134
def data_relevant_changed(self, current_data: Any, new_data: Any) -> bool:
    """Return ``True`` if any unique-constrained field value differs."""
    if not self._unique_fields:
        return False
    return any(
        getattr(new_data, f, None) != getattr(current_data, f, None)
        for f in self._unique_fields
    )