SourcesMixin

SourcesMixin#

class SourcesMixin#

Bases: object

Methods Summary

`delete_source`(project_id, source_id)	Delete an existing Source.
`get_max_inc`(project_id, source_id)	Fetches the maximum incremental value of a source.
`get_preview`(project_id, config)	Preview the source configuration.
`get_preview_post`(project_id, data)	Preview the source configuration using a POST request.
`get_source`(project_id, source_id[, ...])	Get source details.
`get_source_logs`(project_id, source_id, ...)	Get the run logs of a particular source run.
`get_sources`(project_id[, include_config, ...])	Get all sources for the provided project.
`get_sources_v1`(project_id[, search, status, ...])	Get sources for the provided project.
`kill_source`(project_id, source_id)	Try to terminate (SIGTERM) a dataload job if it is already running.
`modify_source`(project_id, source_id[, name, ...])	Modify an existing source.
`new_source`(project_id, name, config[, ...])	Create a new source.
`pause_source`(project_id, source_id)	Pause a source.
`reset_source`(project_id, source_id[, ...])	Resets and run the source.
`resume_source`(project_id, source_id)	Resume a paused source.
`retry_failed_batches`(project_id, source_id)	Move all the failed batches of a given source back for processing.
`run_source`(project_id, source_id)	Runs a source now.
`set_max_inc`(project_id, source_id, max_inc_value)	Sets the maximum incremental value of the incremental column of a source.

Methods Documentation

delete_source(project_id, source_id)#

Delete an existing Source.

Parameters:

project_id – Project identifier.
source_id – Source identifier.

Example:

>>> client.delete_source('Vlg5Z1hOShm0eYmjtsqSqg',
...                      'oTvI6rlaRmKvmYCfCvLwpw')

get_max_inc(project_id, source_id)#

Fetches the maximum incremental value of a source.

Parameters:

project_id – Project identifier.
source_id – Source identifier.

Example:

>>> client.get_max_inc(
...     'Vlg5Z1hOShm0eYmjtsqSqg',
...     'hw8j7LUBRM28-jAellgQdA')

get_preview(project_id, config)#

Preview the source configuration.

Parameters:

project_id – Project identifier.
config – Provider configuration.

Returns:

A dictionary which contains the source preview items.

Example:

>>> client.get_preview(
...     project_id,
...     config={
...         "dataloader_plugin_options": {
...             "source_file": "path:/tmp/test.csv"
...         },
...         "dataloader_options": {
...             "plugin_name": "csv_plugin",
...             "project_id": project_id,
...             "map_title": "title"
...             }
...         })

{
  "count": 2,
  "items": [
    {
      "id": "CTHQDLwzQsOq93zHAUcCRg",
      "name": "name01",
      "title": "title01"
    },
    {
      "id": ",yYNWBDgQQ2Uhuz32boDAg",
      "name": "name02",
      "title": "title02"
    }
  ],
  "data_schema": [
    "name",
    "title"
  ]
}

get_preview_post(project_id, data)#

Preview the source configuration using a POST request.

Unlike the get_preview method, which sends parameters via a GET request, this method uses POST to include data in the request body. This allows sending larger payloads.

Parameters:

project_id – Project identifier.
data – Data with provider configuration.

Returns:

A dictionary which contains the source preview items.

Example:

>>> client.get_preview(
...     project_id,
...     data =
...       "config": {
...           "dataloader_plugin_options": {
...               "source_file": "path:/tmp/test.csv"
...           },
...           "dataloader_options": {
...               "plugin_name": "csv_plugin",
...               "project_id": project_id,
...               "map_title": "title"
...               }
...     }})

{
  "count": 2,
  "items": [
    {
      "id": "CTHQDLwzQsOq93zHAUcCRg",
      "name": "name01",
      "title": "title01"
    },
    {
      "id": ",yYNWBDgQQ2Uhuz32boDAg",
      "name": "name02",
      "title": "title02"
    }
  ],
  "data_schema": [
    "name",
    "title"
  ]
}

get_source(project_id, source_id, include_config=None, include_run_stats=None, include_pipeline_backlog=None)#

Get source details.

Parameters:

project_id – Project identifier.
source_id – Source identifier.
include_config – Bool, whether or not to include the config for the Source.
include_run_stats – Bool, whether or not to include the run stats for the Source.
include_pipeline_backlog – Bool, whether or not to include the the backlog of items in the data pipeline for this source.

Returns:

A dictionary which contains the source.

Example:

>>> client.get_source(
...     project_id='Vlg5Z1hOShm0eYmjtsqSqg',
...     source_id='pqTn4vBZRdS5hYw0TBt0pQ',
...     include_config=True,
...     include_run_stats=True,
...     include_pipeline_backlog=True)

{
  "items_fetched_total": 2,
  "last_error": "",
  "last_check_at": "2019-01-23T10:23:23",
  "last_items_at": "2019-01-23T10:23:23",
  "paused": false,
  "error_count": 0,
  "id": "pqTn4vBZRdS5hYw0TBt0pQ",
  "total_error_count": 0,
  "project_id": "Vlg5Z1hOShm0eYmjtsqSqg",
  "config": {
    "dataloader_plugin_options": {
      "source_file": "path:/tmp/test.csv"
    },
    "dataloader_options": {
      "map_title": "title",
      "project_id": "Vlg5Z1hOShm0eYmjtsqSqg",
      "plugin_name": "csv_plugin"
    }
  },
  "status": "complete",
  "total_runs": 1,
  "pipeline_workflow_id": "S0fVQ-K0TmS1UgT0msZRBA",
  "last_error_at": null,
  "last_update_at": "2019-01-23T10:23:23",
  "last_success_at": "2019-01-23T10:23:23",
  "items_fetched_last_run": 2,
  "tenant": "squirro",
  "next_run_time_at": "2019-01-23T10:52:51",
  "name": "test source",
  "scheduling_options": {
    "repeat": "30m",
    "schedule": true
  },
  "created_at": "2019-01-23T10:23:23",
  "modified_at": "2019-01-23T10:23:23",
  "processed": true,
  "pipeline_backlog": 10
}

Note

Credentials are never returned. Options the platform classifies as credentials are omitted from config, and a flat secret_fields map is returned alongside it reporting which ones are set:

"secret_fields": {"access_key_id": {"is_set": true}}

is_set is presence, not truthiness – a stored empty string still reports true – and no length, digest or member count is exposed. The key is present only on responses that carry config.

config itself is omitted entirely when the source’s plugin manifest cannot be read, so that a client cannot round-trip a partial config back and destroy the stored one. Treat a missing config as “unknown”, not as “empty”.

get_source_logs(project_id, source_id, last_n_log_lines)#

Get the run logs of a particular source run.

Parameters:

project_id – Project identifier.
source_id – Source identifier.
last_n_log_lines – Last n log lines from the last run of the source.

Example:

>>> client.get_source_logs('Vlg5Z1hOShm0eYmjtsqSqg',
...                     'hw8j7LUBRM28-jAellgQdA',
...                     10)

get_sources(project_id, include_config=None, include_run_stats=None, include_pipeline_backlog=None)#

Get all sources for the provided project.

Parameters:

project_id (str) – Project identifier.
include_config (Optional[bool]) – Bool, whether or not to include the config for all the Sources.
include_run_stats (Optional[bool]) – Bool, whether or not to include the run stats for all the Sources.
include_pipeline_backlog (Optional[bool]) – Bool, whether or not to include the the backlog of items in the data pipeline for sources.

Returns:

A list of sources.

Example:

>>> client.get_sources(
...    project_id='Vlg5Z1hOShm0eYmjtsqSqg',
...    include_config=True,
...    include_run_stats=True,
...    include_pipeline_backlog=True)

[
  {
    "items_fetched_total": 2,
    "last_error": "",
    "last_check_at": "2019-01-23T10:23:23",
    "last_items_at": "2019-01-23T10:23:23",
    "paused": false,
    "error_count": 0,
    "id": "pqTn4vBZRdS5hYw0TBt0pQ",
    "total_error_count": 0,
    "project_id": "Vlg5Z1hOShm0eYmjtsqSqg",
    "config": {
      "dataloader_plugin_options": {
        "source_file": "path:/tmp/test.csv"
      },
      "dataloader_options": {
        "map_title": "title",
        "project_id": "Vlg5Z1hOShm0eYmjtsqSqg",
        "plugin_name": "csv_plugin"
      }
    },
    "status": "complete",
    "total_runs": 1,
    "pipeline_workflow_id": "S0fVQ-K0TmS1UgT0msZRBA",
    "last_error_at": null,
    "last_update_at": "2019-01-23T10:23:23",
    "last_success_at": "2019-01-23T10:23:23",
    "items_fetched_last_run": 2,
    "tenant": "squirro",
    "next_run_time_at": "2019-01-23T10:52:51",
    "name": "test source",
    "scheduling_options": {
      "repeat": "30m",
      "schedule": true
    },
    "created_at": "2019-01-23T10:23:23",
    "modified_at": "2019-01-23T10:23:23",
    "processed": true,
    "pipeline_backlog": 10
  }
]

Note

Credentials are never returned. Options the platform classifies as credentials are omitted from config, and a flat secret_fields map is returned alongside it reporting which ones are set:

"secret_fields": {"access_key_id": {"is_set": true}}

is_set is presence, not truthiness – a stored empty string still reports true – and no length, digest or member count is exposed. The key is present only on responses that carry config.

config itself is omitted entirely when the source’s plugin manifest cannot be read, so that a client cannot round-trip a partial config back and destroy the stored one. Treat a missing config as “unknown”, not as “empty”.

get_sources_v1(project_id, search=None, status=None, plugin_name=None, include=None, counts_agg=None, start=0, count=-1)#

Get sources for the provided project.

Parameters:

project_id (str) – Project identifier.
search (Optional[str]) – filter by part of Source name
status (Optional[str]) – filter by Source status
plugin_name (Optional[str]) – filter by Source plugin_name
include (Optional[str]) – comma seperated list of additional Source fields. config - Source config run_stats - Source run stats pipeline_backlog - Source backlog information from ingester items_indexed - number of indexed items from Source
counts_agg (Optional[str]) – whether or not to include Sources field value aggregations. Specified as comma seperated string of aggregated fields. Possible fields: status,plugin_name.
start (int) – Integer. Used for pagination of objects. If set, the objects starting with offset start are returned.
count (int) – Integer. Used for pagination of objects. If set, count number of objects are returned. To return all objects, set to -1.

Returns:

A paginated list of sources.

Example:

>>> client.get_sources_v1(
...    project_id='Vlg5Z1hOShm0eYmjtsqSqg',
...    include="config,run_stats,pipeline_backlog,items_indexed",
...    counts_agg="status,plugin_name",
...    count=1,
... )

{
  "count": 1,
  "sources": [
    {
      "items_fetched_total": 2,
      "last_error": "",
      "last_check_at": "2019-01-23T10:23:23",
      "last_items_at": "2019-01-23T10:23:23",
      "paused": false,
      "error_count": 0,
      "id": "pqTn4vBZRdS5hYw0TBt0pQ",
      "total_error_count": 0,
      "project_id": "Vlg5Z1hOShm0eYmjtsqSqg",
      "config": {
        "dataloader_plugin_options": {
          "source_file": "path:/tmp/test.csv"
        },
        "dataloader_options": {
          "map_title": "title",
          "project_id": "Vlg5Z1hOShm0eYmjtsqSqg",
          "plugin_name": "csv_plugin"
        }
      },
      "status": "complete",
      "total_runs": 1,
      "pipeline_workflow_id": "S0fVQ-K0TmS1UgT0msZRBA",
      "last_error_at": null,
      "last_update_at": "2019-01-23T10:23:23",
      "last_success_at": "2019-01-23T10:23:23",
      "items_fetched_last_run": 2,
      "tenant": "squirro",
      "next_run_time_at": "2019-01-23T10:52:51",
      "name": "test source",
      "scheduling_options": {
        "repeat": "30m",
        "schedule": true
      },
      "created_at": "2019-01-23T10:23:23",
      "modified_at": "2019-01-23T10:23:23",
      "processed": true,
      "pipeline_backlog": 10,
      "items_indexed": 100
    }
  ],
  "counts_agg": {
    "plugin_name": {"values": [{"key": "csv_plugin", "value": 3}]},
    "status": {"values": [{"key": "queued", "value": 2}, {"key": "running", "value": 0}, {"key": "errored", "value": 0}, {"key": "complete", "value": 1}]}
  },
  "total": 3,
  "next_params": {
    "count": 1,
    "start": 1
  }
}

Note

Credentials are never returned. Options the platform classifies as credentials are omitted from config, and a flat secret_fields map is returned alongside it reporting which ones are set:

"secret_fields": {"access_key_id": {"is_set": true}}

is_set is presence, not truthiness – a stored empty string still reports true – and no length, digest or member count is exposed. The key is present only on responses that carry config.

config itself is omitted entirely when the source’s plugin manifest cannot be read, so that a client cannot round-trip a partial config back and destroy the stored one. Treat a missing config as “unknown”, not as “empty”.

kill_source(project_id, source_id)#

Try to terminate (SIGTERM) a dataload job if it is already running. After a fixed timeout, a SIGKILL signal is sent instead.

Parameters:

project_id – Project identifier.
source_id – Source identifier.

Example:

>>> client.kill_source('Vlg5Z1hOShm0eYmjtsqSqg',
...                     'hw8j7LUBRM28-jAellgQdA')

modify_source(project_id, source_id, name=None, config=None, scheduling_options=None, pipeline_workflow_id=None, enable_scheduling=None, validate_schema=None, notify_scheduler=None, execute_rerun=None, priority=None, description=None)#

Modify an existing source.

Parameters:

project_id – Project identifier.
source_id – Source identifier.
name – Name for the Source.
config – Changed config of the source.
scheduling_options – dict, scheduling options for the run of a Source.
pipeline_workflow_id – Optional pipeline workflow id to change the source to.
enable_scheduling – DEPRECATED; Will be removed in a future release. Optional boolean. Indicate whether or not to enable the scheduling of this source.
validate_schema – Optional boolean. Indicate whether or not to validate the provided configuration of the source.
notify_scheduler – Optional boolean. Indicate whether or not to notify the scheduler to immediately start the procedure of loading data from the source.
execute_rerun – Optional boolean. Indicate whether or not to queue for reprocessing the batches (if any) of this source.
priority – Optional string parameter to define the priority for the source.
description – Optional string parameter for a description of the source

Returns:

A dictionary which contains the source.

Example:

>>> client.modify_source(
...     project_id='Vlg5Z1hOShm0eYmjtsqSqg',
...     source_id='601AoqmkSFWGt4sAwaX8ag',
...     name="new name")

{
  "pipeline_workflow_id": "S0fVQ-K0TmS1UgT0msZRBA",
  "name": "new name",
  "scheduling_options": {
    "repeat": "30m",
    "schedule": true
  },
  "created_at": "2019-01-23T10:32:13",
  "modified_at": "2019-01-23T10:34:41",
  "paused": false,
  "processed": true,
  "project_id": "Vlg5Z1hOShm0eYmjtsqSqg",
  "id": "601AoqmkSFWGt4sAwaX8ag",
  "tenant": "squirro"
}

Note

Credential-bearing options are write-only. For any option the platform classifies as a credential, the semantics of config differ from every other field:

omit the key to keep the stored value – this is what lets you edit an unrelated setting without re-entering the credential;
send a value to replace it;
send None to clear it – the explicit removal path;
sending "" is rejected with 400 when a non-empty value is stored. If nothing is stored, or the stored value is itself empty, it is treated as absent and the stored value is untouched;
echoing back a secret_fields entry such as {"is_set": true} in the value position is rejected with 400.

Non-secret options keep the previous behaviour: the option dict you send replaces the stored one wholesale.

A read-modify-write of the config you got from get_source() is therefore safe and preserves credentials.

new_source(project_id, name, config, scheduling_options=None, pipeline_workflow_id=None, source_id=None, paused=False, use_default_options=None, notify_scheduler=None, priority=None, description=None)#

Create a new source.

Parameters:

project_id – Project identifier.
name – Name for the Source.
config – dict, config including dataloader_options and dataloader_plugin_options for the Source.
scheduling_options – dict, scheduling options for the run of a Source.
pipeline_workflow_id – Optional id of the pipeline workflow to apply to the data of this Source. If not specified, then the default workflow of the project with project_id will be applied.
source_id – Optional string parameter to create the source with the provided id. The length of the parameter must be 22 characters. Useful when exporting and importing projects across multiple Squirro servers.
paused – Optional boolean. Indicate whether to immediately start data loading, or rather create the source in a paused state
use_default_options – Optional boolean. Indicate whether or not to use the default mappings for facets, fields, scheduling_options and pipeline workflow provided by the dataloader plugin itself. Setting this to True will throw a 400 HTTP error code if these default mappings are not available for a specific plugin
notify_scheduler – Optional boolean. Indicate whether or not to notify the scheduler to immediately start the procedure of loading data from the source.
priority – Optional string parameter to define the priority for the source.
description – Optional string parameter for a description of the source

Returns:

A dictionary which contains the new source.

Example:

>>> client.new_source(
...     project_id='Vlg5Z1hOShm0eYmjtsqSqg',
...     name='test source',
...     config={
...         "dataloader_plugin_options": {
...             "source_file": "path:/tmp/test.csv"
...         },
...         "dataloader_options": {
...             "plugin_name": "csv_plugin",
...             "project_id": 'Vlg5Z1hOShm0eYmjtsqSqg',
...             "map_title": "title"
...             }
...         },
...     scheduling_options={'schedule': True, 'repeat': '30m'})

{
  "items_fetched_total": 0,
  "last_error": "",
  "last_check_at": null,
  "last_items_at": null,
  "paused": false,
  "error_count": 0,
  "id": "601AoqmkSFWGt4sAwaX8ag",
  "total_error_count": 0,
  "project_id": "Vlg5Z1hOShm0eYmjtsqSqg",
  "status": "queued",
  "total_runs": 0,
  "pipeline_workflow_id": "S0fVQ-K0TmS1UgT0msZRBA",
  "last_error_at": null,
  "last_update_at": null,
  "last_success_at": null,
  "items_fetched_last_run": 0,
  "tenant": "squirro",
  "next_run_time_at": "2019-01-23T10:32:13",
  "name": "test source",
  "scheduling_options": {
    "repeat": "30m",
    "schedule": true
  },
  "created_at": "2019-01-23T10:32:13",
  "modified_at": "2019-01-23T10:32:13",
  "processed": false
}

Note

Credential-bearing options are write-only. They can be written but are never read back. On create there is nothing stored yet, so:

send a value to store it;
omit the key, or send "", and nothing is stored – the option simply has no value;
sending None is a no-op;
echoing back a secret_fields entry such as {"is_set": true} in the value position is rejected with 400.

See modify_source() for the update semantics, which differ: there, omitting a credential keeps the stored one.

pause_source(project_id, source_id)#

Pause a source.

Parameters:

project_id – Project identifier.
source_id – Source identifier.

Example:

>>> client.pause_source('Vlg5Z1hOShm0eYmjtsqSqg',
...                     'hw8j7LUBRM28-jAellgQdA')

reset_source(project_id, source_id, delete_source_data=None)#

Resets and run the source.

Parameters:

project_id – Project identifier.
source_id – Source identifier.
delete_source_data – Bool, to determine whether to delete the data associated with a source or not

Example:

>>> client.reset_source(
...     'Vlg5Z1hOShm0eYmjtsqSqg',
...     'hw8j7LUBRM28-jAellgQdA')

resume_source(project_id, source_id)#

Resume a paused source.

Parameters:

project_id – Project identifier.
source_id – Source identifier.

Example:

>>> client.resume_source(
...     'Vlg5Z1hOShm0eYmjtsqSqg',
...     'hw8j7LUBRM28-jAellgQdA')

retry_failed_batches(project_id, source_id, batch_id=None, batch_priority=None)#

Move all the failed batches of a given source back for processing.

If batch identifier will be specified, only this specific batch will be moved.

If batch priority level will be specified, only batches with this priority level will be moved.

Parameters:

project_id (str) – project identifier
source_id (str) – source identifier
batch_id (Optional[str]) – optional batch identifier
batch_priority (Optional[str]) – optional batch priority level

Returns:

tuple with number of moved batches and items

Example

>>> client.retry_failed_batches(
...     "Oqc-DYVWRGe9gtDYBIiKyA",
...     "uu6nikJZTiWazZJ0-iOXKA",
...     batch_id="1zxAeuo1Syyi5zgZgihwkg",
...     batch_priority="high")

run_source(project_id, source_id)#

Runs a source now.

Parameters:

project_id – Project identifier.
source_id – Source identifier.

Example:

>>> client.run_source(
...     'Vlg5Z1hOShm0eYmjtsqSqg',
...     'hw8j7LUBRM28-jAellgQdA')

set_max_inc(project_id, source_id, max_inc_value)#

Sets the maximum incremental value of the incremental column of a source.

Parameters:

project_id – Project identifier.
source_id – Source identifier.
max_inc_val – The maximum incremental value to be set.

Example:

>>> client.set_max_inc(
...     'Vlg5Z1hOShm0eYmjtsqSqg',
...     'hw8j7LUBRM28-jAellgQdA',
...     '2020-08-17T19:10:33')

SourcesMixin

Contents

SourcesMixin#