token_auth = dmda.DomoTokenAuth(
domo_instance="domo-alpha",
domo_access_token=os.environ["ALPHA_ACCESS_TOKEN"],
)Job
A job refers to a scheduled task in an application.
For the purposes of this beta class, the only Job type implemented has been Watchdog. This class will be heavily revised in a future development sprint.
An additional Job type might be PDP Automation Toolkit.
RemoteDomoStats_Config
RemoteDomoStats_Config (policies:List[__main__.RemoteDomoStats_Config_Po licy]=<factory>)
RemoteDomoStats_Config_Policy
RemoteDomoStats_Config_Policy (type:str, dataset_id:str)
DomoJob_RemoteDomoStats
DomoJob_RemoteDomoStats (auth:domolibrary.client.DomoAuth.DomoAuth, name:str, application_id:str, logs_dataset_id:str=None, id:str=None, user_id:str=None, execution_timeout:int=1440, is_enabled:bool=False, customer_id:str=None, created_dt:datetime.datetime=None, updated_dt:datetime.datetime=None, description:str=None, execution_payload:dict=<factory>, share_state:dict=<factory>, accounts:List[str]=<factory>, triggers:List[domo library.classes.DomoApplication_Job_Base.DomoTri gger]=<factory>, remote_instance:str=None, subscriber_job_id:str=None, Config:__main__.RemoteDomoStats_Config=None)
token_auth = dmda.DomoTokenAuth(
domo_instance="domo-alpha",
domo_access_token=os.environ["ALPHA_ACCESS_TOKEN"],
)
res = await application_routes.get_applications(auth=token_auth)
application_id = next(
(
app["applicationId"]
for app in res.response
if app["name"] == "Toolkit: Remote Domo Stats"
)
)
res = await application_routes.get_application_jobs(
application_id=application_id, auth=token_auth
)
job_id = res.response[1]["jobId"]
test_job = await DomoJob_RemoteDomoStats.get_by_id(
job_id=job_id,
application_id=application_id, auth=token_auth, return_raw=False
)
# pprint(test_job.response)
pprint(test_job)DomoJob_RemoteDomoStats(name='REMOTE_Users',
application_id='50e7230f-d2f2-42e2-a208-d94c8ae9f64c',
logs_dataset_id='c29116fa-c3ad-4bc9-9de5-0f05ce7e7649',
id='b4fd853a-1524-4512-9e4f-1d7a809913db',
user_id=1623162654,
execution_timeout=1440,
is_enabled=False,
customer_id='domo-alpha',
created_dt=datetime.datetime(2024, 3, 8, 21, 35, 46),
updated_dt=datetime.datetime(2024, 9, 19, 21, 30, 25),
description='updated - 2024-03-08 17:50:24.269350',
execution_payload={'metricsDatasetId': 'c29116fa-c3ad-4bc9-9de5-0f05ce7e7649',
'policies': {'People': '7a7caccc-4724-470d-b89f-e5f6d27676a8'},
'remoteInstance': 'domo-alpha.domo.com'},
share_state={'sharedEntities': [{'accessLevel': 'OWNER',
'id': '1623162654',
'type': 'USER'}]},
accounts=[10],
triggers=[],
remote_instance='domo-alpha',
subscriber_job_id=None,
Config=RemoteDomoStats_Config(policies=[RemoteDomoStats_Config_Policy(type='People',
dataset_id='7a7caccc-4724-470d-b89f-e5f6d27676a8')]))pprint(test_job.to_json()){'accounts': [10],
'applicationId': '50e7230f-d2f2-42e2-a208-d94c8ae9f64c',
'customerId': 'domo-alpha',
'executionPayload': {'metricsDatasetId': 'c29116fa-c3ad-4bc9-9de5-0f05ce7e7649',
'policies': {'People': '7a7caccc-4724-470d-b89f-e5f6d27676a8'},
'remoteInstance': 'domo-alpha',
'subscriberJobId': None},
'executionTimeout': 1440,
'jobDescription': 'updated - 2024-03-08 17:50:24.269350',
'jobId': 'b4fd853a-1524-4512-9e4f-1d7a809913db',
'jobName': 'REMOTE_Users',
'shareState': {'sharedEntities': [{'accessLevel': 'OWNER',
'id': '1623162654',
'type': 'USER'}]},
'triggers': [],
'userId': 1623162654}RemoteDomoStats CRUD
DomoJob_RemoteDomoStats.create
DomoJob_RemoteDomoStats.create (auth:domolibrary.client.DomoAuth.DomoAut h, name:str, config:__main__.RemoteDomoStats_Config, application_id:str, logs_dataset_id:str, description:str='created via domolibrary f2025-03-11', remote_instance:str=None, accounts:List[int]=None, triggers:List[do molibrary.classes.DomoApplication_Job_Bas e.DomoTrigger_Schedule]=None, execution_timeout:int=1440, return_raw:bool=False, debug_api:bool=False, debug_num_stacks_to_drop=2, session:Optional[httpx.AsyncClient]=None)
Exported source
@patch_to(DomoJob_RemoteDomoStats, cls_method=True)
async def create(
cls,
auth: dmda.DomoAuth,
name: str,
config: RemoteDomoStats_Config,
application_id: str,
logs_dataset_id: str,
description: str = f"created via domolibrary f{dt.date.today()}",
remote_instance: str = None,
accounts: List[int] = None,
triggers: List[DomoTrigger_Schedule] = None,
execution_timeout: int = 1440,
return_raw: bool = False,
debug_api: bool = False,
debug_num_stacks_to_drop=2,
session: Optional[httpx.AsyncClient] = None,
):
triggers_ls = []
if triggers is not None and len(triggers) > 0:
triggers_ls = [
DomoTrigger(id=None, job_id=None, schedule=schedule)
for schedule in triggers
]
domo_job = cls(
application_id=application_id,
auth=auth,
name=name,
logs_dataset_id=logs_dataset_id,
accounts=accounts,
description=description,
remote_instance=remote_instance,
Config=config,
triggers=triggers_ls,
execution_timeout=execution_timeout,
)
body = domo_job.to_json()
res = await application_routes.create_application_job(
auth=auth,
application_id=application_id,
body=body,
parent_class=cls.__name__,
session=session,
debug_api=debug_api,
debug_num_stacks_to_drop=debug_num_stacks_to_drop,
)
if return_raw:
return res
return cls._from_json(res.response, auth=auth)token_auth = dmda.DomoTokenAuth(
domo_instance=os.environ['DOMO_INSTANCE'],
domo_access_token=os.environ["DOMO_ACCESS_TOKEN"],
)
config = RemoteDomoStats_Config._from_json(
{"AccountPermissions": "b156bfe4-d0e5-4410-8354-b3062973a99c"}
)
# await DomoJob_RemoteDomoStats.create(
# auth = token_auth,
# application_id="50e7230f-d2f2-42e2-a208-d94c8ae9f64c",
# logs_dataset_id = 'da68b3ea-088f-4493-ae69-ee5094feaa29' ,
# name="domo-community v3",
# accounts=[96],
# remote_instance=os.environ['DOMO_INSTANCE'],
# config=config,
# return_raw = True
# )