Skip to content

static_provider

A ready-to-use, in-memory :class:ModelProvider.

Most third parties already have their records in hand (or as plain dicts) and only need to expose them under a source id. :class:StaticModelProvider covers that case without requiring a custom subclass: construct it with already-built records, or use :meth:StaticModelProvider.from_raw to validate plain dicts against each category's record type. Subclass :class:~horde_model_reference.providers.base.ModelProvider directly only when records must be fetched lazily from a live/remote source.

StaticModelProvider

Bases: ModelProvider

A :class:ModelProvider backed by a fixed, in-memory set of records.

This is the lowest-friction way to contribute model records: hand it the records you already have and register it. The provider is read-only and never refreshes (force_refresh is a no-op), which is exactly what you want for a static snapshot.

Example 
from horde_model_reference import (
    ModelReferenceManager,
    MODEL_REFERENCE_CATEGORY,
    StaticModelProvider,
)

provider = StaticModelProvider.from_raw(
    "civitai",
    {
        MODEL_REFERENCE_CATEGORY.image_generation: {
            "my_model": {"baseline": "stable_diffusion_xl", "nsfw": False},
        },
    },
)
ModelReferenceManager().register_provider(provider)
Source code in src/horde_model_reference/providers/static_provider.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class StaticModelProvider(ModelProvider):
    """A :class:`ModelProvider` backed by a fixed, in-memory set of records.

    This is the lowest-friction way to contribute model records: hand it the records
    you already have and register it. The provider is read-only and never refreshes
    (``force_refresh`` is a no-op), which is exactly what you want for a static
    snapshot.

    Example:
        ```python
        from horde_model_reference import (
            ModelReferenceManager,
            MODEL_REFERENCE_CATEGORY,
            StaticModelProvider,
        )

        provider = StaticModelProvider.from_raw(
            "civitai",
            {
                MODEL_REFERENCE_CATEGORY.image_generation: {
                    "my_model": {"baseline": "stable_diffusion_xl", "nsfw": False},
                },
            },
        )
        ModelReferenceManager().register_provider(provider)
        ```
    """

    def __init__(
        self,
        source_id: str,
        records_by_category: Mapping[MODEL_REFERENCE_CATEGORY | str, Mapping[str, GenericModelRecord]],
        *,
        cache_ttl_seconds: int | None = None,
    ) -> None:
        """Store already-built records to serve under *source_id*.

        Args:
            source_id: This provider's stable, unique id (e.g. ``"civitai"``). Must not
                be a reserved id (``"horde"`` / ``"any"``); validated on construction.
            records_by_category: Mapping of category to ``model_name -> record``. The
                records must already be validated instances of the appropriate
                :class:`~horde_model_reference.model_reference_records.GenericModelRecord`
                subclass (use :meth:`from_raw` to build them from plain dicts).
            cache_ttl_seconds: Optional advisory staleness hint reported via
                :meth:`cache_ttl_seconds`.

        Raises:
            ValueError: If *source_id* is empty or reserved.

        """
        self._source_id = source_id
        self._records: dict[MODEL_REFERENCE_CATEGORY | str, dict[str, GenericModelRecord]] = {
            category: dict(records) for category, records in records_by_category.items()
        }
        self._cache_ttl_seconds = cache_ttl_seconds
        self.validate_source_id()

    @classmethod
    def from_raw(
        cls,
        source_id: str,
        raw_by_category: Mapping[MODEL_REFERENCE_CATEGORY | str, Mapping[str, Mapping[str, object]]],
        *,
        cache_ttl_seconds: int | None = None,
    ) -> StaticModelProvider:
        """Build a provider by validating plain dicts against each category's record type.

        Each raw record is validated with the record class registered for its category
        (:data:`~horde_model_reference.model_reference_records.MODEL_RECORD_TYPE_LOOKUP`,
        falling back to ``GenericModelRecord``). The model name from the mapping key is
        injected as the record's ``name`` so callers need not repeat it.

        Args:
            source_id: This provider's stable, unique id (e.g. ``"civitai"``).
            raw_by_category: Mapping of category to ``model_name -> raw field dict``.
            cache_ttl_seconds: Optional advisory staleness hint.

        Returns:
            StaticModelProvider: A provider serving the validated records.

        Raises:
            pydantic.ValidationError: If any raw record fails validation.
            ValueError: If *source_id* is empty or reserved.

        """
        records_by_category: dict[MODEL_REFERENCE_CATEGORY | str, dict[str, GenericModelRecord]] = {}
        for category, raw_records in raw_by_category.items():
            record_cls = _record_type_for(category)
            records_by_category[category] = {
                name: record_cls.model_validate({**raw, "name": name}) for name, raw in raw_records.items()
            }
        return cls(source_id, records_by_category, cache_ttl_seconds=cache_ttl_seconds)

    @property
    @override
    def source_id(self) -> str:
        """Return this provider's source id."""
        return self._source_id

    @override
    def provided_categories(self) -> set[MODEL_REFERENCE_CATEGORY | str]:
        """Return the categories this provider holds records for."""
        return set(self._records)

    @override
    def fetch_category(
        self,
        category: MODEL_REFERENCE_CATEGORY | str,
        *,
        force_refresh: bool = False,
    ) -> dict[str, GenericModelRecord] | None:
        """Return a copy of the stored records for *category*, or ``None`` if absent."""
        records = self._records.get(category)
        if records is None:
            return None
        return dict(records)

    @override
    def cache_ttl_seconds(self) -> int | None:
        """Return the advisory staleness hint set at construction, if any."""
        return self._cache_ttl_seconds

_source_id instance-attribute 

_source_id = source_id

_records instance-attribute 

_records: dict[
    MODEL_REFERENCE_CATEGORY | str,
    dict[str, GenericModelRecord],
] = {
    category: (dict(records))
    for category, records in (items())
}

_cache_ttl_seconds instance-attribute 

_cache_ttl_seconds = cache_ttl_seconds

source_id property 

source_id: str

Return this provider's source id.

__init__ 

__init__(
    source_id: str,
    records_by_category: Mapping[
        MODEL_REFERENCE_CATEGORY | str,
        Mapping[str, GenericModelRecord],
    ],
    *,
    cache_ttl_seconds: int | None = None,
) -> None

Store already-built records to serve under source_id.

Parameters:

  • source_id (str) –

    This provider's stable, unique id (e.g. "civitai"). Must not be a reserved id ("horde" / "any"); validated on construction.

  • records_by_category (Mapping[MODEL_REFERENCE_CATEGORY | str, Mapping[str, GenericModelRecord]]) –

    Mapping of category to model_name -> record. The records must already be validated instances of the appropriate :class:~horde_model_reference.model_reference_records.GenericModelRecord subclass (use :meth:from_raw to build them from plain dicts).

  • cache_ttl_seconds (int | None, default: None ) –

    Optional advisory staleness hint reported via :meth:cache_ttl_seconds.

Raises:

  • ValueError –

    If source_id is empty or reserved.

Source code in src/horde_model_reference/providers/static_provider.py 
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
def __init__(
    self,
    source_id: str,
    records_by_category: Mapping[MODEL_REFERENCE_CATEGORY | str, Mapping[str, GenericModelRecord]],
    *,
    cache_ttl_seconds: int | None = None,
) -> None:
    """Store already-built records to serve under *source_id*.

    Args:
        source_id: This provider's stable, unique id (e.g. ``"civitai"``). Must not
            be a reserved id (``"horde"`` / ``"any"``); validated on construction.
        records_by_category: Mapping of category to ``model_name -> record``. The
            records must already be validated instances of the appropriate
            :class:`~horde_model_reference.model_reference_records.GenericModelRecord`
            subclass (use :meth:`from_raw` to build them from plain dicts).
        cache_ttl_seconds: Optional advisory staleness hint reported via
            :meth:`cache_ttl_seconds`.

    Raises:
        ValueError: If *source_id* is empty or reserved.

    """
    self._source_id = source_id
    self._records: dict[MODEL_REFERENCE_CATEGORY | str, dict[str, GenericModelRecord]] = {
        category: dict(records) for category, records in records_by_category.items()
    }
    self._cache_ttl_seconds = cache_ttl_seconds
    self.validate_source_id()

from_raw classmethod 

from_raw(
    source_id: str,
    raw_by_category: Mapping[
        MODEL_REFERENCE_CATEGORY | str,
        Mapping[str, Mapping[str, object]],
    ],
    *,
    cache_ttl_seconds: int | None = None,
) -> StaticModelProvider

Build a provider by validating plain dicts against each category's record type.

Each raw record is validated with the record class registered for its category (:data:~horde_model_reference.model_reference_records.MODEL_RECORD_TYPE_LOOKUP, falling back to GenericModelRecord). The model name from the mapping key is injected as the record's name so callers need not repeat it.

Parameters:

Returns:

Raises:

  • ValidationError –

    If any raw record fails validation.

  • ValueError –

    If source_id is empty or reserved.

Source code in src/horde_model_reference/providers/static_provider.py 
 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
@classmethod
def from_raw(
    cls,
    source_id: str,
    raw_by_category: Mapping[MODEL_REFERENCE_CATEGORY | str, Mapping[str, Mapping[str, object]]],
    *,
    cache_ttl_seconds: int | None = None,
) -> StaticModelProvider:
    """Build a provider by validating plain dicts against each category's record type.

    Each raw record is validated with the record class registered for its category
    (:data:`~horde_model_reference.model_reference_records.MODEL_RECORD_TYPE_LOOKUP`,
    falling back to ``GenericModelRecord``). The model name from the mapping key is
    injected as the record's ``name`` so callers need not repeat it.

    Args:
        source_id: This provider's stable, unique id (e.g. ``"civitai"``).
        raw_by_category: Mapping of category to ``model_name -> raw field dict``.
        cache_ttl_seconds: Optional advisory staleness hint.

    Returns:
        StaticModelProvider: A provider serving the validated records.

    Raises:
        pydantic.ValidationError: If any raw record fails validation.
        ValueError: If *source_id* is empty or reserved.

    """
    records_by_category: dict[MODEL_REFERENCE_CATEGORY | str, dict[str, GenericModelRecord]] = {}
    for category, raw_records in raw_by_category.items():
        record_cls = _record_type_for(category)
        records_by_category[category] = {
            name: record_cls.model_validate({**raw, "name": name}) for name, raw in raw_records.items()
        }
    return cls(source_id, records_by_category, cache_ttl_seconds=cache_ttl_seconds)

provided_categories 

provided_categories() -> set[
    MODEL_REFERENCE_CATEGORY | str
]

Return the categories this provider holds records for.

Source code in src/horde_model_reference/providers/static_provider.py 
134
135
136
137
@override
def provided_categories(self) -> set[MODEL_REFERENCE_CATEGORY | str]:
    """Return the categories this provider holds records for."""
    return set(self._records)

fetch_category 

fetch_category(
    category: MODEL_REFERENCE_CATEGORY | str,
    *,
    force_refresh: bool = False,
) -> dict[str, GenericModelRecord] | None

Return a copy of the stored records for category, or None if absent.

Source code in src/horde_model_reference/providers/static_provider.py 
139
140
141
142
143
144
145
146
147
148
149
150
@override
def fetch_category(
    self,
    category: MODEL_REFERENCE_CATEGORY | str,
    *,
    force_refresh: bool = False,
) -> dict[str, GenericModelRecord] | None:
    """Return a copy of the stored records for *category*, or ``None`` if absent."""
    records = self._records.get(category)
    if records is None:
        return None
    return dict(records)

cache_ttl_seconds 

cache_ttl_seconds() -> int | None

Return the advisory staleness hint set at construction, if any.

Source code in src/horde_model_reference/providers/static_provider.py 
152
153
154
155
@override
def cache_ttl_seconds(self) -> int | None:
    """Return the advisory staleness hint set at construction, if any."""
    return self._cache_ttl_seconds

fetch_category_async async 

fetch_category_async(
    category: MODEL_REFERENCE_CATEGORY | str,
    *,
    force_refresh: bool = False,
) -> dict[str, GenericModelRecord] | None

Asynchronously fetch records for category.

The default implementation runs :meth:fetch_category in a worker thread. Override when a native async data path is available.

Parameters:

Returns:

  • dict[str, GenericModelRecord] | None –

    dict[str, GenericModelRecord] | None: Validated records keyed by model name, or None when the provider cannot serve this category.

Source code in src/horde_model_reference/providers/base.py 
 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
async def fetch_category_async(
    self,
    category: MODEL_REFERENCE_CATEGORY | str,
    *,
    force_refresh: bool = False,
) -> dict[str, GenericModelRecord] | None:
    """Asynchronously fetch records for *category*.

    The default implementation runs :meth:`fetch_category` in a worker thread.
    Override when a native async data path is available.

    Args:
        category: The category to fetch.
        force_refresh: If ``True``, bypass any provider-side cache.

    Returns:
        dict[str, GenericModelRecord] | None: Validated records keyed by model
            name, or ``None`` when the provider cannot serve this category.

    """
    return await asyncio.to_thread(
        self.fetch_category,
        category,
        force_refresh=force_refresh,
    )

serves_category 

serves_category(
    category: MODEL_REFERENCE_CATEGORY | str,
) -> bool

Return whether this provider advertises support for category.

Source code in src/horde_model_reference/providers/base.py 
104
105
106
def serves_category(self, category: MODEL_REFERENCE_CATEGORY | str) -> bool:
    """Return whether this provider advertises support for *category*."""
    return category in self.provided_categories()

supports_writes 

supports_writes() -> bool

Return whether this provider supports write operations.

Always False: third-party write semantics are explicitly out of scope. The hook exists so consumers can branch on capability uniformly.

Source code in src/horde_model_reference/providers/base.py 
108
109
110
111
112
113
114
def supports_writes(self) -> bool:
    """Return whether this provider supports write operations.

    Always ``False``: third-party write semantics are explicitly out of scope.
    The hook exists so consumers can branch on capability uniformly.
    """
    return False

validate_source_id 

validate_source_id() -> None

Raise ValueError if :attr:source_id is empty or reserved.

Called by the registry at registration time.

Source code in src/horde_model_reference/providers/base.py 
124
125
126
127
128
129
130
131
132
133
134
def validate_source_id(self) -> None:
    """Raise ``ValueError`` if :attr:`source_id` is empty or reserved.

    Called by the registry at registration time.
    """
    if not self.source_id:
        raise ValueError("Provider source_id must be a non-empty string.")
    if self.source_id in RESERVED_SOURCE_IDS:
        raise ValueError(
            f"Provider source_id {self.source_id!r} is reserved. Reserved ids: {sorted(RESERVED_SOURCE_IDS)}.",
        )

_record_type_for 

_record_type_for(
    category: MODEL_REFERENCE_CATEGORY | str,
) -> type[GenericModelRecord]

Return the record class registered for category, or GenericModelRecord.

Unknown category strings (a provider may define its own) fall back to :class:~horde_model_reference.model_reference_records.GenericModelRecord.

Source code in src/horde_model_reference/providers/static_provider.py 
21
22
23
24
25
26
27
28
29
30
31
def _record_type_for(category: MODEL_REFERENCE_CATEGORY | str) -> type[GenericModelRecord]:
    """Return the record class registered for *category*, or ``GenericModelRecord``.

    Unknown category strings (a provider may define its own) fall back to
    :class:`~horde_model_reference.model_reference_records.GenericModelRecord`.
    """
    try:
        enum_category = MODEL_REFERENCE_CATEGORY(category)
    except ValueError:
        return GenericModelRecord
    return MODEL_RECORD_TYPE_LOOKUP.get(enum_category, GenericModelRecord)