Hide navigation sidebar
Hide table of contents sidebar
Skip to content
Back to top
View this page
Edit this page
Toggle table of contents sidebar

# 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, bool | None]) – 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