added create_index support to sqlite

Author: dantownsend

piccolo/query/methods/create_index.py

@@ -35,7 +35,7 @@ def column_names(self) -> t.List[str]:
         ]
 
     @property
-    def querystrings(self) -> t.Sequence[QueryString]:
+    def postgres_querystrings(self) -> t.Sequence[QueryString]:
         column_names = self.column_names
         index_name = self.table._get_index_name(column_names)
         tablename = self.table._meta.tablename
@@ -47,3 +47,21 @@ def querystrings(self) -> t.Sequence[QueryString]:
                 f" ({column_names_str})"
             )
         ]
+
+    @property
+    def sqlite_querystrings(self) -> t.Sequence[QueryString]:
+        column_names = self.column_names
+        index_name = self.table._get_index_name(column_names)
+        tablename = self.table._meta.tablename
+
+        method_name = self.method.value
+        if method_name != "btree":
+            raise ValueError("SQLite only support btree indexes.")
+
+        column_names_str = ", ".join(column_names)
+        return [
+            QueryString(
+                f"CREATE INDEX {index_name} ON {tablename} "
+                f"({column_names_str})"
+            )
+        ]

tests/table/test_create_index.py

@@ -1,19 +1,27 @@
-from ..base import DBTestCase, postgres_only
+from unittest import TestCase
+
+from ..base import DBTestCase
 from ..example_project.tables import Manager
 
 
-@postgres_only
 class TestCreateIndex(DBTestCase):
     def test_create_index(self):
         """
         Test single column and multi column indexes
         """
         for columns in [[Manager.name], [Manager.id, Manager.name]]:
-            query = Manager.raw(
-                "SELECT * FROM pg_indexes WHERE indexname = {}",
-                Manager._get_index_name([i._meta.name for i in columns]),
+
+            index_name = Manager._get_index_name(
+                [i._meta.name for i in columns]
             )
 
+            if Manager._meta.db.engine_type == "postgres":
+                query = Manager.raw(
+                    "SELECT * FROM pg_indexes WHERE indexname = {}", index_name
+                )
+            else:
+                query = Manager.raw(f"PRAGMA index_list(manager)")
+
             Manager.create_index(columns).run_sync()
             response = query.run_sync()
             self.assertEqual(len(response), 1)
@@ -22,6 +30,8 @@ def test_create_index(self):
             response = query.run_sync()
             self.assertEqual(len(response), 0)
 
+
+class TestIndexName(TestCase):
     def test_index_name(self):
         self.assertEqual(
             Manager._get_index_name(["name", "id"]), "manager_name_id"