Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from pydomo.DomoAPIClient import DomoAPIClient
from pydomo.Transport import HTTPMethod
import requests
"""
Group Client
- Programmatically manage Domo User Groups
- Docs: https://developer.domo.com/docs/domo-apis/group-apis
"""
class GroupClient(DomoAPIClient):
def __init__(self, transport, logger):
super(GroupClient, self).__init__(transport, logger)
self.urlBase = '/v1/groups/'
self.groupDesc = "Group"
"""
Create a Group
"""
def create(self, group_request):
return self._create(self.urlBase, group_request, {}, self.groupDesc)
"""
Get a Group
"""
def get(self, group_id):
return self._get(self._base(group_id), self.groupDesc)
from pydomo.DomoAPIClient import DomoAPIClient
from pydomo.Transport import HTTPMethod
"""
Streams
- Programmatically manage Domo Streams
- A Domo Stream is a specialized upload pipeline pointing to a single Domo DataSet
- Use Streams for massive, constantly changing, or rapidly growing data sources
- Streams support uploading data sources in parts, in parallel
- Use plain DataSets for data sources that only require occasional updates via replacement
- Usage: Create a Stream, create an Execution, upload data via the Execution, then 'commit' the Execution
- Docs: https://developer.domo.com/docs/data-apis/data
"""
class StreamClient(DomoAPIClient):
def __init__(self, transport, logger):
super(StreamClient, self).__init__(transport, logger)
self.urlBase = '/v1/streams/'
self.streamDesc = "Stream"
self.executionDesc = "Execution"
"""
Create a Stream and DataSet
- A Stream is an upload pipeline attached to the newly created DataSet
- Currently, Streams cannot be applied to existing DataSets
- Deleting a Stream does not delete the associated DataSet
"""
def create(self, stream_request):
return self._create(self.urlBase, stream_request, {}, self.streamDesc)
"""
from pydomo.Transport import HTTPMethod
"""
DataSets
- Programmatically manage Domo DataSets
- Use DataSets for fairly static data sources that only require occasional updates via data replacement
- Use Streams if your data source is massive, constantly changing, or rapidly growing
- Docs: https://developer.domo.com/docs/data-apis/data
"""
DATA_SET_DESC = "DataSet"
PDP_DESC = "Personalized Data Policy (PDP)"
URL_BASE = '/v1/datasets'
class DataSetClient(DomoAPIClient):
def __init__(self, transport, logger):
super(DataSetClient, self).__init__(transport, logger)
"""
Create a DataSet
"""
def create(self, dataset_request):
return self._create(URL_BASE, dataset_request, {}, DATA_SET_DESC)
"""
Get a DataSet
"""
def get(self, dataset_id):
url = '{base}/{dataset_id}'.format(
base=URL_BASE, dataset_id=dataset_id)
return self._get(url, DATA_SET_DESC)