Core

Base classes and routes for authentication, and dealing with api responses.

DomoAuth encapsulates authentication. Can init with username/password or access_token
For consistency, use domo_api_request to handle API requests in route functions. domo_api_requests and route functions should always return an instance of ResponseGetData class. For streaming responses, use domo_api_stream_request to automagically download API responses in route functions.
Route Functions should always throw an error if not ResponseGetData.is_success

Features or classes are implemented as separate notebooks and then exported into their own module folder (see default_exp line in each .ipynb file)

each feature should be implemented as series of route functions (‘GET’, ‘PUT’, ‘POST’ requests), then a @dataclass class may be used to represent the feature with methods that wrap and provide logic around route functions.
- ex. an UPSERT method might wrap logic to test if the entity already exists (in which case call the PUT route) or create the entity if it doesn’t already exist (‘POST’)
@classmethods return an instance of the class
- ex. each class should have a get_by_id method which returns an instance of the entity (typically the result of the search datacenter API)

Implementations, take a set of features and create a use case or project - ex.back up code engine and custom app (features) into appdb collections (feature)

source

get_jupyter_account

 get_jupyter_account (account_name, domojupyter_fn:Callable)

Exported source

def get_jupyter_account(account_name, domojupyter_fn: Callable):
    creds_json = None
    i = 0

    while not creds_json:
        try:
            creds_json = json.loads(
                domojupyter_fn.get_account_property_value(account_name, "credentials")
            )

        except Exception as e:
            print(f"account error - retrying {e}")
            i = i + 1
            if i > 12:
                return
            time.sleep(5)

    return creds_json

handling authentication

source

DomoAuth

 DomoAuth (domo_instance:str, username:str=None, password:str=None,
           access_token:str=None, session_token:str=None)

Exported source

@dataclass
class DomoAuth:
    domo_instance: str

    username: str = None
    password: str = field(repr=False, default=None)
    access_token: str = field(repr=False, default=None)
    session_token: str = field(repr=False, default=None)

    @classmethod
    def from_creds_json(
        cls,
        creds_json: dict,
    ):
        username = creds_json.get("DOMO_USERNAME")
        password = creds_json.get("DOMO_PASSWORD")

        access_token = creds_json.get("accessToken") or creds_json.get("ACCESS_TOKEN")

        domo_instance = creds_json.get("instance_url")

        if username and password:
            return cls(
                username=username,
                password=password,
                access_token=access_token,
                domo_instance=domo_instance,
            )

    def generate_request_headers(self):
        headers = {
            "Accept": "application/json",
            "Content-Type": "application/json",
        }

        if self.session_token:
            headers.update({"x-domo-authentication": self.session_token})
            return headers

        if self.access_token:
            headers.update({"x-domo-developer-token": self.access_token})
            return headers

        raise Exception(
            "generate_request_headers: unable to authenticate request with provided Auth"
        )

    def who_am_i(self, return_raw: bool = False):
        """identify which credentials are being used in this Auth Object (useful for access_token based authentication)"""

        url = f"https://{self.domo_instance}.domo.com/api/content/v2/users/me"

        res = json.loads(
            requests.request(
                method="GET", headers=self.generate_request_headers(), url=url
            ).text
        )

        if return_raw:
            return res

        self.username = res["emailAddress"]

        return res

sample DomoAuth

auth = DomoAuth(
    domo_instance=os.environ["DOMO_INSTANCE"],
    access_token=os.environ["DOMO_ACCESS_TOKEN"],
)

print(auth.who_am_i(return_raw=False))
auth

{'id': 1893952720, 'invitorUserId': 587894148, 'displayName': 'Jae Wilson1', 'department': 'Business Improvement', 'userName': 'jae@onyxreporting.com', 'emailAddress': 'jae@datacrew.space', 'avatarKey': 'c605f478-0cd2-4451-9fd4-d82090b71e66', 'accepted': True, 'userType': 'USER', 'modified': 1721847526952, 'created': 1588960518, 'active': True, 'pending': False, 'anonymous': False, 'systemUser': False}

DomoAuth(domo_instance='domo-community', username='jae@datacrew.space')

sending requests

source

ResponseGetData

 ResponseGetData (auth:__main__.DomoAuth, response:dict, is_success:bool,
                  status:int, download_path:str=None)

Exported source

class API_Exception(Exception):

    def __init__(self, res, message: str = None):
        message = message or ""
        base = f" || {str(res.status)} - {res.response.get('message') or res.response.get('statusReason')} || {res.auth.domo_instance}"
        message += base

        super().__init__(message)


class Class_Exception(Exception):

    def __init__(self, cls, auth=None, message: str = None):
        cls_instance = cls

        if hasattr(cls, "__cls__"):
            cls_instance = cls
            cls = cls.__cls__

        domo_instance = {
            (cls_instance and cls_instance.auth.domo_instance)
            or (auth and auth.domo_instance)
        }

        message = f"{message or 'error'} || {cls.__name__}"

        if domo_instance:
            message = f"{message} || {domo_instance}"

        super().__init__(message)


@dataclass
class ResponseGetData:
    auth: DomoAuth = field(repr=False)
    response: dict
    is_success: bool
    status: int
    download_path: str = None

    @classmethod
    def from_response(cls, res, auth: DomoAuth):

        try:
            if res.text == "" or not res.text:
                return cls(
                    response=res.text,
                    is_success=res.ok,
                    status=res.status_code,
                    auth=auth,
                )

            return cls(
                response=json.loads(res.text),
                is_success=res.ok,
                status=res.status_code,
                auth=auth,
            )

        except Exception as e:
            print({"rgd.from_response": {"text": res.text, "err": e}})

            raise e

    @staticmethod
    def _write_stream(res: requests.Response, file_path: str, stream_chunks=8192):

        dmut.upsert_folder(file_path)

        with open(file_path, "wb") as fd:
            for chunk in res.iter_content(stream_chunks):
                fd.write(chunk)

        print("done writing stream")
        return True

    @staticmethod
    def read_stream(download_path):
        with open(download_path, "rb") as f:
            return f.read()

    @classmethod
    def from_stream(cls, res: requests.Response, download_path: str, auth: DomoAuth):

        if not res.ok:
            return cls.from_response(res=res, auth=auth)

        cls._write_stream(res, download_path)

        return cls(
            response=True if cls.read_stream(download_path) else False,
            is_success=res.ok,
            status=res.status_code,
            auth=auth,
            download_path=download_path,
        )

source

Class_Exception

 Class_Exception (cls, auth=None, message:str=None)

Common base class for all non-exit exceptions.

source

API_Exception

 API_Exception (res, message:str=None)

Common base class for all non-exit exceptions.

source

domo_api_request

 domo_api_request (auth:__main__.DomoAuth, endpoint, request_type,
                   params=None, headers=None, body=None,
                   return_raw:bool=False, debug_api:bool=False, timeout=3)

Exported source

def domo_api_request(
    auth: DomoAuth,
    endpoint,
    request_type,
    params=None,
    headers=None,
    body=None,
    return_raw: bool = False,
    debug_api: bool = False,
    timeout=3,
):

    url = f"https://{auth.domo_instance}.domo.com{endpoint}"

    headers = headers or {}

    headers = {**auth.generate_request_headers(), **headers}

    if debug_api:

        print("🐛 debugging domo_api_request")
        pprint(
            {
                "url": url,
                "headers": headers,
                "request_type": request_type,
                "params": params,
                "body": body,
            }
        )

    if request_type.lower() == "post":
        response = requests.post(
            url, json=body, headers=headers, params=params, timeout=timeout
        )

    elif request_type.lower() == "get":
        response = requests.get(url, headers=headers, params=params, timeout=timeout)

    else:
        raise Exception(
            f'domo_api_request method "{request_type.lower()}" not implemented yet.'
        )

    if return_raw:
        return response

    return ResponseGetData.from_response(res=response, auth=auth)

source

looper

 looper (auth:__main__.DomoAuth, arr_fn:Callable, endpoint:str,
         request_type='GET', params:dict=None, body:dict=None, offset=0,
         limit=50, offset_params:dict=None,
         offset_params_is_header:bool=False, debug_api:bool=False,
         debug_loop:bool=False, return_raw:bool=False)

	Type	Default	Details
auth	DomoAuth
arr_fn	Callable
endpoint	str
request_type	str	GET
params	dict	None
body	dict	None
offset	int	0
limit	int	50
offset_params	dict	None	format {“offset” : <> , “limit” : <>}
offset_params_is_header	bool	False	should offset parameters be passed in the header or body?
debug_api	bool	False
debug_loop	bool	False
return_raw	bool	False	will break the looper after the first request and ignore the array processing step.

Exported source

looper_offset_params = {
    "offset": "offset",
    "limit": "limit",
}  # what are the offset parameters called that handle pagination?


def looper(
    auth: DomoAuth,
    arr_fn: Callable,
    endpoint: str,
    request_type="GET",
    params: dict = None,
    body: dict = None,
    offset=0,
    limit=50,
    offset_params: dict = None,  # format {"offset" : <<value>> , "limit" : <<value>>}
    offset_params_is_header: bool = False,  # should offset parameters be passed in the header or body?
    debug_api: bool = False,
    debug_loop: bool = False,
    return_raw: bool = False,  # will break the looper after the first request and ignore the array processing step.
):
    params = params or {}
    body = body or {}
    offset_params = offset_params or looper_offset_params

    final_array = []
    keep_looping = True

    while keep_looping:
        new_offset = {
            offset_params["offset"]: offset,
            offset_params["limit"]: limit,
        }
        if offset_params_is_header:
            params = deepcopy({**params, **new_offset})

        else:
            body = {**body, **new_offset}

        res = domo_api_request(
            auth=auth,
            endpoint=endpoint,
            request_type=request_type,
            params=params,
            debug_api=debug_api,
            body=body,
        )

        if res.status == 429:
            print("sleeping in timeout")
            time.sleep(10)
            debug_loop = True
            keep_looping = True

        elif not res.is_success or return_raw:
            return res

        else:
            new_array = arr_fn(res)

            if debug_loop:
                pprint("🔁 debug_loop")
                pprint(
                    {
                        "params": params,
                        "body": body,
                        # "new_array": new_array[0:1]
                    }
                )

            if not new_array or len(new_array) == 0:
                keep_looping = False

            if len(new_array) < limit:
                keep_looping = False

            final_array += new_array

            offset += limit

    res.response = final_array

    return res

source

domo_api_stream_request

 domo_api_stream_request (auth:__main__.DomoAuth, endpoint, request_type,
                          download_path, params=None, headers=None,
                          debug_api:bool=False, timeout=3)

Exported source

def domo_api_stream_request(
    auth: DomoAuth,
    endpoint,
    request_type,
    download_path,
    params=None,
    headers=None,
    debug_api: bool = False,
    timeout=3,
):

    url = f"https://{auth.domo_instance}.domo.com{endpoint}"

    headers = headers or {}

    headers = {**auth.generate_request_headers(), **headers}

    if debug_api:

        print("debugging domo_api_request")

        pprint(
            {
                "url": url,
                "headers": headers,
                "request_type": request_type,
                "params": params,
            }
        )

    res = requests.get(
        url=url, headers=headers, params=params, timeout=timeout, stream=True
    )

    if not res.ok:
        return ResponseGetData.from_response(res=res, auth=auth)

    return ResponseGetData.from_stream(res=res, download_path=download_path, auth=auth)