Repository: kayak/pypika
Pull request #600 (2021-05-22 16:27:04)

sqlite: insert or replace
#442


Patch:
diff --git a/pypika/dialects.py b/pypika/dialects.py
index ec5a4e4..3f35b0f 100644
--- a/pypika/dialects.py
+++ b/pypika/dialects.py
@@ -705,10 +705,10 @@ class ClickHouseQueryBuilder(QueryBuilder):
 
 
 class SQLLiteValueWrapper(ValueWrapper):
-    def get_value_sql(self, *args: Any, **kwargs: Any) -> str:
+    def get_value_sql(self, **kwargs: Any) -> str:
         if isinstance(self.value, bool):
             return "1" if self.value else "0"
-        return super().get_value_sql(*args, **kwargs)
+        return super().get_value_sql(**kwargs)
 
 
 class SQLLiteQuery(Query):
@@ -718,8 +718,22 @@ class SQLLiteQuery(Query):
 
     @classmethod
     def _builder(cls, **kwargs: Any) -> "SQLLiteQueryBuilder":
-        return SQLLiteQueryBuilder(dialect=Dialects.SQLLITE, wrapper_cls=SQLLiteValueWrapper, **kwargs)
+        return SQLLiteQueryBuilder(**kwargs)
 
 
 class SQLLiteQueryBuilder(QueryBuilder):
     QUERY_CLS = SQLLiteQuery
+
+    def __init__(self, **kwargs: Any) -> None:
+        super().__init__(dialect=Dialects.SQLLITE, wrapper_cls=SQLLiteValueWrapper, **kwargs)
+        self._insert_or_replace = False
+
+    @builder
+    def insert_or_replace(self, *terms: Any) -> "SQLLiteQueryBuilder":
+        self._apply_terms(*terms)
+        self._replace = True
+        self._insert_or_replace = True
+
+    def _replace_sql(self, **kwargs: Any) -> str:
+        prefix = "INSERT OR " if self._insert_or_replace else ""
+        return prefix + super()._replace_sql(**kwargs)
diff --git a/pypika/queries.py b/pypika/queries.py
index 36397eb..3ea70ab 100644
--- a/pypika/queries.py
+++ b/pypika/queries.py
@@ -837,24 +837,12 @@ class QueryBuilder(Selectable, Term):
 
     @builder
     def insert(self, *terms: Any) -> "QueryBuilder":
-        if self._insert_table is None:
-            raise AttributeError("'Query' object has no attribute '%s'" % "insert")
-
-        if not terms:
-            return
-        else:
-            self._validate_terms_and_append(*terms)
+        self._apply_terms(*terms)
         self._replace = False
 
     @builder
     def replace(self, *terms: Any) -> "QueryBuilder":
-        if self._insert_table is None:
-            raise AttributeError("'Query' object has no attribute '%s'" % "insert")
-
-        if not terms:
-            return
-        else:
-            self._validate_terms_and_append(*terms)
+        self._apply_terms(*terms)
         self._replace = True
 
     @builder
@@ -1140,11 +1128,17 @@ class QueryBuilder(Selectable, Term):
         subquery.alias = "sq%d" % self._subquery_count
         self._subquery_count += 1
 
-    def _validate_terms_and_append(self, *terms: Any) -> None:
+    def _apply_terms(self, *terms: Any) -> None:
         """
         Handy function for INSERT and REPLACE statements in order to check if
         terms are introduced and how append them to `self._values`
         """
+        if self._insert_table is None:
+            raise AttributeError("'Query' object has no attribute '%s'" % "insert")
+
+        if not terms:
+            return
+
         if not isinstance(terms[0], (list, tuple, set)):
             terms = [terms]
 


Test patch:
diff --git a/pypika/tests/dialects/test_sqllite.py b/pypika/tests/dialects/test_sqllite.py
index b3b269a..104a723 100644
--- a/pypika/tests/dialects/test_sqllite.py
+++ b/pypika/tests/dialects/test_sqllite.py
@@ -16,3 +16,20 @@ class SelectTests(unittest.TestCase):
         q = SQLLiteQuery.from_("abc").select(False)
 
         self.assertEqual('SELECT 0 FROM "abc"', str(q))
+
+
+class ReplaceTests(unittest.TestCase):
+    def test_normal_replace(self):
+        query = SQLLiteQuery.into("abc").replace("v1", "v2", "v3")
+        expected_output = "REPLACE INTO \"abc\" VALUES ('v1','v2','v3')"
+        self.assertEqual(expected_output, str(query))
+
+    def test_replace_subquery(self):
+        query = SQLLiteQuery.into("abc").replace(SQLLiteQuery.from_("def").select("f1", "f2"))
+        expected_output = 'REPLACE INTO "abc" VALUES ((SELECT "f1","f2" FROM "def"))'
+        self.assertEqual(expected_output, str(query))
+
+    def test_insert_or_replace(self):
+        query = SQLLiteQuery.into("abc").insert_or_replace("v1", "v2", "v3")
+        expected_output = "INSERT OR REPLACE INTO \"abc\" VALUES ('v1','v2','v3')"
+        self.assertEqual(expected_output, str(query))