pontos.helper package¶

class pontos.helper.AsyncDownloadProgressIterable(*, content_iterator, url, length)¶

An async iterator to iterate over a downloadable content and the progress.

Example

from pontos.helper import AsyncDownloadProgressIterable

it = AsyncDownloadProgressIterable(...)
async for content, progress in it:
    file.write(content)
    print(progress)

Create a new AsyncDownloadProgressIterable instance

Parameters:

content_iterator (AsyncIterator[T]) – An async iterator to call for getting the content. Should be a stream of bytes or strings.
url (SupportsStr) – The URL where the content gets downloaded.
length (int | None) – Length of the content.

property length: int | None¶: Size in bytes of the to be downloaded file or None if the size is not available

property url: str¶: The URL where the content gets downloaded from

class pontos.helper.DownloadProgressIterable(*, content_iterator, url, destination, length)¶

An synchronous iterator to iterate over a download progress.

Example

from pontos.helper import DownloadProgressIterable

it = DownloadProgressIterable(...)
for progress in it:
    print(progress)

Create a new DownloadProgressIterable instance

Parameters:

content_iterator (Iterator[bytes]) – An iterator of bytes to write to destination path
url (str) – A URL where the content will be downloaded from
destination (Path) – Path to write the downloaded content to
length (int | None) – Length of the content to be downloaded

property length: int | None¶: Size in bytes of the to be downloaded file or None if the size is not available

property destination: Path¶: Destination path of the to be downloaded file

run()¶: Just run the download without caring about the progress

pontos.helper.add_sys_path(directory)¶

Context Manager to add a directory path to the module search path aka. sys.path. The directory path is removed when the context manager is left.

Parameters:: directory (str | PathLike) – A os.PathLike directory to add to sys.path
Return type:: Generator[None, None, None]

Example

from pontos.helper import add_sys_path

with add_sys_path("/tmp/test-modules"):
    import mymodule

pontos.helper.deprecated(_func_or_cls=None, *, since=None, reason=None)¶

A decorator to mark functions, classes and methods as deprecated

Parameters:

since (str | None) – An optional version since the referenced item is deprecated.
reason (str | None) – An optional reason why the references item is deprecated.

Examples

from pontos.helper import deprecated

@deprecated
def my_function(*args, **kwargs):
    ...

@deprecated("The function is obsolete. Please use my_func instead.")
def my_function(*args, **kwargs):
    ...

@deprecated(
    since="1.2.3",
    reason="The function is obsolete. Please use my_func instead."
)
def my_function(*args, **kwargs):
    ...

@deprecated(reason="The class will be removed in version 3.4.5")
class Foo:
    ...

class Foo:
    @deprecated(since="2.3.4")
    def bar(self, *args, **kwargs):
        ...

pontos.helper.download_async(stream, *, content_length=None, chunk_size=4096, url=None)¶

An async context manager that returns an AsyncDownloadProgressIterable.

It ensures that the stream is closed automatically via the context manager.

Parameters:

stream (AsyncContextManager[Response]) – An async context manager providing a streaming response.
content_length (int | None) – Optional length of the content to download. If now provided it is determined from the response if available.
chunk_size (int) – Download the content in chunks of this size.
url (str | None) – Use a specific URL. If not set the URL of the response is used.

Returns:

A context manager containing an AsyncDownloadProgressIterable

Raises:

HTTPStatusError – If the request was invalid

Return type:

AsyncIterator[AsyncDownloadProgressIterable[bytes]]

Example

import httpx
from pontos.helper import download_async

client = httpx.AsyncClient(...)
stream = client.stream("GET, "https://foo.bar/baz.zip)

async with download_async(stream) as download:
    async for content, progress in download:
        file.write(content)
        print(progress)

pontos.helper.download(url, destination=None, *, headers=None, params=None, chunk_size=4096, timeout=1000)¶

Download file in url to filename

Parameters:

url (str) – The url of the file we want to download
destination (Path | str | None) – Path of the file to store the download in. If set it will be derived from the passed URL.
headers (Dict[str, Any] | None) – HTTP headers to use for the download
params (Dict[str, Any] | None) – HTTP request parameters to use for the download
chunk_size (int) – Download file in chunks of this size
timeout (int) – Connection timeout

Raises:

HTTPStatusError – If the request was invalid

Returns:

A DownloadProgressIterator that yields the progress of the download in percent for each downloaded chunk or None for each chunk if the progress is unknown.

Return type:

Generator[DownloadProgressIterable, None, None]

Example

from pontos.helper import download

with download("https://example.com/some/file") as progress_it:
    for progress in progress_it:
        print(progress)

pontos.helper.ensure_unload_module(module)¶

A context manager to ensure that a module gets removed even if an error occurs

Parameters:: module (str | ModuleType) – Module instance or name of the Python module to unload. For example: foo.bar
Return type:: Generator[None, None, None]

Example

from pontos.helper import ensure_unload_module

with ensure_unload_module("foo.bar"):
    do_something()

pontos.helper.enum_or_value(value)¶

Return the value of an Enum or the value if it isn’t an Enum

Return type:: Any

pontos.helper.parse_timedelta(time_str)¶

Parse a timedelta from a string

Examples

from pontos.helper import parse_timedelta

parse_timedelta("1.5h")
parse_timedelta("1w2d4h5m6s")

Return type:: timedelta

pontos.helper.snake_case(value)¶

Convert a string to snake case/underscore naming scheme

Parameters:: value (str) – String to convert into snake case
Return type:: str

Example

from pontos.helper import snake_case

snake_case("CamelCase")

will return “camel_case”

pontos.helper.unload_module(module)¶

Unload a Python module

Parameters:: module (str | ModuleType) – Module instance or name of the Python module to unload. For example: foo.bar