Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def streams(self, domo):
'''Streams are useful for uploading massive data sources in
chunks, in parallel. They are also useful with data sources that
are constantly changing/growing.
Streams Docs: https://developer.domo.com/docs/data-apis/data
'''
domo.logger.info("\n**** Domo API - Stream Examples ****\n")
streams = domo.streams
# Define a DataSet Schema to populate the Stream Request
dsr = DataSetRequest()
dsr.name = 'Leonhard Euler Party'
dsr.description = 'Mathematician Guest List'
dsr.schema = Schema([Column(ColumnType.STRING, 'Friend'),
Column(ColumnType.STRING, 'Attending')])
# Build a Stream Request
stream_request = CreateStreamRequest(dsr, UpdateMethod.APPEND)
# Create a Stream w/DataSet
stream = streams.create(stream_request)
domo.logger.info("Created Stream {} containing the new DataSet {}"
.format(stream['id'], stream['dataSet']['id']))
# Get a Stream's metadata
retrieved_stream = streams.get(stream['id'])
domo.logger.info("Retrieved Stream {} containing DataSet {}".format(
retrieved_stream['id'], retrieved_stream['dataSet']['id']))
# List Streams
limit = 1000
# Get a DataSets's metadata
retrieved_dataset = datasets.get(dataset['id'])
domo.logger.info("Retrieved DataSet " + retrieved_dataset['id'])
# List DataSets
dataset_list = list(datasets.list(sort=Sorting.NAME, limit=10))
domo.logger.info("Retrieved a list containing {} DataSet(s)".format(
len(dataset_list)))
# Update a DataSets's metadata
update = DataSetRequest()
update.name = 'Leonhard Euler Party - Update'
update.description = 'Mathematician Guest List - Update'
update.schema = Schema([Column(ColumnType.STRING, 'Friend'),
Column(ColumnType.STRING, 'Attending')])
updated_dataset = datasets.update(dataset['id'], update)
domo.logger.info("Updated DataSet {}: {}".format(updated_dataset['id'],
updated_dataset['name']))
# Import Data from a string
csv_upload = '"Pythagoras","FALSE"\n"Alan Turing","TRUE"\n' \
'"George Boole","TRUE"'
datasets.data_import(dataset['id'], csv_upload)
domo.logger.info("Uploaded data to DataSet " + dataset['id'])
# Export Data to a string
include_csv_header = True
csv_download = datasets.data_export(dataset['id'], include_csv_header)
domo.logger.info("Downloaded data from DataSet {}:\n{}".format(
dataset['id'], csv_download))