airflow.providers.common.sql.operators.sql

Module Contents

Classes

BaseSQLOperator

This is a base class for generic SQL Operator to get a DB Hook.

SQLExecuteQueryOperator

Executes SQL code in a specific database.

SQLColumnCheckOperator

Performs one or more of the templated checks in the column_checks dictionary.

SQLTableCheckOperator

Performs one or more of the checks provided in the checks dictionary.

SQLCheckOperator

Performs checks against a db.

SQLValueCheckOperator

Performs a simple value check using sql code.

SQLIntervalCheckOperator

Check that metrics given as SQL expressions are within tolerance of the ones from days_back before.

SQLThresholdCheckOperator

Performs a value check using sql code against a minimum threshold and a maximum threshold.

BranchSQLOperator

Allows a DAG to "branch" or follow a specified path based on the results of a SQL query.

class airflow.providers.common.sql.operators.sql.BaseSQLOperator(*, conn_id=None, database=None, hook_params=None, retry_on_failure=True, **kwargs)[source]

Bases: airflow.models.BaseOperator

This is a base class for generic SQL Operator to get a DB Hook.

The provided method is .get_db_hook(). The default behavior will try to retrieve the DB hook based on connection type. You can customize the behavior by overriding the .get_db_hook() method.

Parameters

conn_id (str | None) – reference to a specific database

conn_id_field = 'conn_id'[source]
get_db_hook()[source]

Get the database hook for the connection.

Returns

the database hook object.

Return type

airflow.providers.common.sql.hooks.sql.DbApiHook

class airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator(*, sql, autocommit=False, parameters=None, handler=fetch_all_handler, conn_id=None, database=None, split_statements=None, return_last=True, show_return_value_in_logs=False, **kwargs)[source]

Bases: BaseSQLOperator

Executes SQL code in a specific database.

When implementing a specific Operator, you can also implement _process_output method in the hook to perform additional processing of values returned by the DB Hook of yours. For example, you can join description retrieved from the cursors of your statements with returned values, or save the output of your operator to a file.

Parameters
  • sql (str | list[str]) – the SQL code or string pointing to a template file to be executed (templated). File must have a ‘.sql’ extension.

  • autocommit (bool) – (optional) if True, each command is automatically committed (default: False).

  • parameters (Mapping | Iterable | None) – (optional) the parameters to render the SQL query with.

  • handler (Callable[[Any], Any]) – (optional) the function that will be applied to the cursor (default: fetch_all_handler).

  • split_statements (bool | None) – (optional) if split single SQL string into statements. By default, defers to the default value in the run method of the configured hook.

  • conn_id (str | None) – the connection ID used to connect to the database

  • database (str | None) – name of database which overwrite the defined one in connection

  • return_last (bool) – (optional) return the result of only last statement (default: True).

  • show_return_value_in_logs (bool) – (optional) if true operator output will be printed to the task log. Use with caution. It’s not recommended to dump large datasets to the log. (default: False).

See also

For more information on how to use this operator, take a look at the guide: Execute SQL query

template_fields: Sequence[str] = ('conn_id', 'sql', 'parameters')[source]
template_ext: Sequence[str] = ('.sql', '.json')[source]
template_fields_renderers[source]
ui_color = '#cdaaed'[source]
execute(context)[source]

Derive when creating an operator.

Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

prepare_template()[source]

Parse template file for attribute parameters.

get_openlineage_facets_on_start()[source]
get_openlineage_facets_on_complete(task_instance)[source]
class airflow.providers.common.sql.operators.sql.SQLColumnCheckOperator(*, table, column_mapping, partition_clause=None, conn_id=None, database=None, accept_none=True, **kwargs)[source]

Bases: BaseSQLOperator

Performs one or more of the templated checks in the column_checks dictionary.

Checks are performed on a per-column basis specified by the column_mapping.

Each check can take one or more of the following options:

  • equal_to: an exact value to equal, cannot be used with other comparison options

  • greater_than: value that result should be strictly greater than

  • less_than: value that results should be strictly less than

  • geq_to: value that results should be greater than or equal to

  • leq_to: value that results should be less than or equal to

  • tolerance: the percentage that the result may be off from the expected value

  • partition_clause: an extra clause passed into a WHERE statement to partition data

Parameters
  • table (str) – the table to run checks on

  • column_mapping (dict[str, dict[str, Any]]) –

    the dictionary of columns and their associated checks, e.g.

    {
        "col_name": {
            "null_check": {
                "equal_to": 0,
                "partition_clause": "foreign_key IS NOT NULL",
            },
            "min": {
                "greater_than": 5,
                "leq_to": 10,
                "tolerance": 0.2,
            },
            "max": {"less_than": 1000, "geq_to": 10, "tolerance": 0.01},
        }
    }
    

  • partition_clause (str | None) –

    a partial SQL statement that is added to a WHERE clause in the query built by the operator that creates partition_clauses for the checks to run on, e.g.

    "date = '1970-01-01'"
    

  • conn_id (str | None) – the connection ID used to connect to the database

  • database (str | None) – name of database which overwrite the defined one in connection

  • accept_none (bool) – whether or not to accept None values returned by the query. If true, converts None to 0.

See also

For more information on how to use this operator, take a look at the guide: Check SQL Table Columns

template_fields: Sequence[str] = ('partition_clause', 'table', 'sql')[source]
template_fields_renderers[source]
sql_check_template = Multiline-String[source]
Show Value
"""
        SELECT '{column}' AS col_name, '{check}' AS check_type, {column}_{check} AS check_result
        FROM (SELECT {check_statement} AS {column}_{check} FROM {table} {partition_clause}) AS sq
    """
column_checks[source]
execute(context)[source]

Derive when creating an operator.

Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

class airflow.providers.common.sql.operators.sql.SQLTableCheckOperator(*, table, checks, partition_clause=None, conn_id=None, database=None, **kwargs)[source]

Bases: BaseSQLOperator

Performs one or more of the checks provided in the checks dictionary.

Checks should be written to return a boolean result.

Parameters
  • table (str) – the table to run checks on

  • checks (dict[str, dict[str, Any]]) –

    the dictionary of checks, where check names are followed by a dictionary containing at least a check statement, and optionally a partition clause, e.g.:

    {
        "row_count_check": {"check_statement": "COUNT(*) = 1000"},
        "column_sum_check": {"check_statement": "col_a + col_b < col_c"},
        "third_check": {"check_statement": "MIN(col) = 1", "partition_clause": "col IS NOT NULL"},
    }
    

  • partition_clause (str | None) –

    a partial SQL statement that is added to a WHERE clause in the query built by the operator that creates partition_clauses for the checks to run on, e.g.

    "date = '1970-01-01'"
    

  • conn_id (str | None) – the connection ID used to connect to the database

  • database (str | None) – name of database which overwrite the defined one in connection

See also

For more information on how to use this operator, take a look at the guide: Check SQL Table Values

template_fields: Sequence[str] = ('partition_clause', 'table', 'sql', 'conn_id')[source]
template_fields_renderers[source]
sql_check_template = Multiline-String[source]
Show Value
"""
    SELECT '{check_name}' AS check_name, MIN({check_name}) AS check_result
    FROM (SELECT CASE WHEN {check_statement} THEN 1 ELSE 0 END AS {check_name}
          FROM {table} {partition_clause}) AS sq
    """
execute(context)[source]

Derive when creating an operator.

Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

class airflow.providers.common.sql.operators.sql.SQLCheckOperator(*, sql, conn_id=None, database=None, parameters=None, **kwargs)[source]

Bases: BaseSQLOperator

Performs checks against a db.

The SQLCheckOperator expects a sql query that will return a single row. Each value on that first row is evaluated using python bool casting. If any of the values return False the check is failed and errors out. If a Python dict is returned, and any values in the Python dict are False, the check is failed and errors out.

Note that Python bool casting evals the following as False:

  • False

  • 0

  • Empty string ("")

  • Empty list ([])

  • Empty dictionary or set ({})

  • Dictionary with value = False ({'DUPLICATE_ID_CHECK': False})

Given a query like SELECT COUNT(*) FROM foo, it will fail only if the count == 0. You can craft much more complex query that could, for instance, check that the table has the same number of rows as the source table upstream, or that the count of today’s partition is greater than yesterday’s partition, or that a set of metrics are less than 3 standard deviation for the 7 day average.

This operator can be used as a data quality check in your pipeline, and depending on where you put it in your DAG, you have the choice to stop the critical path, preventing from publishing dubious data, or on the side and receive email alerts without stopping the progress of the DAG.

Parameters
  • sql (str) – the sql to be executed. (templated)

  • conn_id (str | None) – the connection ID used to connect to the database.

  • database (str | None) – name of database which overwrite the defined one in connection

  • parameters (Iterable | Mapping[str, Any] | None) – (optional) the parameters to render the SQL query with.

template_fields: Sequence[str] = ('sql',)[source]
template_ext: Sequence[str] = ('.hql', '.sql')[source]
template_fields_renderers[source]
ui_color = '#fff7e6'[source]
execute(context)[source]

Derive when creating an operator.

Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

class airflow.providers.common.sql.operators.sql.SQLValueCheckOperator(*, sql, pass_value, tolerance=None, conn_id=None, database=None, **kwargs)[source]

Bases: BaseSQLOperator

Performs a simple value check using sql code.

Parameters
  • sql (str) – the sql to be executed. (templated)

  • conn_id (str | None) – the connection ID used to connect to the database.

  • database (str | None) – name of database which overwrite the defined one in connection

__mapper_args__[source]
template_fields: Sequence[str] = ('sql', 'pass_value')[source]
template_ext: Sequence[str] = ('.hql', '.sql')[source]
template_fields_renderers[source]
ui_color = '#fff7e6'[source]
check_value(records)[source]
execute(context)[source]

Derive when creating an operator.

Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

class airflow.providers.common.sql.operators.sql.SQLIntervalCheckOperator(*, table, metrics_thresholds, date_filter_column='ds', days_back=-7, ratio_formula='max_over_min', ignore_zero=True, conn_id=None, database=None, **kwargs)[source]

Bases: BaseSQLOperator

Check that metrics given as SQL expressions are within tolerance of the ones from days_back before.

Parameters
  • table (str) – the table name

  • conn_id (str | None) – the connection ID used to connect to the database.

  • database (str | None) – name of database which will overwrite the defined one in connection

  • days_back (SupportsAbs[int]) – number of days between ds and the ds we want to check against. Defaults to 7 days

  • date_filter_column (str | None) – The column name for the dates to filter on. Defaults to ‘ds’

  • ratio_formula (str | None) –

    which formula to use to compute the ratio between the two metrics. Assuming cur is the metric of today and ref is the metric to today - days_back. Default: ‘max_over_min’

    • max_over_min: computes max(cur, ref) / min(cur, ref)

    • relative_diff: computes abs(cur-ref) / ref

  • ignore_zero (bool) – whether we should ignore zero metrics

  • metrics_thresholds (dict[str, int]) – a dictionary of ratios indexed by metrics

__mapper_args__[source]
template_fields: Sequence[str] = ('sql1', 'sql2')[source]
template_ext: Sequence[str] = ('.hql', '.sql')[source]
template_fields_renderers[source]
ui_color = '#fff7e6'[source]
ratio_formulas[source]
execute(context)[source]

Derive when creating an operator.

Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

class airflow.providers.common.sql.operators.sql.SQLThresholdCheckOperator(*, sql, min_threshold, max_threshold, conn_id=None, database=None, **kwargs)[source]

Bases: BaseSQLOperator

Performs a value check using sql code against a minimum threshold and a maximum threshold.

Thresholds can be in the form of a numeric value OR a sql statement that results a numeric.

Parameters
  • sql (str) – the sql to be executed. (templated)

  • conn_id (str | None) – the connection ID used to connect to the database.

  • database (str | None) – name of database which overwrite the defined one in connection

  • min_threshold (Any) – numerical value or min threshold sql to be executed (templated)

  • max_threshold (Any) – numerical value or max threshold sql to be executed (templated)

template_fields: Sequence[str] = ('sql', 'min_threshold', 'max_threshold')[source]
template_ext: Sequence[str] = ('.hql', '.sql')[source]
template_fields_renderers[source]
execute(context)[source]

Derive when creating an operator.

Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

push(meta_data)[source]

Send data check info and metadata to an external database.

Default functionality will log metadata.

class airflow.providers.common.sql.operators.sql.BranchSQLOperator(*, sql, follow_task_ids_if_true, follow_task_ids_if_false, conn_id='default_conn_id', database=None, parameters=None, **kwargs)[source]

Bases: BaseSQLOperator, airflow.models.SkipMixin

Allows a DAG to “branch” or follow a specified path based on the results of a SQL query.

Parameters
  • sql (str) – The SQL code to be executed, should return true or false (templated) Template reference are recognized by str ending in ‘.sql’. Expected SQL query to return a boolean (True/False), integer (0 = False, Otherwise = 1) or string (true/y/yes/1/on/false/n/no/0/off).

  • follow_task_ids_if_true (list[str]) – task id or task ids to follow if query returns true

  • follow_task_ids_if_false (list[str]) – task id or task ids to follow if query returns false

  • conn_id (str) – the connection ID used to connect to the database.

  • database (str | None) – name of database which overwrite the defined one in connection

  • parameters (Iterable | Mapping[str, Any] | None) – (optional) the parameters to render the SQL query with.

template_fields: Sequence[str] = ('sql',)[source]
template_ext: Sequence[str] = ('.sql',)[source]
template_fields_renderers[source]
ui_color = '#a22034'[source]
ui_fgcolor = '#F7F7F7'[source]
execute(context)[source]

Derive when creating an operator.

Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

Was this entry helpful?