Stream

A stream refers to how data gets into Domo BUT does not necessarily refer to streaming data / low latency data ingest. In this context it is more accurate to think of a stream as a path for data to travel down to get into Domo’s data storage environment Vault (Amazon S3)

auth = dmda.DomoTokenAuth(
    domo_instance=os.environ["DOMO_INSTANCE"],
    domo_access_token=os.environ["DOMO_ACCESS_TOKEN"],
)

await auth.print_is_token()


res = await stream_routes.get_streams(auth=auth, maximum=5, loop_until_end=False)

stream = res.response[0]
stream_id = stream["id"]
stream

🎉 token_auth token retrieved from domo-community ⚙️

{'id': 1,
 'valid': True,
 'invalidExecutionId': None,
 'transport': {'type': 'GLOBAL', 'description': None, 'version': '1'},
 'updateMethod': 'REPLACE',
 'dataProvider': {'id': 0,
  'key': 'publicsampledata',
  'name': None,
  'url': None,
  'defaultConnectorId': None,
  'defaultConnectorLabel': None,
  'authenticationScheme': None,
  'connectorValidatorPresent': False,
  'resourceBundlePresent': False,
  'moduleHandler': None,
  'iconPicker': False,
  'unsafe': False,
  'scope': None,
  'producer': None,
  'version': None,
  'dateCreated': None,
  'dateUpdated': None,
  'visibility': 'WITH_CONNECTOR',
  'type': 'STANDARD',
  'customerOAuthConfigurationDataProviderType': None,
  'authenticationSchemeConfiguration': None},
 'account': None,
 'accounts': [],
 'accountTemplate': None,
 'dataSource': {'id': 'd40b1f41-8847-4d2a-884b-d45f3ba1aaec',
  'adc': False,
  'adcExternal': False,
  'hidden': False},
 'schemaDefinition': None,
 'scheduleExpression': None,
 'scheduleStartDate': None,
 'advancedScheduleJson': None,
 'scheduleRetryExpression': None,
 'scheduleRetryCount': 0,
 'lastExecution': None,
 'lastSuccessfulExecution': None,
 'currentExecution': None,
 'currentExecutionState': 'IDLE',
 'createdAt': 1580425528.0,
 'createdBy': 27,
 'modifiedAt': 1580425528.0,
 'modifiedBy': 27,
 'scheduleState': 'ACTIVE',
 'scheduleAssertion': False,
 'inactiveScheduleCode': None,
 'configuration': [{'streamId': 1,
   'category': 'METADATA',
   'name': 'isSampleData',
   'type': 'string',
   'value': 'true'}]}

---

source

StreamConfig_Mappings

 StreamConfig_Mappings (value, names=None, module=None, qualname=None,
                        type=None, start=1)

An enumeration.

---

source

StreamConfig_Mapping

 StreamConfig_Mapping (data_provider_type:str, sql:str=None,
                       warehouse:str=None, database_name:str=None,
                       s3_bucket_category:str=None, is_default:bool=False,
                       table_name:str=None, src_url:str=None,
                       google_sheets_file_name:str=None,
                       adobe_report_suite_id:str=None,
                       qualtrics_survey_id:str=None)

data_provider_type = "aws-athena"
StreamConfig_Mappings.search(data_provider_type)
# StreamConfig_Mappings('hello world')

<StreamConfig_Mappings.aws_athena: StreamConfig_Mapping(data_provider_type='aws-athena', sql='query', warehouse=None, database_name='databaseName', s3_bucket_category=None, is_default=False, table_name='tableName', src_url=None, google_sheets_file_name=None, adobe_report_suite_id=None, qualtrics_survey_id=None)>

---

source

StreamConfig

 StreamConfig (stream_category:str, name:str, type:str, value:str,
               value_clean:str=None, parent:Any=None)

---

source

DomoStream

 DomoStream (auth:domolibrary.client.DomoAuth.DomoAuth, id:str,
             dataset_id:str, parent:Any=None,
             transport_description:str=None, transport_version:int=None,
             update_method:str=None, data_provider_name:str=None,
             data_provider_key:str=None, account_id:str=None,
             account_display_name:str=None, account_userid:str=None,
             has_mapping:bool=False,
             configuration:List[__main__.StreamConfig]=<factory>,
             configuration_tables:List[str]=<factory>,
             configuration_query:str=None)

---

source

Dataset_Stream_GET_Error

 Dataset_Stream_GET_Error (cls_instance, message)

base exception

# StreamConfig.from_json(obj = stream['configuration'],
#                        data_provider_type= stream['dataProvider']['key'])

await DomoStream.get_stream_by_id(auth=auth, stream_id=stream_id)

DomoStream(id=1, dataset_id='d40b1f41-8847-4d2a-884b-d45f3ba1aaec', transport_description=None, transport_version='1', update_method='REPLACE', data_provider_name='Public Sample Data', data_provider_key='publicsampledata', account_id=None, account_display_name=None, account_userid=None, has_mapping=True, configuration=[StreamConfig(stream_category=None, name='isSampleData', type='string', value='true', value_clean=None)], configuration_tables=[], configuration_query=None)