A very basic interface to ESO’s COP API

copout.py

"""A very basic interface to the COP API.
See https://www.eso.org/copdemo/apidoc/ for API details.

To use:
    >>> login('52052', 'tutorial')
    >>> runs, __ = request('GET', '/obsRuns')
    >>> len(runs)
    31
    >>> muse_run = [r for r in runs if r['instrument'] == 'MUSE'][0]
    >>> ob, obEtag = request('POST',
    ...    '/containers/%d/items' % muse_run['containerId'],
    ...    {'itemType': 'OB', 'name': 'created by script'})
    >>> str(ob['name'])
    'created by script'
    >>> __ = request('DELETE', '/obsBlocks/%d' % ob['obId'], etag=obEtag)
"""
import json
try:
    from urllib.parse import urlencode
    from urllib.request import urlopen, Request
except ImportError:  # Python 2
    from urllib import urlencode
    import urllib2
    from contextlib import closing

    def urlopen(req):
        return closing(urllib2.urlopen(req))

    class Request(urllib2.Request):
        def get_method(self):
            return self.method

PUBLIC_DEMO = 'https://www.eso.org/copdemo'
PRODUCTION = 'https://www.eso.org/cop'

_baseurl = None
_auth = None


def login(username, password, serverurl=PUBLIC_DEMO):
    assert (serverurl.startswith('https://') or
            serverurl.startswith('http://')), "must use https or http scheme"
    global _baseurl, _auth
    _baseurl, _auth = serverurl, None
    h = {'Content-Type': 'application/x-www-form-urlencoded'}
    d = urlencode([('username', username), ('password', password)])
    req = Request(_baseurl + '/api/login', data=d.encode('utf8'), headers=h)
    req.method = 'POST'
    with urlopen(req) as res:
        body = res.read()
        j = json.loads(body.decode('utf8'))
        _auth = 'Bearer ' + j['access_token']


def request(method, endpoint, data=None, etag=None, headers={}):
    assert _auth is not None, "login() not called"
    h = {'Authorization': _auth, 'Content-Type': 'application/json'}
    h.update(headers)
    if etag:
        h['If-Match'] = etag
    if data is not None:
        if h['Content-Type'].startswith('application/json'):
            data = json.dumps(data, allow_nan=False).encode('utf8')
        elif h['Content-Type'].startswith('text/'):
            data = data.encode('utf8')
    req = Request(_baseurl + '/api/v1' + endpoint, data=data, headers=h)
    req.method = method
    with urlopen(req) as res:
        status, headers, body = res.getcode(), res.info(), res.read()
        etag = headers.get('ETag', None)
        if method in ('GET', 'POST', 'PUT') and status != 204:
            if headers.get('Content-Type', '').startswith('application/json'):
                body = json.loads(body.decode('utf8'))
            elif headers.get('Content-Type', '').startswith('text/'):
                body = body.decode('utf8')
            return body, etag
        else:
            return headers, etag
    return None, etag


if __name__ == '__main__':
    import doctest
    doctest.testmod()