fast_program_launch.ipynb

service.py

import requests
import json
import subprocess

def get_token(email, password, env):
    envs = {
        "dev": "https://atlas.dev.zengrc.io",
        "stg": "https://atlas.staging.zengrc.io",
        "prod": "https://id.roc.reciprocity.com",
    }
    result = subprocess.run(
        [
            "atlastokengen",
            "-email",
            email,
            "-password",
            password,
            "--authbackUrl",
            envs[env],
            "--hydraUrl",
            envs[env]
        ],
        stdout=subprocess.PIPE
    )
    return result.stdout.decode("utf-8")[:-1]


class Service:
    def __init__(self, url, token):
        self.url = url
        self.token = token

    def _headers(self):
        return {"Authorization": f"Bearer {self.token}"}

    def get(self, path):
        resp = requests.get(self.url + path, headers=self._headers())
        if resp.status_code not in {200, 201}:
            raise Exception(f"GET Request ({path}) failed ({resp.status_code}): {resp.text}")
        return resp.json()

    def post(self, path, body):
        resp = requests.post(self.url + path, headers=self._headers(), data=json.dumps(body))
        if resp.status_code not in {200, 201}:
            raise Exception(f"POST Request ({path}) failed ({resp.status_code}): {resp.text}")
        return resp.json()

    def put(self, path, body):
        resp = requests.put(self.url + path, headers=self._headers(), data=json.dumps(body))
        if resp.status_code not in {200, 201}:
            raise Exception(f"PUT Request ({path}) failed ({resp.status_code}): {resp.text}")
        return resp.json()

    def patch(self, path, body):
        resp = requests.patch(self.url + path, headers=self._headers(), data=json.dumps(body))
        if resp.status_code not in {200, 201}:
            raise Exception(f"POST Request ({path}) failed ({resp.status_code}): {resp.text}")
        return resp.json()

    def delete(self, path, body=None):
        if body:
            resp = requests.delete(self.url + path, headers=self._headers(), data=json.dumps(body))
        else:
            resp = requests.delete(self.url + path, headers=self._headers())
        if resp.status_code not in {200, 201}:
            raise Exception(f"DELETE Request ({path}) failed ({resp.status_code}): {resp.text}")
        return resp.json()