Source code for pyapp.conf.loaders.http_loader

"""
HTTP Loader
~~~~~~~~~~~

Loads settings from an HTTP endpoint (HTTPS is recommended)

"""

import contextlib
import ssl
import tempfile
from typing import TextIO, Tuple
from urllib.error import ContentTooShortError
from urllib.request import urlopen

from yarl import URL

from pyapp.conf.loaders.base import Loader
from pyapp.conf.loaders.content_types import content_type_from_url, registry
from pyapp.exceptions import InvalidConfiguration


def retrieve_file(url: URL) -> Tuple[TextIO, str]:
    """Fetch a file from a URL (handling SSL).

    This is based off `urllib.request.urlretrieve`.

    """
    if url.scheme not in ("http", "https"):
        raise InvalidConfiguration("Illegal scheme.")

    context = (
        ssl.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT)
        if url.scheme == "https"
        else None
    )

    with contextlib.closing(
        urlopen(url, context=context)  # noqa: S310 - Completed above
    ) as response:
        block_size = 1024 * 8
        size = -1
        read = 0

        headers = response.info()
        if "Content-Length" in headers:
            size = int(headers["Content-Length"])

        content_type = (
            headers["Content-Type"]
            if "Content-Type" in headers
            else content_type_from_url(url)
        )

        tfp = tempfile.TemporaryFile()
        while True:
            block = response.read(block_size)
            if not block:
                break
            read += len(block)
            tfp.write(block)

        # Seek to start
        tfp.seek(0)

    if size >= 0 and read < size:
        tfp.close()

        raise ContentTooShortError(
            f"retrieval incomplete: got only {read} out of {size} bytes", headers
        )

    return tfp, content_type


[docs] class HttpLoader(Loader): """ Load settings from a file. Usage:: >>> loader = HttpLoader(URL("https://hostname/path/to/settings.json")) >>> settings = dict(loader) """ scheme = ("http", "https") @classmethod def from_url(cls, url: URL) -> Loader: """Create an instance of :class:`HttpLoader` from :class:`urllib.parse.ParseResult`.""" return HttpLoader(url) def __init__(self, url: URL): self.url = url self._fp = None self.content_type = None def __del__(self): self.close() def __iter__(self): try: self._fp, self.content_type = retrieve_file(self.url) except OSError as ex: raise InvalidConfiguration(f"Unable to load settings: {self}\n{ex}") from ex try: data = registry.parse_file(self._fp, self.content_type) except ValueError as ex: raise InvalidConfiguration( f"Unable to parse JSON file: {self}\n{ex}" ) from ex # Check we have a valid container object if not isinstance(data, dict): raise InvalidConfiguration( f"Invalid root object, expected a JSON Object: {self}" ) return ((k, v) for k, v in data.items() if k.isupper()) def __str__(self): return str(self.url) def close(self): """ Ensure the file pointer is closed """ if self._fp: self._fp.close() self._fp = None