<jupyter_start><jupyter_text>First we register a test user in SQuaSH and get an access token You may be running the [squash-restful-api](https://github.com/lsst-sqre/squash-rest-api) app locally, in this case use `SQUASH_API_URL=http://localhost:5000`, or you could use the deployed demo at https://squash-restful-api-demo.lsst.codes if available.You can check the API documentation at https://sqr-009.lsst.io/the-squash-restful-api<jupyter_code>#SQUASH_API_URL = "http://localhost:5000"
SQUASH_API_URL = "https://squash-restful-api-demo.lsst.codes"
import requests

user={'username': 'testuser', 'password': '<PASSWORD>'}
r = requests.post(SQUASH_API_URL + "/register", json=user)
r.json()

r = requests.post(SQUASH_API_URL + "/auth", json=user)
r.json()<jupyter_output><empty_output><jupyter_text>Add the access token into the authorization header for the next requests.<jupyter_code>headers = dict()
headers['Authorization'] = 'JWT ' + r.json()['access_token']<jupyter_output><empty_output><jupyter_text>We'll use a test verification job You can reproduce this verification job document with the following command:`$ dispatch_verify.py --test --env jenkins --lsstsw $(pwd) Cfht_output_r.json --write verify_job.json`where `Cfht_output_r.json` is an output from `validate_drp`.<jupyter_code>import json
with open("data/verify_job.json", "r") as f:
     data = json.load(f)<jupyter_output><empty_output><jupyter_text>A verification job is a container for measurements, metadata, data blobs as well as metric definitions and specifications. Metric definitions and specifications are pre-loaded to SQuaSH and only measurements, metadata and data blobs must be sent to SQuaSH each time a new verification job is registered. For the purpose of testing the API we will exercise loading metric definitions and specifications here too.<jupyter_code>data.keys()
data['blobs'][0].keys()<jupyter_output><empty_output><jupyter_text>Loading metric definitions<jupyter_code>metrics={'metrics': data['metrics']}
r = requests.post(SQUASH_API_URL + "/metrics", json=metrics, headers=headers)
r.json()<jupyter_output><empty_output><jupyter_text>Updating a metric definition<jupyter_code>metric = data['metrics'][0]
metric
r = requests.put(SQUASH_API_URL + "/metric/{}".format(metric['name']), json=metric, headers=headers)
r.json()
r = requests.get(SQUASH_API_URL + "/metric/{}".format(metric['name']))
r.json()<jupyter_output><empty_output><jupyter_text>Loading specifications<jupyter_code>specs={'specs': data['specs']}
r = requests.post(SQUASH_API_URL + "/specs", json=specs, headers=headers)
r.json()<jupyter_output><empty_output><jupyter_text>Update a specification<jupyter_code>spec = data['specs'][0]
spec
r = requests.put(SQUASH_API_URL + "/spec/{}".format(spec['name']), json=spec, headers=headers)
r.json()
r = requests.get(SQUASH_API_URL + "/spec/{}".format(spec['name']))
r.json()<jupyter_output><empty_output><jupyter_text>Create a job In addition to measurements and blobs, add job metatada which consists of (arbitrary) metadata, packages and environment metadata. Specify the dataset for this job<jupyter_code>data['meta']['env']['ci_dataset']='validation_data_cfht'

%%time
r = requests.post(SQUASH_API_URL + "/auth", json=user)
headers['Authorization'] = 'JWT ' + r.json()['access_token']
job = {'measurements': data['measurements'], 'meta': data['meta'], 'blobs': data['blobs']}
r = requests.post(SQUASH_API_URL + "/job", json=job, headers=headers)
r.json()
%%time
r = requests.post(SQUASH_API_URL + "/auth", json=user)
headers['Authorization'] = 'JWT ' + r.json()['access_token']
job = {'measurements': data['measurements'], 'meta': data['meta'], 'blobs': []}
r = requests.post(SQUASH_API_URL + "/job", json=job, headers=headers)
r.json()<jupyter_output><empty_output><jupyter_text>Finally we generate some verification jobs for testing varying the enviroment metadata as needed Job metatada consists of (arbitrary) metadata, packages and environment metadata<jupyter_code>import random
import time
datasets=['cfht', 'decam', 'hsc']
for ci_id in range(10):
    data['meta']['env']['ci_id']=str(ci_id)
    data['meta']['env']['ci_url']="https://example.com/{}".format(ci_id)
    data['meta']['env']['ci_dataset']=random.choice(datasets)
    job = {'measurements': data['measurements'], 'meta': data['meta'], 'blobs': data['blobs']}
    # sleep 15s to have the jobs created at different times
    time.sleep(15)
    r = requests.post(SQUASH_API_URL + "/job", json=job, headers=headers)
    r.json()<jupyter_output><empty_output><jupyter_text>You can retrieve a job from SQuaSH using<jupyter_code>r = requests.get(SQUASH_API_URL + "/job/1")
#r.json()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15210425122837, 'fraction_numerical': 0.004486220892971587, 'mean_word_length': 7.185314685314685, 'pattern_counts': {'":': 0, '<': 46, '<?xml version=': 0, '>': 46, 'https://': 5, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '21312', 'license': 'MIT', 'n_tokens_mistral': 1589, 'n_tokens_neox': 1569, 'n_words': 363, 'path': 'tests/test_api.ipynb', 'repo_name': 'lsst-sqre/squash-rest-api', 'size': '4678'}