Source code for radiant_mlhub.session

"""
Methods and classes to simplify constructing and authenticating requests to the MLHub API.

It is generally recommended that you use the :func:`get_session` function to create sessions, since this will propertly handle resolution
of the API key from function arguments, environment variables, and profiles as described in :ref:`Authentication`. See the
:func:`get_session` docs for usage examples.
"""

import configparser
import os
import platform
import urllib.parse
from pathlib import Path
from typing import Any, Dict, Iterator, Optional

import requests
import requests.adapters

from .__version__ import __version__
from .exceptions import APIKeyNotFound, AuthenticationError
from .retry_config import config as retry_config

ANONYMOUS_PROFILE = "__anonymous__"


[docs]class Session(requests.Session): """Custom class inheriting from :class:`requests.Session` with some additional conveniences: * Adds the API key as a ``key`` query parameter * Adds an ``Accept: application/json`` header * Adds a ``User-Agent`` header that contains the package name and version, plus basic system information like the OS name * Prepends the MLHub root URL (``https://api.radiant.earth/mlhub/v1/``) to any request paths without a domain * Raises a :exc:`radiant_mlhub.exceptions.AuthenticationError` for ``401 (UNAUTHORIZED)`` responses * Calls :meth:`requests.Response.raise_for_status` after all requests to raise exceptions for any status codes above 400. """ MLHUB_HOME_ENV_VARIABLE = 'MLHUB_HOME' API_KEY_ENV_VARIABLE = 'MLHUB_API_KEY' PROFILE_ENV_VARIABLE = 'MLHUB_PROFILE' ROOT_URL_ENV_VARIABLE = 'MLHUB_ROOT_URL' DEFAULT_ROOT_URL = 'https://api.radiant.earth/mlhub/v1/' def __init__(self, *, api_key: Optional[str]): super().__init__() self.root_url = os.getenv(self.ROOT_URL_ENV_VARIABLE, self.DEFAULT_ROOT_URL) # Add the API key query parameter if api_key is not None: self.params.update({'key': api_key}) # type: ignore [union-attr] # Set the default headers self.headers.update({ 'Accept': 'application/json', # Add the package name + version and the system info to the user-agent header 'User-Agent': f'{__name__.split(".")[0]}/{__version__} ({platform.version()})' }) adapter = requests.adapters.HTTPAdapter(max_retries=retry_config()) for prefix in 'http://', 'https://': self.mount(prefix, adapter)
[docs] def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: # type: ignore[override] """Overwrites the default :meth:`requests.Session.request` method to prepend the MLHub root URL if the given ``url`` does not include a scheme. This will raise an :exc:`~radiant_mlhub.exceptions.AuthenticationError` if a 401 response is returned by the server, and a :class:`~requests.exceptions.HTTPError` if any other status code of 400 or above is returned. Parameters ---------- method : str The request method to use. Passed directly to the ``method`` argument of :meth:`requests.Session.request` url : str Either a full URL or a path relative to the :attr:`ROOT_URL`. For example, to make a request to the Radiant MLHub API ``/collections`` endpoint, you could use ``session.get('collections')``. **kwargs All other keyword arguments are passed directly to :meth:`requests.Session.request` (see that documentation for an explanation of these keyword arguments). Raises ------ AuthenticationError If the response status code is 401 HTTPError For all other response status codes at or above 400 """ # Parse the url argument and substitute the base URL if this is a relative path parsed_url = urllib.parse.urlsplit(url) if not parsed_url.scheme: parsed_root = urllib.parse.urlsplit(self.root_url) url = urllib.parse.SplitResult( parsed_root.scheme, parsed_root.netloc, # Remove leading slashes so urljoin appends the path to our root path urllib.parse.urljoin(parsed_root.path, parsed_url.path.lstrip('/')), parsed_url.query, parsed_url.fragment, ).geturl() response = super().request(method, url, **kwargs) # Handle authentication errors if response.status_code == 401: msg = "Authentication failed. " request_qs = str(urllib.parse.urlsplit(response.request.url).query) request_params = urllib.parse.parse_qs(request_qs) if "key" in request_params: msg += f"API Key: {request_params['key'][0]}" else: msg += "No API key provided." raise AuthenticationError(msg) # Raise exceptions for any HTTP codes >=400 response.raise_for_status() return response
[docs] @classmethod def from_env(cls) -> 'Session': """Create a session object from an API key from the environment variable. Returns ------- session : Session Raises ------ APIKeyNotFound If the API key cannot be found in the environment """ api_key = os.getenv(cls.API_KEY_ENV_VARIABLE) if not api_key: raise APIKeyNotFound(f'No "{cls.API_KEY_ENV_VARIABLE}" variable found in environment.') return cls(api_key=api_key)
[docs] @classmethod def from_config(cls, profile: Optional[str] = None) -> 'Session': """Create a session object by reading an API key from the given profile in the ``profiles`` file. By default, the client will look for the ``profiles`` file in a ``.mlhub`` directory in the user's home directory (as determined by :meth:`Path.home <pathlib.Path.home>`). However, if an ``MLHUB_HOME`` environment variable is present, the client will look in that directory instead. Parameters ---------- profile: str, optional The name of a profile configured in the ``profiles`` file. Returns ------- session : Session Raises ------ APIKeyNotFound If the given config file does not exist, the given profile cannot be found, or there is no ``api_key`` property in the given profile section. """ mlhub_home = Path(os.getenv(Session.MLHUB_HOME_ENV_VARIABLE, Path.home() / '.mlhub')) config_path = mlhub_home / 'profiles' if not config_path.exists(): raise APIKeyNotFound(f'No file found at {config_path}') config = configparser.ConfigParser() config.read(config_path) profile = profile or 'default' # Use the default profile if the given profile is None or empty if profile not in config.sections(): raise APIKeyNotFound(f'Could not find "{profile}" section in {config_path}') api_key = config.get(profile, 'api_key', fallback=None) if not api_key: raise APIKeyNotFound(f'Could not find "api_key" value in "{profile}" section of {config_path}') return cls(api_key=api_key)
[docs] def paginate(self, url: str, **kwargs: Any) -> Iterator[Dict[str, Any]]: """Makes a GET request to the given ``url`` and paginates through all results by looking for a link in each response with a ``rel`` type of ``"next"``. Any additional keyword arguments are passed directly to :meth:`requests.Session.get`. Parameters ---------- url : str The URL to which the initial request will be made. Note that this may either be a full URL or a path relative to the :attr:`ROOT_URL` as described in :meth:`Session.request`. Yields ------ page : dict An individual response as a dictionary. """ current_url: Optional[str] = str(url) while True: if current_url is None: break page = self.get(current_url, **kwargs).json() yield page current_url = dict(next(( link for link in page.get('links', []) if link['rel'] == 'next'), {} )).get('href')
[docs]def get_session(*, api_key: Optional[str] = None, profile: Optional[str] = None) -> Session: """Gets a :class:`Session` object that uses the given ``api_key`` for all requests. Resolves an API key by trying each of the following (in this order): 1. Use the `api_key` argument provided (Optional). 2. Use an `MLHUB_API_KEY` environment variable. 3. Use the profile argument provided (Optional). 4. Use the `MLHUB_PROFILE` environment variable. 5. Use the default profile If none of the above strategies results in a valid API key, then an APIKeyNotFound exception is raised. See Using Profiles section for details. Parameters ---------- api_key : str, optional The API key to use for all requests from the session. See description above for how the API key is resolved if not provided as an argument. profile : str, optional The name of a profile configured in the ``.mlhub/profiles`` file. This will be passed directly to :func:`~Session.from_config`. Returns ------- session : Session Raises ------ APIKeyNotFound If no API key can be resolved. Examples -------- >>> from radiant_mlhub import get_session # Get the API from the "default" profile >>> session = get_session() # Get the session from the "project1" profile # Alternatively, you could set the MLHUB_PROFILE environment variable to "project1" >>> session = get_session(profile='project1') # Pass an API key directly to the session # Alternatively, you could set the MLHUB_API_KEY environment variable to "some-api-key" >>> session = get_session(api_key='some-api-key') """ # 1. Use the `api_key` argument provided (Optional) if api_key: return Session(api_key=api_key) # 2. Use an `MLHUB_API_KEY` environment variable if Session.API_KEY_ENV_VARIABLE in os.environ: return Session.from_env() # 3. Use the profile argument provided (Optional, see Using Profiles section for details) # 4. Use the `MLHUB_PROFILE` environment variable (see Using Profiles section for details) # 5. Use the default profile (see Using Profiles section for details) try: if profile == ANONYMOUS_PROFILE: # For the special case of the "__anonymous__" profile, create a Session with no API key return Session(api_key=None) return Session.from_config(profile=profile or os.getenv(Session.PROFILE_ENV_VARIABLE)) except APIKeyNotFound: raise APIKeyNotFound('Could not resolve an API key from arguments, the environment, or a config file.') from None