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 (str) – The URL where the content gets downloaded.
length (Optional[int]) – Length of the content.
- property length: Optional[int]#
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 (Optional[int]) – Length of the content to be downloaded
- property length: Optional[int]#
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 (Union[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 (Optional[str]) – An optional version since the referenced item is deprecated.
reason (Optional[str]) – 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)#
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 (Optional[int]) – 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.
- 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 (Optional[Union[Path, str]]) – Path of the file to store the download in. If set it will be derived from the passed URL.
headers (Optional[Dict[str, Any]]) – HTTP headers to use for the download
params (Optional[Dict[str, Any]]) – 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 (Union[str, module]) – 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 (Union[str, module]) – Module instance or name of the Python module to unload. For example: foo.bar