from abc import ABC, abstractmethod from collections.abc import AsyncGenerator, Awaitable, Callable from dataclasses import dataclass from typing import Optional, Type, TypeVar, BinaryIO, Union T = TypeVar("T") Fetcher = Callable[[str], Awaitable[Optional[Union[str, BinaryIO]]]] # Like public Fetcher type above but does not allow 'str' outputs. IOFetcher = Callable[[str], Awaitable[Optional[BinaryIO]]] ICON_FOLDER = "📂" ICON_PACKAGE = "📦" ICON_OPTICAL = "📀" ICON_QCOW = "🐮" ICON_OTHER = " " class ContentError(Exception): """An error raised when indexed content appears to be invalid. Errors of this type are raised when repo-autoindex is able to successfully retrieve content and determine a repository type but fails to parse repository metadata. For example, a corrupt yum repository may cause this error to be raised. """ class FetcherError(Exception): # Internal-only error used to separate exceptions raised by fetchers from # exceptions raised by anything else. pass @dataclass class GeneratedIndex: """A single HTML index page generated by repo-autoindex.""" content: str """The content of this index page (an HTML document).""" relative_dir: str = "." """The directory of this index page, relative to the root of the indexed repository. """ @dataclass class IndexEntry: href: str text: str time: str = "" size: str = "" padding: str = "" icon: str = ICON_OTHER @property def sort_key(self): # Returns a suggested sort key for displaying entries in # a UI. priority = 0 # Folders should come first if self.href.endswith("/"): priority -= 1 # And special folders like ".." even earlier if self.href.startswith("."): priority -= 1 # Entries sort by the priority we calculated, and then by name return (priority, self.href) class Repo(ABC): def __init__( self, base_url: str, entry_point_content: str, fetcher: IOFetcher, ): self.base_url = base_url self.entry_point_content = entry_point_content self.fetcher = fetcher @abstractmethod def render_index( self, index_href_suffix: str ) -> AsyncGenerator[GeneratedIndex, None]: pass # pragma: no cover @classmethod @abstractmethod async def probe(cls: Type[T], fetcher: IOFetcher, url: str) -> Optional[T]: """Determine if a specified URL seems to point at a repository of this type. If so, returns an initialized Repo of a concrete subtype. If not, returns None. """ pass # pragma: no cover