Repository: apache/airflow
Pull request: Fix sqlite hook - insert and replace functions
closes: #17632 

override _generate_insert_sql()  in sqlite hook to use '?' instead of the standard '%' for replacing.
Added unit tests and example of insert/replace in example dag.

Problem statement:
SqliteHook invalid insert syntax
<!--
Welcome to Apache Airflow!

Please complete the next sections or the issue will be closed.
-->

**Apache Airflow version**: 2.1.0

<!-- AIRFLOW VERSION IS MANDATORY -->

**OS**: Raspbian GNU/Linux 10 (buster)

<!-- MANDATORY! You can get it via `cat /etc/oss-release` for example -->

**Apache Airflow Provider versions**: apache-airflow-providers-sqlite==1.0.2

<!-- You can use `pip freeze | grep apache-airflow-providers` (you can leave only relevant ones)-->

**Deployment**:  Virtualenv

<!-- e.g. Virtualenv / VM / Docker-compose / K8S / Helm Chart / Managed Airflow Service -->

<!-- Please include your deployment tools and versions: docker-compose, k8s, helm, etc -->

**What happened**:

When using the insert_rows function of the sqlite hook the generated parametrized sql query has an invalid syntax. It will generate queries with the %s placeholder which result in the following error:

*Query*: INSERT INTO example_table (col1, col2) VALUES (%s,%s)

*Error*:
sqlite3.OperationalError: near "%": syntax error

*Stacktrace*:
  File "/home/airflow/.pyenv/versions/3.8.10/envs/airflow_3.8.10/lib/python3.8/site-packages/airflow/hooks/dbapi.py", line 307, in insert_rows
    cur.execute(sql, values)

I replaced the placeholder with "?" in the _generate_insert_sql function DbApiHook as a test. This solved the issue.

<!-- Please include exact error messages if you can -->

**What you expected to happen**:

Every inherited method of the DbApiHook should work for the SqliteHook. Using the insert_rows method should generate the correct parametrized query and insert rows as expected.

<!-- What do you think went wrong? -->

**How to reproduce it**:

1. Create an instance of the SqliteHook
2. Use the insert_rows method of the SqliteHook

<!--
As minimally and precisely as possible. Keep in mind we do not have access to your cluster or dags.
If this is a UI bug, please provide a screenshot of the bug or a link to a youtube video of the bug in action
You can include images/screen-casts etc. by drag-dropping the image here.
-->

Hints from issue discussion:
Thanks for opening your first issue here! Be sure to follow the issue template!

Would you be willing to provide a fix for that @bart-art ? That sounds like rather easy fix - good for first-time contribution. Happy to help with it if needed.
@Bart-art if you are not working on it, please let me know I can also take it on.


Assigned you @subkanthi -> @Bart-art if you want you can join as well (at least as reviewer :)

Commit message: Merge branch 'master' of github.com:subkanthi/airflow into fix_sqlite_doc_typo

Patch to review:
diff --git a/airflow/hooks/dbapi.py b/airflow/hooks/dbapi.py
index 4156500085ada..f0bd85fdabfef 100644
--- a/airflow/hooks/dbapi.py
+++ b/airflow/hooks/dbapi.py
@@ -324,6 +324,7 @@ def insert_rows(self, table, rows, target_fields=None, commit_every=1000, replac
                         lst.append(self._serialize_cell(cell, conn))
                     values = tuple(lst)
                     sql = self._generate_insert_sql(table, values, target_fields, replace, **kwargs)
+                    self.log.info(f"Generated sql: {sql}")
                     cur.execute(sql, values)
                     if commit_every and i % commit_every == 0:
                         conn.commit()
diff --git a/airflow/providers/sqlite/example_dags/example_sqlite.py b/airflow/providers/sqlite/example_dags/example_sqlite.py
index 88b4ba2293469..2ff5114b80c19 100644
--- a/airflow/providers/sqlite/example_dags/example_sqlite.py
+++ b/airflow/providers/sqlite/example_dags/example_sqlite.py
@@ -24,6 +24,8 @@
 """
 
 from airflow import DAG
+from airflow.operators.python import PythonOperator
+from airflow.providers.sqlite.hooks.sqlite import SqliteHook
 from airflow.providers.sqlite.operators.sqlite import SqliteOperator
 from airflow.utils.dates import days_ago
 
@@ -52,6 +54,28 @@
 
 # [END howto_operator_sqlite]
 
+
+def insert_sqlite_hook():
+    sqlite_hook = SqliteHook("sqlite_default")
+    sqlite_hook.get_conn()
+
+    rows = [('James', '11'), ('James', '22'), ('James', '33')]
+    target_fields = ['first_name', 'last_name']
+    sqlite_hook.insert_rows(table='Customer', rows=rows, target_fields=target_fields)
+
+
+def replace_sqlite_hook():
+    sqlite_hook = SqliteHook("sqlite_default")
+    sqlite_hook.get_conn()
+
+    rows = [('James', '11'), ('James', '22'), ('James', '33')]
+    target_fields = ['first_name', 'last_name']
+    sqlite_hook.insert_rows(table='Customer', rows=rows, target_fields=target_fields, replace=True)
+
+
+insert_sqlite_task = PythonOperator(task_id="insert_sqlite_task", python_callable=insert_sqlite_hook)
+replace_sqlite_task = PythonOperator(task_id="replace_sqlite_task", python_callable=replace_sqlite_hook)
+
 # [START howto_operator_sqlite_external_file]
 
 # Example of creating a task that calls an sql command from an external file.
@@ -64,4 +88,4 @@
 
 # [END howto_operator_sqlite_external_file]
 
-create_table_sqlite_task >> external_create_table_sqlite_task
+create_table_sqlite_task >> external_create_table_sqlite_task >> insert_sqlite_task >> replace_sqlite_task
diff --git a/airflow/providers/sqlite/hooks/sqlite.py b/airflow/providers/sqlite/hooks/sqlite.py
index 82c9dd0468edc..e4a43317859b5 100644
--- a/airflow/providers/sqlite/hooks/sqlite.py
+++ b/airflow/providers/sqlite/hooks/sqlite.py
@@ -35,3 +35,37 @@ def get_conn(self) -> sqlite3.dbapi2.Connection:
         airflow_conn = self.get_connection(conn_id)
         conn = sqlite3.connect(airflow_conn.host)
         return conn
+
+    @staticmethod
+    def _generate_insert_sql(table, values, target_fields, replace, **kwargs):
+        """
+        Static helper method that generate the INSERT SQL statement.
+        The REPLACE variant is specific to MySQL syntax.
+
+        :param table: Name of the target table
+        :type table: str
+        :param values: The row to insert into the table
+        :type values: tuple of cell values
+        :param target_fields: The names of the columns to fill in the table
+        :type target_fields: iterable of strings
+        :param replace: Whether to replace instead of insert
+        :type replace: bool
+        :return: The generated INSERT or REPLACE SQL statement
+        :rtype: str
+        """
+        placeholders = [
+            "?",
+        ] * len(values)
+
+        if target_fields:
+            target_fields = ", ".join(target_fields)
+            target_fields = f"({target_fields})"
+        else:
+            target_fields = ''
+
+        if not replace:
+            sql = "INSERT INTO "
+        else:
+            sql = "REPLACE INTO "
+        sql += f"{table} {target_fields} VALUES ({','.join(placeholders)})"
+        return sql
diff --git a/docs/apache-airflow-providers-sqlite/operators.rst b/docs/apache-airflow-providers-sqlite/operators.rst
index 100156e9fcf29..6cbe5f6d3062e 100644
--- a/docs/apache-airflow-providers-sqlite/operators.rst
+++ b/docs/apache-airflow-providers-sqlite/operators.rst
@@ -39,7 +39,7 @@ the connection metadata is structured as follows:
    * - Parameter
      - Input
    * - Host: string
-     - MySql hostname
+     - Sqlite database file
    * - Schema: string
      - Set schema to execute Sql operations on by default
    * - Login: string
diff --git a/tests/providers/sqlite/hooks/test_sqlite.py b/tests/providers/sqlite/hooks/test_sqlite.py
index d618fed97af74..3c5d977037499 100644
--- a/tests/providers/sqlite/hooks/test_sqlite.py
+++ b/tests/providers/sqlite/hooks/test_sqlite.py
@@ -106,3 +106,23 @@ def test_run_log(self):
         statement = 'SQL'
         self.db_hook.run(statement)
         assert self.db_hook.log.info.call_count == 2
+
+    def test_generate_insert_sql_replace_false(self):
+        expected_sql = "INSERT INTO Customer (first_name, last_name) VALUES (?,?)"
+        rows = ('James', '1')
+        target_fields = ['first_name', 'last_name']
+        sql = self.db_hook._generate_insert_sql(
+            table='Customer', values=rows, target_fields=target_fields, replace=False
+        )
+
+        assert sql == expected_sql
+
+    def test_generate_insert_sql_replace_true(self):
+        expected_sql = "REPLACE INTO Customer (first_name, last_name) VALUES (?,?)"
+        rows = ('James', '1')
+        target_fields = ['first_name', 'last_name']
+        sql = self.db_hook._generate_insert_sql(
+            table='Customer', values=rows, target_fields=target_fields, replace=True
+        )
+
+        assert sql == expected_sql


Code review comments:
diff_hunk: @@ -324,6 +324,7 @@ def insert_rows(self, table, rows, target_fields=None, commit_every=1000, replac
                         lst.append(self._serialize_cell(cell, conn))
                     values = tuple(lst)
                     sql = self._generate_insert_sql(table, values, target_fields, replace, **kwargs)
+                    self.log.info(f"Generated sql: {sql}")
path: airflow/hooks/dbapi.py
text: @user1:
This might be a bit too agressive. Please change it to debug, also use "%" format of log.* rather than f-string for performance. 

@author:
Fixed. Thanks, @user1