Skip to content

apply

Applies approved pending changes to the live model reference dataset via the backend.

BackendUpdateCallable

Bases: Protocol

Protocol for backend update operations, supporting both legacy and canonical formats.

Source code in src/horde_model_reference/pending_queue/apply.py 
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
class BackendUpdateCallable(Protocol):
    """Protocol for backend update operations, supporting both legacy and canonical formats."""

    def __call__(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        model_name: str,
        record_dict: dict[str, Any],
        *,
        logical_user_id: str | None = None,
        request_id: str | None = None,
    ) -> None:
        """Protocol for backend update operations, supporting both legacy and canonical formats.

        The callable should perform the necessary update or create operation for the given category and model name,
        using the provided record dictionary as the source of truth for the model's fields. The callable must also
        accept optional parameters for logical user ID and request ID to support auditing and traceability of changes
        through the pending change application process.

        Args:
            category (MODEL_REFERENCE_CATEGORY): The category of the model being updated (e.g., image
                generation, text generation, etc.).
            model_name (str): The unique name of the model to update.
            record_dict (dict[str, Any]): A dictionary representing the model's fields and their new values.
            logical_user_id (str | None): An optional immutable user ID for auditing purposes, representing
                the user on whose behalf the change is being applied.
            request_id (str | None): An optional identifier for the request or job performing the update
                (e.g., a batch apply job ID or CLI invocation ID) to support traceability in logs and audits.

        """

__call__ 

__call__(
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_dict: dict[str, Any],
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Protocol for backend update operations, supporting both legacy and canonical formats.

The callable should perform the necessary update or create operation for the given category and model name, using the provided record dictionary as the source of truth for the model's fields. The callable must also accept optional parameters for logical user ID and request ID to support auditing and traceability of changes through the pending change application process.

Parameters:

  • category (MODEL_REFERENCE_CATEGORY) –

    The category of the model being updated (e.g., image generation, text generation, etc.).

  • model_name (str) –

    The unique name of the model to update.

  • record_dict (dict[str, Any]) –

    A dictionary representing the model's fields and their new values.

  • logical_user_id (str | None, default: None ) –

    An optional immutable user ID for auditing purposes, representing the user on whose behalf the change is being applied.

  • request_id (str | None, default: None ) –

    An optional identifier for the request or job performing the update (e.g., a batch apply job ID or CLI invocation ID) to support traceability in logs and audits.

Source code in src/horde_model_reference/pending_queue/apply.py 
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
def __call__(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    record_dict: dict[str, Any],
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Protocol for backend update operations, supporting both legacy and canonical formats.

    The callable should perform the necessary update or create operation for the given category and model name,
    using the provided record dictionary as the source of truth for the model's fields. The callable must also
    accept optional parameters for logical user ID and request ID to support auditing and traceability of changes
    through the pending change application process.

    Args:
        category (MODEL_REFERENCE_CATEGORY): The category of the model being updated (e.g., image
            generation, text generation, etc.).
        model_name (str): The unique name of the model to update.
        record_dict (dict[str, Any]): A dictionary representing the model's fields and their new values.
        logical_user_id (str | None): An optional immutable user ID for auditing purposes, representing
            the user on whose behalf the change is being applied.
        request_id (str | None): An optional identifier for the request or job performing the update
            (e.g., a batch apply job ID or CLI invocation ID) to support traceability in logs and audits.

    """

BackendDeleteCallable

Bases: Protocol

Protocol for backend delete operations, supporting both legacy and canonical formats.

Source code in src/horde_model_reference/pending_queue/apply.py 
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
class BackendDeleteCallable(Protocol):
    """Protocol for backend delete operations, supporting both legacy and canonical formats."""

    def __call__(
        self,
        category: MODEL_REFERENCE_CATEGORY,
        model_name: str,
        *,
        logical_user_id: str | None = None,
        request_id: str | None = None,
    ) -> None:
        """Protocol for backend delete operations, supporting both legacy and canonical formats.

        The callable should perform the necessary delete operation for the given category and model name. The callable
        must also accept optional parameters for logical user ID and request ID to support auditing and traceability
        of changes through the pending change application process.

        Args:
            category (MODEL_REFERENCE_CATEGORY): The category of the model being deleted (e.g., image
                generation, text generation, etc.).
            model_name (str): The unique name of the model to delete.
            logical_user_id (str | None): An optional immutable user ID for auditing purposes, representing
                the user on whose behalf the change is being applied.
            request_id (str | None): An optional identifier for the request or job performing the delete
                (e.g., a batch apply job ID or CLI invocation ID) to support traceability in logs and audits.

        """

__call__ 

__call__(
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None

Protocol for backend delete operations, supporting both legacy and canonical formats.

The callable should perform the necessary delete operation for the given category and model name. The callable must also accept optional parameters for logical user ID and request ID to support auditing and traceability of changes through the pending change application process.

Parameters:

  • category (MODEL_REFERENCE_CATEGORY) –

    The category of the model being deleted (e.g., image generation, text generation, etc.).

  • model_name (str) –

    The unique name of the model to delete.

  • logical_user_id (str | None, default: None ) –

    An optional immutable user ID for auditing purposes, representing the user on whose behalf the change is being applied.

  • request_id (str | None, default: None ) –

    An optional identifier for the request or job performing the delete (e.g., a batch apply job ID or CLI invocation ID) to support traceability in logs and audits.

Source code in src/horde_model_reference/pending_queue/apply.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def __call__(
    self,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    *,
    logical_user_id: str | None = None,
    request_id: str | None = None,
) -> None:
    """Protocol for backend delete operations, supporting both legacy and canonical formats.

    The callable should perform the necessary delete operation for the given category and model name. The callable
    must also accept optional parameters for logical user ID and request ID to support auditing and traceability
    of changes through the pending change application process.

    Args:
        category (MODEL_REFERENCE_CATEGORY): The category of the model being deleted (e.g., image
            generation, text generation, etc.).
        model_name (str): The unique name of the model to delete.
        logical_user_id (str | None): An optional immutable user ID for auditing purposes, representing
            the user on whose behalf the change is being applied.
        request_id (str | None): An optional identifier for the request or job performing the delete
            (e.g., a batch apply job ID or CLI invocation ID) to support traceability in logs and audits.

    """

PendingChangeApplyError

Bases: RuntimeError

Base class for pending change apply failures.

Source code in src/horde_model_reference/pending_queue/apply.py 
81
82
class PendingChangeApplyError(RuntimeError):
    """Base class for pending change apply failures."""

PendingChangeApplyResult dataclass

Return value when applying a pending change via helper APIs.

Source code in src/horde_model_reference/pending_queue/apply.py 
85
86
87
88
89
90
@dataclass(slots=True)
class PendingChangeApplyResult:
    """Return value when applying a pending change via helper APIs."""

    record: PendingChangeRecord
    batch_split: BatchSplitInfo | None = None

record instance-attribute 

record: PendingChangeRecord

batch_split class-attribute instance-attribute 

batch_split: BatchSplitInfo | None = None

__init__ 

__init__(
    record: PendingChangeRecord,
    batch_split: BatchSplitInfo | None = None,
) -> None

PendingChangeApplyManyResult dataclass

Return value when applying multiple pending changes sequentially.

Source code in src/horde_model_reference/pending_queue/apply.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
@dataclass(slots=True)
class PendingChangeApplyManyResult:
    """Return value when applying multiple pending changes sequentially."""

    applied_records: list[PendingChangeRecord]
    failed_change_id: int | None = None
    failed_error: PendingChangeApplyError | None = None

    # Batch split information (populated if any apply triggered a batch split)
    batch_split_occurred: bool = False
    batch_split_original_batch_id: int | None = None
    batch_split_new_batch_id: int | None = None
    batch_split_reassigned_count: int | None = None

applied_records instance-attribute 

applied_records: list[PendingChangeRecord]

failed_change_id class-attribute instance-attribute 

failed_change_id: int | None = None

failed_error class-attribute instance-attribute 

failed_error: PendingChangeApplyError | None = None

batch_split_occurred class-attribute instance-attribute 

batch_split_occurred: bool = False

batch_split_original_batch_id class-attribute instance-attribute 

batch_split_original_batch_id: int | None = None

batch_split_new_batch_id class-attribute instance-attribute 

batch_split_new_batch_id: int | None = None

batch_split_reassigned_count class-attribute instance-attribute 

batch_split_reassigned_count: int | None = None

__init__ 

__init__(
    applied_records: list[PendingChangeRecord],
    failed_change_id: int | None = None,
    failed_error: PendingChangeApplyError | None = None,
    batch_split_occurred: bool = False,
    batch_split_original_batch_id: int | None = None,
    batch_split_new_batch_id: int | None = None,
    batch_split_reassigned_count: int | None = None,
) -> None

PendingChangeNotFoundError

Bases: PendingChangeApplyError

Raised when a requested pending change cannot be found.

Source code in src/horde_model_reference/pending_queue/apply.py 
108
109
class PendingChangeNotFoundError(PendingChangeApplyError):
    """Raised when a requested pending change cannot be found."""

PendingChangeStateError

Bases: PendingChangeApplyError

Raised when a pending change is in an invalid state for apply.

Source code in src/horde_model_reference/pending_queue/apply.py 
112
113
class PendingChangeStateError(PendingChangeApplyError):
    """Raised when a pending change is in an invalid state for apply."""

PendingChangePayloadError

Bases: PendingChangeApplyError

Raised when a change lacks the payload required for application.

Source code in src/horde_model_reference/pending_queue/apply.py 
116
117
class PendingChangePayloadError(PendingChangeApplyError):
    """Raised when a change lacks the payload required for application."""

PendingChangeBackendError

Bases: PendingChangeApplyError

Raised when the backend fails to persist the applied change.

Source code in src/horde_model_reference/pending_queue/apply.py 
120
121
class PendingChangeBackendError(PendingChangeApplyError):
    """Raised when the backend fails to persist the applied change."""

apply_pending_change 

apply_pending_change(
    *,
    manager: ModelReferenceManager,
    queue_service: PendingQueueService,
    change_id: int,
    applied_by: str,
    applied_username: str,
    job_id: str | None = None,
) -> PendingChangeApplyResult

Apply an approved pending change through the write-capable backend.

Parameters:

  • manager (ModelReferenceManager) –

    Singleton manager exposing the write-capable backend.

  • queue_service (PendingQueueService) –

    Pending queue service used for persistence updates.

  • change_id (int) –

    Identifier of the pending change to apply.

  • applied_by (str) –

    Immutable Horde user id for auditing purposes.

  • applied_username (str) –

    Username corresponding to applied_by.

  • job_id (str | None, default: None ) –

    Optional identifier for the job or CLI invocation performing the apply.

Returns:

Raises:

Source code in src/horde_model_reference/pending_queue/apply.py 
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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def apply_pending_change(
    *,
    manager: ModelReferenceManager,
    queue_service: PendingQueueService,
    change_id: int,
    applied_by: str,
    applied_username: str,
    job_id: str | None = None,
) -> PendingChangeApplyResult:
    """Apply an approved pending change through the write-capable backend.

    Args:
        manager: Singleton manager exposing the write-capable backend.
        queue_service: Pending queue service used for persistence updates.
        change_id: Identifier of the pending change to apply.
        applied_by: Immutable Horde user id for auditing purposes.
        applied_username: Username corresponding to ``applied_by``.
        job_id: Optional identifier for the job or CLI invocation performing the apply.

    Returns:
        PendingChangeApplyResult containing the updated record (now marked as applied).

    Raises:
        PendingChangeNotFoundError: If the change cannot be located.
        PendingChangeStateError: If the change is not approved yet.
        PendingChangePayloadError: When the change operation requires a payload but none exists.
        PendingChangeBackendError: If the backend rejects or fails to persist the write.

    """
    reservation_id = job_id or f"apply-{change_id}-{uuid4().hex}"

    record = queue_service.get_change(change_id)
    if record is None:
        raise PendingChangeNotFoundError(f"Change {change_id} not found.")

    try:
        record = queue_service.reserve_for_apply(change_id=change_id, reservation_id=reservation_id)
    except ValueError as exc:
        if "does not exist" in str(exc):
            raise PendingChangeNotFoundError(f"Change {change_id} not found.") from exc
        raise PendingChangeStateError(str(exc)) from exc

    backend = manager.backend
    canonical_format = horde_model_reference_settings.canonical_format
    if canonical_format == CanonicalFormat.LEGACY:
        if not backend.supports_legacy_writes():
            raise PendingChangeBackendError(
                "Backend does not support legacy write operations in this deployment.",
            )
        backend_update = backend.update_model_legacy
        backend_delete = backend.delete_model_legacy
    else:
        if not backend.supports_writes():
            raise PendingChangeBackendError(
                "Backend does not support write operations in this deployment.",
            )
        backend_update = backend.update_model
        backend_delete = backend.delete_model

    try:
        _apply_change_to_backend(
            record,
            backend_update=backend_update,
            backend_delete=backend_delete,
            logical_user_id=applied_by,
            request_id=reservation_id,
        )
    except PendingChangePayloadError:
        queue_service.clear_apply_reservation(change_id=change_id, reservation_id=reservation_id)
        raise
    except Exception as exc:  # pragma: no cover - defensive log for backend errors
        logger.error("Failed to apply pending change %s: %s", change_id, exc)
        queue_service.clear_apply_reservation(change_id=change_id, reservation_id=reservation_id)
        raise PendingChangeBackendError(str(exc)) from exc

    # The backend's mark_stale() may have already fired during the write, but
    # this ensures stale data is never served even if the callback chain is
    # delayed or skipped.
    manager.invalidate_category_cache(record.category)

    try:
        mark_result = queue_service.mark_applied(
            change_id=change_id,
            applied_by=applied_by,
            applied_username=applied_username,
            job_id=reservation_id,
        )
    except Exception as exc:  # pragma: no cover - defensive log for store errors
        logger.error("Failed to mark pending change %s applied: %s", change_id, exc)
        queue_service.clear_apply_reservation(change_id=change_id, reservation_id=reservation_id)
        raise PendingChangeBackendError(str(exc)) from exc
    return PendingChangeApplyResult(record=mark_result.record, batch_split=mark_result.batch_split)

validate_batch_cohesion 

validate_batch_cohesion(
    *,
    change_ids: Sequence[int],
    queue_service: PendingQueueService,
) -> None

Validate that all change_ids belong to the same batch.

Parameters:

Raises:

Source code in src/horde_model_reference/pending_queue/apply.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def validate_batch_cohesion(
    *,
    change_ids: Sequence[int],
    queue_service: PendingQueueService,
) -> None:
    """Validate that all change_ids belong to the same batch.

    Args:
        change_ids: List of change IDs to validate
        queue_service: The active pending queue service

    Raises:
        ValueError: If changes belong to different batches or have no batch_id
        PendingChangeNotFoundError: If any change_id is not found

    """
    if not change_ids:
        return

    batch_ids: set[int | None] = set()
    for change_id in change_ids:
        change = queue_service.get_change(change_id)
        if change is None:
            raise PendingChangeNotFoundError(f"Change {change_id} not found")
        batch_ids.add(change.batch_id)

    if None in batch_ids:
        raise ValueError("Cannot apply changes that have not been approved in a batch")

    if len(batch_ids) > 1:
        # All batch_ids are non-None at this point
        sorted_ids = sorted(bid for bid in batch_ids if bid is not None)
        raise ValueError(f"All changes must belong to the same batch. Found batch IDs: {sorted_ids}")

apply_pending_changes 

apply_pending_changes(
    *,
    manager: ModelReferenceManager,
    queue_service: PendingQueueService,
    change_ids: Sequence[int],
    applied_by: str,
    applied_username: str,
    job_id: str | None = None,
    enforce_batch_cohesion: bool = True,
) -> PendingChangeApplyManyResult

Apply multiple approved changes sequentially, stopping on first failure.

Parameters:

  • manager (ModelReferenceManager) –

    The active model reference manager

  • queue_service (PendingQueueService) –

    The active pending queue service

  • change_ids (Sequence[int]) –

    List of change IDs to apply

  • applied_by (str) –

    The user ID applying the changes

  • applied_username (str) –

    The username applying the changes

  • job_id (str | None, default: None ) –

    An optional job identifier for tracking the apply job

  • enforce_batch_cohesion (bool, default: True ) –

    If True, all changes must belong to the same batch

Returns:

Raises:

Source code in src/horde_model_reference/pending_queue/apply.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
def apply_pending_changes(
    *,
    manager: ModelReferenceManager,
    queue_service: PendingQueueService,
    change_ids: Sequence[int],
    applied_by: str,
    applied_username: str,
    job_id: str | None = None,
    enforce_batch_cohesion: bool = True,
) -> PendingChangeApplyManyResult:
    """Apply multiple approved changes sequentially, stopping on first failure.

    Args:
        manager: The active model reference manager
        queue_service: The active pending queue service
        change_ids: List of change IDs to apply
        applied_by: The user ID applying the changes
        applied_username: The username applying the changes
        job_id: An optional job identifier for tracking the apply job
        enforce_batch_cohesion: If True, all changes must belong to the same batch

    Returns:
        PendingChangeApplyManyResult: Summary of the apply operation, including
        batch split information if a partial apply triggered reassignment.

    Raises:
        ValueError: If enforce_batch_cohesion=True and changes belong to different batches
        PendingChangeNotFoundError: If any change_id is not found

    """
    if not change_ids:
        raise ValueError("change_ids must contain at least one id")

    if enforce_batch_cohesion:
        validate_batch_cohesion(change_ids=change_ids, queue_service=queue_service)

    applied_records: list[PendingChangeRecord] = []
    last_batch_split: BatchSplitInfo | None = None

    for change_id in change_ids:
        try:
            result = apply_pending_change(
                manager=manager,
                queue_service=queue_service,
                change_id=change_id,
                applied_by=applied_by,
                applied_username=applied_username,
                job_id=job_id,
            )
        except PendingChangeApplyError as exc:
            # On failure, include any batch split info from previous applies
            return PendingChangeApplyManyResult(
                applied_records=applied_records,
                failed_change_id=change_id,
                failed_error=exc,
                batch_split_occurred=last_batch_split is not None,
                batch_split_original_batch_id=last_batch_split.original_batch_id if last_batch_split else None,
                batch_split_new_batch_id=last_batch_split.new_batch_id if last_batch_split else None,
                batch_split_reassigned_count=len(last_batch_split.reassigned_change_ids) if last_batch_split else None,
            )

        applied_records.append(result.record)
        # Track the last batch split (typically only the last apply in a batch triggers it)
        if result.batch_split is not None:
            last_batch_split = result.batch_split

    return PendingChangeApplyManyResult(
        applied_records=applied_records,
        batch_split_occurred=last_batch_split is not None,
        batch_split_original_batch_id=last_batch_split.original_batch_id if last_batch_split else None,
        batch_split_new_batch_id=last_batch_split.new_batch_id if last_batch_split else None,
        batch_split_reassigned_count=len(last_batch_split.reassigned_change_ids) if last_batch_split else None,
    )

_apply_change_to_backend 

_apply_change_to_backend(
    record: PendingChangeRecord,
    *,
    backend_update: BackendUpdateCallable,
    backend_delete: BackendDeleteCallable,
    logical_user_id: str,
    request_id: str | None,
) -> None

Execute the backend mutation for the given pending change.

Source code in src/horde_model_reference/pending_queue/apply.py 
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
def _apply_change_to_backend(
    record: PendingChangeRecord,
    *,
    backend_update: BackendUpdateCallable,
    backend_delete: BackendDeleteCallable,
    logical_user_id: str,
    request_id: str | None,
) -> None:
    """Execute the backend mutation for the given pending change."""
    category = record.category
    model_name = record.model_name

    if record.operation in {AuditOperation.CREATE, AuditOperation.UPDATE}:
        payload = record.payload
        if payload is None:
            raise PendingChangePayloadError(
                f"Change {record.change_id} ({record.operation}) is missing payload data.",
            )
        backend_update(
            category,
            model_name,
            payload,
            logical_user_id=logical_user_id,
            request_id=request_id,
        )
        return

    if record.operation is AuditOperation.DELETE:
        backend_delete(
            category,
            model_name,
            logical_user_id=logical_user_id,
            request_id=request_id,
        )
        return

    raise PendingChangeBackendError(f"Unsupported operation {record.operation} for change {record.change_id}.")