service.py

# WARNING THIS FILE IS AUTOGENERATED
#
# Databricks CLI
# Copyright 2017 Databricks, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"), except
# that the use of services to which certain application programming
# interfaces (each, an "API") connect requires that the user first obtain
# a license for the use of the APIs from Databricks, Inc. ("Databricks"),
# by creating an account at www.databricks.com and agreeing to either (a)
# the Community Edition Terms of Service, (b) the Databricks Terms of
# Service, or (c) another written agreement between Licensee and Databricks
# for the use of the APIs.
#
# You may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import os
import re

from six.moves.urllib.parse import urlparse


class JobsService(object):
    def __init__(self, client):
        self.client = client

    def create_job(
        self,
        name=None,
        existing_cluster_id=None,
        new_cluster=None,
        libraries=None,
        email_notifications=None,
        timeout_seconds=None,
        max_retries=None,
        min_retry_interval_millis=None,
        retry_on_timeout=None,
        schedule=None,
        notebook_task=None,
        spark_jar_task=None,
        spark_python_task=None,
        spark_submit_task=None,
        max_concurrent_runs=None,
        tasks=None,
        headers=None,
        version=None,
        git_source=None,
        job_clusters=None,
        tags=None,
        format=None,
        dbt_task=None,
        access_control_list=None,
        pipeline_task=None,
        python_wheel_task=None,
        sql_task=None,
        webhook_notifications=None,
        continuous=None,
    ):
        _data = {}
        if name is not None:
            _data['name'] = name
        if existing_cluster_id is not None:
            _data['existing_cluster_id'] = existing_cluster_id
        if new_cluster is not None:
            _data['new_cluster'] = new_cluster
            if not isinstance(new_cluster, dict):
                raise TypeError('Expected databricks.NewCluster() or dict for field new_cluster')
        if libraries is not None:
            _data['libraries'] = libraries
        if email_notifications is not None:
            _data['email_notifications'] = email_notifications
            if not isinstance(email_notifications, dict):
                raise TypeError(
                    'Expected databricks.JobEmailNotifications() or dict for field email_notifications'
                )
        if timeout_seconds is not None:
            _data['timeout_seconds'] = timeout_seconds
        if max_retries is not None:
            _data['max_retries'] = max_retries
        if min_retry_interval_millis is not None:
            _data['min_retry_interval_millis'] = min_retry_interval_millis
        if retry_on_timeout is not None:
            _data['retry_on_timeout'] = retry_on_timeout
        if schedule is not None:
            _data['schedule'] = schedule
            if not isinstance(schedule, dict):
                raise TypeError('Expected databricks.CronSchedule() or dict for field schedule')
        if notebook_task is not None:
            _data['notebook_task'] = notebook_task
            if not isinstance(notebook_task, dict):
                raise TypeError(
                    'Expected databricks.NotebookTask() or dict for field notebook_task'
                )
        if spark_jar_task is not None:
            _data['spark_jar_task'] = spark_jar_task
            if not isinstance(spark_jar_task, dict):
                raise TypeError(
                    'Expected databricks.SparkJarTask() or dict for field spark_jar_task'
                )
        if spark_python_task is not None:
            _data['spark_python_task'] = spark_python_task
            if not isinstance(spark_python_task, dict):
                raise TypeError(
                    'Expected databricks.SparkPythonTask() or dict for field spark_python_task'
                )
        if spark_submit_task is not None:
            _data['spark_submit_task'] = spark_submit_task
            if not isinstance(spark_submit_task, dict):
                raise TypeError(
                    'Expected databricks.SparkSubmitTask() or dict for field spark_submit_task'
                )
        if max_concurrent_runs is not None:
            _data['max_concurrent_runs'] = max_concurrent_runs
        if tasks is not None:
            _data['tasks'] = tasks
        if git_source is not None:
            _data['git_source'] = git_source
            if not isinstance(git_source, dict):
                raise TypeError('Expected databricks.GitSource() or dict for field git_source')
        if job_clusters is not None:
            _data['job_clusters'] = job_clusters
        if tags is not None:
            _data['tags'] = tags
        if format is not None:
            _data['format'] = format
        if dbt_task is not None:
            _data['dbt_task'] = dbt_task
            if not isinstance(dbt_task, dict):
                raise TypeError('Expected databricks.DbtTask() or dict for field dbt_task')
        if access_control_list is not None:
            _data['access_control_list'] = access_control_list
        if pipeline_task is not None:
            _data['pipeline_task'] = pipeline_task
            if not isinstance(pipeline_task, dict):
                raise TypeError(
                    'Expected databricks.PipelineTask() or dict for field pipeline_task'
                )
        if python_wheel_task is not None:
            _data['python_wheel_task'] = python_wheel_task
            if not isinstance(python_wheel_task, dict):
                raise TypeError(
                    'Expected databricks.PythonWheelTask() or dict for field python_wheel_task'
                )
        if sql_task is not None:
            _data['sql_task'] = sql_task
            if not isinstance(sql_task, dict):
                raise TypeError('Expected databricks.SqlTask() or dict for field sql_task')
        if webhook_notifications is not None:
            _data['webhook_notifications'] = webhook_notifications
            if not isinstance(webhook_notifications, dict):
                raise TypeError(
                    'Expected databricks.WebhookNotifications() or dict for field webhook_notifications'
                )
        if continuous is not None:
            _data['continuous'] = continuous
            if not isinstance(continuous, dict):
                raise TypeError(
                    'Expected databricks.ContinuousSettings() or dict for field continuous'
                )
        return self.client.perform_query(
            'POST', '/jobs/create', data=_data, headers=headers, version=version
        )

    def submit_run(
        self,
        run_name=None,
        existing_cluster_id=None,
        new_cluster=None,
        libraries=None,
        notebook_task=None,
        spark_jar_task=None,
        spark_python_task=None,
        spark_submit_task=None,
        timeout_seconds=None,
        tasks=None,
        headers=None,
        version=None,
        idempotency_token=None,
        job_clusters=None,
        git_source=None,
        dbt_task=None,
        access_control_list=None,
        pipeline_task=None,
        python_wheel_task=None,
        sql_task=None,
        webhook_notifications=None,
    ):
        _data = {}
        if run_name is not None:
            _data['run_name'] = run_name
        if existing_cluster_id is not None:
            _data['existing_cluster_id'] = existing_cluster_id
        if new_cluster is not None:
            _data['new_cluster'] = new_cluster
            if not isinstance(new_cluster, dict):
                raise TypeError('Expected databricks.NewCluster() or dict for field new_cluster')
        if libraries is not None:
            _data['libraries'] = libraries
        if notebook_task is not None:
            _data['notebook_task'] = notebook_task
            if not isinstance(notebook_task, dict):
                raise TypeError(
                    'Expected databricks.NotebookTask() or dict for field notebook_task'
                )
        if spark_jar_task is not None:
            _data['spark_jar_task'] = spark_jar_task
            if not isinstance(spark_jar_task, dict):
                raise TypeError(
                    'Expected databricks.SparkJarTask() or dict for field spark_jar_task'
                )
        if spark_python_task is not None:
            _data['spark_python_task'] = spark_python_task
            if not isinstance(spark_python_task, dict):
                raise TypeError(
                    'Expected databricks.SparkPythonTask() or dict for field spark_python_task'
                )
        if spark_submit_task is not None:
            _data['spark_submit_task'] = spark_submit_task
            if not isinstance(spark_submit_task, dict):
                raise TypeError(
                    'Expected databricks.SparkSubmitTask() or dict for field spark_submit_task'
                )
        if timeout_seconds is not None:
            _data['timeout_seconds'] = timeout_seconds
        if tasks is not None:
            _data['tasks'] = tasks
        if idempotency_token is not None:
            _data['idempotency_token'] = idempotency_token
        if job_clusters is not None:
            _data['job_clusters'] = job_clusters
        if git_source is not None:
            _data['git_source'] = git_source
            if not isinstance(git_source, dict):
                raise TypeError('Expected databricks.GitSource() or dict for field git_source')
        if dbt_task is not None:
            _data['dbt_task'] = dbt_task
            if not isinstance(dbt_task, dict):
                raise TypeError('Expected databricks.DbtTask() or dict for field dbt_task')
        if access_control_list is not None:
            _data['access_control_list'] = access_control_list
        if pipeline_task is not None:
            _data['pipeline_task'] = pipeline_task
            if not isinstance(pipeline_task, dict):
                raise TypeError(
                    'Expected databricks.PipelineTask() or dict for field pipeline_task'
                )
        if python_wheel_task is not None:
            _data['python_wheel_task'] = python_wheel_task
            if not isinstance(python_wheel_task, dict):
                raise TypeError(
                    'Expected databricks.PythonWheelTask() or dict for field python_wheel_task'
                )
        if sql_task is not None:
            _data['sql_task'] = sql_task
            if not isinstance(sql_task, dict):
                raise TypeError('Expected databricks.SqlTask() or dict for field sql_task')
        if webhook_notifications is not None:
            _data['webhook_notifications'] = webhook_notifications
            if not isinstance(webhook_notifications, dict):
                raise TypeError(
                    'Expected databricks.WebhookNotifications() or dict for field webhook_notifications'
                )
        return self.client.perform_query(
            'POST', '/jobs/runs/submit', data=_data, headers=headers, version=version
        )

    def reset_job(self, job_id, new_settings, headers=None, version=None):
        _data = {}
        if job_id is not None:
            _data['job_id'] = job_id
        if new_settings is not None:
            _data['new_settings'] = new_settings
            if not isinstance(new_settings, dict):
                raise TypeError('Expected databricks.JobSettings() or dict for field new_settings')
        return self.client.perform_query(
            'POST', '/jobs/reset', data=_data, headers=headers, version=version
        )

    def update_job(
        self, job_id, new_settings=None, fields_to_remove=None, headers=None, version=None
    ):
        _data = {}
        if job_id is not None:
            _data['job_id'] = job_id
        if new_settings is not None:
            _data['new_settings'] = new_settings
            if not isinstance(new_settings, dict):
                raise TypeError('Expected databricks.JobSettings() or dict for field new_settings')
        if fields_to_remove is not None:
            _data['fields_to_remove'] = fields_to_remove
        return self.client.perform_query(
            'POST', '/jobs/update', data=_data, headers=headers, version=version
        )

    def delete_job(self, job_id, headers=None, version=None):
        _data = {}
        if job_id is not None:
            _data['job_id'] = job_id
        return self.client.perform_query(
            'POST', '/jobs/delete', data=_data, headers=headers, version=version
        )

    def get_job(self, job_id, headers=None, version=None, include_trigger_history=None):
        _data = {}
        if job_id is not None:
            _data['job_id'] = job_id
        if include_trigger_history is not None:
            _data['include_trigger_history'] = include_trigger_history
        return self.client.perform_query(
            'GET', '/jobs/get', data=_data, headers=headers, version=version
        )

    def list_jobs(
        self,
        job_type=None,
        expand_tasks=None,
        limit=None,
        offset=None,
        headers=None,
        version=None,
        name=None,
    ):
        _data = {}
        if expand_tasks is not None:
            _data['expand_tasks'] = expand_tasks
        if limit is not None:
            _data['limit'] = limit
        if offset is not None:
            _data['offset'] = offset
        if name is not None:
            _data['name'] = name
        return self.client.perform_query(
            'GET', '/jobs/list', data=_data, headers=headers, version=version
        )

    def run_now(
        self,
        job_id=None,
        jar_params=None,
        notebook_params=None,
        python_params=None,
        spark_submit_params=None,
        python_named_params=None,
        idempotency_token=None,
        headers=None,
        version=None,
        dbt_commands=None,
        pipeline_params=None,
    ):
        _data = {}
        if job_id is not None:
            _data['job_id'] = job_id
        if jar_params is not None:
            _data['jar_params'] = jar_params
        if notebook_params is not None:
            _data['notebook_params'] = notebook_params
        if python_params is not None:
            _data['python_params'] = python_params
        if spark_submit_params is not None:
            _data['spark_submit_params'] = spark_submit_params
        if python_named_params is not None:
            _data['python_named_params'] = python_named_params
        if idempotency_token is not None:
            _data['idempotency_token'] = idempotency_token
        if dbt_commands is not None:
            _data['dbt_commands'] = dbt_commands
        if pipeline_params is not None:
            _data['pipeline_params'] = pipeline_params
            if not isinstance(pipeline_params, dict):
                raise TypeError(
                    'Expected databricks.PipelineParameters() or dict for field pipeline_params'
                )
        return self.client.perform_query(
            'POST', '/jobs/run-now', data=_data, headers=headers, version=version
        )

    def repair(
        self,
        run_id,
        latest_repair_id=None,
        rerun_tasks=None,
        jar_params=None,
        notebook_params=None,
        python_params=None,
        spark_submit_params=None,
        python_named_params=None,
        headers=None,
        version=None,
        dbt_commands=None,
        pipeline_params=None,
        rerun_all_failed_tasks=None,
        rerun_dependent_tasks=None,
    ):
        _data = {}
        if run_id is not None:
            _data['run_id'] = run_id
        if latest_repair_id is not None:
            _data['latest_repair_id'] = latest_repair_id
        if rerun_tasks is not None:
            _data['rerun_tasks'] = rerun_tasks
        if jar_params is not None:
            _data['jar_params'] = jar_params
        if notebook_params is not None:
            _data['notebook_params'] = notebook_params
        if python_params is not None:
            _data['python_params'] = python_params
        if spark_submit_params is not None:
            _data['spark_submit_params'] = spark_submit_params
        if python_named_params is not None:
            _data['python_named_params'] = python_named_params
        if dbt_commands is not None:
            _data['dbt_commands'] = dbt_commands
        if pipeline_params is not None:
            _data['pipeline_params'] = pipeline_params
            if not isinstance(pipeline_params, dict):
                raise TypeError(
                    'Expected databricks.PipelineParameters() or dict for field pipeline_params'
                )
        if rerun_all_failed_tasks is not None:
            _data['rerun_all_failed_tasks'] = rerun_all_failed_tasks
        if rerun_dependent_tasks is not None:
            _data['rerun_dependent_tasks'] = rerun_dependent_tasks
        return self.client.perform_query(
            'POST', '/jobs/runs/repair', data=_data, headers=headers, version=version
        )

    def list_runs(
        self,
        job_id=None,
        active_only=None,
        completed_only=None,
        offset=None,
        limit=None,
        headers=None,
        version=None,
        run_type=None,
        expand_tasks=None,
        start_time_from=None,
        start_time_to=None,
        page_token=None,
    ):
        _data = {}
        if job_id is not None:
            _data['job_id'] = job_id
        if active_only is not None:
            _data['active_only'] = active_only
        if completed_only is not None:
            _data['completed_only'] = completed_only
        if offset is not None:
            _data['offset'] = offset
        if limit is not None:
            _data['limit'] = limit
        if run_type is not None:
            _data['run_type'] = run_type
        if expand_tasks is not None:
            _data['expand_tasks'] = expand_tasks
        if start_time_from is not None:
            _data['start_time_from'] = start_time_from
        if start_time_to is not None:
            _data['start_time_to'] = start_time_to
        if page_token is not None:
            _data['page_token'] = page_token
        return self.client.perform_query(
            'GET', '/jobs/runs/list', data=_data, headers=headers, version=version
        )

    def get_run(self, run_id=None, headers=None, version=None, include_history=None):
        _data = {}
        if run_id is not None:
            _data['run_id'] = run_id
        if include_history is not None:
            _data['include_history'] = include_history
        return self.client.perform_query(
            'GET', '/jobs/runs/get', data=_data, headers=headers, version=version
        )

    def delete_run(self, run_id=None, headers=None, version=None):
        _data = {}
        if run_id is not None:
            _data['run_id'] = run_id
        return self.client.perform_query(
            'POST', '/jobs/runs/delete', data=_data, headers=headers, version=version
        )

    def cancel_run(self, run_id, headers=None, version=None):
        _data = {}
        if run_id is not None:
            _data['run_id'] = run_id
        return self.client.perform_query(
            'POST', '/jobs/runs/cancel', data=_data, headers=headers, version=version
        )

    def cancel_all_runs(self, job_id=None, headers=None, version=None, all_queued_runs=None):
        _data = {}
        if job_id is not None:
            _data['job_id'] = job_id
        if all_queued_runs is not None:
            _data['all_queued_runs'] = all_queued_runs
        return self.client.perform_query(
            'POST', '/jobs/runs/cancel-all', data=_data, headers=headers, version=version
        )

    def get_run_output(self, run_id, headers=None, version=None):
        _data = {}
        if run_id is not None:
            _data['run_id'] = run_id
        return self.client.perform_query(
            'GET', '/jobs/runs/get-output', data=_data, headers=headers, version=version
        )

    def export_run(self, run_id, views_to_export=None, headers=None, version=None):
        _data = {}
        if run_id is not None:
            _data['run_id'] = run_id
        if views_to_export is not None:
            _data['views_to_export'] = views_to_export
        return self.client.perform_query(
            'GET', '/jobs/runs/export', data=_data, headers=headers, version=version
        )


class ClusterService(object):
    def __init__(self, client):
        self.client = client

    def list_clusters(self, headers=None):
        _data = {}
        return self.client.perform_query('GET', '/clusters/list', data=_data, headers=headers)

    def create_cluster(
        self,
        num_workers=None,
        autoscale=None,
        cluster_name=None,
        spark_version=None,
        spark_conf=None,
        aws_attributes=None,
        node_type_id=None,
        driver_node_type_id=None,
        ssh_public_keys=None,
        custom_tags=None,
        cluster_log_conf=None,
        spark_env_vars=None,
        autotermination_minutes=None,
        enable_elastic_disk=None,
        cluster_source=None,
        instance_pool_id=None,
        headers=None,
        azure_attributes=None,
        gcp_attributes=None,
        policy_id=None,
        enable_local_disk_encryption=None,
        driver_instance_pool_id=None,
        apply_policy_default_values=None,
    ):
        _data = {}
        if num_workers is not None:
            _data['num_workers'] = num_workers
        if autoscale is not None:
            _data['autoscale'] = autoscale
            if not isinstance(autoscale, dict):
                raise TypeError('Expected databricks.AutoScale() or dict for field autoscale')
        if cluster_name is not None:
            _data['cluster_name'] = cluster_name
        if spark_version is not None:
            _data['spark_version'] = spark_version
        if spark_conf is not None:
            _data['spark_conf'] = spark_conf
        if aws_attributes is not None:
            _data['aws_attributes'] = aws_attributes
            if not isinstance(aws_attributes, dict):
                raise TypeError(
                    'Expected databricks.AwsAttributes() or dict for field aws_attributes'
                )
        if node_type_id is not None:
            _data['node_type_id'] = node_type_id
        if driver_node_type_id is not None:
            _data['driver_node_type_id'] = driver_node_type_id
        if ssh_public_keys is not None:
            _data['ssh_public_keys'] = ssh_public_keys
        if custom_tags is not None:
            _data['custom_tags'] = custom_tags
        if cluster_log_conf is not None:
            _data['cluster_log_conf'] = cluster_log_conf
            if not isinstance(cluster_log_conf, dict):
                raise TypeError(
                    'Expected databricks.ClusterLogConf() or dict for field cluster_log_conf'
                )
        if spark_env_vars is not None:
            _data['spark_env_vars'] = spark_env_vars
        if autotermination_minutes is not None:
            _data['autotermination_minutes'] = autotermination_minutes
        if enable_elastic_disk is not None:
            _data['enable_elastic_disk'] = enable_elastic_disk
        if cluster_source is not None:
            _data['cluster_source'] = cluster_source
        if instance_pool_id is not None:
            _data['instance_pool_id'] = instance_pool_id
        if azure_attributes is not None:
            _data['azure_attributes'] = azure_attributes
            if not isinstance(azure_attributes, dict):
                raise TypeError(
                    'Expected databricks.AzureAttributes() or dict for field azure_attributes'
                )
        if gcp_attributes is not None:
            _data['gcp_attributes'] = gcp_attributes
            if not isinstance(gcp_attributes, dict):
                raise TypeError(
                    'Expected databricks.GcpAttributes() or dict for field gcp_attributes'
                )
        if policy_id is not None:
            _data['policy_id'] = policy_id
        if enable_local_disk_encryption is not None:
            _data['enable_local_disk_encryption'] = enable_local_disk_encryption
        if driver_instance_pool_id is not None:
            _data['driver_instance_pool_id'] = driver_instance_pool_id
        if apply_policy_default_values is not None:
            _data['apply_policy_default_values'] = apply_policy_default_values
        return self.client.perform_query('POST', '/clusters/create', data=_data, headers=headers)

    def start_cluster(self, cluster_id, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        return self.client.perform_query('POST', '/clusters/start', data=_data, headers=headers)

    def list_spark_versions(self, headers=None):
        _data = {}
        return self.client.perform_query(
            'GET', '/clusters/spark-versions', data=_data, headers=headers
        )

    def delete_cluster(self, cluster_id, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        return self.client.perform_query('POST', '/clusters/delete', data=_data, headers=headers)

    def permanent_delete_cluster(self, cluster_id, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        return self.client.perform_query(
            'POST', '/clusters/permanent-delete', data=_data, headers=headers
        )

    def restart_cluster(self, cluster_id, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        return self.client.perform_query('POST', '/clusters/restart', data=_data, headers=headers)

    def resize_cluster(self, cluster_id, num_workers=None, autoscale=None, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        if num_workers is not None:
            _data['num_workers'] = num_workers
        if autoscale is not None:
            _data['autoscale'] = autoscale
            if not isinstance(autoscale, dict):
                raise TypeError('Expected databricks.AutoScale() or dict for field autoscale')
        return self.client.perform_query('POST', '/clusters/resize', data=_data, headers=headers)

    def edit_cluster(
        self,
        cluster_id,
        num_workers=None,
        autoscale=None,
        cluster_name=None,
        spark_version=None,
        spark_conf=None,
        aws_attributes=None,
        node_type_id=None,
        driver_node_type_id=None,
        ssh_public_keys=None,
        custom_tags=None,
        cluster_log_conf=None,
        spark_env_vars=None,
        autotermination_minutes=None,
        enable_elastic_disk=None,
        cluster_source=None,
        instance_pool_id=None,
        headers=None,
        azure_attributes=None,
        gcp_attributes=None,
        policy_id=None,
        enable_local_disk_encryption=None,
        driver_instance_pool_id=None,
        apply_policy_default_values=None,
    ):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        if num_workers is not None:
            _data['num_workers'] = num_workers
        if autoscale is not None:
            _data['autoscale'] = autoscale
            if not isinstance(autoscale, dict):
                raise TypeError('Expected databricks.AutoScale() or dict for field autoscale')
        if cluster_name is not None:
            _data['cluster_name'] = cluster_name
        if spark_version is not None:
            _data['spark_version'] = spark_version
        if spark_conf is not None:
            _data['spark_conf'] = spark_conf
        if aws_attributes is not None:
            _data['aws_attributes'] = aws_attributes
            if not isinstance(aws_attributes, dict):
                raise TypeError(
                    'Expected databricks.AwsAttributes() or dict for field aws_attributes'
                )
        if node_type_id is not None:
            _data['node_type_id'] = node_type_id
        if driver_node_type_id is not None:
            _data['driver_node_type_id'] = driver_node_type_id
        if ssh_public_keys is not None:
            _data['ssh_public_keys'] = ssh_public_keys
        if custom_tags is not None:
            _data['custom_tags'] = custom_tags
        if cluster_log_conf is not None:
            _data['cluster_log_conf'] = cluster_log_conf
            if not isinstance(cluster_log_conf, dict):
                raise TypeError(
                    'Expected databricks.ClusterLogConf() or dict for field cluster_log_conf'
                )
        if spark_env_vars is not None:
            _data['spark_env_vars'] = spark_env_vars
        if autotermination_minutes is not None:
            _data['autotermination_minutes'] = autotermination_minutes
        if enable_elastic_disk is not None:
            _data['enable_elastic_disk'] = enable_elastic_disk
        if cluster_source is not None:
            _data['cluster_source'] = cluster_source
        if instance_pool_id is not None:
            _data['instance_pool_id'] = instance_pool_id
        if azure_attributes is not None:
            _data['azure_attributes'] = azure_attributes
            if not isinstance(azure_attributes, dict):
                raise TypeError(
                    'Expected databricks.AzureAttributes() or dict for field azure_attributes'
                )
        if gcp_attributes is not None:
            _data['gcp_attributes'] = gcp_attributes
            if not isinstance(gcp_attributes, dict):
                raise TypeError(
                    'Expected databricks.GcpAttributes() or dict for field gcp_attributes'
                )
        if policy_id is not None:
            _data['policy_id'] = policy_id
        if enable_local_disk_encryption is not None:
            _data['enable_local_disk_encryption'] = enable_local_disk_encryption
        if driver_instance_pool_id is not None:
            _data['driver_instance_pool_id'] = driver_instance_pool_id
        if apply_policy_default_values is not None:
            _data['apply_policy_default_values'] = apply_policy_default_values
        return self.client.perform_query('POST', '/clusters/edit', data=_data, headers=headers)

    def get_cluster(self, cluster_id, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        return self.client.perform_query('GET', '/clusters/get', data=_data, headers=headers)

    def pin_cluster(self, cluster_id, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        return self.client.perform_query('POST', '/clusters/pin', data=_data, headers=headers)

    def unpin_cluster(self, cluster_id, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        return self.client.perform_query('POST', '/clusters/unpin', data=_data, headers=headers)

    def list_node_types(self, headers=None):
        _data = {}
        return self.client.perform_query(
            'GET', '/clusters/list-node-types', data=_data, headers=headers
        )

    def list_available_zones(self, headers=None):
        _data = {}
        return self.client.perform_query('GET', '/clusters/list-zones', data=_data, headers=headers)

    def get_events(
        self,
        cluster_id,
        start_time=None,
        end_time=None,
        order=None,
        event_types=None,
        offset=None,
        limit=None,
        headers=None,
    ):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        if start_time is not None:
            _data['start_time'] = start_time
        if end_time is not None:
            _data['end_time'] = end_time
        if order is not None:
            _data['order'] = order
        if event_types is not None:
            _data['event_types'] = event_types
        if offset is not None:
            _data['offset'] = offset
        if limit is not None:
            _data['limit'] = limit
        return self.client.perform_query('POST', '/clusters/events', data=_data, headers=headers)


class PolicyService(object):
    def __init__(self, client):
        self.client = client

    def get_policy(self, policy_id=None, headers=None):
        _data = {}
        if policy_id is not None:
            _data['policy_id'] = policy_id
        return self.client.perform_query(
            'GET', '/policies/clusters/get', data=_data, headers=headers
        )

    def list_policies(self, headers=None):
        _data = {}
        return self.client.perform_query(
            'GET', '/policies/clusters/list', data=_data, headers=headers
        )

    def create_policy(self, policy_name=None, definition=None, headers=None, name=None):
        _data = {}
        if definition is not None:
            _data['definition'] = definition
        if name is not None:
            _data['name'] = name
        return self.client.perform_query(
            'POST', '/policies/clusters/create', data=_data, headers=headers
        )

    def edit_policy(self, policy_id, policy_name=None, definition=None, headers=None, name=None):
        _data = {}
        if policy_id is not None:
            _data['policy_id'] = policy_id
        if definition is not None:
            _data['definition'] = definition
        if name is not None:
            _data['name'] = name
        return self.client.perform_query(
            'POST', '/policies/clusters/edit', data=_data, headers=headers
        )

    def delete_policy(self, policy_id, headers=None):
        _data = {}
        if policy_id is not None:
            _data['policy_id'] = policy_id
        return self.client.perform_query(
            'POST', '/policies/clusters/delete', data=_data, headers=headers
        )


class ManagedLibraryService(object):
    def __init__(self, client):
        self.client = client

    def cluster_status(self, cluster_id, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        return self.client.perform_query(
            'GET', '/libraries/cluster-status', data=_data, headers=headers
        )

    def all_cluster_statuses(self, headers=None):
        _data = {}
        return self.client.perform_query(
            'GET', '/libraries/all-cluster-statuses', data=_data, headers=headers
        )

    def install_libraries(self, cluster_id, libraries=None, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        if libraries is not None:
            _data['libraries'] = libraries
        return self.client.perform_query('POST', '/libraries/install', data=_data, headers=headers)

    def uninstall_libraries(self, cluster_id, libraries=None, headers=None):
        _data = {}
        if cluster_id is not None:
            _data['cluster_id'] = cluster_id
        if libraries is not None:
            _data['libraries'] = libraries
        return self.client.perform_query(
            'POST', '/libraries/uninstall', data=_data, headers=headers
        )


class DbfsService(object):
    def __init__(self, client):
        self.client = client

    def read(self, path, offset=None, length=None, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        if offset is not None:
            _data['offset'] = offset
        if length is not None:
            _data['length'] = length
        return self.client.perform_query('GET', '/dbfs/read', data=_data, headers=headers)

    def get_status(self, path, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        return self.client.perform_query('GET', '/dbfs/get-status', data=_data, headers=headers)

    def list(self, path, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        return self.client.perform_query('GET', '/dbfs/list', data=_data, headers=headers)

    def put(self, path, contents=None, overwrite=None, headers=None, src_path=None):
        _data = {}
        _files = None
        if path is not None:
            _data['path'] = path
        if contents is not None:
            _data['contents'] = contents
        if overwrite is not None:
            _data['overwrite'] = overwrite
        if src_path is not None:
            headers = {'Content-Type': None}
            filename = os.path.basename(src_path)
            _files = {'file': (filename, open(src_path, 'rb'), 'multipart/form-data')}
        return self.client.perform_query(
            'POST', '/dbfs/put', data=_data, headers=headers, files=_files
        )

    def mkdirs(self, path, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        return self.client.perform_query('POST', '/dbfs/mkdirs', data=_data, headers=headers)

    def move(self, source_path, destination_path, headers=None):
        _data = {}
        if source_path is not None:
            _data['source_path'] = source_path
        if destination_path is not None:
            _data['destination_path'] = destination_path
        return self.client.perform_query('POST', '/dbfs/move', data=_data, headers=headers)

    def delete(self, path, recursive=None, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        if recursive is not None:
            _data['recursive'] = recursive
        return self.client.perform_query('POST', '/dbfs/delete', data=_data, headers=headers)

    def create(self, path, overwrite=None, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        if overwrite is not None:
            _data['overwrite'] = overwrite
        return self.client.perform_query('POST', '/dbfs/create', data=_data, headers=headers)

    def add_block(self, handle, data, headers=None):
        _data = {}
        if handle is not None:
            _data['handle'] = handle
        if data is not None:
            _data['data'] = data
        return self.client.perform_query('POST', '/dbfs/add-block', data=_data, headers=headers)

    def close(self, handle, headers=None):
        _data = {}
        if handle is not None:
            _data['handle'] = handle
        return self.client.perform_query('POST', '/dbfs/close', data=_data, headers=headers)


class WorkspaceService(object):
    def __init__(self, client):
        self.client = client

    def mkdirs(self, path, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        return self.client.perform_query('POST', '/workspace/mkdirs', data=_data, headers=headers)

    def list(self, path, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        return self.client.perform_query('GET', '/workspace/list', data=_data, headers=headers)

    def import_workspace(
        self, path, format=None, language=None, content=None, overwrite=None, headers=None
    ):
        _data = {}
        if path is not None:
            _data['path'] = path
        if format is not None:
            _data['format'] = format
        if language is not None:
            _data['language'] = language
        if content is not None:
            _data['content'] = content
        if overwrite is not None:
            _data['overwrite'] = overwrite
        return self.client.perform_query('POST', '/workspace/import', data=_data, headers=headers)

    def export_workspace(self, path, format=None, direct_download=None, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        if format is not None:
            _data['format'] = format
        if direct_download is not None:
            _data['direct_download'] = direct_download
        return self.client.perform_query('GET', '/workspace/export', data=_data, headers=headers)

    def delete(self, path, recursive=None, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        if recursive is not None:
            _data['recursive'] = recursive
        return self.client.perform_query('POST', '/workspace/delete', data=_data, headers=headers)

    def get_status(self, path, headers=None):
        _data = {}
        if path is not None:
            _data['path'] = path
        return self.client.perform_query(
            'GET', '/workspace/get-status', data=_data, headers=headers
        )


class SecretService(object):
    def __init__(self, client):
        self.client = client

    def create_scope(
        self,
        scope,
        initial_manage_principal=None,
        scope_backend_type=None,
        backend_azure_keyvault=None,
        headers=None,
    ):
        _data = {}
        if scope is not None:
            _data['scope'] = scope
        if initial_manage_principal is not None:
            _data['initial_manage_principal'] = initial_manage_principal
        if scope_backend_type is not None:
            _data['scope_backend_type'] = scope_backend_type
        if backend_azure_keyvault is not None:
            _data['backend_azure_keyvault'] = backend_azure_keyvault
            if not isinstance(backend_azure_keyvault, dict):
                raise TypeError(
                    'Expected databricks.AzureKeyVaultSecretScopeMetadata() or dict for field backend_azure_keyvault'
                )
        return self.client.perform_query(
            'POST', '/secrets/scopes/create', data=_data, headers=headers
        )

    def delete_scope(self, scope, headers=None):
        _data = {}
        if scope is not None:
            _data['scope'] = scope
        return self.client.perform_query(
            'POST', '/secrets/scopes/delete', data=_data, headers=headers
        )

    def list_scopes(self, headers=None):
        _data = {}
        return self.client.perform_query('GET', '/secrets/scopes/list', data=_data, headers=headers)

    def put_secret(self, scope, key, string_value=None, bytes_value=None, headers=None):
        _data = {}
        if scope is not None:
            _data['scope'] = scope
        if key is not None:
            _data['key'] = key
        if string_value is not None:
            _data['string_value'] = string_value
        if bytes_value is not None:
            _data['bytes_value'] = bytes_value
        return self.client.perform_query('POST', '/secrets/put', data=_data, headers=headers)

    def delete_secret(self, scope, key, headers=None):
        _data = {}
        if scope is not None:
            _data['scope'] = scope
        if key is not None:
            _data['key'] = key
        return self.client.perform_query('POST', '/secrets/delete', data=_data, headers=headers)

    def list_secrets(self, scope, headers=None):
        _data = {}
        if scope is not None:
            _data['scope'] = scope
        return self.client.perform_query('GET', '/secrets/list', data=_data, headers=headers)

    def put_acl(self, scope, principal, permission, headers=None):
        _data = {}
        if scope is not None:
            _data['scope'] = scope
        if principal is not None:
            _data['principal'] = principal
        if permission is not None:
            _data['permission'] = permission
        return self.client.perform_query('POST', '/secrets/acls/put', data=_data, headers=headers)

    def delete_acl(self, scope, principal, headers=None):
        _data = {}
        if scope is not None:
            _data['scope'] = scope
        if principal is not None:
            _data['principal'] = principal
        return self.client.perform_query(
            'POST', '/secrets/acls/delete', data=_data, headers=headers
        )

    def list_acls(self, scope, headers=None):
        _data = {}
        if scope is not None:
            _data['scope'] = scope
        return self.client.perform_query('GET', '/secrets/acls/list', data=_data, headers=headers)

    def get_acl(self, scope, principal, headers=None):
        _data = {}
        if scope is not None:
            _data['scope'] = scope
        if principal is not None:
            _data['principal'] = principal
        return self.client.perform_query('GET', '/secrets/acls/get', data=_data, headers=headers)


class GroupsService(object):
    def __init__(self, client):
        self.client = client

    def create_group(self, group_name, headers=None):
        _data = {}
        if group_name is not None:
            _data['group_name'] = group_name
        return self.client.perform_query('POST', '/groups/create', data=_data, headers=headers)

    def add_to_group(self, parent_name, user_name=None, group_name=None, headers=None):
        _data = {}
        if parent_name is not None:
            _data['parent_name'] = parent_name
        if user_name is not None:
            _data['user_name'] = user_name
        if group_name is not None:
            _data['group_name'] = group_name
        return self.client.perform_query('POST', '/groups/add-member', data=_data, headers=headers)

    def remove_from_group(self, parent_name, user_name=None, group_name=None, headers=None):
        _data = {}
        if parent_name is not None:
            _data['parent_name'] = parent_name
        if user_name is not None:
            _data['user_name'] = user_name
        if group_name is not None:
            _data['group_name'] = group_name
        return self.client.perform_query(
            'POST', '/groups/remove-member', data=_data, headers=headers
        )

    def get_groups(self, headers=None):
        _data = {}
        return self.client.perform_query('GET', '/groups/list', data=_data, headers=headers)

    def get_group_members(self, group_name, headers=None):
        _data = {}
        if group_name is not None:
            _data['group_name'] = group_name
        return self.client.perform_query('GET', '/groups/list-members', data=_data, headers=headers)

    def remove_group(self, group_name, headers=None):
        _data = {}
        if group_name is not None:
            _data['group_name'] = group_name
        return self.client.perform_query('POST', '/groups/delete', data=_data, headers=headers)

    def get_groups_for_principal(self, user_name=None, group_name=None, headers=None):
        _data = {}
        if user_name is not None:
            _data['user_name'] = user_name
        if group_name is not None:
            _data['group_name'] = group_name
        return self.client.perform_query('GET', '/groups/list-parents', data=_data, headers=headers)


class TokenService(object):
    def __init__(self, client):
        self.client = client

    def create_token(self, lifetime_seconds=None, comment=None, headers=None):
        _data = {}
        if lifetime_seconds is not None:
            _data['lifetime_seconds'] = lifetime_seconds
        if comment is not None:
            _data['comment'] = comment
        return self.client.perform_query('POST', '/token/create', data=_data, headers=headers)

    def list_tokens(self, headers=None):
        _data = {}
        return self.client.perform_query('GET', '/token/list', data=_data, headers=headers)

    def revoke_token(self, token_id, headers=None):
        _data = {}
        if token_id is not None:
            _data['token_id'] = token_id
        return self.client.perform_query('POST', '/token/delete', data=_data, headers=headers)


class InstancePoolService(object):
    def __init__(self, client):
        self.client = client

    def create_instance_pool(
        self,
        instance_pool_name=None,
        min_idle_instances=None,
        max_capacity=None,
        aws_attributes=None,
        node_type_id=None,
        custom_tags=None,
        idle_instance_autotermination_minutes=None,
        enable_elastic_disk=None,
        disk_spec=None,
        preloaded_spark_versions=None,
        headers=None,
        preloaded_docker_images=None,
        azure_attributes=None,
        gcp_attributes=None,
    ):
        _data = {}
        if instance_pool_name is not None:
            _data['instance_pool_name'] = instance_pool_name
        if min_idle_instances is not None:
            _data['min_idle_instances'] = min_idle_instances
        if max_capacity is not None:
            _data['max_capacity'] = max_capacity
        if aws_attributes is not None:
            _data['aws_attributes'] = aws_attributes
            if not isinstance(aws_attributes, dict):
                raise TypeError(
                    'Expected databricks.InstancePoolAwsAttributes() or dict for field aws_attributes'
                )
        if node_type_id is not None:
            _data['node_type_id'] = node_type_id
        if custom_tags is not None:
            _data['custom_tags'] = custom_tags
        if idle_instance_autotermination_minutes is not None:
            _data['idle_instance_autotermination_minutes'] = idle_instance_autotermination_minutes
        if enable_elastic_disk is not None:
            _data['enable_elastic_disk'] = enable_elastic_disk
        if disk_spec is not None:
            _data['disk_spec'] = disk_spec
            if not isinstance(disk_spec, dict):
                raise TypeError('Expected databricks.DiskSpec() or dict for field disk_spec')
        if preloaded_spark_versions is not None:
            _data['preloaded_spark_versions'] = preloaded_spark_versions
        if preloaded_docker_images is not None:
            _data['preloaded_docker_images'] = preloaded_docker_images
        if azure_attributes is not None:
            _data['azure_attributes'] = azure_attributes
            if not isinstance(azure_attributes, dict):
                raise TypeError(
                    'Expected databricks.InstancePoolAzureAttributes() or dict for field azure_attributes'
                )
        if gcp_attributes is not None:
            _data['gcp_attributes'] = gcp_attributes
            if not isinstance(gcp_attributes, dict):
                raise TypeError(
                    'Expected databricks.InstancePoolGcpAttributes() or dict for field gcp_attributes'
                )
        return self.client.perform_query(
            'POST', '/instance-pools/create', data=_data, headers=headers
        )

    def delete_instance_pool(self, instance_pool_id=None, headers=None):
        _data = {}
        if instance_pool_id is not None:
            _data['instance_pool_id'] = instance_pool_id
        return self.client.perform_query(
            'POST', '/instance-pools/delete', data=_data, headers=headers
        )

    def edit_instance_pool(
        self,
        instance_pool_id,
        instance_pool_name=None,
        min_idle_instances=None,
        max_capacity=None,
        aws_attributes=None,
        node_type_id=None,
        custom_tags=None,
        idle_instance_autotermination_minutes=None,
        enable_elastic_disk=None,
        disk_spec=None,
        preloaded_spark_versions=None,
        headers=None,
        preloaded_docker_images=None,
        azure_attributes=None,
        gcp_attributes=None,
    ):
        _data = {}
        if instance_pool_id is not None:
            _data['instance_pool_id'] = instance_pool_id
        if instance_pool_name is not None:
            _data['instance_pool_name'] = instance_pool_name
        if min_idle_instances is not None:
            _data['min_idle_instances'] = min_idle_instances
        if max_capacity is not None:
            _data['max_capacity'] = max_capacity
        if aws_attributes is not None:
            _data['aws_attributes'] = aws_attributes
            if not isinstance(aws_attributes, dict):
                raise TypeError(
                    'Expected databricks.InstancePoolAwsAttributes() or dict for field aws_attributes'
                )
        if node_type_id is not None:
            _data['node_type_id'] = node_type_id
        if custom_tags is not None:
            _data['custom_tags'] = custom_tags
        if idle_instance_autotermination_minutes is not None:
            _data['idle_instance_autotermination_minutes'] = idle_instance_autotermination_minutes
        if enable_elastic_disk is not None:
            _data['enable_elastic_disk'] = enable_elastic_disk
        if disk_spec is not None:
            _data['disk_spec'] = disk_spec
            if not isinstance(disk_spec, dict):
                raise TypeError('Expected databricks.DiskSpec() or dict for field disk_spec')
        if preloaded_spark_versions is not None:
            _data['preloaded_spark_versions'] = preloaded_spark_versions
        if preloaded_docker_images is not None:
            _data['preloaded_docker_images'] = preloaded_docker_images
        if azure_attributes is not None:
            _data['azure_attributes'] = azure_attributes
            if not isinstance(azure_attributes, dict):
                raise TypeError(
                    'Expected databricks.InstancePoolAzureAttributes() or dict for field azure_attributes'
                )
        if gcp_attributes is not None:
            _data['gcp_attributes'] = gcp_attributes
            if not isinstance(gcp_attributes, dict):
                raise TypeError(
                    'Expected databricks.InstancePoolGcpAttributes() or dict for field gcp_attributes'
                )
        return self.client.perform_query(
            'POST', '/instance-pools/edit', data=_data, headers=headers
        )

    def get_instance_pool(self, instance_pool_id=None, headers=None):
        _data = {}
        if instance_pool_id is not None:
            _data['instance_pool_id'] = instance_pool_id
        return self.client.perform_query('GET', '/instance-pools/get', data=_data, headers=headers)

    def list_instance_pools(self, headers=None):
        _data = {}
        return self.client.perform_query('GET', '/instance-pools/list', data=_data, headers=headers)


class DeltaPipelinesService(object):
    def __init__(self, client):
        self.client = client

    def create(
        self,
        id=None,
        name=None,
        storage=None,
        configuration=None,
        clusters=None,
        libraries=None,
        trigger=None,
        filters=None,
        target=None,
        continuous=None,
        development=None,
        allow_duplicate_names=None,
        headers=None,
    ):
        _data = {}
        if id is not None:
            _data['id'] = id
        if name is not None:
            _data['name'] = name
        if storage is not None:
            _data['storage'] = storage
        if configuration is not None:
            _data['configuration'] = configuration
        if clusters is not None:
            _data['clusters'] = clusters
        if libraries is not None:
            _data['libraries'] = libraries
        if trigger is not None:
            _data['trigger'] = trigger
            if not isinstance(trigger, dict):
                raise TypeError('Expected databricks.PipelineTrigger() or dict for field trigger')
        if filters is not None:
            _data['filters'] = filters
            if not isinstance(filters, dict):
                raise TypeError('Expected databricks.Filters() or dict for field filters')
        if target is not None:
            _data['target'] = target
        if continuous is not None:
            _data['continuous'] = continuous
        if development is not None:
            _data['development'] = development
        if allow_duplicate_names is not None:
            _data['allow_duplicate_names'] = allow_duplicate_names
        return self.client.perform_query('POST', '/pipelines', data=_data, headers=headers)

    def deploy(
        self,
        pipeline_id=None,
        id=None,
        name=None,
        storage=None,
        configuration=None,
        clusters=None,
        libraries=None,
        trigger=None,
        filters=None,
        target=None,
        continuous=None,
        development=None,
        allow_duplicate_names=None,
        headers=None,
    ):
        _data = {}
        if id is not None:
            _data['id'] = id
        if name is not None:
            _data['name'] = name
        if storage is not None:
            _data['storage'] = storage
        if configuration is not None:
            _data['configuration'] = configuration
        if clusters is not None:
            _data['clusters'] = clusters
        if libraries is not None:
            _data['libraries'] = libraries
        if trigger is not None:
            _data['trigger'] = trigger
            if not isinstance(trigger, dict):
                raise TypeError('Expected databricks.PipelineTrigger() or dict for field trigger')
        if filters is not None:
            _data['filters'] = filters
            if not isinstance(filters, dict):
                raise TypeError('Expected databricks.Filters() or dict for field filters')
        if target is not None:
            _data['target'] = target
        if continuous is not None:
            _data['continuous'] = continuous
        if development is not None:
            _data['development'] = development
        if allow_duplicate_names is not None:
            _data['allow_duplicate_names'] = allow_duplicate_names
        return self.client.perform_query(
            'PUT',
            '/pipelines/{pipeline_id}'.format(pipeline_id=pipeline_id),
            data=_data,
            headers=headers,
        )

    def delete(self, pipeline_id=None, headers=None):
        _data = {}

        return self.client.perform_query(
            'DELETE',
            '/pipelines/{pipeline_id}'.format(pipeline_id=pipeline_id),
            data=_data,
            headers=headers,
        )

    def get(self, pipeline_id=None, headers=None):
        _data = {}

        return self.client.perform_query(
            'GET',
            '/pipelines/{pipeline_id}'.format(pipeline_id=pipeline_id),
            data=_data,
            headers=headers,
        )

    def list(self, pagination=None, headers=None):
        _data = {}
        if pagination is not None:
            _data['pagination'] = pagination
            if not isinstance(pagination, dict):
                raise TypeError('Expected databricks.Pagination() or dict for field pagination')
        return self.client.perform_query('GET', '/pipelines', data=_data, headers=headers)

    def reset(self, pipeline_id=None, headers=None):
        _data = {}

        return self.client.perform_query(
            'POST',
            '/pipelines/{pipeline_id}/reset'.format(pipeline_id=pipeline_id),
            data=_data,
            headers=headers,
        )

    def run(self, pipeline_id=None, headers=None):
        _data = {}

        return self.client.perform_query(
            'POST',
            '/pipelines/{pipeline_id}/run'.format(pipeline_id=pipeline_id),
            data=_data,
            headers=headers,
        )

    def start_update(self, pipeline_id=None, full_refresh=None, headers=None):
        _data = {}
        if full_refresh:
            _data['full_refresh'] = full_refresh
        _data['cause'] = 'USER_ACTION'
        return self.client.perform_query(
            'POST',
            '/pipelines/{pipeline_id}/updates'.format(pipeline_id=pipeline_id),
            data=_data,
            headers=headers,
        )

    def stop(self, pipeline_id=None, headers=None):
        _data = {}

        return self.client.perform_query(
            'POST',
            '/pipelines/{pipeline_id}/stop'.format(pipeline_id=pipeline_id),
            data=_data,
            headers=headers,
        )


class ReposService(object):
    __git_providers__ = {
        "github.com": "gitHub",
        "dev.azure.com": "azureDevOpsServices",
        "gitlab.com": "gitLab",
        "bitbucket.org": "bitbucketCloud",
    }
    __aws_code_commit_regexp__ = re.compile(r"^git-codecommit\.[^.]+\.amazonaws.com$")

    def __init__(self, client):
        self.client = client

    @staticmethod
    def detect_repo_provider(url):
        provider = None
        try:
            netloc = urlparse(url).netloc
            idx = netloc.rfind("@")
            if idx != -1:
                netloc = netloc[(idx + 1) :]
            provider = ReposService.__git_providers__.get(netloc.lower())
            if provider is None and ReposService.__aws_code_commit_regexp__.match(netloc):
                provider = "awsCodeCommit"
        except:
            pass
        return provider

    def list_repos(self, path_prefix=None, next_page_token=None, headers=None):
        _data = {}
        if path_prefix is not None:
            _data['path_prefix'] = path_prefix
        if next_page_token is not None:
            _data['next_page_token'] = next_page_token
        return self.client.perform_query('GET', '/repos', data=_data, headers=headers)

    def get_repo(self, id, headers=None):
        _data = {}

        return self.client.perform_query(
            'GET', '/repos/{id}'.format(id=id), data=_data, headers=headers
        )

    def update_repo(self, id, branch=None, tag=None, headers=None):
        _data = {}
        if branch is not None:
            _data['branch'] = branch
        if tag is not None:
            _data['tag'] = tag
        return self.client.perform_query(
            'PATCH', '/repos/{id}'.format(id=id), data=_data, headers=headers
        )

    def create_repo(self, url, provider, path=None, headers=None):
        _data = {}
        if url is not None:
            _data['url'] = url
        if provider is None or provider.strip() == "":
            provider = self.detect_repo_provider(url)
        if provider is not None:
            _data['provider'] = provider
        else:
            raise ValueError(
                "The Git provider parameter wasn't specified and we can't detect it "
                "from URL. Please pass 'provider' option"
            )
        if path is not None:
            _data['path'] = path
        return self.client.perform_query('POST', '/repos', data=_data, headers=headers)

    def delete_repo(self, id, headers=None):
        _data = {}

        return self.client.perform_query(
            'DELETE', '/repos/{id}'.format(id=id), data=_data, headers=headers
        )