`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.

Functions¶

default_output_processor(results, descriptions)

airflow.providers.common.sql.operators.sql.default_output_processor(results, descriptions)[source]¶

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]¶

template_fields: collections.abc.Sequence[str] = ('conn_id', 'database', 'hook_params')[source]¶

classmethod get_hook(conn_id, hook_params=None)[source]¶

Return default hook for this connection id.

Parameters

conn_id (str) – connection id
hook_params (dict | None) – hook parameters

Returns

default hook for this connection

Return type

airflow.hooks.base.BaseHook

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, output_processor=None, 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 (collections.abc.Mapping | collections.abc.Iterable | None) – (optional) the parameters to render the SQL query with.
handler (Callable[[Any], list[tuple] | None]) – (optional) the function that will be applied to the cursor (default: fetch_all_handler).
output_processor (Callable[[list[Any], list[collections.abc.Sequence[collections.abc.Sequence] | None]], list[Any] | tuple[list[collections.abc.Sequence[collections.abc.Sequence] | None], list]] | None) – (optional) the function that will be applied to the result (default: default_output_processor).
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: collections.abc.Sequence[str] = ('sql', 'parameters')[source]¶

template_ext: collections.abc.Sequence[str] = ('.sql', '.json')[source]¶

template_fields_renderers: ClassVar[dict][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: collections.abc.Sequence[str] = ('table', 'partition_clause', 'sql')[source]¶

template_fields_renderers: ClassVar[dict][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: collections.abc.Sequence[str] = ('table', 'partition_clause', 'sql')[source]¶

template_fields_renderers: ClassVar[dict][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 (collections.abc.Iterable | collections.abc.Mapping[str, Any] | None) – (optional) the parameters to render the SQL query with.

template_fields: collections.abc.Sequence[str] = ('sql',)[source]¶

template_ext: collections.abc.Sequence[str] = ('.hql', '.sql')[source]¶

template_fields_renderers: ClassVar[dict][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: collections.abc.Sequence[str] = ('sql', 'pass_value')[source]¶

template_ext: collections.abc.Sequence[str] = ('.hql', '.sql')[source]¶

template_fields_renderers: ClassVar[dict][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: collections.abc.Sequence[str] = ('sql1', 'sql2')[source]¶

template_ext: collections.abc.Sequence[str] = ('.hql', '.sql')[source]¶

template_fields_renderers: ClassVar[dict][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)

See also

For more information on how to use this operator, take a look at the guide: Check values against a threshold

template_fields: collections.abc.Sequence[str] = ('sql', 'min_threshold', 'max_threshold')[source]¶

template_ext: collections.abc.Sequence[str] = ('.hql', '.sql')[source]¶

template_fields_renderers: ClassVar[dict][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 (collections.abc.Iterable | collections.abc.Mapping[str, Any] | None) – (optional) the parameters to render the SQL query with.

template_fields: collections.abc.Sequence[str] = ('sql',)[source]¶

template_ext: collections.abc.Sequence[str] = ('.sql',)[source]¶

template_fields_renderers: ClassVar[dict][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.

airflow.providers.common.sql.operators.sql¶

Module Contents¶

Classes¶

Functions¶

`airflow.providers.common.sql.operators.sql`¶