Source code for eodms_api_client.auth

import os
from getpass import getpass
from netrc import netrc

from requests import Session
from requests.adapters import HTTPAdapter
from requests.auth import HTTPBasicAuth
from urllib3.util.retry import Retry

# establish a retry strategy with backoff
RETRY_STRAT = Retry(
    total=5, # retry a max of 5 times
    backoff_factor=2, # seconds
    status_forcelist=[500, 502, 503, 504] # only retry on these HTTP codes
)
EODMS_HOSTNAME = 'data.eodms-sgdot.nrcan-rncan.gc.ca'

[docs]def create_session(username=None, password=None): ''' Create a persistent session object for the EODMS REST API using the given username and password. If neither is provided, attempt to use the .netrc file and fallback to console entry as a last-resort Inputs: - username: EODMS username - password: EODMS password Outputs: - session: requests.Session object for the EODMS REST API ''' if username is None and password is None: if os.name == 'posix': netrc_file = os.path.join(os.path.expanduser('~'), '.netrc') elif os.name == 'nt': netrc_file = os.path.join(os.path.expanduser('~'), '_netrc') if not os.path.exists(netrc_file): netrc_file = os.path.join(os.path.expanduser('~'), '.netrc') if not os.path.exists(netrc_file): raise FileNotFoundError('Cannot locate netrc file in expected location: %s' % os.path.expanduser('~')) else: raise NotImplementedError('Unsupported OS: %s' % os.name) try: hosts = netrc(netrc_file).hosts try: username, _, password = hosts.get(EODMS_HOSTNAME) except TypeError as no_eodms_host_defined: raise ValueError('Cannot locate credentials for EODMS server. Check your netrc file') from no_eodms_host_defined except (FileNotFoundError, TypeError): username = input('Enter EODMS username: ') password = getpass('Enter EODMS password: ') elif username is None and password is not None: username = input('Enter EODMS username: ') elif username is not None and password is None: password = getpass('Enter EODMS password: ') session = Session() session.auth = HTTPBasicAuth(username, password) session.mount('https://', HTTPAdapter(max_retries=RETRY_STRAT)) return session