Skip to content

shared

_direct_write_allowlist 

_direct_write_allowlist() -> set[str]

Get the allowlist for direct (non-queue) write operations.

Uses the approver allowlist since direct writes bypass the review workflow.

Source code in src/horde_model_reference/service/v1/routers/shared.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
def _direct_write_allowlist() -> set[str]:
    """Get the allowlist for direct (non-queue) write operations.

    Uses the approver allowlist since direct writes bypass the review workflow.
    """
    from horde_model_reference import horde_model_reference_settings
    from horde_model_reference.service.shared import _normalize_ids

    settings = horde_model_reference_settings.pending_queue
    allowlist = _normalize_ids(settings.approver_ids)
    if not allowlist:
        logger.warning("No approver IDs configured; direct writes will be rejected")
    return allowlist

_check_legacy_model_exists 

_check_legacy_model_exists(
    manager: ModelReferenceManager,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
) -> bool

Check if a model exists in the given category.

Source code in src/horde_model_reference/service/v1/routers/shared.py 
50
51
52
53
54
55
56
57
def _check_legacy_model_exists(
    manager: ModelReferenceManager,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
) -> bool:
    """Check if a model exists in the given category."""
    existing_models = manager.backend.get_legacy_json(category)
    return existing_models is not None and model_name in existing_models

_create_or_update_legacy_model async 

_create_or_update_legacy_model(
    manager: ModelReferenceManager,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    model_record: LegacyGenericRecord
    | LegacyStableDiffusionRecord
    | LegacyTextGenerationRecord
    | LegacyBlipRecord
    | LegacyClipRecord
    | LegacyCodeformerRecord
    | LegacyControlnetRecord
    | LegacyEsrganRecord
    | LegacyGfpganRecord
    | LegacySafetyCheckerRecord
    | LegacyMiscellaneousRecord,
    operation: Operation,
    apikey: str,
    route_name: str,
) -> JSONResponse

Create or update a legacy model record.

When pending queue is enabled, this enqueues the change and returns HTTP 202. When pending queue is disabled, this writes directly to backend and returns HTTP 200/201.

Parameters:

Returns:

  • JSONResponse –

    JSONResponse with either PendingChangeRecord (202) or the model record (200/201).

Raises:

  • HTTPException –

    On validation failure or backend error.

Source code in src/horde_model_reference/service/v1/routers/shared.py 
 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
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
async def _create_or_update_legacy_model(
    manager: ModelReferenceManager,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    model_record: (
        LegacyGenericRecord
        | LegacyStableDiffusionRecord
        | LegacyTextGenerationRecord
        | LegacyBlipRecord
        | LegacyClipRecord
        | LegacyCodeformerRecord
        | LegacyControlnetRecord
        | LegacyEsrganRecord
        | LegacyGfpganRecord
        | LegacySafetyCheckerRecord
        | LegacyMiscellaneousRecord
    ),
    operation: Operation,
    apikey: str,
    route_name: str,
) -> JSONResponse:
    """Create or update a legacy model record.

    When pending queue is enabled, this enqueues the change and returns HTTP 202.
    When pending queue is disabled, this writes directly to backend and returns HTTP 200/201.

    Args:
        manager: The model reference manager.
        category: The model reference category.
        model_name: The name of the model.
        model_record: The model record data.
        operation: Description of operation for logging (e.g., "create", "update").
        apikey: The API key for authentication.
        route_name: The route name for audit metadata.

    Returns:
        JSONResponse with either PendingChangeRecord (202) or the model record (200/201).

    Raises:
        HTTPException: On validation failure or backend error.
    """
    validate_model_name(model_name)

    # Reject backend-prefixed names for text_generation: server auto-generates duplicates
    if category == MODEL_REFERENCE_CATEGORY.text_generation:
        from horde_model_reference.text_backend_names import has_legacy_text_backend_prefix

        if has_legacy_text_backend_prefix(model_name):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=(
                    f"Model name '{model_name}' contains a backend prefix (aphrodite/, koboldcpp/). "
                    "Submit only the base model name — backend duplicates are generated automatically."
                ),
            )

    # Check if pending queue is enabled
    queue_service = manager.pending_queue_service

    # Validate model existence before proceeding
    model_exists = _check_legacy_model_exists(manager, category, model_name)

    if operation == Operation.create and model_exists:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"Model '{model_name}' already exists in category '{category}'. Use PUT to update existing models.",
        )
    if operation == Operation.update and not model_exists:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Model '{model_name}' does not exist in category '{category}'. Use POST to create new models.",
        )

    # Route to queue or direct write based on queue availability
    if queue_service is not None:
        # Pending queue is enabled - enqueue the change
        requestor = await authenticate_queue_requestor(apikey)

        # Convert operation to AuditOperation
        audit_operation = AuditOperation.CREATE if operation == Operation.create else AuditOperation.UPDATE

        # Compute related_models for text_generation so UI can display affected variants
        related_models: list[str] | None = None
        if category == MODEL_REFERENCE_CATEGORY.text_generation:
            from horde_model_reference.text_model_duplicates import TextModelDuplicateManager

            related_models = TextModelDuplicateManager.get_variant_names(model_name)

        # Enqueue the change
        change_record = queue_service.enqueue_change(
            category=category,
            model_name=model_name,
            operation=audit_operation,
            payload=model_record.model_dump(mode="json"),
            requestor_id=requestor.user_id,
            requestor_username=requestor.username,
            notes=None,
            request_metadata={"route": route_name},
            related_models=related_models,
        )

        # Return 202 with the pending change record
        return JSONResponse(
            status_code=status.HTTP_202_ACCEPTED,
            content=change_record.model_dump(mode="json", exclude_none=True),
        )

    # Pending queue is disabled - write directly to backend
    auth_context = await auth_against_horde(
        apikey,
        httpx_client,
        allowed_user_ids=_direct_write_allowlist(),
    )

    if auth_context is None:
        raise APIKeyInvalidException()

    try:
        manager.backend.update_model_legacy_from_base_model(
            category,
            model_name,
            model_record,
            logical_user_id=auth_context.user_id,
        )
        logger.info(f"{operation.capitalize()} legacy model '{model_name}' in category '{category}'")
    except Exception as e:
        logger.exception(f"Error {operation}ing legacy model '{model_name}': {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to {operation} model: {e!s}",
        ) from e

    # Return appropriate success status
    if operation == Operation.create:
        location = route_registry.url_for(
            RouteNames.delete_model,
            {PathVariables.model_category_name: category.value, PathVariables.model_name: model_name},
            prefix=v1_prefix,
        )
        return JSONResponse(
            status_code=status.HTTP_201_CREATED,
            content=model_record.model_dump(),
            headers={"Location": location},
        )
    return JSONResponse(status_code=status.HTTP_200_OK, content=model_record.model_dump())

_delete_legacy_model async 

_delete_legacy_model(
    manager: ModelReferenceManager,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    apikey: str,
    route_name: str,
) -> JSONResponse | Response

Delete a legacy model record.

When pending queue is enabled, this enqueues the deletion and returns HTTP 202. When pending queue is disabled, this deletes directly from backend and returns HTTP 204.

Parameters:

  • manager (ModelReferenceManager) –

    The model reference manager.

  • category (MODEL_REFERENCE_CATEGORY) –

    The model reference category.

  • model_name (str) –

    The name of the model to delete.

  • apikey (str) –

    The API key for authentication.

  • route_name (str) –

    The route name for audit metadata.

Returns:

  • JSONResponse | Response –

    JSONResponse with either PendingChangeRecord (202) or empty response (204).

Raises:

  • HTTPException –

    On validation failure or backend error.

Source code in src/horde_model_reference/service/v1/routers/shared.py 
207
208
209
210
211
212
213
214
215
216
217
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
251
252
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
async def _delete_legacy_model(
    manager: ModelReferenceManager,
    category: MODEL_REFERENCE_CATEGORY,
    model_name: str,
    apikey: str,
    route_name: str,
) -> JSONResponse | Response:
    """Delete a legacy model record.

    When pending queue is enabled, this enqueues the deletion and returns HTTP 202.
    When pending queue is disabled, this deletes directly from backend and returns HTTP 204.

    Args:
        manager: The model reference manager.
        category: The model reference category.
        model_name: The name of the model to delete.
        apikey: The API key for authentication.
        route_name: The route name for audit metadata.

    Returns:
        JSONResponse with either PendingChangeRecord (202) or empty response (204).

    Raises:
        HTTPException: On validation failure or backend error.
    """
    validate_model_name(model_name)

    # Reject backend-prefixed names for text_generation: server auto-generates duplicates
    if category == MODEL_REFERENCE_CATEGORY.text_generation:
        from horde_model_reference.text_backend_names import has_legacy_text_backend_prefix

        if has_legacy_text_backend_prefix(model_name):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=(
                    f"Model name '{model_name}' contains a backend prefix (aphrodite/, koboldcpp/). "
                    "Submit only the base model name — backend duplicates are deleted automatically."
                ),
            )

    # Check if pending queue is enabled
    queue_service = manager.pending_queue_service

    # Validate model exists
    existing_models = manager.backend.get_legacy_json(category)
    if existing_models is None or model_name not in existing_models:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Model '{model_name}' not found in category '{category}'",
        )

    # Route to queue or direct delete based on queue availability
    if queue_service is not None:
        # Pending queue is enabled - enqueue the deletion
        requestor = await authenticate_queue_requestor(apikey)

        # Compute related_models for text_generation so UI can display affected variants
        related_models: list[str] | None = None
        if category == MODEL_REFERENCE_CATEGORY.text_generation:
            from horde_model_reference.text_model_duplicates import TextModelDuplicateManager

            related_models = TextModelDuplicateManager.get_variant_names(model_name)

        # Enqueue the deletion with the existing model data as payload
        change_record = queue_service.enqueue_change(
            category=category,
            model_name=model_name,
            operation=AuditOperation.DELETE,
            payload=existing_models[model_name],
            requestor_id=requestor.user_id,
            requestor_username=requestor.username,
            notes=None,
            request_metadata={"route": route_name},
            related_models=related_models,
        )

        # Return 202 with the pending change record
        return JSONResponse(
            status_code=status.HTTP_202_ACCEPTED,
            content=change_record.model_dump(mode="json", exclude_none=True),
        )

    # Pending queue is disabled - delete directly from backend
    auth_context = await auth_against_horde(
        apikey,
        httpx_client,
        allowed_user_ids=_direct_write_allowlist(),
    )

    if auth_context is None:
        raise APIKeyInvalidException()

    try:
        manager.backend.delete_model_legacy(
            category,
            model_name,
            logical_user_id=auth_context.user_id,
        )
        logger.info(f"Deleted legacy model '{model_name}' from category '{category}'")
    except KeyError as e:
        logger.warning(f"Model '{model_name}' not found during deletion: {e}")
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Model '{model_name}' not found in category '{category}'",
        ) from e
    except Exception as e:
        logger.exception(f"Error deleting legacy model '{model_name}': {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to delete model: {e!s}",
        ) from e

    return Response(status_code=status.HTTP_204_NO_CONTENT)