airflow/contrib/hooks/bigquery_hook.py
def run_query(self,
                  sql,
                  destination_dataset_table=None,
                  write_disposition='WRITE_EMPTY',
                  allow_large_results=False,
                  flatten_results=None,
                  udf_config=None,
                  use_legacy_sql=None,
                  maximum_billing_tier=None,
                  maximum_bytes_billed=None,
                  create_disposition='CREATE_IF_NEEDED',
                  query_params=None,
                  labels=None,
                  schema_update_options=(),
                  priority='INTERACTIVE',
                  time_partitioning=None,
                  api_resource_configs=None,
                  cluster_fields=None,
                  location=None):
        """
        Executes a BigQuery SQL query. Optionally persists results in a BigQuery
        table. See here:

        https://cloud.google.com/bigquery/docs/reference/v2/jobs

        For more details about these parameters.

        :param sql: The BigQuery SQL to execute.
        :type sql: str
        :param destination_dataset_table: The dotted ``<dataset>.<table>``
            BigQuery table to save the query results.
        :type destination_dataset_table: str
        :param write_disposition: What to do if the table already exists in
            BigQuery.
        :type write_disposition: str
        :param allow_large_results: Whether to allow large results.
        :type allow_large_results: bool
        :param flatten_results: If true and query uses legacy SQL dialect, flattens
            all nested and repeated fields in the query results. ``allowLargeResults``
            must be true if this is set to false. For standard SQL queries, this
            flag is ignored and results are never flattened.
        :type flatten_results: bool
        :param udf_config: The User Defined Function configuration for the query.
            See https://cloud.google.com/bigquery/user-defined-functions for details.
        :type udf_config: list
        :param use_legacy_sql: Whether to use legacy SQL (true) or standard SQL (false).
            If `None`, defaults to `self.use_legacy_sql`.
        :type use_legacy_sql: bool
        :param api_resource_configs: a dictionary that contain params
            'configuration' applied for Google BigQuery Jobs API:
            https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs
            for example, {'query': {'useQueryCache': False}}. You could use it
            if you need to provide some params that are not supported by the
            BigQueryHook like args.
        :type api_resource_configs: dict
        :param maximum_billing_tier: Positive integer that serves as a
            multiplier of the basic price.
        :type maximum_billing_tier: int
        :param maximum_bytes_billed: Limits the bytes billed for this job.
            Queries that will have bytes billed beyond this limit will fail
            (without incurring a charge). If unspecified, this will be
            set to your project default.
        :type maximum_bytes_billed: float
        :param create_disposition: Specifies whether the job is allowed to
            create new tables.
        :type create_disposition: str
        :param query_params: a list of dictionary containing query parameter types and
            values, passed to BigQuery
        :type query_params: list
        :param labels: a dictionary containing labels for the job/query,
            passed to BigQuery
        :type labels: dict
        :param schema_update_options: Allows the schema of the destination
            table to be updated as a side effect of the query job.
        :type schema_update_options: tuple
        :param priority: Specifies a priority for the query.
            Possible values include INTERACTIVE and BATCH.
            The default value is INTERACTIVE.
        :type priority: str
        :param time_partitioning: configure optional time partitioning fields i.e.
            partition by field, type and expiration as per API specifications.
        :type time_partitioning: dict
        :param cluster_fields: Request that the result of this query be stored sorted
            by one or more columns. This is only available in combination with
            time_partitioning. The order of columns given determines the sort order.
        :type cluster_fields: list[str]
        :param location: The geographic location of the job. Required except for
            US and EU. See details at
            https://cloud.google.com/bigquery/docs/locations#specifying_your_location
        :type location: str
        """

        if time_partitioning is None:
            time_partitioning = {}

        if location:
            self.location = location

        if not api_resource_configs:
            api_resource_configs = self.api_resource_configs
        else:
            _validate_value('api_resource_configs',
                            api_resource_configs, dict)
        configuration = deepcopy(api_resource_configs)
        if 'query' not in configuration:
            configuration['query'] = {}

        else:
            _validate_value("api_resource_configs['query']",
                            configuration['query'], dict)

        if sql is None and not configuration['query'].get('query', None):
            raise TypeError('`BigQueryBaseCursor.run_query` '
                            'missing 1 required positional argument: `sql`')

        # BigQuery also allows you to define how you want a table's schema to change
        # as a side effect of a query job
        # for more details:
        #   https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.query.schemaUpdateOptions

        allowed_schema_update_options = [
            'ALLOW_FIELD_ADDITION', "ALLOW_FIELD_RELAXATION"
        ]

        if not set(allowed_schema_update_options
                   ).issuperset(set(schema_update_options)):
            raise ValueError("{0} contains invalid schema update options. "
                             "Please only use one or more of the following "
                             "options: {1}"
                             .format(schema_update_options,
                                     allowed_schema_update_options))

        if schema_update_options:
            if write_disposition not in ["WRITE_APPEND", "WRITE_TRUNCATE"]:
                raise ValueError("schema_update_options is only "
                                 "allowed if write_disposition is "
                                 "'WRITE_APPEND' or 'WRITE_TRUNCATE'.")

        if destination_dataset_table:
            destination_project, destination_dataset, destination_table = \
                _split_tablename(table_input=destination_dataset_table,
                                 default_project_id=self.project_id)

            destination_dataset_table = {
                'projectId': destination_project,
                'datasetId': destination_dataset,
                'tableId': destination_table,
            }

        if cluster_fields:
            cluster_fields = {'fields': cluster_fields}

        query_param_list = [
            (sql, 'query', None, six.string_types),
            (priority, 'priority', 'INTERACTIVE', six.string_types),
            (use_legacy_sql, 'useLegacySql', self.use_legacy_sql, bool),
            (query_params, 'queryParameters', None, list),
            (udf_config, 'userDefinedFunctionResources', None, list),
            (maximum_billing_tier, 'maximumBillingTier', None, int),
            (maximum_bytes_billed, 'maximumBytesBilled', None, float),
            (time_partitioning, 'timePartitioning', {}, dict),
            (schema_update_options, 'schemaUpdateOptions', None, tuple),
            (destination_dataset_table, 'destinationTable', None, dict),
            (cluster_fields, 'clustering', None, dict),
        ]

        for param_tuple in query_param_list:

            param, param_name, param_default, param_type = param_tuple

            if param_name not in configuration['query'] and param in [None, {}, ()]:
                if param_name == 'timePartitioning':
                    param_default = _cleanse_time_partitioning(
                        destination_dataset_table, time_partitioning)
                param = param_default

            if param not in [None, {}, ()]:
                _api_resource_configs_duplication_check(
                    param_name, param, configuration['query'])

                configuration['query'][param_name] = param

                # check valid type of provided param,
                # it last step because we can get param from 2 sources,
                # and first of all need to find it

                _validate_value(param_name, configuration['query'][param_name],
                                param_type)

                if param_name == 'schemaUpdateOptions' and param:
                    self.log.info("Adding experimental 'schemaUpdateOptions': "
                                  "%s", schema_update_options)

                if param_name == 'destinationTable':
                    for key in ['projectId', 'datasetId', 'tableId']:
                        if key not in configuration['query']['destinationTable']:
                            raise ValueError(
                                "Not correct 'destinationTable' in "
                                "api_resource_configs. 'destinationTable' "
                                "must be a dict with {'projectId':'', "
                                "'datasetId':'', 'tableId':''}")

                    configuration['query'].update({
                        'allowLargeResults': allow_large_results,
                        'flattenResults': flatten_results,
                        'writeDisposition': write_disposition,
                        'createDisposition': create_disposition,
                    })

        if 'useLegacySql' in configuration['query'] and configuration['query']['useLegacySql'] and\
                'queryParameters' in configuration['query']:
            raise ValueError("Query parameters are not allowed "
                             "when using legacy SQL")

        if labels:
            _api_resource_configs_duplication_check(
                'labels', labels, configuration)
            configuration['labels'] = labels

        return self.run_with_configuration(configuration)
def run_query(self,
                  sql,
                  destination_dataset_table=None,
                  write_disposition='WRITE_EMPTY',
                  allow_large_results=False,
                  flatten_results=None,
                  udf_config=None,
                  use_legacy_sql=None,
                  maximum_billing_tier=None,
                  maximum_bytes_billed=None,
                  create_disposition='CREATE_IF_NEEDED',
                  query_params=None,
                  labels=None,
                  schema_update_options=(),
                  priority='INTERACTIVE',
                  time_partitioning=None,
                  api_resource_configs=None,
                  cluster_fields=None,
                  location=None):
        """
        Executes a BigQuery SQL query. Optionally persists results in a BigQuery
        table. See here:

        https://cloud.google.com/bigquery/docs/reference/v2/jobs

        For more details about these parameters.

        :param sql: The BigQuery SQL to execute.
        :type sql: str
        :param destination_dataset_table: The dotted ``<dataset>.<table>``
            BigQuery table to save the query results.
        :type destination_dataset_table: str
        :param write_disposition: What to do if the table already exists in
            BigQuery.
        :type write_disposition: str
        :param allow_large_results: Whether to allow large results.
        :type allow_large_results: bool
        :param flatten_results: If true and query uses legacy SQL dialect, flattens
            all nested and repeated fields in the query results. ``allowLargeResults``
            must be true if this is set to false. For standard SQL queries, this
            flag is ignored and results are never flattened.
        :type flatten_results: bool
        :param udf_config: The User Defined Function configuration for the query.
            See https://cloud.google.com/bigquery/user-defined-functions for details.
        :type udf_config: list
        :param use_legacy_sql: Whether to use legacy SQL (true) or standard SQL (false).
            If `None`, defaults to `self.use_legacy_sql`.
        :type use_legacy_sql: bool
        :param api_resource_configs: a dictionary that contain params
            'configuration' applied for Google BigQuery Jobs API:
            https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs
            for example, {'query': {'useQueryCache': False}}. You could use it
            if you need to provide some params that are not supported by the
            BigQueryHook like args.
        :type api_resource_configs: dict
        :param maximum_billing_tier: Positive integer that serves as a
            multiplier of the basic price.
        :type maximum_billing_tier: int
        :param maximum_bytes_billed: Limits the bytes billed for this job.
            Queries that will have bytes billed beyond this limit will fail
            (without incurring a charge). If unspecified, this will be
            set to your project default.
        :type maximum_bytes_billed: float
        :param create_disposition: Specifies whether the job is allowed to
            create new tables.
        :type create_disposition: str
        :param query_params: a list of dictionary containing query parameter types and
            values, passed to BigQuery
        :type query_params: list
        :param labels: a dictionary containing labels for the job/query,
            passed to BigQuery
        :type labels: dict
        :param schema_update_options: Allows the schema of the destination
            table to be updated as a side effect of the query job.
        :type schema_update_options: tuple
        :param priority: Specifies a priority for the query.
            Possible values include INTERACTIVE and BATCH.
            The default value is INTERACTIVE.
        :type priority: str
        :param time_partitioning: configure optional time partitioning fields i.e.
            partition by field, type and expiration as per API specifications.
        :type time_partitioning: dict
        :param cluster_fields: Request that the result of this query be stored sorted
            by one or more columns. This is only available in combination with
            time_partitioning. The order of columns given determines the sort order.
        :type cluster_fields: list[str]
        :param location: The geographic location of the job. Required except for
            US and EU. See details at
            https://cloud.google.com/bigquery/docs/locations#specifying_your_location
        :type location: str
        """

        if time_partitioning is None:
            time_partitioning = {}

        if location:
            self.location = location

        if not api_resource_configs:
            api_resource_configs = self.api_resource_configs
        else:
            _validate_value('api_resource_configs',
                            api_resource_configs, dict)
        configuration = deepcopy(api_resource_configs)
        if 'query' not in configuration:
            configuration['query'] = {}

        else:
            _validate_value("api_resource_configs['query']",
                            configuration['query'], dict)

        if sql is None and not configuration['query'].get('query', None):
            raise TypeError('`BigQueryBaseCursor.run_query` '
                            'missing 1 required positional argument: `sql`')

        # BigQuery also allows you to define how you want a table's schema to change
        # as a side effect of a query job
        # for more details:
        #   https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs#configuration.query.schemaUpdateOptions

        allowed_schema_update_options = [
            'ALLOW_FIELD_ADDITION', "ALLOW_FIELD_RELAXATION"
        ]

        if not set(allowed_schema_update_options
                   ).issuperset(set(schema_update_options)):
            raise ValueError("{0} contains invalid schema update options. "
                             "Please only use one or more of the following "
                             "options: {1}"
                             .format(schema_update_options,
                                     allowed_schema_update_options))

        if schema_update_options:
            if write_disposition not in ["WRITE_APPEND", "WRITE_TRUNCATE"]:
                raise ValueError("schema_update_options is only "
                                 "allowed if write_disposition is "
                                 "'WRITE_APPEND' or 'WRITE_TRUNCATE'.")

        if destination_dataset_table:
            destination_project, destination_dataset, destination_table = \
                _split_tablename(table_input=destination_dataset_table,
                                 default_project_id=self.project_id)

            destination_dataset_table = {
                'projectId': destination_project,
                'datasetId': destination_dataset,
                'tableId': destination_table,
            }

        if cluster_fields:
            cluster_fields = {'fields': cluster_fields}

        query_param_list = [
            (sql, 'query', None, six.string_types),
            (priority, 'priority', 'INTERACTIVE', six.string_types),
            (use_legacy_sql, 'useLegacySql', self.use_legacy_sql, bool),
            (query_params, 'queryParameters', None, list),
            (udf_config, 'userDefinedFunctionResources', None, list),
            (maximum_billing_tier, 'maximumBillingTier', None, int),
            (maximum_bytes_billed, 'maximumBytesBilled', None, float),
            (time_partitioning, 'timePartitioning', {}, dict),
            (schema_update_options, 'schemaUpdateOptions', None, tuple),
            (destination_dataset_table, 'destinationTable', None, dict),
            (cluster_fields, 'clustering', None, dict),
        ]

        for param_tuple in query_param_list:

            param, param_name, param_default, param_type = param_tuple

            if param_name not in configuration['query'] and param in [None, {}, ()]:
                if param_name == 'timePartitioning':
                    param_default = _cleanse_time_partitioning(
                        destination_dataset_table, time_partitioning)
                param = param_default

            if param not in [None, {}, ()]:
                _api_resource_configs_duplication_check(
                    param_name, param, configuration['query'])

                configuration['query'][param_name] = param

                # check valid type of provided param,
                # it last step because we can get param from 2 sources,
                # and first of all need to find it

                _validate_value(param_name, configuration['query'][param_name],
                                param_type)

                if param_name == 'schemaUpdateOptions' and param:
                    self.log.info("Adding experimental 'schemaUpdateOptions': "
                                  "%s", schema_update_options)

                if param_name == 'destinationTable':
                    for key in ['projectId', 'datasetId', 'tableId']:
                        if key not in configuration['query']['destinationTable']:
                            raise ValueError(
                                "Not correct 'destinationTable' in "
                                "api_resource_configs. 'destinationTable' "
                                "must be a dict with {'projectId':'', "
                                "'datasetId':'', 'tableId':''}")

                    configuration['query'].update({
                        'allowLargeResults': allow_large_results,
                        'flattenResults': flatten_results,
                        'writeDisposition': write_disposition,
                        'createDisposition': create_disposition,
                    })

        if 'useLegacySql' in configuration['query'] and configuration['query']['useLegacySql'] and\
                'queryParameters' in configuration['query']:
            raise ValueError("Query parameters are not allowed "
                             "when using legacy SQL")

        if labels:
            _api_resource_configs_duplication_check(
                'labels', labels, configuration)
            configuration['labels'] = labels

        return self.run_with_configuration(configuration)
Executes a BigQuery SQL query. Optionally persists results in a BigQuery
        table. See here:

        https://cloud.google.com/bigquery/docs/reference/v2/jobs

        For more details about these parameters.

        :param sql: The BigQuery SQL to execute.
        :type sql: str
        :param destination_dataset_table: The dotted ``<dataset>.<table>``
            BigQuery table to save the query results.
        :type destination_dataset_table: str
        :param write_disposition: What to do if the table already exists in
            BigQuery.
        :type write_disposition: str
        :param allow_large_results: Whether to allow large results.
        :type allow_large_results: bool
        :param flatten_results: If true and query uses legacy SQL dialect, flattens
            all nested and repeated fields in the query results. ``allowLargeResults``
            must be true if this is set to false. For standard SQL queries, this
            flag is ignored and results are never flattened.
        :type flatten_results: bool
        :param udf_config: The User Defined Function configuration for the query.
            See https://cloud.google.com/bigquery/user-defined-functions for details.
        :type udf_config: list
        :param use_legacy_sql: Whether to use legacy SQL (true) or standard SQL (false).
            If `None`, defaults to `self.use_legacy_sql`.
        :type use_legacy_sql: bool
        :param api_resource_configs: a dictionary that contain params
            'configuration' applied for Google BigQuery Jobs API:
            https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs
            for example, {'query': {'useQueryCache': False}}. You could use it
            if you need to provide some params that are not supported by the
            BigQueryHook like args.
        :type api_resource_configs: dict
        :param maximum_billing_tier: Positive integer that serves as a
            multiplier of the basic price.
        :type maximum_billing_tier: int
        :param maximum_bytes_billed: Limits the bytes billed for this job.
            Queries that will have bytes billed beyond this limit will fail
            (without incurring a charge). If unspecified, this will be
            set to your project default.
        :type maximum_bytes_billed: float
        :param create_disposition: Specifies whether the job is allowed to
            create new tables.
        :type create_disposition: str
        :param query_params: a list of dictionary containing query parameter types and
            values, passed to BigQuery
        :type query_params: list
        :param labels: a dictionary containing labels for the job/query,
            passed to BigQuery
        :type labels: dict
        :param schema_update_options: Allows the schema of the destination
            table to be updated as a side effect of the query job.
        :type schema_update_options: tuple
        :param priority: Specifies a priority for the query.
            Possible values include INTERACTIVE and BATCH.
            The default value is INTERACTIVE.
        :type priority: str
        :param time_partitioning: configure optional time partitioning fields i.e.
            partition by field, type and expiration as per API specifications.
        :type time_partitioning: dict
        :param cluster_fields: Request that the result of this query be stored sorted
            by one or more columns. This is only available in combination with
            time_partitioning. The order of columns given determines the sort order.
        :type cluster_fields: list[str]
        :param location: The geographic location of the job. Required except for
            US and EU. See details at
            https://cloud.google.com/bigquery/docs/locations#specifying_your_location
        :type location: str