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

READ_ERROR_RESPONSES module-attribute 

READ_ERROR_RESPONSES: dict[int | str, dict[str, object]] = {
    404: {
        "model": ErrorResponse,
        "description": "The requested category or model was not found.",
    },
    422: {
        "model": ErrorResponse,
        "description": "The category name or query parameters were invalid.",
    },
    503: {
        "model": ErrorResponse,
        "description": "A required capability (e.g. metadata) is unavailable in this deployment mode.",
    },
}

WRITE_ERROR_RESPONSES module-attribute 

WRITE_ERROR_RESPONSES: dict[
    int | str, dict[str, object]
] = {
    400: {
        "model": ErrorResponse,
        "description": "The request was malformed or violated a state precondition.",
    },
    401: {
        "model": ErrorResponse,
        "description": "Missing or invalid ``apikey``, or the user lacks the required role.",
    },
    404: {
        "model": ErrorResponse,
        "description": "The target category or model was not found.",
    },
    409: {
        "model": ErrorResponse,
        "description": "The model already exists, or conflicts with another entry.",
    },
    422: {
        "model": ErrorResponse,
        "description": "The request body failed validation.",
    },
    503: {
        "model": ErrorResponse,
        "description": "Writes are unavailable: REPLICA mode, or the canonical format does not match this API.",
    },
}

_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 
33
34
35
36
37
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 
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
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 
 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
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 
90
91
92
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 
 94
 95
 96
 97
 98
 99
100
101
102
103
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 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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 
133
134
135
136
137
138
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 
145
146
147
148
149
150
@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 
367
368
369
370
371
372
373
374
375
376
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 
370
371
372
373
374
375
376
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 
379
380
381
382
383
384
385
386
387
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 
390
391
392
393
394
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 
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
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 
211
212
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 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
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 
234
235
236
237
238
239
240
241
242
243
244
245
246
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 
249
250
251
252
253
254
255
256
257
258
259
260
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.

Distinguishes invalid credentials (401) from a valid key that lacks the role (403) so clients can message each case correctly.

Raises:

  • APIKeyInvalidException –

    If no allowlist is configured or the key is invalid.

  • HTTPException –

    403 if the user is authenticated but not on the requestor allowlist; 503 if the Horde auth service is unreachable.

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

    Distinguishes invalid credentials (401) from a valid key that lacks the
    role (403) so clients can message each case correctly.

    Raises:
        APIKeyInvalidException: If no allowlist is configured or the key is invalid.
        HTTPException: 403 if the user is authenticated but not on the requestor
            allowlist; 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=None)
    if context is None:
        raise APIKeyInvalidException()
    if context.user_id not in allowlist:
        raise HTTPException(
            status_code=403,
            detail="You are not on the pending queue requestor list.",
        )
    return context

authenticate_queue_approver async 

authenticate_queue_approver(
    apikey: str,
) -> HordeUserContext

Authenticate a queue approver using the configured allowlist.

Distinguishes invalid credentials (401) from a valid key that lacks the role (403) so clients can message each case correctly.

Raises:

  • APIKeyInvalidException –

    If no allowlist is configured or the key is invalid.

  • HTTPException –

    403 if the user is authenticated but not on the approver allowlist; 503 if the Horde auth service is unreachable.

Source code in src/horde_model_reference/service/shared.py 
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
async def authenticate_queue_approver(apikey: str) -> HordeUserContext:
    """Authenticate a queue approver using the configured allowlist.

    Distinguishes invalid credentials (401) from a valid key that lacks the
    role (403) so clients can message each case correctly.

    Raises:
        APIKeyInvalidException: If no allowlist is configured or the key is invalid.
        HTTPException: 403 if the user is authenticated but not on the approver
            allowlist; 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=None)
    if context is None:
        raise APIKeyInvalidException()
    if context.user_id not in allowlist:
        raise HTTPException(
            status_code=403,
            detail="You are not on the pending queue approver list.",
        )
    logger.debug(f"Approver authenticated: user_id={context.user_id}")
    return context

authenticate_queue_reader async 

authenticate_queue_reader(apikey: str) -> HordeUserContext

Authenticate any valid Horde API key for read-only pending queue access.

The pending queue is a transparency surface: any authenticated user may view queued changes and audit history, while approve/reject/apply/purge operations remain gated by the approver allowlist.

Raises:

  • APIKeyInvalidException –

    If the key does not resolve to a Horde user.

  • HTTPException –

    503 if the Horde auth service is unreachable.

Source code in src/horde_model_reference/service/shared.py 
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
async def authenticate_queue_reader(apikey: str) -> HordeUserContext:
    """Authenticate any valid Horde API key for read-only pending queue access.

    The pending queue is a transparency surface: any authenticated user may view
    queued changes and audit history, while approve/reject/apply/purge operations
    remain gated by the approver allowlist.

    Raises:
        APIKeyInvalidException: If the key does not resolve to a Horde user.
        HTTPException: 503 if the Horde auth service is unreachable.
    """
    context = await auth_against_horde(apikey, httpx_client, allowed_user_ids=None)
    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 
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
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 
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
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 
447
448
449
def get_model_reference_manager() -> ModelReferenceManager:
    """Dependency helper that returns the singleton model reference manager."""
    return ModelReferenceManager()

assert_primary_mode 

assert_primary_mode(manager: ModelReferenceManager) -> None

Ensure the backend is in PRIMARY mode (supports writes), regardless of canonical format.

Use for metadata-only write operations (schemas, aliases, families) that don't modify model records and are not gated by the canonical format setting.

Source code in src/horde_model_reference/service/shared.py 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
def assert_primary_mode(manager: ModelReferenceManager) -> None:
    """Ensure the backend is in PRIMARY mode (supports writes), regardless of canonical format.

    Use for metadata-only write operations (schemas, aliases, families) that don't
    modify model records and are not gated by the canonical format setting.
    """
    if not manager.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 modify data."
            ),
        )

_category_is_v2_only 

_category_is_v2_only(
    category: MODEL_REFERENCE_CATEGORY | str | None,
) -> bool

Return whether category has no legacy representation, so v2 is its only possible write path.

Such a category (has_legacy_format=False, e.g. controlnet_annotator) has no v1 endpoint, so gating its writes behind canonical_format='v2' would make it permanently unwritable on a legacy-canonical PRIMARY. These categories are therefore writable via v2 regardless of the deployment's canonical format.

Source code in src/horde_model_reference/service/shared.py 
468
469
470
471
472
473
474
475
476
477
478
479
480
def _category_is_v2_only(category: MODEL_REFERENCE_CATEGORY | str | None) -> bool:
    """Return whether *category* has no legacy representation, so v2 is its only possible write path.

    Such a category (``has_legacy_format=False``, e.g. ``controlnet_annotator``) has no v1 endpoint, so gating
    its writes behind ``canonical_format='v2'`` would make it permanently unwritable on a legacy-canonical
    PRIMARY. These categories are therefore writable via v2 regardless of the deployment's canonical format.
    """
    if category is None:
        return False
    try:
        return get_category_descriptor(category).has_legacy_format is False
    except KeyError:
        return False

assert_canonical_write_enabled 

assert_canonical_write_enabled(
    manager: ModelReferenceManager,
    *,
    canonical_format: CanonicalFormat,
    category: MODEL_REFERENCE_CATEGORY | str | None = None,
) -> None

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

A category with no legacy representation (has_legacy_format=False) is exempt from the v2 canonical-format requirement: v2 is its only write path, so it is writable on any PRIMARY.

Source code in src/horde_model_reference/service/shared.py 
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
def assert_canonical_write_enabled(
    manager: ModelReferenceManager,
    *,
    canonical_format: CanonicalFormat,
    category: MODEL_REFERENCE_CATEGORY | str | None = None,
) -> None:
    """Ensure that writes are attempted only when the canonical format allows them.

    A category with no legacy representation (``has_legacy_format=False``) is exempt from the v2
    canonical-format requirement: v2 is its only write path, so it is writable on any PRIMARY.
    """
    assert_primary_mode(manager)

    backend = manager.backend
    expected_format = horde_model_reference_settings.canonical_format
    if canonical_format == CanonicalFormat.v2:
        if expected_format != canonical_format and not _category_is_v2_only(category):
            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 
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
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}'"
        ),
    )