omnipy.components.remote.tasks

Tasks for fetching remote resources and building URL datasets.

CLASS	DESCRIPTION
GithubRepoContext	Describe a GitHub repository location used for URL generation.

FUNCTION	DESCRIPTION
async_get_github_repo_urls	Asynchronously create raw GitHub content URLs for repository files.
async_load_urls_into_new_dataset	Asynchronously load remote URLs into a newly created dataset instance.
get_auto_from_api_endpoint	Fetch an API endpoint and decode the response from its MIME type.
get_bytes_from_api_endpoint	Fetch an API endpoint and decode the response body as raw bytes.
get_github_repo_urls	Create raw GitHub content URLs for one file or matching files in a repository path.
get_json_from_api_endpoint	Fetch a JSON API endpoint and decode the response body as JSON.
get_retry_client	Build a retry-enabled HTTP client wrapper.
get_str_from_api_endpoint	Fetch an API endpoint and decode the response body as text.
load_urls_into_new_dataset	Load remote URLs into a newly created dataset instance.

GithubRepoContext dataclass

Describe a GitHub repository location used for URL generation.

PARAMETER	DESCRIPTION
owner	GitHub repository owner or organization. TYPE: str
repo	Repository name. TYPE: str
branch	Branch or reference to read from. TYPE: str
path	File or directory path inside the repository. TYPE: str | Path

RETURNS	DESCRIPTION
GithubRepoContext	Repository context container.

RAISES	DESCRIPTION
TypeError	If field values do not match declared types.

Example

ctx = GithubRepoContext('octocat', 'hello-world', 'main', 'docs') ctx.repo 'hello-world'

METHOD	DESCRIPTION
__init__

ATTRIBUTE	DESCRIPTION
branch	TYPE: str
owner	TYPE: str
path	TYPE: str | Path
repo	TYPE: str

Source code in src/omnipy/components/remote/tasks.py

@dataclass
class GithubRepoContext:
    """Describe a GitHub repository location used for URL generation.

    Args:
        owner: GitHub repository owner or organization.
        repo: Repository name.
        branch: Branch or reference to read from.
        path: File or directory path inside the repository.

    Returns:
        GithubRepoContext: Repository context container.

    Raises:
        TypeError: If field values do not match declared types.

    Example:
        >>> ctx = GithubRepoContext('octocat', 'hello-world', 'main', 'docs')
        >>> ctx.repo
        'hello-world'
    """

    owner: str
    repo: str
    branch: str
    path: str | Path

branch instance-attribute

branch: str

owner instance-attribute

owner: str

path instance-attribute

path: str | Path

repo instance-attribute

repo: str

__init__

__init__(owner: str, repo: str, branch: str, path: str | Path) -> None

async_get_github_repo_urls async

async_get_github_repo_urls(
    owner: str, repo: str, branch: str, path: str | Path, file_suffix: str | None = None
) -> HttpUrlDataset

Asynchronously create raw GitHub content URLs for repository files.

PARAMETER	DESCRIPTION
owner	GitHub repository owner or organization. TYPE: str
repo	Repository name. TYPE: str
branch	Branch or ref to read from. TYPE: str
path	File or directory path inside the repository. TYPE: str | Path
file_suffix	Optional suffix filter when path points to a directory. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
HttpUrlDataset	A dataset of raw-content URLs keyed by file name.

RAISES	DESCRIPTION
TypeError	If any input cannot be converted to expected model types.

Example

urls = await async_get_github_repo_urls('octocat', 'hello-world', 'main', 'README.md')

isinstance(urls, HttpUrlDataset)

True

Source code in src/omnipy/components/remote/tasks.py

@TaskTemplate()
async def async_get_github_repo_urls(
    owner: str,
    repo: str,
    branch: str,
    path: str | Path,
    file_suffix: str | None = None,
) -> HttpUrlDataset:
    """Asynchronously create raw GitHub content URLs for repository files.

    Args:
        owner: GitHub repository owner or organization.
        repo: Repository name.
        branch: Branch or ref to read from.
        path: File or directory path inside the repository.
        file_suffix: Optional suffix filter when ``path`` points to a directory.

    Returns:
        A dataset of raw-content URLs keyed by file name.

    Raises:
        TypeError: If any input cannot be converted to expected model types.

    Example:
        >>> # urls = await async_get_github_repo_urls('octocat', 'hello-world', 'main', 'README.md')
        >>> # isinstance(urls, HttpUrlDataset)
        >>> True
    """

    repo_context = GithubRepoContext(owner=owner, repo=repo, branch=branch, path=path)

    if file_suffix:
        return await _async_get_urls_for_files_in_dir_with_suffix(repo_context, file_suffix)
    else:
        return _get_url_for_single_file(repo_context)

async_load_urls_into_new_dataset async

async_load_urls_into_new_dataset(
    urls: HttpUrlDataset,
    dataset_cls: type[_JsonDatasetT] = JsonDataset,
    as_mime_type: str | None = None,
) -> _JsonDatasetT

Asynchronously load remote URLs into a newly created dataset instance.

PARAMETER	DESCRIPTION
urls	Dataset containing the URLs to load. TYPE: HttpUrlDataset
dataset_cls	Dataset type to populate from the fetched content. TYPE: type[_JsonDatasetT] DEFAULT: JsonDataset
as_mime_type	Optional MIME type override for response decoding. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
_JsonDatasetT	A newly loaded dataset of type dataset_cls.

RAISES	DESCRIPTION
TypeError	If fetched data cannot be parsed by dataset_cls.

Example

loaded = await async_load_urls_into_new_dataset(

HttpUrlDataset({'a': 'https://example.com'}),

)

isinstance(loaded, JsonDataset)

True

Source code in src/omnipy/components/remote/tasks.py

@TaskTemplate()
async def async_load_urls_into_new_dataset(
    urls: HttpUrlDataset,
    dataset_cls: type[_JsonDatasetT] = JsonDataset,
    as_mime_type: str | None = None,
) -> _JsonDatasetT:
    """Asynchronously load remote URLs into a newly created dataset instance.

    Args:
        urls: Dataset containing the URLs to load.
        dataset_cls: Dataset type to populate from the fetched content.
        as_mime_type: Optional MIME type override for response decoding.

    Returns:
        A newly loaded dataset of type ``dataset_cls``.

    Raises:
        TypeError: If fetched data cannot be parsed by ``dataset_cls``.

    Example:
        >>> # loaded = await async_load_urls_into_new_dataset(
        >>> #     HttpUrlDataset({'a': 'https://example.com'}),
        >>> # )
        >>> # isinstance(loaded, JsonDataset)
        >>> True
    """
    return await dataset_cls.load(urls, as_mime_type=as_mime_type)

get_auto_from_api_endpoint async

get_auto_from_api_endpoint(
    url: HttpUrlModel, retry_client: RetryClient | None = None, as_mime_type: str | None = None
) -> AutoResponseContentModel

Fetch an API endpoint and decode the response from its MIME type.

PARAMETER	DESCRIPTION
url	HTTP URL to request. TYPE: HttpUrlModel
client_session	Optional shared aiohttp client session.
retry_http_statuses	HTTP status codes that trigger retries.
retry_attempts	Maximum number of retry attempts.
retry_backoff_strategy	Backoff policy used between retries.
as_mime_type	Optional MIME type override for response decoding. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
AutoResponseContentModel	The response content wrapped together with its effective content type.

RAISES	DESCRIPTION
AssertionError	If the response has no Content-Type header and no override is given.
ConnectionError	If the endpoint response status is not 200.

Example

data = await get_auto_from_api_endpoint(HttpUrlModel('https://example.com/data'))

isinstance(data, AutoResponseContentModel)

True

Source code in src/omnipy/components/remote/tasks.py

@TaskTemplate(iterate_over_data_files=True, output_dataset_cls=AutoResponseContentDataset)
async def get_auto_from_api_endpoint(
    url: HttpUrlModel,
    retry_client: 'RetryClient | None' = None,
    as_mime_type: str | None = None,
) -> AutoResponseContentModel:
    """Fetch an API endpoint and decode the response from its MIME type.

    Args:
        url: HTTP URL to request.
        client_session: Optional shared aiohttp client session.
        retry_http_statuses: HTTP status codes that trigger retries.
        retry_attempts: Maximum number of retry attempts.
        retry_backoff_strategy: Backoff policy used between retries.
        as_mime_type: Optional MIME type override for response decoding.

    Returns:
        The response content wrapped together with its effective content type.

    Raises:
        AssertionError: If the response has no ``Content-Type`` header and no override is given.
        ConnectionError: If the endpoint response status is not ``200``.

    Example:
        >>> # data = await get_auto_from_api_endpoint(HttpUrlModel('https://example.com/data'))
        >>> # isinstance(data, AutoResponseContentModel)
        >>> True
    """
    from .lazy_import import ClientSession, CONTENT_TYPE

    async for retry_session in _ensure_retry_session(retry_client):
        async for response in _call_get(url, cast(ClientSession, retry_session)):
            _check_response_status(response)
            if as_mime_type:
                content_type = content_type_header = as_mime_type
            else:
                assert CONTENT_TYPE in response.headers
                content_type_header = response.headers[CONTENT_TYPE]
                content_type = response.content_type
            match content_type:
                case 'application/json':
                    content = await response.json(content_type=None)
                case 'text/plain':
                    content = await response.text()
                case 'application/octet-stream' | _:
                    content = await response.read()

            model = AutoResponseContentModel(
                ResponseContentPydModel(content_type=content_type_header, response=content))
            return model

    raise ShouldNotOccurException('Other exception should have been raised before this point.')

get_bytes_from_api_endpoint async

get_bytes_from_api_endpoint(
    url: HttpUrlModel, retry_client: RetryClient | None = None
) -> BytesModel

Fetch an API endpoint and decode the response body as raw bytes.

PARAMETER	DESCRIPTION
url	HTTP URL to request. TYPE: HttpUrlModel
client_session	Optional shared aiohttp client session.
retry_http_statuses	HTTP status codes that trigger retries.
retry_attempts	Maximum number of retry attempts.
retry_backoff_strategy	Backoff policy used between retries.

RETURNS	DESCRIPTION
BytesModel	The response body as bytes.

RAISES	DESCRIPTION
ConnectionError	If the endpoint response status is not 200.

Example

blob = await get_bytes_from_api_endpoint(HttpUrlModel('https://example.com/file'))

isinstance(blob, BytesModel)

True

Source code in src/omnipy/components/remote/tasks.py

@TaskTemplate(iterate_over_data_files=True, output_dataset_cls=BytesDataset)
async def get_bytes_from_api_endpoint(
    url: HttpUrlModel,
    retry_client: 'RetryClient | None' = None,
) -> BytesModel:
    """Fetch an API endpoint and decode the response body as raw bytes.

    Args:
        url: HTTP URL to request.
        client_session: Optional shared aiohttp client session.
        retry_http_statuses: HTTP status codes that trigger retries.
        retry_attempts: Maximum number of retry attempts.
        retry_backoff_strategy: Backoff policy used between retries.

    Returns:
        The response body as bytes.

    Raises:
        ConnectionError: If the endpoint response status is not ``200``.

    Example:
        >>> # blob = await get_bytes_from_api_endpoint(HttpUrlModel('https://example.com/file'))
        >>> # isinstance(blob, BytesModel)
        >>> True
    """
    from .lazy_import import ClientSession

    async for retry_session in _ensure_retry_session(retry_client,):
        async for response in _call_get(url, cast(ClientSession, retry_session)):
            _check_response_status(response)
            return BytesModel(await response.read())

    raise ShouldNotOccurException('Other exception should have been raised before this point.')

get_github_repo_urls

get_github_repo_urls(
    owner: str, repo: str, branch: str, path: str | Path, file_suffix: str | None = None
) -> HttpUrlDataset

Create raw GitHub content URLs for one file or matching files in a repository path.

PARAMETER	DESCRIPTION
owner	GitHub repository owner or organization. TYPE: str
repo	Repository name. TYPE: str
branch	Branch or ref to read from. TYPE: str
path	File or directory path inside the repository. TYPE: str | Path
file_suffix	Optional suffix filter when path points to a directory. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
HttpUrlDataset	A dataset of raw-content URLs keyed by file name.

RAISES	DESCRIPTION
TypeError	If any input cannot be converted to expected model types.

Example

urls = get_github_repo_urls('octocat', 'hello-world', 'main', 'README.md')

isinstance(urls, HttpUrlDataset)

True

Source code in src/omnipy/components/remote/tasks.py

@TaskTemplate()
def get_github_repo_urls(
    owner: str,
    repo: str,
    branch: str,
    path: str | Path,
    file_suffix: str | None = None,
) -> HttpUrlDataset:
    """Create raw GitHub content URLs for one file or matching files in a repository path.

    Args:
        owner: GitHub repository owner or organization.
        repo: Repository name.
        branch: Branch or ref to read from.
        path: File or directory path inside the repository.
        file_suffix: Optional suffix filter when ``path`` points to a directory.

    Returns:
        A dataset of raw-content URLs keyed by file name.

    Raises:
        TypeError: If any input cannot be converted to expected model types.

    Example:
        >>> # urls = get_github_repo_urls('octocat', 'hello-world', 'main', 'README.md')
        >>> # isinstance(urls, HttpUrlDataset)
        >>> True
    """

    repo_context = GithubRepoContext(owner=owner, repo=repo, branch=branch, path=path)

    if file_suffix:
        return _get_urls_for_files_in_dir_with_suffix(repo_context, file_suffix)
    else:
        return _get_url_for_single_file(repo_context)

get_json_from_api_endpoint async

get_json_from_api_endpoint(url: HttpUrlModel, retry_client: RetryClient | None = None) -> JsonModel

Fetch a JSON API endpoint and decode the response body as JSON.

PARAMETER	DESCRIPTION
url	HTTP URL to request. TYPE: HttpUrlModel
client_session	Optional shared aiohttp client session.
retry_http_statuses	HTTP status codes that trigger retries.
retry_attempts	Maximum number of retry attempts.
retry_backoff_strategy	Backoff policy used between retries.

RETURNS	DESCRIPTION
JsonModel	The decoded JSON response for the requested URL.

RAISES	DESCRIPTION
ConnectionError	If the endpoint response status is not 200.
ValueError	If the response body cannot be decoded as JSON.

Example

result = await get_json_from_api_endpoint(HttpUrlModel('https://api.example.com'))

isinstance(result, JsonModel)

True

Source code in src/omnipy/components/remote/tasks.py

@TaskTemplate(iterate_over_data_files=True, output_dataset_cls=JsonDataset)
async def get_json_from_api_endpoint(
    url: HttpUrlModel,
    retry_client: 'RetryClient | None' = None,
) -> JsonModel:
    """Fetch a JSON API endpoint and decode the response body as JSON.

    Args:
        url: HTTP URL to request.
        client_session: Optional shared aiohttp client session.
        retry_http_statuses: HTTP status codes that trigger retries.
        retry_attempts: Maximum number of retry attempts.
        retry_backoff_strategy: Backoff policy used between retries.

    Returns:
        The decoded JSON response for the requested URL.

    Raises:
        ConnectionError: If the endpoint response status is not ``200``.
        ValueError: If the response body cannot be decoded as JSON.

    Example:
        >>> # result = await get_json_from_api_endpoint(HttpUrlModel('https://api.example.com'))
        >>> # isinstance(result, JsonModel)
        >>> True
    """
    from .lazy_import import ClientSession

    async for retry_session in _ensure_retry_session(retry_client,):
        async for response in _call_get(url, cast(ClientSession, retry_session)):
            _check_response_status(response)
            return JsonModel(await response.json(content_type=None))

    raise ShouldNotOccurException('Other exception should have been raised before this point.')

get_retry_client

get_retry_client(
    client_session: ClientSession | None = None,
    retry_http_statuses: tuple[int, ...] = DEFAULT_RETRY_STATUSES,
    retry_attempts: int = DEFAULT_RETRIES,
    retry_backoff_strategy: BackoffStrategy.Literals = DEFAULT_BACKOFF_STRATEGY,
) -> RetryClient

Build a retry-enabled HTTP client wrapper.

PARAMETER	DESCRIPTION
client_session	Existing client session to wrap. TYPE: ClientSession | None DEFAULT: None
retry_http_statuses	Status codes that should trigger retries. TYPE: tuple[int, ...] DEFAULT: DEFAULT_RETRY_STATUSES
retry_attempts	Maximum number of retry attempts. TYPE: int DEFAULT: DEFAULT_RETRIES
retry_backoff_strategy	Backoff strategy identifier. TYPE: BackoffStrategy.Literals DEFAULT: DEFAULT_BACKOFF_STRATEGY

RETURNS	DESCRIPTION
RetryClient	Configured retry client instance. TYPE: RetryClient

RAISES	DESCRIPTION
KeyError	If retry_backoff_strategy is not registered.

Example

retry_client = _get_retry_client(session, (429, 503), 5, 'exponential')

type(retry_client).name

'RetryClient'

Source code in src/omnipy/components/remote/tasks.py

def get_retry_client(
    client_session: 'ClientSession | None' = None,
    retry_http_statuses: tuple[int, ...] = DEFAULT_RETRY_STATUSES,
    retry_attempts: int = DEFAULT_RETRIES,
    retry_backoff_strategy: BackoffStrategy.Literals = DEFAULT_BACKOFF_STRATEGY,
) -> 'RetryClient':
    """Build a retry-enabled HTTP client wrapper.

    Args:
        client_session: Existing client session to wrap.
        retry_http_statuses: Status codes that should trigger retries.
        retry_attempts: Maximum number of retry attempts.
        retry_backoff_strategy: Backoff strategy identifier.

    Returns:
        RetryClient: Configured retry client instance.

    Raises:
        KeyError: If ``retry_backoff_strategy`` is not registered.

    Example:
        >>> # retry_client = _get_retry_client(session, (429, 503), 5, 'exponential')
        >>> # type(retry_client).__name__
        >>> 'RetryClient'
    """
    from .helpers import BACKOFF_STRATEGY_2_RETRY_CLS
    from .lazy_import import RetryClient

    retry_cls = BACKOFF_STRATEGY_2_RETRY_CLS[retry_backoff_strategy]
    retry_options = retry_cls(
        attempts=retry_attempts,
        statuses=retry_http_statuses,
        retry_all_server_errors=False,
    )
    return RetryClient(
        client_session=client_session,
        retry_options=retry_options,
    )

get_str_from_api_endpoint async

get_str_from_api_endpoint(url: HttpUrlModel, retry_client: RetryClient | None = None) -> StrModel

Fetch an API endpoint and decode the response body as text.

PARAMETER	DESCRIPTION
url	HTTP URL to request. TYPE: HttpUrlModel
client_session	Optional shared aiohttp client session.
retry_http_statuses	HTTP status codes that trigger retries.
retry_attempts	Maximum number of retry attempts.
retry_backoff_strategy	Backoff policy used between retries.

RETURNS	DESCRIPTION
StrModel	The response body as plain text.

RAISES	DESCRIPTION
ConnectionError	If the endpoint response status is not 200.

Example

text = await get_str_from_api_endpoint(HttpUrlModel('https://example.com'))

isinstance(text, StrModel)

True

Source code in src/omnipy/components/remote/tasks.py

@TaskTemplate(iterate_over_data_files=True, output_dataset_cls=StrDataset)
async def get_str_from_api_endpoint(
    url: HttpUrlModel,
    retry_client: 'RetryClient | None' = None,
) -> StrModel:
    """Fetch an API endpoint and decode the response body as text.

    Args:
        url: HTTP URL to request.
        client_session: Optional shared aiohttp client session.
        retry_http_statuses: HTTP status codes that trigger retries.
        retry_attempts: Maximum number of retry attempts.
        retry_backoff_strategy: Backoff policy used between retries.

    Returns:
        The response body as plain text.

    Raises:
        ConnectionError: If the endpoint response status is not ``200``.

    Example:
        >>> # text = await get_str_from_api_endpoint(HttpUrlModel('https://example.com'))
        >>> # isinstance(text, StrModel)
        >>> True
    """
    from .lazy_import import ClientSession

    async for retry_session in _ensure_retry_session(retry_client,):
        async for response in _call_get(url, cast(ClientSession, retry_session)):
            _check_response_status(response)
            return StrModel(await response.text())

    raise ShouldNotOccurException('Other exception should have been raised before this point.')

load_urls_into_new_dataset

load_urls_into_new_dataset(
    urls: HttpUrlDataset,
    dataset_cls: type[_JsonDatasetT] = JsonDataset,
    as_mime_type: str | None = None,
) -> _JsonDatasetT

Load remote URLs into a newly created dataset instance.

PARAMETER	DESCRIPTION
urls	Dataset containing the URLs to load. TYPE: HttpUrlDataset
dataset_cls	Dataset type to populate from the fetched content. TYPE: type[_JsonDatasetT] DEFAULT: JsonDataset
as_mime_type	Optional MIME type override for response decoding. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
_JsonDatasetT	A newly loaded dataset of type dataset_cls.

RAISES	DESCRIPTION
TypeError	If fetched data cannot be parsed by dataset_cls.

Example

loaded = load_urls_into_new_dataset(HttpUrlDataset({'a': 'https://example.com'}))

isinstance(loaded, JsonDataset)

True

Source code in src/omnipy/components/remote/tasks.py

@TaskTemplate()
def load_urls_into_new_dataset(
    urls: HttpUrlDataset,
    dataset_cls: type[_JsonDatasetT] = JsonDataset,
    as_mime_type: str | None = None,
) -> _JsonDatasetT:
    """Load remote URLs into a newly created dataset instance.

    Args:
        urls: Dataset containing the URLs to load.
        dataset_cls: Dataset type to populate from the fetched content.
        as_mime_type: Optional MIME type override for response decoding.

    Returns:
        A newly loaded dataset of type ``dataset_cls``.

    Raises:
        TypeError: If fetched data cannot be parsed by ``dataset_cls``.

    Example:
        >>> # loaded = load_urls_into_new_dataset(HttpUrlDataset({'a': 'https://example.com'}))
        >>> # isinstance(loaded, JsonDataset)
        >>> True
    """
    return dataset_cls.load(urls, as_mime_type=as_mime_type)