0
# Apache Airflow SQLite Provider
1

2
A provider package that integrates SQLite databases with Apache Airflow for workflow orchestration and data pipeline operations. This package extends Airflow's SQL capabilities with SQLite-specific connection handling, enabling seamless integration of SQLite databases into data workflows.
3

4
## Package Information
5

6
- **Package Name**: apache-airflow-providers-sqlite
7
- **Package Type**: pypi
8
- **Language**: Python
9
- **Installation**: `pip install apache-airflow-providers-sqlite`
10
- **Minimum Requirements**: Python >=3.10, Apache Airflow >=2.10.0, apache-airflow-providers-common-sql >=1.26.0
11

12
## Core Imports
13

14
```python
15
# Main hook import
16
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
17

18
# For type hints and advanced usage
19
import sqlite3
20
from airflow.models import Connection
21
from sqlalchemy.engine import Engine, Inspector
22
from sqlalchemy.engine.url import URL
23
```
24

25
Additional imports for DataFrame operations:
26

27
```python
28
# For pandas DataFrames (optional dependency)
29
from pandas import DataFrame as PandasDataFrame
30

31
# For polars DataFrames (optional dependency) 
32
from polars import DataFrame as PolarsDataFrame
33
```
34

35
## Basic Usage
36

37
```python
38
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
39

40
# Initialize hook with connection ID
41
hook = SqliteHook(sqlite_conn_id='sqlite_default')
42

43
# Execute a query and get all results
44
results = hook.get_records("SELECT * FROM users WHERE active = ?", parameters=[True])
45

46
# Execute a query and get first result only  
47
first_result = hook.get_first("SELECT COUNT(*) FROM users")
48

49
# Run SQL commands (INSERT, UPDATE, DELETE)
50
hook.run("INSERT INTO users (name, email) VALUES (?, ?)", parameters=["John Doe", "john@example.com"])
51

52
# Get results as pandas DataFrame (requires pandas)
53
df = hook.get_df("SELECT * FROM users", df_type="pandas")
54

55
# Get results as polars DataFrame (requires polars)
56
df = hook.get_df("SELECT * FROM users", df_type="polars")
57

58
# Bulk insert multiple rows
59
rows = [("Alice", "alice@example.com"), ("Bob", "bob@example.com")]
60
hook.insert_rows(table="users", rows=rows, target_fields=["name", "email"])
61

62
# Test connection
63
status, message = hook.test_connection()
64
if status:
65
    print("Connection successful")
66
else:
67
    print(f"Connection failed: {message}")
68
```
69

70
## Connection Configuration
71

72
SQLite connections support various URI formats:
73

74
```python
75
# File-based database (relative path)
76
sqlite:///path/to/database.db
77

78
# File-based database (absolute path)
79
sqlite:////absolute/path/to/database.db
80

81
# In-memory database
82
sqlite:///:memory:
83

84
# With query parameters
85
sqlite:///path/to/db.sqlite?mode=ro
86
sqlite:///path/to/db.sqlite?mode=rw
87
sqlite:///path/to/db.sqlite?cache=shared
88
```
89

90
## Capabilities
91

92
### Hook Class Definition
93

94
The SqliteHook class provides SQLite database integration.
95

96
```python { .api }
97
class SqliteHook(DbApiHook):
98
    """
99
    Interact with SQLite databases.
100
    
101
    Class Attributes:
102
        conn_name_attr: str = "sqlite_conn_id"
103
        default_conn_name: str = "sqlite_default"
104
        conn_type: str = "sqlite" 
105
        hook_name: str = "Sqlite"
106
    """
107
    
108
    def __init__(self, *args, schema: str | None = None, log_sql: bool = True, **kwargs):
109
        """
110
        Initialize SQLite hook.
111
        
112
        Args:
113
            *args: If single positional arg provided, used as connection ID
114
            schema (str, optional): Database schema (typically not used with SQLite)
115
            log_sql (bool): Whether to log SQL statements (default: True)
116
            **kwargs: Additional keyword arguments, including connection ID via conn_name_attr
117
        """
118
```
119

120
### Connection Management
121

122
Establish and manage SQLite database connections with proper URI handling.
123

124
```python { .api }
125
def get_conn(self) -> sqlite3.dbapi2.Connection:
126
    """
127
    Return SQLite connection object with proper URI conversion.
128
    
129
    Converts SQLAlchemy URI format to sqlite3-compatible file URI format.
130
    Handles file paths, in-memory databases, and query parameters.
131
    
132
    Returns:
133
        sqlite3.dbapi2.Connection: SQLite database connection
134
    """
135

136
def get_uri(self) -> str:
137
    """
138
    Override DbApiHook get_uri method for SQLAlchemy engine compatibility.
139
    
140
    Transforms Airflow connection URI to SQLAlchemy-compatible format,
141
    handling SQLite-specific URI requirements.
142
    
143
    Returns:
144
        str: SQLAlchemy-compatible URI string  
145
    """
146

147
def get_conn_id(self) -> str:
148
    """
149
    Get the connection ID used by this hook.
150
    
151
    Returns:
152
        str: Connection ID
153
    """
154

155
def get_cursor(self):
156
    """
157
    Get database cursor for executing SQL statements.
158
    
159
    Returns:
160
        sqlite3.Cursor: Database cursor object
161
    """
162

163
def test_connection(self):
164
    """
165
    Test the SQLite database connection.
166
    
167
    Returns:
168
        tuple[bool, str]: (connection_success, status_message)
169
    """
170
```
171

172
### SQL Execution
173

174
Execute SQL statements with parameter binding and transaction control.
175

176
```python { .api }
177
def run(self, sql, autocommit: bool = False, parameters=None, handler=None, 
178
        split_statements: bool = False, return_last: bool = True):
179
    """
180
    Execute SQL statement(s) with optional parameter binding.
181
    
182
    Args:
183
        sql (str | list[str]): SQL statement(s) to execute
184
        autocommit (bool): Enable autocommit mode (default: False)
185
        parameters (list | dict, optional): Query parameters for binding
186
        handler (callable, optional): Result handler function
187
        split_statements (bool): Split multiple statements (default: False)
188
        return_last (bool): Return result from last statement only (default: True)
189
        
190
    Returns:
191
        any: Query results based on handler, or None for non-SELECT statements
192
    """
193

194
def get_records(self, sql: str, parameters=None) -> list[tuple]:
195
    """
196
    Execute SQL query and return all records.
197
    
198
    Args:
199
        sql (str): SQL query to execute
200
        parameters (list | dict, optional): Query parameters for binding
201
        
202
    Returns:
203
        list[tuple]: List of result tuples
204
    """
205

206
def get_first(self, sql: str, parameters=None):
207
    """
208
    Execute SQL query and return first record.
209
    
210
    Args:
211
        sql (str): SQL query to execute  
212
        parameters (list | dict, optional): Query parameters for binding
213
        
214
    Returns:
215
        tuple | None: First result tuple or None if no results
216
    """
217
```
218

219
### DataFrame Operations
220

221
Convert query results to pandas or polars DataFrames for data analysis.
222

223
```python { .api }
224
def get_df(self, sql: str, parameters=None, *, df_type: str = "pandas", **kwargs):
225
    """
226
    Execute SQL query and return results as DataFrame.
227
    
228
    Args:
229
        sql (str): SQL query to execute
230
        parameters (list | dict, optional): Query parameters for binding
231
        df_type (str): DataFrame type - "pandas" or "polars" (default: "pandas")
232
        **kwargs: Additional arguments passed to DataFrame constructor
233
        
234
    Returns:
235
        PandasDataFrame | PolarsDataFrame: DataFrame with query results
236
    """
237

238
def get_df_by_chunks(self, sql: str, parameters=None, *, chunksize: int, 
239
                     df_type: str = "pandas", **kwargs):
240
    """
241
    Execute SQL query and return results as DataFrame chunks.
242
    
243
    Args:
244
        sql (str): SQL query to execute
245
        parameters (list | dict, optional): Query parameters for binding
246
        chunksize (int): Number of rows per chunk (required)
247
        df_type (str): DataFrame type - "pandas" or "polars" (default: "pandas")
248
        **kwargs: Additional arguments passed to DataFrame constructor
249
        
250
    Yields:
251
        PandasDataFrame | PolarsDataFrame: Iterator of DataFrame chunks
252
    """
253
```
254

255
### Bulk Operations
256

257
Efficiently insert multiple rows with batching and transaction control.
258

259
```python { .api }
260
def insert_rows(self, table: str, rows, target_fields=None, commit_every: int = 1000, 
261
                replace: bool = False, *, executemany: bool = False, 
262
                fast_executemany: bool = False, autocommit: bool = False, **kwargs) -> None:
263
    """
264
    Insert multiple rows into table with batching and optional replacement.
265
    
266
    Args:
267
        table (str): Target table name
268
        rows (Iterable): Collection of row tuples to insert
269
        target_fields (list[str], optional): Column names for insertion
270
        commit_every (int): Commit transaction every N rows (default: 1000)
271
        replace (bool): Use REPLACE INTO instead of INSERT INTO (default: False)
272
        executemany (bool): Use cursor.executemany() for batch insertion (default: False)
273
        fast_executemany (bool): Use fast executemany if supported (default: False)
274
        autocommit (bool): Enable autocommit mode (default: False) 
275
        **kwargs: Additional arguments for customization
276
    """
277
```
278

279
### SQLAlchemy Integration
280

281
Access SQLAlchemy engines and metadata for advanced database operations.
282

283
```python { .api }
284
def get_sqlalchemy_engine(self, engine_kwargs=None) -> Engine:
285
    """
286
    Get SQLAlchemy engine for advanced database operations.
287
    
288
    Args:
289
        engine_kwargs (dict, optional): Additional engine configuration parameters
290
        
291
    Returns:
292
        Engine: SQLAlchemy engine instance
293
    """
294

295
@property
296
def sqlalchemy_url(self) -> URL:
297
    """
298
    SQLAlchemy URL object for this connection.
299
    
300
    Returns:
301
        URL: SQLAlchemy URL object
302
    """
303

304
@property  
305
def inspector(self) -> Inspector:
306
    """
307
    SQLAlchemy Inspector for database metadata.
308
    
309
    Returns:
310
        Inspector: Database inspector instance
311
    """
312
```
313

314
### Transaction Control
315

316
Manage database transactions and autocommit behavior.
317

318
```python { .api }
319
def get_autocommit(self, conn) -> bool:
320
    """
321
    Get autocommit setting for connection.
322
    
323
    Args:
324
        conn: Database connection object
325
        
326
    Returns:
327
        bool: Current autocommit status
328
    """
329

330
def set_autocommit(self, conn, autocommit: bool) -> None:
331
    """
332
    Set autocommit flag on connection.
333
    
334
    Args:
335
        conn: Database connection object
336
        autocommit (bool): Autocommit setting to apply
337
    """
338
```
339

340
### Properties and Utilities
341

342
Helper methods and properties for SQL operations and metadata access.
343

344
```python { .api }
345
@property
346
def placeholder(self) -> str:
347
    """
348
    SQL parameter placeholder character for SQLite.
349
    
350
    Returns:
351
        str: "?" (question mark placeholder)
352
    """
353

354
@property
355
def connection(self) -> Connection:
356
    """
357
    Airflow connection object for this hook.
358
    
359
    Returns:
360
        Connection: Connection object instance
361
    """
362

363
@property
364
def connection_extra(self) -> dict:
365
    """
366
    Connection extra parameters as dictionary.
367
    
368
    Returns:
369
        dict: Extra connection parameters from connection configuration
370
    """
371

372
@property
373
def last_description(self) -> list:
374
    """
375
    Description from last executed cursor.
376
    
377
    Returns:
378
        list: Cursor description with column metadata
379
    """
380

381
@staticmethod
382
def split_sql_string(sql: str, strip_semicolon: bool = False) -> list[str]:
383
    """
384
    Split SQL string into individual statements.
385
    
386
    Args:
387
        sql (str): SQL string with multiple statements
388
        strip_semicolon (bool): Remove trailing semicolons (default: False)
389
        
390
    Returns:
391
        list[str]: List of individual SQL statements
392
    """
393

394
@staticmethod
395
def strip_sql_string(sql: str) -> str:
396
    """
397
    Strip whitespace and comments from SQL string.
398
    
399
    Args:
400
        sql (str): SQL string to clean
401
        
402
    Returns:
403
        str: Cleaned SQL string
404
    """
405
```
406

407
### Provider Metadata
408

409
Access provider configuration and metadata.
410

411
```python { .api }
412
# From airflow.providers.sqlite.get_provider_info
413
def get_provider_info() -> dict:
414
    """
415
    Get provider metadata including integrations and connection types.
416
    
417
    Returns:
418
        dict: Provider metadata containing:
419
            - package-name: "apache-airflow-providers-sqlite"
420
            - name: "SQLite" 
421
            - description: SQLite provider description
422
            - integrations: List of SQLite integration info
423
            - hooks: List of available hook modules
424
            - connection-types: List of supported connection types
425
    """
426
```
427

428
## Types
429

430
```python { .api }
431
# Type aliases for clarity
432
PandasDataFrame = "pandas.DataFrame"
433
PolarsDataFrame = "polars.DataFrame" 
434
Connection = "airflow.models.Connection"
435
Engine = "sqlalchemy.engine.Engine"
436
Inspector = "sqlalchemy.engine.Inspector"
437
URL = "sqlalchemy.engine.URL"
438
```
439

440
## Error Handling
441

442
The SQLite hook handles common database errors and connection issues:
443

444
- **Connection errors**: Invalid file paths, permission issues, database corruption
445
- **SQL errors**: Syntax errors, constraint violations, table/column not found  
446
- **Transaction errors**: Deadlocks, lock timeouts, rollback scenarios
447
- **URI format errors**: Invalid connection string formats, parameter parsing
448

449
Common error patterns:
450

451
```python
452
import sqlite3
453
from airflow.exceptions import AirflowException
454

455
try:
456
    hook = SqliteHook(sqlite_conn_id='my_sqlite_conn')
457
    results = hook.get_records("SELECT * FROM users")
458
except sqlite3.Error as e:
459
    # Handle SQLite-specific errors
460
    print(f"Database error: {e}")
461
except AirflowException as e:
462
    # Handle Airflow-specific errors (connection not found, etc.)
463
    print(f"Airflow error: {e}")
464
except Exception as e:
465
    # Handle other errors
466
    print(f"General error: {e}")
467
```
468

469
## Usage Examples
470

471
### Working with In-Memory Databases
472

473
```python
474
# Connection URI: sqlite:///:memory:
475
hook = SqliteHook(sqlite_conn_id='sqlite_memory')
476
hook.run("CREATE TABLE temp_data (id INTEGER, value TEXT)")
477
hook.insert_rows("temp_data", [(1, "test"), (2, "data")])
478
results = hook.get_records("SELECT * FROM temp_data")
479
```
480

481
### File Database with Custom Parameters
482

483
```python
484
# Connection URI: sqlite:///path/to/db.sqlite?mode=rw&cache=shared
485
hook = SqliteHook(sqlite_conn_id='sqlite_file')
486
results = hook.get_df("SELECT * FROM large_table", df_type="pandas")
487
```
488

489
### Batch Processing with Chunked DataFrames
490

491
```python
492
# Process large results in chunks to manage memory
493
for chunk_df in hook.get_df_by_chunks("SELECT * FROM big_table", chunksize=1000):
494
    # Process each chunk
495
    processed_chunk = chunk_df.groupby('category').sum()
496
    # Save or further process results
497
    print(f"Processed chunk with {len(chunk_df)} rows")
498
```
499

500
### Transaction Management
501

502
```python
503
# Manual transaction control
504
with hook._create_autocommit_connection(autocommit=False) as conn:
505
    cursor = conn.cursor()
506
    try:
507
        cursor.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
508
        cursor.execute("UPDATE users SET active = 1 WHERE name = ?", ("Alice",))
509
        conn.commit()
510
    except Exception:
511
        conn.rollback()
512
        raise
513
```
514

515
### Using SQLAlchemy Engine
516

517
```python
518
# Get SQLAlchemy engine for advanced operations
519
engine = hook.get_sqlalchemy_engine()
520
with engine.connect() as conn:
521
    result = conn.execute("SELECT * FROM users")
522
    for row in result:
523
        print(row)
524
```
525

526
## Connection Configuration
527

528
- **Connection Type**: `sqlite`
529
- **Hook Class**: `airflow.providers.sqlite.hooks.sqlite.SqliteHook`
530
- **Supported URI Schemes**: `sqlite://`
531
- **Default Connection ID**: `sqlite_default`
532

533
## Dependencies
534

535
- **Required**: `apache-airflow>=2.10.0`, `apache-airflow-providers-common-sql>=1.26.0`
536
- **Python**: >=3.10
537
- **Optional**: `pandas` (for pandas DataFrame support), `polars` (for polars DataFrame support)
538

539
## Provider Information
540

541
- **Package Name**: apache-airflow-providers-sqlite
542
- **Provider Name**: SQLite
543
- **Integration**: SQLite database integration
544
- **External Documentation**: https://www.sqlite.org/index.html