The following
Python programs demonstrates how the
OAuth2 protocol can be used to access Frost.
#!/usr/bin/python
"""
Usage:
./example.py --help (and take it from there)
The program has been tested on the following platform:
- Python 2.7.12 on Ubuntu 16.04 Xenial
"""
import sys, os, json, argparse
import requests # See http://docs.python-requests.org/
from oauthlib.oauth2 import BackendApplicationClient
from requests_oauthlib import OAuth2Session
from traceback import format_exc
def parse_cmd_line_args():
parser = argparse.ArgumentParser()
parser.add_argument(
'-c', required = False, dest = 'client_id', default = os.environ.get('CLIENTID', None), metavar = '',
help = 'client ID (optionally set via CLIENTID environment variable)')
parser.add_argument(
'-x', required = False, dest = 'client_secret', default = os.environ.get('CLIENTSECRET', None), metavar = '',
help = 'client secret (optionally set via CLIENTSECRET environment variable)')
parser.add_argument(
'-t', required = False, dest = 'access_token', default = os.environ.get('ACCESSTOKEN', None), metavar = '',
help = 'access token (optionally set via ACCESSTOKEN environment variable)')
parser.add_argument('-s', required = True, dest = 'sources', metavar = '', help = 'query parameter sources')
parser.add_argument('-e', required = True, dest = 'elements', metavar = '', help = 'query parameter elements')
parser.add_argument('-r', required = True, dest = 'referencetime', metavar = '', help = 'query parameter referencetime')
return parser.parse_args(sys.argv[1:])
def ensure(var, name):
if var == None:
sys.stderr.write('error: {} not found\n'.format(name))
sys.exit(1)
return var
def get_basic_session(client_id):
session = requests.Session()
session.auth = (client_id, '')
return session
def get_oauth2_session(client_id, client_secret, access_token):
if access_token == None:
session = OAuth2Session(client = BackendApplicationClient(client_id = client_id))
ensure(client_secret, 'client secret')
try:
session.fetch_token(token_url = 'https://frost.met.no/auth/accessToken', client_id = client_id, client_secret = client_secret)
except:
raise Exception('failed to get access token: {}'.format(format_exc()))
else:
session = OAuth2Session(token = { 'access_token': access_token })
return session
def get_session(clargs):
if (clargs.access_token == None) and (clargs.client_secret == None):
session = get_basic_session(ensure(clargs.client_id, 'client ID')) # basic access is the only option
else:
# oauth2 access, either by creating a new access token or using an existing one
session = get_oauth2_session(clargs.client_id, clargs.client_secret, clargs.access_token)
return session
def get_observations(session, sources, elements, referencetime):
r = session.get(
'https://frost.met.no/observations/v0.jsonld?sources={}&elements={}&referencetime={}'.format(sources, elements, referencetime)
)
if r.status_code == 200:
print json.dumps(r.json()['data'], indent = 4, separators = (',', ': '))
else:
sys.stdout.write('error:\n')
sys.stdout.write('\tstatus code: {}\n'.format(r.status_code))
if 'error' in r.json():
assert(r.json()['error']['code'] == r.status_code)
sys.stdout.write('\tmessage: {}\n'.format(r.json()['error']['message']))
sys.stdout.write('\treason: {}\n'.format(r.json()['error']['reason']))
else:
sys.stdout.write('\tother error\n')
if __name__ == "__main__":
clargs = parse_cmd_line_args()
get_observations(get_session(clargs), clargs.sources, clargs.elements, clargs.referencetime)