0
# Apache Airflow Common SQL Provider
1

2
A comprehensive provider package offering common SQL functionality for Apache Airflow, providing hooks, operators, sensors, and triggers for SQL database operations. This package serves as a foundational component for database-related workflows in Airflow, offering reusable SQL utilities that can be extended by specific database provider packages.
3

4
## Package Information
5

6
- **Package Name**: apache-airflow-providers-common-sql
7
- **Language**: Python
8
- **Installation**: `pip install apache-airflow-providers-common-sql`
9
- **Version**: 1.27.5
10

11
## Core Imports
12

13
```python
14
from airflow.providers.common.sql.hooks.sql import DbApiHook
15
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
16
from airflow.providers.common.sql.sensors.sql import SqlSensor
17
from airflow.providers.common.sql.triggers.sql import SQLExecuteQueryTrigger
18
```
19

20
## Basic Usage
21

22
```python
23
from airflow import DAG
24
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
25
from datetime import datetime
26

27
# Create a simple SQL execution task
28
with DAG(
29
    'sql_example',
30
    start_date=datetime(2023, 1, 1),
31
    schedule_interval=None,
32
) as dag:
33
    
34
    sql_task = SQLExecuteQueryOperator(
35
        task_id='run_sql_query',
36
        conn_id='my_database_conn',
37
        sql='SELECT COUNT(*) FROM users WHERE active = true;',
38
        autocommit=True
39
    )
40
```
41

42
## Architecture
43

44
The Common SQL Provider follows Airflow's standard provider architecture:
45

46
- **Hooks**: Handle database connections and low-level operations (DbApiHook as base class)
47
- **Operators**: Execute SQL tasks within DAGs (query execution, data validation, transfers)
48
- **Sensors**: Monitor database states and conditions (SqlSensor for periodic checks)
49
- **Triggers**: Enable asynchronous database operations (SQLExecuteQueryTrigger)
50
- **Dialects**: Provide database-specific SQL formatting and operations
51

52
This design enables database-agnostic workflows while supporting specialized database providers that extend these base components.
53

54
## Capabilities
55

56
### Database Hooks
57

58
Core hook functionality for establishing database connections, executing queries, and managing database operations. The DbApiHook serves as the foundation for all database interactions.
59

60
```python { .api }
61
class DbApiHook:
62
    def get_conn(self): ...
63
    def get_df(self, sql, parameters=None, **kwargs): ...
64
    def get_records(self, sql, parameters=None): ...
65
    def run(self, sql, autocommit=False, parameters=None, handler=None): ...
66
    def insert_rows(self, table, rows, target_fields=None, commit_every=1000): ...
67
    def bulk_dump(self, table, tmp_file): ...
68
    def bulk_load(self, table, tmp_file): ...
69
    def test_connection(self): ...
70
```
71

72
[Database Hooks](./hooks.md)
73

74
### SQL Operators
75

76
Task operators for executing SQL queries, performing data validation, and transferring data between databases. Includes specialized operators for data quality checks and conditional workflows.
77

78
```python { .api }
79
class SQLExecuteQueryOperator:
80
    def __init__(self, sql, conn_id, autocommit=False, parameters=None, **kwargs): ...
81

82
class SQLCheckOperator:
83
    def __init__(self, sql, conn_id, **kwargs): ...
84

85
class GenericTransfer:
86
    def __init__(self, sql, destination_table, source_conn_id, destination_conn_id, **kwargs): ...
87
```
88

89
[SQL Operators](./operators.md)
90

91
### SQL Sensors
92

93
Monitoring sensors that periodically check database conditions and states. Enables data-driven workflow orchestration based on SQL query results.
94

95
```python { .api }
96
class SqlSensor:
97
    def __init__(self, conn_id, sql, parameters=None, success=None, failure=None, **kwargs): ...
98
```
99

100
[SQL Sensors](./sensors.md)
101

102
### SQL Triggers
103

104
Asynchronous triggers for executing SQL operations without blocking the Airflow scheduler. Enables efficient handling of long-running database operations.
105

106
```python { .api }
107
class SQLExecuteQueryTrigger:
108
    def __init__(self, sql, conn_id, hook_params=None, **kwargs): ...
109
    def serialize(self): ...
110
    def get_hook(self): ...
111
```
112

113
[SQL Triggers](./triggers.md)
114

115
### SQL Dialects
116

117
Database-specific SQL formatting and operations. Provides abstraction layer for handling differences between SQL databases including query formatting and data type handling.
118

119
```python { .api }
120
class Dialect:
121
    def escape_word(self, word): ...
122
    def generate_insert_sql(self, table, values, target_fields, replace=False): ...
123
    def get_column_names(self, table): ...
124
    def get_primary_keys(self, table): ...
125
```
126

127
[SQL Dialects](./dialects.md)
128

129
## Types
130

131
```python { .api }
132
from typing import Any, Dict, List, Optional, Union, Callable, Protocol, Mapping, Iterable, Sequence
133
from operator import itemgetter
134

135
# Connection protocol for database connectors
136
class ConnectorProtocol(Protocol):
137
    def connect(self, host: str, port: int, username: str, schema: str) -> Any: ...
138

139
# Common type aliases
140
SQL = Union[str, List[str]]
141
Parameters = Optional[Union[Mapping[str, Any], Iterable]]
142
Handler = Optional[Callable[[Any], Any]]
143
ResultProcessor = Callable[[Any], Any]
144
SuccessCriteria = Optional[Callable[[Any], bool]]
145
FailureCriteria = Optional[Callable[[Any], bool]]
146
Selector = Callable[[tuple], Any]
147

148
# Version compatibility flags
149
AIRFLOW_V_3_0_PLUS: bool
150
AIRFLOW_V_3_1_PLUS: bool
151

152
# SQL placeholders
153
SQL_PLACEHOLDERS: frozenset[str]  # {"%s", "?"}
154
```