Skip to content

base

Abstract interface for third-party model providers.

A :class:ModelProvider lets a third party contribute model records for one or more :class:~horde_model_reference.meta_consts.MODEL_REFERENCE_CATEGORY values without going through the canonical write loop. Providers are read-only from the library's perspective: any persistence, validation-on-write, or moderation is the third party's own responsibility.

Providers return already-validated Pydantic records (subclasses of :class:~horde_model_reference.model_reference_records.GenericModelRecord). A provider may reuse the built-in record type for a category (:data:~horde_model_reference.model_reference_records.MODEL_RECORD_TYPE_LOOKUP) or register and return its own subclass via :func:~horde_model_reference.model_reference_records.register_record_type.

ModelProvider

Bases: ABC

Read-only third-party source of model records for one or more categories.

Subclasses must implement :attr:source_id, :meth:provided_categories, and :meth:fetch_category. The async variant has a thread-pool default and only needs overriding when a native async implementation is available.

Implementations should:

  • Return a mapping of model_name -> record from :meth:fetch_category, or None when the category cannot be served right now.
  • Validate/construct records themselves (the library does not re-validate provider output against any schema).
  • Be resilient: raising from :meth:fetch_category is tolerated by the manager (the error is logged and that provider is skipped), but returning None is the preferred way to signal "no data".
Source code in src/horde_model_reference/providers/base.py 
 27
 28
 29
 30
 31
 32
 33
 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
class ModelProvider(ABC):
    """Read-only third-party source of model records for one or more categories.

    Subclasses must implement :attr:`source_id`, :meth:`provided_categories`, and
    :meth:`fetch_category`. The async variant has a thread-pool default and only
    needs overriding when a native async implementation is available.

    Implementations should:

    * Return a mapping of ``model_name -> record`` from :meth:`fetch_category`,
      or ``None`` when the category cannot be served right now.
    * Validate/construct records themselves (the library does not re-validate
      provider output against any schema).
    * Be resilient: raising from :meth:`fetch_category` is tolerated by the
      manager (the error is logged and that provider is skipped), but returning
      ``None`` is the preferred way to signal "no data".
    """

    @property
    @abstractmethod
    def source_id(self) -> str:
        """Return this provider's stable, unique source id.

        Must not be one of :data:`~horde_model_reference.source_consts.RESERVED_SOURCE_IDS`
        (``"horde"`` / ``"any"``). The id is how consumers select this provider in
        queries, so it should be stable across versions (e.g. ``"civitai"``).
        """

    @abstractmethod
    def provided_categories(self) -> set[MODEL_REFERENCE_CATEGORY | str]:
        """Return the set of categories this provider can serve."""

    @abstractmethod
    def fetch_category(
        self,
        category: MODEL_REFERENCE_CATEGORY | str,
        *,
        force_refresh: bool = False,
    ) -> dict[str, GenericModelRecord] | None:
        """Return ``model_name -> record`` for *category*, or ``None`` if unavailable.

        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.

        """

    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,
        )

    def serves_category(self, category: MODEL_REFERENCE_CATEGORY | str) -> bool:
        """Return whether this provider advertises support for *category*."""
        return category in self.provided_categories()

    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

    def cache_ttl_seconds(self) -> int | None:
        """Return a staleness hint in seconds, or ``None`` for no hint.

        This is advisory metadata for consumers; the library does not currently
        cache provider output itself.
        """
        return None

    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)}.",
            )

source_id abstractmethod property 

source_id: str

Return this provider's stable, unique source id.

Must not be one of :data:~horde_model_reference.source_consts.RESERVED_SOURCE_IDS ("horde" / "any"). The id is how consumers select this provider in queries, so it should be stable across versions (e.g. "civitai").

provided_categories abstractmethod 

provided_categories() -> set[
    MODEL_REFERENCE_CATEGORY | str
]

Return the set of categories this provider can serve.

Source code in src/horde_model_reference/providers/base.py 
55
56
57
@abstractmethod
def provided_categories(self) -> set[MODEL_REFERENCE_CATEGORY | str]:
    """Return the set of categories this provider can serve."""

fetch_category abstractmethod 

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

Return model_name -> record for category, or None if unavailable.

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 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@abstractmethod
def fetch_category(
    self,
    category: MODEL_REFERENCE_CATEGORY | str,
    *,
    force_refresh: bool = False,
) -> dict[str, GenericModelRecord] | None:
    """Return ``model_name -> record`` for *category*, or ``None`` if unavailable.

    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.

    """

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

cache_ttl_seconds 

cache_ttl_seconds() -> int | None

Return a staleness hint in seconds, or None for no hint.

This is advisory metadata for consumers; the library does not currently cache provider output itself.

Source code in src/horde_model_reference/providers/base.py 
116
117
118
119
120
121
122
def cache_ttl_seconds(self) -> int | None:
    """Return a staleness hint in seconds, or ``None`` for no hint.

    This is advisory metadata for consumers; the library does not currently
    cache provider output itself.
    """
    return None

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)}.",
        )