Skip to content

shared

header_auth_scheme module-attribute 

header_auth_scheme = APIKeyHeader(name='apikey')

DEFAULT_AUTH_TIMEOUT_SECONDS module-attribute 

DEFAULT_AUTH_TIMEOUT_SECONDS = 10.0

httpx_client module-attribute 

httpx_client = AsyncClient(
    timeout=Timeout(DEFAULT_AUTH_TIMEOUT_SECONDS)
)

v1_prefix module-attribute 

v1_prefix = '/model_references/v1'

v2_prefix module-attribute 

v2_prefix = '/model_references/v2'

statistics_prefix module-attribute 

statistics_prefix = '/model_references/statistics'

route_registry module-attribute 

route_registry = RouteRegistry()

_requestor_fallback_logged module-attribute 

_requestor_fallback_logged = False

_approver_fallback_logged module-attribute 

_approver_fallback_logged = False

_INVALID_MODEL_NAME_CHARS module-attribute 

_INVALID_MODEL_NAME_CHARS = frozenset('\\')

PathVariables

Bases: StrEnum

Path variables used in the router.

Source code in src/horde_model_reference/service/shared.py 
32
33
34
35
36
class PathVariables(StrEnum):
    """Path variables used in the router."""

    model_category_name = auto()
    model_name = auto()

model_category_name class-attribute instance-attribute 

model_category_name = auto()

model_name class-attribute instance-attribute 

model_name = auto()

RouteNames

Bases: StrEnum

Route names used in the router.

Source code in src/horde_model_reference/service/shared.py 
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
class RouteNames(StrEnum):
    """Route names used in the router."""

    get_reference_info = auto()
    get_reference_names = auto()
    get_reference_by_category = auto()
    get_text_generation_reference = auto()
    get_single_model = auto()
    image_generation_model = auto()
    text_generation_model = auto()
    blip_model = auto()
    clip_model = auto()
    codeformer_model = auto()
    controlnet_model = auto()
    esrgan_model = auto()
    gfpgan_model = auto()
    safety_checker_model = auto()
    miscellaneous_model = auto()
    create_model = auto()
    update_model = auto()
    update_image_generation_model = auto()
    update_text_generation_model = auto()
    update_controlnet_model = auto()
    delete_model = auto()
    get_models_with_stats = auto()
    get_category_statistics = auto()
    get_category_deletion_risk = auto()

    # V1 metadata routes
    get_legacy_last_updated = auto()
    get_legacy_category_last_updated = auto()
    get_all_legacy_metadata = auto()
    get_legacy_category_metadata = auto()

    # V2 metadata routes
    get_v2_last_updated = auto()
    get_v2_category_last_updated = auto()
    get_all_v2_metadata = auto()
    get_v2_category_metadata = auto()

get_reference_info class-attribute instance-attribute 

get_reference_info = auto()

get_reference_names class-attribute instance-attribute 

get_reference_names = auto()

get_reference_by_category class-attribute instance-attribute 

get_reference_by_category = auto()

get_text_generation_reference class-attribute instance-attribute 

get_text_generation_reference = auto()

get_single_model class-attribute instance-attribute 

get_single_model = auto()

image_generation_model class-attribute instance-attribute 

image_generation_model = auto()

text_generation_model class-attribute instance-attribute 

text_generation_model = auto()

blip_model class-attribute instance-attribute 

blip_model = auto()

clip_model class-attribute instance-attribute 

clip_model = auto()

codeformer_model class-attribute instance-attribute 

codeformer_model = auto()

controlnet_model class-attribute instance-attribute 

controlnet_model = auto()

esrgan_model class-attribute instance-attribute 

esrgan_model = auto()

gfpgan_model class-attribute instance-attribute 

gfpgan_model = auto()

safety_checker_model class-attribute instance-attribute 

safety_checker_model = auto()

miscellaneous_model class-attribute instance-attribute 

miscellaneous_model = auto()

create_model class-attribute instance-attribute 

create_model = auto()

update_model class-attribute instance-attribute 

update_model = auto()

update_image_generation_model class-attribute instance-attribute 

update_image_generation_model = auto()

update_text_generation_model class-attribute instance-attribute 

update_text_generation_model = auto()

update_controlnet_model class-attribute instance-attribute 

update_controlnet_model = auto()

delete_model class-attribute instance-attribute 

delete_model = auto()

get_models_with_stats class-attribute instance-attribute 

get_models_with_stats = auto()

get_category_statistics class-attribute instance-attribute 

get_category_statistics = auto()

get_category_deletion_risk class-attribute instance-attribute 

get_category_deletion_risk = auto()

get_legacy_last_updated class-attribute instance-attribute 

get_legacy_last_updated = auto()

get_legacy_category_last_updated class-attribute instance-attribute 

get_legacy_category_last_updated = auto()

get_all_legacy_metadata class-attribute instance-attribute 

get_all_legacy_metadata = auto()

get_legacy_category_metadata class-attribute instance-attribute 

get_legacy_category_metadata = auto()

get_v2_last_updated class-attribute instance-attribute 

get_v2_last_updated = auto()

get_v2_category_last_updated class-attribute instance-attribute 

get_v2_category_last_updated = auto()

get_all_v2_metadata class-attribute instance-attribute 

get_all_v2_metadata = auto()

get_v2_category_metadata class-attribute instance-attribute 

get_v2_category_metadata = auto()

RouteRegistry

Registry for routes.

Routes are stored with a composite key of (prefix, route_name) to support multiple API versions (v1, v2) with overlapping route names.

Source code in src/horde_model_reference/service/shared.py 
 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
class RouteRegistry:
    """Registry for routes.

    Routes are stored with a composite key of (prefix, route_name) to support
    multiple API versions (v1, v2) with overlapping route names.
    """

    _routes: dict[tuple[str, RouteNames], str]

    def __init__(self) -> None:
        """Initialize the route registry."""
        self._routes = {}

    def register_route(self, prefix: str, route_name: RouteNames, path: str) -> None:
        """Register a route with its path.

        Args:
            prefix (str): The prefix to use for the route (e.g., "/model_references/v1").
            route_name (RouteNames): The name of the route to register.
            path (str): The path of the route.
        """
        key = (prefix, route_name)
        self._routes[key] = f"{prefix}{path}"

    def url_for(self, route_name: RouteNames, path_params: dict[str, str], prefix: str) -> str:
        """Get the URL for a registered route.

        Args:
            route_name (RouteNames): The name of the route to get the URL for.
            path_params (dict[str, str]): The path parameters to include in the URL.
            prefix (str | None): The API prefix (e.g., "/model_references/v2").
                If None, defaults to v2_prefix for backward compatibility.

        Returns:
            str: The complete URL path with parameters substituted.

        Raises:
            ValueError: If the route is not registered.
        """
        key = (prefix, route_name)
        path = self._routes.get(key)
        if path is None:
            raise ValueError(f"Route {route_name} with prefix {prefix} is not registered.")

        for key_name, value in path_params.items():
            path = path.replace(f"{{{key_name}}}", value)
        return path

_routes instance-attribute 

_routes: dict[tuple[str, RouteNames], str] = {}

__init__ 

__init__() -> None

Initialize the route registry.

Source code in src/horde_model_reference/service/shared.py 
89
90
91
def __init__(self) -> None:
    """Initialize the route registry."""
    self._routes = {}

register_route 

register_route(
    prefix: str, route_name: RouteNames, path: str
) -> None

Register a route with its path.

Parameters:

  • prefix (str) –

    The prefix to use for the route (e.g., "/model_references/v1").

  • route_name (RouteNames) –

    The name of the route to register.

  • path (str) –

    The path of the route.

Source code in src/horde_model_reference/service/shared.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
def register_route(self, prefix: str, route_name: RouteNames, path: str) -> None:
    """Register a route with its path.

    Args:
        prefix (str): The prefix to use for the route (e.g., "/model_references/v1").
        route_name (RouteNames): The name of the route to register.
        path (str): The path of the route.
    """
    key = (prefix, route_name)
    self._routes[key] = f"{prefix}{path}"

url_for 

url_for(
    route_name: RouteNames,
    path_params: dict[str, str],
    prefix: str,
) -> str

Get the URL for a registered route.

Parameters:

  • route_name (RouteNames) –

    The name of the route to get the URL for.

  • path_params (dict[str, str]) –

    The path parameters to include in the URL.

  • prefix (str | None) –

    The API prefix (e.g., "/model_references/v2"). If None, defaults to v2_prefix for backward compatibility.

Returns:

  • str ( str ) –

    The complete URL path with parameters substituted.

Raises:

Source code in src/horde_model_reference/service/shared.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
def url_for(self, route_name: RouteNames, path_params: dict[str, str], prefix: str) -> str:
    """Get the URL for a registered route.

    Args:
        route_name (RouteNames): The name of the route to get the URL for.
        path_params (dict[str, str]): The path parameters to include in the URL.
        prefix (str | None): The API prefix (e.g., "/model_references/v2").
            If None, defaults to v2_prefix for backward compatibility.

    Returns:
        str: The complete URL path with parameters substituted.

    Raises:
        ValueError: If the route is not registered.
    """
    key = (prefix, route_name)
    path = self._routes.get(key)
    if path is None:
        raise ValueError(f"Route {route_name} with prefix {prefix} is not registered.")

    for key_name, value in path_params.items():
        path = path.replace(f"{{{key_name}}}", value)
    return path

Operation

Bases: StrEnum

CRUD operation types.

Source code in src/horde_model_reference/service/shared.py 
132
133
134
135
136
137
class Operation(StrEnum):
    """CRUD operation types."""

    create = "create"
    update = "update"
    delete = "delete"

create class-attribute instance-attribute 

create = 'create'

update class-attribute instance-attribute 

update = 'update'

delete class-attribute instance-attribute 

delete = 'delete'

HordeUserContext dataclass

Immutable details about the authenticated Horde user.

Source code in src/horde_model_reference/service/shared.py 
144
145
146
147
148
149
@dataclass(frozen=True)
class HordeUserContext:
    """Immutable details about the authenticated Horde user."""

    user_id: str
    username: str

user_id instance-attribute 

user_id: str

username instance-attribute 

username: str

__init__ 

__init__(user_id: str, username: str) -> None

APIKeyInvalidException

Bases: HTTPException

Exception raised when an API key is invalid.

Source code in src/horde_model_reference/service/shared.py 
332
333
334
335
336
337
338
339
340
341
class APIKeyInvalidException(HTTPException):
    """Exception raised when an API key is invalid."""

    def __init__(self, headers: dict[str, str] | None = None) -> None:
        """Initialize the exception."""
        super().__init__(
            status_code=401,
            detail="Invalid API key",
            headers=headers,
        )

__init__ 

__init__(headers: dict[str, str] | None = None) -> None

Initialize the exception.

Source code in src/horde_model_reference/service/shared.py 
335
336
337
338
339
340
341
def __init__(self, headers: dict[str, str] | None = None) -> None:
    """Initialize the exception."""
    super().__init__(
        status_code=401,
        detail="Invalid API key",
        headers=headers,
    )

ErrorDetail

Bases: BaseModel

Detail about a specific error.

Source code in src/horde_model_reference/service/shared.py 
344
345
346
347
348
349
350
351
352
class ErrorDetail(BaseModel):
    """Detail about a specific error."""

    loc: list[str | int] | None = None
    """Location of the error (for validation errors)."""
    msg: str
    """Error message."""
    type: str | None = None
    """Error type."""

loc class-attribute instance-attribute 

loc: list[str | int] | None = None

Location of the error (for validation errors).

msg instance-attribute 

msg: str

Error message.

type class-attribute instance-attribute 

type: str | None = None

Error type.

ErrorResponse

Bases: BaseModel

Standardized error response.

Source code in src/horde_model_reference/service/shared.py 
355
356
357
358
359
class ErrorResponse(BaseModel):
    """Standardized error response."""

    detail: str | list[ErrorDetail]
    """Error details - either a string message or list of validation errors."""

detail instance-attribute 

detail: str | list[ErrorDetail]

Error details - either a string message or list of validation errors.

auth_against_horde async 

auth_against_horde(
    apikey: str,
    client: AsyncClient,
    *,
    allowed_user_ids: Collection[str] | None = None,
) -> HordeUserContext | None

Authenticate the provided API key against the AI-Horde.

This uses the endpoint defined by AI_HORDE_URL by AIHordeClientSettings in haidra_core.

Parameters:

  • apikey (str) –

    The API key to authenticate.

  • client (AsyncClient) –

    The HTTP client to use for the request.

  • allowed_user_ids (Collection[str] | None, default: None ) –

    Optional allowlist of Horde user IDs permitted for the caller.

Returns:

  • HordeUserContext | None –

    HordeUserContext | None: User details if authentication is successful, None otherwise.

Raises:

  • HTTPException –

    503 if the Horde auth service is unreachable or times out.

Source code in src/horde_model_reference/service/shared.py 
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
async def auth_against_horde(
    apikey: str,
    client: httpx.AsyncClient,
    *,
    allowed_user_ids: Collection[str] | None = None,
) -> HordeUserContext | None:
    """Authenticate the provided API key against the AI-Horde.

    This uses the endpoint defined by AI_HORDE_URL by AIHordeClientSettings in haidra_core.

    Args:
        apikey (str): The API key to authenticate.
        client (httpx.AsyncClient): The HTTP client to use for the request.
        allowed_user_ids (Collection[str] | None): Optional allowlist of Horde user IDs permitted for the caller.

    Returns:
        HordeUserContext | None: User details if authentication is successful, None otherwise.

    Raises:
        HTTPException: 503 if the Horde auth service is unreachable or times out.
    """
    find_user_subpath = "v2/find_user"
    url = urllib.parse.urljoin(
        str(ai_horde_worker_settings.ai_horde_url),
        find_user_subpath,
    )

    try:
        response = await client.get(
            url,
            headers={"apikey": f"{apikey}"},
        )
    except httpx.TimeoutException:
        logger.warning("Horde auth service timed out")
        raise HTTPException(status_code=503, detail="Auth service timed out") from None
    except httpx.HTTPError as exc:
        logger.warning(f"Horde auth service unreachable: {exc}")
        raise HTTPException(status_code=503, detail="Auth service unavailable") from None

    if response.status_code != 200:
        return None

    user_data = response.json()
    user_name = user_data.get("username", "")

    if "#" not in user_name:
        logger.warning(f"Unknown apikey: {user_data}")
        return None

    user_id = user_name.split("#")[-1]

    if allowed_user_ids and user_id not in allowed_user_ids:
        logger.warning(f"Unauthorized user ID: {user_id}")
        return None

    return HordeUserContext(user_id=user_id, username=user_name)

_normalize_ids 

_normalize_ids(values: Collection[str]) -> set[str]
Source code in src/horde_model_reference/service/shared.py 
210
211
def _normalize_ids(values: Collection[str]) -> set[str]:
    return {value.strip() for value in values if value and value.strip()}

_fallback_allowed_users 

_fallback_allowed_users(
    context: Literal["requestor", "approver"],
) -> set[str]

Return an empty set and log a warning when no allowlist is configured.

Fails closed: if no allowlist is configured, no users are authorized.

Source code in src/horde_model_reference/service/shared.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def _fallback_allowed_users(context: Literal["requestor", "approver"]) -> set[str]:
    """Return an empty set and log a warning when no allowlist is configured.

    Fails closed: if no allowlist is configured, no users are authorized.
    """
    global _requestor_fallback_logged, _approver_fallback_logged
    already_logged = _requestor_fallback_logged if context == "requestor" else _approver_fallback_logged
    if not already_logged:
        logger.warning(
            f"Pending queue {context} allowlist is not configured; all {context} requests will be rejected",
        )
        if context == "requestor":
            _requestor_fallback_logged = True
        else:
            _approver_fallback_logged = True

    return set()

_queue_requestor_allowlist 

_queue_requestor_allowlist() -> set[str]
Source code in src/horde_model_reference/service/shared.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
def _queue_requestor_allowlist() -> set[str]:
    settings = horde_model_reference_settings.pending_queue

    logger.debug("Building requestor allowlist from settings")

    allowlist = _normalize_ids(settings.requestor_ids)
    allowlist.update(_normalize_ids(settings.approver_ids))

    logger.debug(f"Combined requestor allowlist (including approvers): {allowlist}")

    if allowlist:
        return allowlist
    return _fallback_allowed_users("requestor")

_queue_approver_allowlist 

_queue_approver_allowlist() -> set[str]
Source code in src/horde_model_reference/service/shared.py 
248
249
250
251
252
253
254
255
256
257
258
259
def _queue_approver_allowlist() -> set[str]:
    settings = horde_model_reference_settings.pending_queue

    logger.debug("Building approver allowlist from settings")

    allowlist = _normalize_ids(settings.approver_ids)

    logger.debug(f"Approver allowlist: {allowlist}")

    if allowlist:
        return allowlist
    return _fallback_allowed_users("approver")

authenticate_queue_requestor async 

authenticate_queue_requestor(
    apikey: str,
) -> HordeUserContext

Authenticate a queue requestor using the configured allowlist.

Raises:

  • APIKeyInvalidException –

    If no allowlist is configured or the user is not authorized.

  • HTTPException –

    503 if the Horde auth service is unreachable.

Source code in src/horde_model_reference/service/shared.py 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
async def authenticate_queue_requestor(apikey: str) -> HordeUserContext:
    """Authenticate a queue requestor using the configured allowlist.

    Raises:
        APIKeyInvalidException: If no allowlist is configured or the user is not authorized.
        HTTPException: 503 if the Horde auth service is unreachable.
    """
    allowlist = _queue_requestor_allowlist()
    if not allowlist:
        raise APIKeyInvalidException()

    context = await auth_against_horde(apikey, httpx_client, allowed_user_ids=allowlist)
    if context is None:
        raise APIKeyInvalidException()
    return context

authenticate_queue_approver async 

authenticate_queue_approver(
    apikey: str,
) -> HordeUserContext

Authenticate a queue approver using the configured allowlist.

Raises:

  • APIKeyInvalidException –

    If no allowlist is configured or the user is not authorized.

  • HTTPException –

    503 if the Horde auth service is unreachable.

Source code in src/horde_model_reference/service/shared.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
async def authenticate_queue_approver(apikey: str) -> HordeUserContext:
    """Authenticate a queue approver using the configured allowlist.

    Raises:
        APIKeyInvalidException: If no allowlist is configured or the user is not authorized.
        HTTPException: 503 if the Horde auth service is unreachable.
    """
    allowlist = _queue_approver_allowlist()
    if not allowlist:
        raise APIKeyInvalidException()

    context = await auth_against_horde(apikey, httpx_client, allowed_user_ids=allowlist)
    if context is not None:
        logger.debug(f"Approver authenticated: user_id={context.user_id}")
    if context is None:
        raise APIKeyInvalidException()
    return context

get_user_roles async 

get_user_roles(
    apikey: str,
) -> tuple[HordeUserContext | None, set[str]]

Authenticate a user and determine their roles based on configured allowlists.

This function authenticates the user without enforcing any specific role requirement, then checks which roles the user has been granted.

Parameters:

  • apikey (str) –

    The API key to authenticate.

Returns:

  • HordeUserContext | None –

    A tuple of (user_context, roles) where:

  • set[str] –
    • user_context: The authenticated user details, or None if authentication failed.
  • tuple[HordeUserContext | None, set[str]] –
    • roles: A set of role names the user has (e.g., {'approver', 'requestor'}).
Source code in src/horde_model_reference/service/shared.py 
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
326
327
328
329
async def get_user_roles(apikey: str) -> tuple[HordeUserContext | None, set[str]]:
    """Authenticate a user and determine their roles based on configured allowlists.

    This function authenticates the user without enforcing any specific role requirement,
    then checks which roles the user has been granted.

    Args:
        apikey: The API key to authenticate.

    Returns:
        A tuple of (user_context, roles) where:
        - user_context: The authenticated user details, or None if authentication failed.
        - roles: A set of role names the user has (e.g., {'approver', 'requestor'}).
    """
    # Authenticate without any allowlist restriction first
    context = await auth_against_horde(apikey, httpx_client, allowed_user_ids=None)
    if context is None:
        return None, set()

    roles: set[str] = set()

    # Check approver status
    approver_allowlist = _queue_approver_allowlist()
    if context.user_id in approver_allowlist:
        roles.add("approver")

    # Check requestor status
    requestor_allowlist = _queue_requestor_allowlist()
    if context.user_id in requestor_allowlist:
        roles.add("requestor")

    return context, roles

validate_model_name 

validate_model_name(model_name: str) -> None

Reject model names that are empty, whitespace-only, or contain path separators.

Raises:

  • HTTPException –

    422 if the model name is invalid.

Source code in src/horde_model_reference/service/shared.py 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
def validate_model_name(model_name: str) -> None:
    """Reject model names that are empty, whitespace-only, or contain path separators.

    Raises:
        HTTPException: 422 if the model name is invalid.
    """
    if not model_name or not model_name.strip():
        raise HTTPException(
            status_code=422,
            detail="Model name must not be empty or whitespace-only.",
        )
    if _INVALID_MODEL_NAME_CHARS & set(model_name):
        raise HTTPException(
            status_code=422,
            detail=f"Model name must not contain invalid characters {''.join(_INVALID_MODEL_NAME_CHARS)}: "
            f"'{model_name}'",
        )

get_model_reference_manager 

get_model_reference_manager() -> ModelReferenceManager

Dependency helper that returns the singleton model reference manager.

Source code in src/horde_model_reference/service/shared.py 
384
385
386
def get_model_reference_manager() -> ModelReferenceManager:
    """Dependency helper that returns the singleton model reference manager."""
    return ModelReferenceManager()

assert_canonical_write_enabled 

assert_canonical_write_enabled(
    manager: ModelReferenceManager,
    *,
    canonical_format: CanonicalFormat,
) -> None

Ensure that writes are attempted only when the canonical format allows them.

Source code in src/horde_model_reference/service/shared.py 
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
def assert_canonical_write_enabled(
    manager: ModelReferenceManager,
    *,
    canonical_format: CanonicalFormat,
) -> None:
    """Ensure that writes are attempted only when the canonical format allows them."""
    backend = manager.backend
    expected_format = horde_model_reference_settings.canonical_format
    if canonical_format == CanonicalFormat.v2:
        if not backend.supports_writes():
            raise HTTPException(
                status_code=503,
                detail=(
                    "This instance is in REPLICA mode and does not support write operations. "
                    "Only PRIMARY instances can queue model changes."
                ),
            )
        if expected_format != canonical_format:
            raise HTTPException(
                status_code=503,
                detail=(
                    "This deployment does not expose write operations for this API. "
                    f"Expected canonical_format='{canonical_format}', got '{expected_format}'."
                ),
            )
        return

    if not backend.supports_legacy_writes():
        raise HTTPException(
            status_code=503,
            detail=(
                "This instance cannot process legacy writes. PRIMARY deployments with legacy canonical format "
                "must enable legacy write support."
            ),
        )

    if expected_format != canonical_format:
        raise HTTPException(
            status_code=503,
            detail=(
                "This deployment does not expose write operations for this API. "
                f"Expected canonical_format='{canonical_format}', got '{expected_format}'."
            ),
        )

assert_pending_queue_write_enabled 

assert_pending_queue_write_enabled(
    manager: ModelReferenceManager,
) -> None

Ensure pending-queue operations are allowed for the active canonical format.

Source code in src/horde_model_reference/service/shared.py 
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
def assert_pending_queue_write_enabled(manager: ModelReferenceManager) -> None:
    """Ensure pending-queue operations are allowed for the active canonical format."""
    backend = manager.backend
    canonical_format = horde_model_reference_settings.canonical_format

    if canonical_format == CanonicalFormat.v2:
        if not backend.supports_writes():
            raise HTTPException(
                status_code=503,
                detail=(
                    "This instance is in REPLICA mode and does not support write operations. "
                    "Only PRIMARY instances can queue model changes."
                ),
            )
        return

    if canonical_format == CanonicalFormat.LEGACY:
        if not backend.supports_legacy_writes():
            raise HTTPException(
                status_code=503,
                detail=(
                    "This instance cannot process legacy writes. PRIMARY deployments with legacy canonical "
                    "format must enable legacy write support."
                ),
            )
        return

    raise HTTPException(
        status_code=503,
        detail=(
            "Pending queue writes are not available for the configured canonical format. "
            f"canonical_format='{canonical_format}'"
        ),
    )