Library Provider API

def __eq__(self, other: object) -> bool:
    """Compare entities by provider namespace and key."""
    if other.__class__ is not self.__class__:
        return NotImplemented
    other_ent = cast(LibraryEntity, other)
    return (
        getattr(self._provider, "NAMESPACE", None)
        == getattr(other_ent._provider, "NAMESPACE", None)
        and self.key == other_ent.key
    )

def __hash__(self) -> int:
    """Compute a hash based on provider namespace, class name, and key."""
    return hash((self._provider.NAMESPACE, self.__class__.__name__, self.key))

def __repr__(self) -> str:
    """Return a short representation for debugging."""
    return f"<{self.__class__.__name__}:{self.key}:{self.title[:32]}>"

def provider(self) -> ProviderT:
    """Return the provider associated with this entity."""
    return self._provider

def __eq__(self, other: object) -> bool:
    """Compare entities by provider namespace and key."""
    if other.__class__ is not self.__class__:
        return NotImplemented
    other_ent = cast(LibraryEntity, other)
    return (
        getattr(self._provider, "NAMESPACE", None)
        == getattr(other_ent._provider, "NAMESPACE", None)
        and self.key == other_ent.key
    )

def __hash__(self) -> int:
    """Compute a hash based on provider namespace, class name, and key."""
    return hash((self._provider.NAMESPACE, self.__class__.__name__, self.key))

def __repr__(self) -> str:
    """Return a short representation for debugging."""
    return f"<{self.__class__.__name__}:{self.key}:{self.title[:32]}>"

@abstractmethod
async def history(self) -> Sequence[HistoryEntry]:
    """Return user history entries for this media item (tz-aware timestamps)."""
    ...

@abstractmethod
def mapping_descriptors(self) -> Sequence[MappingDescriptor]:
    """Return possible mapping descriptors for the media item.

    The returned descriptors refer to providers in the mapping database. They are
    not related to the library provider of this item (although, they might match).
    """
    ...

@abstractmethod
def media(self) -> LibraryMedia[ProviderT]:
    """Return the media item associated with this entry."""
    ...

def provider(self) -> ProviderT:
    """Return the provider associated with this entity."""
    return self._provider

@abstractmethod
def section(self) -> LibrarySection[LibraryProviderT]:
    """Return the parent library section for this media item."""
    ...

def __eq__(self, other: object) -> bool:
    """Compare entities by provider namespace and key."""
    if other.__class__ is not self.__class__:
        return NotImplemented
    other_ent = cast(LibraryEntity, other)
    return (
        getattr(self._provider, "NAMESPACE", None)
        == getattr(other_ent._provider, "NAMESPACE", None)
        and self.key == other_ent.key
    )

def __hash__(self) -> int:
    """Compute a hash based on provider namespace, class name, and key."""
    return hash((self._provider.NAMESPACE, self.__class__.__name__, self.key))

def __repr__(self) -> str:
    """Short representation including show title, season and episode indexes."""
    return (
        f"<{self.__class__.__name__}:{self.key}:{self.show().title[:32]}:"
        f"S{self.season_index:02d}E{self.index:02d}>"
    )

@abstractmethod
async def history(self) -> Sequence[HistoryEntry]:
    """Return user history entries for this media item (tz-aware timestamps)."""
    ...

@abstractmethod
def mapping_descriptors(self) -> Sequence[MappingDescriptor]:
    """Return possible mapping descriptors for the media item.

    The returned descriptors refer to providers in the mapping database. They are
    not related to the library provider of this item (although, they might match).
    """
    ...

@abstractmethod
def media(self) -> LibraryMedia[ProviderT]:
    """Return the media item associated with this entry."""
    ...

def provider(self) -> ProviderT:
    """Return the provider associated with this entity."""
    return self._provider

@abstractmethod
def season(self) -> LibrarySeason[LibraryProviderT]:
    """Get the parent season of the episode.

    Returns:
        LibrarySeason: The parent season.
    """
    ...

@abstractmethod
def section(self) -> LibrarySection[LibraryProviderT]:
    """Return the parent library section for this media item."""
    ...

@abstractmethod
def show(self) -> LibraryShow[LibraryProviderT]:
    """Get the parent show of the episode.

    Returns:
        LibraryShow: The parent show.
    """
    ...

def __eq__(self, other: object) -> bool:
    """Compare entities by provider namespace and key."""
    if other.__class__ is not self.__class__:
        return NotImplemented
    other_ent = cast(LibraryEntity, other)
    return (
        getattr(self._provider, "NAMESPACE", None)
        == getattr(other_ent._provider, "NAMESPACE", None)
        and self.key == other_ent.key
    )

def __hash__(self) -> int:
    """Compute a hash based on provider namespace, class name, and key."""
    return hash((self._provider.NAMESPACE, self.__class__.__name__, self.key))

def __repr__(self) -> str:
    """Return a short representation for debugging."""
    return f"<{self.__class__.__name__}:{self.key}:{self.title[:32]}>"

def provider(self) -> ProviderT:
    """Return the provider associated with this entity."""
    return self._provider

def __eq__(self, other: object) -> bool:
    """Compare entities by provider namespace and key."""
    if other.__class__ is not self.__class__:
        return NotImplemented
    other_ent = cast(LibraryEntity, other)
    return (
        getattr(self._provider, "NAMESPACE", None)
        == getattr(other_ent._provider, "NAMESPACE", None)
        and self.key == other_ent.key
    )

def __hash__(self) -> int:
    """Compute a hash based on provider namespace, class name, and key."""
    return hash((self._provider.NAMESPACE, self.__class__.__name__, self.key))

def __repr__(self) -> str:
    """Return a short representation for debugging."""
    return f"<{self.__class__.__name__}:{self.key}:{self.title[:32]}>"

@abstractmethod
async def history(self) -> Sequence[HistoryEntry]:
    """Return user history entries for this media item (tz-aware timestamps)."""
    ...

@abstractmethod
def mapping_descriptors(self) -> Sequence[MappingDescriptor]:
    """Return possible mapping descriptors for the media item.

    The returned descriptors refer to providers in the mapping database. They are
    not related to the library provider of this item (although, they might match).
    """
    ...

@abstractmethod
def media(self) -> LibraryMedia[ProviderT]:
    """Return the media item associated with this entry."""
    ...

def provider(self) -> ProviderT:
    """Return the provider associated with this entity."""
    return self._provider

@abstractmethod
def section(self) -> LibrarySection[LibraryProviderT]:
    """Return the parent library section for this media item."""
    ...

def __init__(self, *, config: dict | None = None) -> None:
    """Initialize the provider with optional configuration.

    Args:
        config (dict | None): Any configuration options that were detected with the
            provider's namespace as a prefix.
    """
    return None

async def clear_cache(self) -> None:
    """Clear any cached data held by the provider.

    For more efficient implementations, it is a good idea to cache data
    fetched from the provider to minimize network requests. AniBridge uses
    this method occasionally to clear such caches.
    """
    return None

async def close(self) -> None:
    """Close the provider and release resources."""
    return None

@abstractmethod
async def get_sections(self) -> Sequence[LibrarySection[LibraryProviderT]]:
    """Return available library sections for the provider.

    Returns:
        Sequence[LibrarySection[LibraryProviderT]]: The available library sections.
    """
    ...

async def initialize(self) -> None:
    """Asynchronous initialization hook.

    Put any async logic that should be run after construction here.
    """
    return None

@abstractmethod
async def list_items(
    self,
    section: LibrarySection[LibraryProviderT],
    *,
    min_last_modified: datetime | None = None,
    require_watched: bool = False,
    keys: Sequence[str] | None = None,
) -> Sequence[LibraryEntry[LibraryProviderT]]:
    """List items in a library section with optional filtering.

    Args:
        section (LibrarySection[LibraryProviderT]): The library section to list
            items from.
        min_last_modified (datetime | None): If provided, only include items
            modified after this timezone-aware timestamp.
        require_watched (bool): If True, only include items marked as watched.
        keys (Sequence[str] | None): If provided, only include items whose keys are
            in this sequence.

    Returns:
        Sequence[LibraryEntry[LibraryProviderT]]: The list of library items.
    """
    ...

async def parse_webhook(self, request: Request) -> tuple[bool, Sequence[str]]:
    """Parse an incoming webhook and return the affected item keys.

    Args:
        request (Request): The incoming HTTP request.

    Returns:
        tuple[bool, Sequence[str]]: A tuple where the first element indicates
            whether the webhook applies to the current provider, and the second
            element is a sequence of affected item keys.
    """
    return False, ()

@abstractmethod
def user(self) -> LibraryUser | None:
    """Return the associated user object, if any.

    Returns:
        LibraryUser | None: The associated user object, if any.
    """
    ...

def __contains__(self, namespace: object) -> bool:
    """Check if a provider is registered under `namespace`."""
    return isinstance(namespace, str) and namespace in self._providers

def __init__(self) -> None:
    """Initialize an empty registry."""
    self._providers: dict[str, type[LibraryProvider]] = {}

def __iter__(self) -> Iterator[tuple[str, type[LibraryProvider]]]:
    """Iterate over `(namespace, provider_class)` pairs."""
    return iter(self._providers.items())

def clear(self) -> None:
    """Remove all provider registrations."""
    self._providers.clear()

def create(self, namespace: str, *, config: dict | None = None) -> LibraryProvider:
    """Instantiate the provider registered under `namespace`.

    Args:
        namespace (str): The namespace identifier to create.
        config (dict | None): Optional configuration dictionary to pass to the
            provider constructor.

    Returns:
        LibraryProvider: An instance of the registered provider.
    """
    provider_cls = self.get(namespace)
    return provider_cls(config=config)

def get(self, namespace: str) -> type[LibraryProvider]:
    """Return the provider class registered under `namespace`.

    Args:
        namespace (str): The namespace identifier to look up.

    Returns:
        type[LibraryProvider]: The registered provider class.
    """
    try:
        return self._providers[namespace]
    except KeyError as exc:
        raise LookupError(
            f"No provider registered for namespace '{namespace}'."
        ) from exc

def namespaces(self) -> tuple[str, ...]:
    """Return a tuple of registered namespace identifiers.

    Returns:
        tuple[str, ...]: The registered namespace identifiers.
    """
    return tuple(self._providers)

def register(
    self,
    provider_cls: type[LP] | None = None,
    *,
    namespace: str | None = None,
) -> type[LP] | Callable[[type[LP]], type[LP]]:
    """Register a provider class, optionally used as a decorator.

    Args:
        provider_cls (type[LP] | None): The provider class to register. If `None`,
            the method acts as a decorator factory.
        namespace (str | None): Explicit namespace override. Defaults to the class'
            `NAMESPACE` attribute.

    Returns:
        type[LP] | Callable[[type[LP]], type[LP]]: The registered provider class, or
            a decorator that registers the class.
    """

    def decorator(cls: type[LP]) -> type[LP]:
        """Register `cls` as a provider."""
        resolved_namespace = namespace or getattr(cls, "NAMESPACE", None)
        if not isinstance(resolved_namespace, str) or not resolved_namespace:
            raise ValueError(
                "Library providers must define a non-empty string `NAMESPACE` "
                "attribute or pass `namespace=` when registering."
            )

        existing = self._providers.get(resolved_namespace)
        if existing is not None and existing is not cls:
            raise ValueError(
                f"A provider is already registered for namespace "
                f"'{resolved_namespace}'."
            )

        self._providers[resolved_namespace] = cls
        return cls

    if provider_cls is None:
        return decorator
    return decorator(provider_cls)