URS Credential Manager

urscred.py

__all__ = ['URSSession', 'asdc_downloader']


class URSSession:
    """
    Class to store URS credentials:

    * for use with requests
    * for use with netrc and dodsrc
    """
    def __init__(self):
        import requests
        import getpass
        self._urs_username = getpass.getpass('URS User:')
        self._urs_password = getpass.getpass('URS PW:')
        self._session = requests.Session()
        self._session.auth = (self._urs_username, self._urs_password)

    def get(self, url, params=None, request_kwds=None, **get_kwds):
        """Thin wrapper around r1 = session.request and r = session.get(r1.url)"""
        if request_kwds is None:
            request_kwds = {}
        session = self._session
        r1 = session.request('get', url, **request_kwds)
        r = session.get(r1.url, params=params, **get_kwds)
        return r

    def to_netrc(self, netrcpath=None, dodsrcpath=None, servers=None):
        import os
        if servers is None:
            servers = ['urs.earthdata.nasa.gov']
        else:
            servers = ['urs.earthdata.nasa.gov'] + servers
        if netrcpath is None:
            netrcpath = os.path.join(os.path.expanduser('~'), '.netrc')
        if dodsrcpath is None:
            dodsrcpath = os.path.join(os.path.expanduser('~'), '.dodsrc')

        _go_ro_open=lambda path, flags: os.open(path, flags, 0o600)

        with open(netrcpath, 'a', opener=_go_ro_open) as nrc:
            nrc.seek(0, 2)
            for server in servers:
                nrc.write(f"""
machine {server}
    login {self._urs_username}
    password {self._urs_password}
""")
        with open(dodsrcpath, 'a', opener=_go_ro_open) as drc:
            drc.seek(0, 2)
            drc.write(f"""
HTTP.COOKIEJAR={netrcpath}
HTTP.NETRC={netrcpath}
""")

class asdc_downloader:
    def __init__(self, username=None, password=None):
        """
        Object to facilitate downloading from ASDC. For more info See:
        https://forum.earthdata.nasa.gov/viewtopic.php?t=2330

        Arguments
        ---------
        username : str
          URS Username; if not supplied will be requested
        password : str
          URS password; if not supplied will be requested
        """
        from getpass import getpass
        if username is None:
            username = getpass('URS Username: ')
        if password is None:
            password = getpass('URS Password: ')
        self._urs_username = username
        self._urs_password = password

    def get(self, url, **kwds):
        """
        Thin wrapper around requests.Session.get that handles redirection
        for downloading from ASDC. See:
        https://forum.earthdata.nasa.gov/viewtopic.php?t=2330

        Arguments
        ---------
        url : str
            url to download
        kwds : dict
            Must be accepted by get

        Returns
        -------
        out : object
          Same as requests.get 
        """
        from requests import Session
        with Session() as session:
            #session.auth = (username, password)
            _redirect = session.get(url, **kwds)
            _response = session.get(
                _redirect.url,
                auth=(self._urs_username, self._urs_password),
                **kwds
            )
        return _response

    def download(self, url, keepdir=True, exists='warn', **kwds):
        """
        download url to local file using just the file name.
        """
        import os
        # This code downloads the file to your current working directory (where you ran python from)
        import warnings
        from pathlib import Path
        p = Path(url)
        if keepdir:
            file_name = os.path.join(*p.parts[1:])
            os.makedirs(os.path.dirname(file_name), exist_ok=True)
        else:
            file_name = p.name
        
        if Path(file_name).exists():
            if exists == 'keep':
                return file_name
            elif exists == 'warn':
                warnings.warn(f'{file_name} exists; using cached')
                return file_name
            elif exists == 'error':
                raise IOError(f'{file_name} exists; delete to redownload')
            elif exists == 'overwrite':
                pass
            else:

                raise ValueError(
                    'exists must be warn, error, keep or overwrite; '
                    + f'you supplied {exists}'
                  )
        _response = self.get(url, **kwds)
        with open(file_name, 'wb') as file:
          file.write(_response._content)
        return file_name