feat: implement schemas reflection method (#27)

Author: Ilya Gurov

google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py

@@ -310,7 +310,9 @@ def get_indexes(self, connection, table_name, schema=None, **kw):
 FROM information_schema.indexes as i
 JOIN information_schema.index_columns AS ic
     ON ic.index_name = i.index_name AND ic.table_name = i.table_name
-WHERE i.table_name="{table_name}"
+WHERE
+    i.table_name="{table_name}"
+    AND i.index_type != 'PRIMARY_KEY'
 GROUP BY i.index_name, i.is_unique
 """.format(
             table_name=table_name
@@ -372,6 +374,33 @@ def get_pk_constraint(self, connection, table_name, schema=None, **kw):
 
         return {"constrained_columns": cols}
 
+    def get_schema_names(self, connection, **kw):
+        """Get all the schemas in the database.
+
+        Args:
+            connection (Union[
+                sqlalchemy.engine.base.Connection,
+                sqlalchemy.engine.Engine
+            ]):
+                SQLAlchemy connection or engine object.
+
+        Returns:
+            list: Schema names.
+        """
+        if isinstance(connection, Engine):
+            connection = connection.connect()
+
+        schemas = []
+        with connection.connection.database.snapshot() as snap:
+            rows = snap.execute_sql(
+                "SELECT schema_name FROM information_schema.schemata"
+            )
+
+            for row in rows:
+                schemas.append(row[0])
+
+        return schemas
+
     def get_foreign_keys(self, connection, table_name, schema=None, **kw):
         """Get the table foreign key constraints.
 

test/test_suite.py

@@ -45,6 +45,8 @@
 from sqlalchemy.testing.suite.test_select import ExistsTest as _ExistsTest
 from sqlalchemy.testing.suite.test_types import BooleanTest as _BooleanTest
 
+config.test_schema = ""
+
 
 from sqlalchemy.testing.suite.test_types import (  # noqa: F401, F403
     DateTest as _DateTest,