Separate View and Table classes

Both of these subclass a common Queryable class.

Also updated documentation to cover the new View class.

And added view.drop() method.

docs/python-api.rst

@@ -45,6 +45,8 @@ You can also access tables using the ``.table()`` method like so:
 
 Using this factory function allows you to set :ref:`python_api_table_configuration`.
 
+.. _python_api_tables:
+
 Listing tables
 ==============
 
@@ -60,6 +62,31 @@ You can also iterate through the table objects themselves using the ``.tables``
     >>> db.tables
     [<Table dogs>]
 
+.. _python_api_views:
+
+Listing views
+=============
+
+``.table_views()`` shows you a list of views in the database::
+
+    >>> db.table_views()
+    ['good_dogs']
+
+You can iterate through view objects using the ``.views`` property::
+
+    >>> db.views
+    [<View good_dogs>]
+
+View objects are similar to Table objects, except that any attempts to insert or update data will throw an error. The full list of methods and properties available on a view object is as follows:
+
+* ``columns``
+* ``columns_dict``
+* ``count``
+* ``schema``
+* ``rows``
+* ``rows_where(where, where_args)``
+* ``drop()``
+
 .. _python_api_rows:
 
 Listing rows
@@ -682,10 +709,10 @@ If you want to ensure that every foreign key column in your database has a corre
 
 .. _python_api_drop:
 
-Dropping a table
+Dropping a table or view
-================
+========================
 
-You can drop a table by using the ``.drop()`` method:
+You can drop a table or view using the ``.drop()`` method:
 
 .. code-block:: python
 
@@ -764,7 +791,7 @@ For example:
 Introspection
 =============
 
-If you have loaded an existing table, you can use introspection to find out more about it::
+If you have loaded an existing table or view, you can use introspection to find out more about it::
 
     >>> db["PlantType"]
     <Table PlantType (id, value)>
@@ -776,7 +803,7 @@ The ``.count`` property shows the current number of rows (``select count(*) from
     >>> db["Street_Tree_List"].count
     189144
 
-The ``.columns`` property shows the columns in the table::
+The ``.columns`` property shows the columns in the table or view::
 
     >>> db["PlantType"].columns
     [Column(cid=0, name='id', type='INTEGER', notnull=0, default_value=None, is_pk=1),
@@ -787,7 +814,9 @@ The ``.columns_dict`` property returns a dictionary version of this with just th
     >>> db["PlantType"].columns_dict
     {'id': <class 'int'>, 'value': <class 'str'>}
 
-The ``.foreign_keys`` property shows if the table has any foreign key relationships::
+The ``.foreign_keys`` property shows if the table has any foreign key relationships. It is not available on views.
+
+::
 
     >>> db["Street_Tree_List"].foreign_keys
     [ForeignKey(table='Street_Tree_List', column='qLegalStatus', other_table='qLegalStatus', other_column='id'),
@@ -827,7 +856,9 @@ The ``.schema`` property outputs the table's schema as a SQL string::
         FOREIGN KEY ("qCareAssistant") REFERENCES [qCareAssistant](id),
         FOREIGN KEY ("qLegalStatus") REFERENCES [qLegalStatus](id))
 
-The ``.indexes`` property shows you all indexes created for a table::
+The ``.indexes`` property shows you all indexes created for a table. It is not available on views.
+
+::
 
     >>> db["Street_Tree_List"].indexes
     [Index(seq=0, name='"Street_Tree_List_qLegalStatus"', unique=0, origin='c', partial=0, columns=['qLegalStatus']),

sqlite_utils/db.py

@@ -63,6 +63,7 @@ if np:
 
 
 REVERSE_COLUMN_TYPE_MAPPING = {
+    "": str,  # Columns in views sometimes have type = ''
     "TEXT": str,
     "BLOB": bytes,
     "INTEGER": int,
@@ -107,7 +108,8 @@ class Database:
         return "<Database {}>".format(self.conn)
 
     def table(self, table_name, **kwargs):
-        return Table(self, table_name, **kwargs)
+        klass = View if table_name in self.view_names() else Table
+        return klass(self, table_name, **kwargs)
 
     def escape(self, value):
         # Normally we would use .execute(sql, [params]) for escaping, but
@@ -128,10 +130,22 @@ class Database:
         sql = "select name from sqlite_master where {}".format(" AND ".join(where))
         return [r[0] for r in self.conn.execute(sql).fetchall()]
 
+    def view_names(self):
+        return [
+            r[0]
+            for r in self.conn.execute(
+                "select name from sqlite_master where type = 'view'"
+            ).fetchall()
+        ]
+
     @property
     def tables(self):
         return [self[name] for name in self.table_names()]
 
+    @property
+    def views(self):
+        return [self[name] for name in self.view_names()]
+
     def execute_returning_dicts(self, sql, params=None):
         cursor = self.conn.execute(sql, params or tuple())
         keys = [d[0] for d in cursor.description]
@@ -385,7 +399,59 @@ class Database:
         self.conn.execute("VACUUM;")
 
 
-class Table:
+class Queryable:
+    exists = False
+
+    def __init__(self, db, name):
+        self.db = db
+        self.name = name
+
+    @property
+    def count(self):
+        return self.db.conn.execute(
+            "select count(*) from [{}]".format(self.name)
+        ).fetchone()[0]
+
+    @property
+    def rows(self):
+        return self.rows_where()
+
+    def rows_where(self, where=None, where_args=None):
+        if not self.exists:
+            return []
+        sql = "select * from [{}]".format(self.name)
+        if where is not None:
+            sql += " where " + where
+        cursor = self.db.conn.execute(sql, where_args or [])
+        columns = [c[0] for c in cursor.description]
+        for row in cursor:
+            yield dict(zip(columns, row))
+
+    @property
+    def columns(self):
+        if not self.exists:
+            return []
+        rows = self.db.conn.execute(
+            "PRAGMA table_info([{}])".format(self.name)
+        ).fetchall()
+        return [Column(*row) for row in rows]
+
+    @property
+    def columns_dict(self):
+        "Returns {column: python-type} dictionary"
+        return {
+            column.name: REVERSE_COLUMN_TYPE_MAPPING[column.type]
+            for column in self.columns
+        }
+
+    @property
+    def schema(self):
+        return self.db.conn.execute(
+            "select sql from sqlite_master where name = ?", (self.name,)
+        ).fetchone()[0]
+
+
+class Table(Queryable):
     def __init__(
         self,
         db,
@@ -402,8 +468,7 @@ class Table:
         ignore=False,
         extracts=None,
     ):
-        self.db = db
+        super().__init__(db, name)
-        self.name = name
         self.exists = self.name in self.db.table_names()
         self._defaults = dict(
             pk=pk,
@@ -427,44 +492,6 @@ class Table:
             else " ({})".format(", ".join(c.name for c in self.columns)),
         )
 
-    @property
-    def count(self):
-        return self.db.conn.execute(
-            "select count(*) from [{}]".format(self.name)
-        ).fetchone()[0]
-
-    @property
-    def columns(self):
-        if not self.exists:
-            return []
-        rows = self.db.conn.execute(
-            "PRAGMA table_info([{}])".format(self.name)
-        ).fetchall()
-        return [Column(*row) for row in rows]
-
-    @property
-    def columns_dict(self):
-        "Returns {column: python-type} dictionary"
-        return {
-            column.name: REVERSE_COLUMN_TYPE_MAPPING[column.type]
-            for column in self.columns
-        }
-
-    @property
-    def rows(self):
-        return self.rows_where()
-
-    def rows_where(self, where=None, where_args=None):
-        if not self.exists:
-            return []
-        sql = "select * from [{}]".format(self.name)
-        if where is not None:
-            sql += " where " + where
-        cursor = self.db.conn.execute(sql, where_args or [])
-        columns = [c[0] for c in cursor.description]
-        for row in cursor:
-            yield dict(zip(columns, row))
-
     @property
     def pks(self):
         names = [column.name for column in self.columns if column.is_pk]
@@ -511,12 +538,6 @@ class Table:
                 )
         return fks
 
-    @property
-    def schema(self):
-        return self.db.conn.execute(
-            "select sql from sqlite_master where name = ?", (self.name,)
-        ).fetchone()[0]
-
     @property
     def indexes(self):
         sql = 'PRAGMA index_list("{}")'.format(self.name)
@@ -1114,6 +1135,18 @@ class Table:
         return self
 
 
+class View(Queryable):
+    exists = True
+
+    def __repr__(self):
+        return "<View {} ({})>".format(
+            self.name, ", ".join(c.name for c in self.columns)
+        )
+
+    def drop(self):
+        self.db.conn.execute("DROP VIEW {}".format(self.name))
+
+
 def chunks(sequence, size):
     iterator = iter(sequence)
     for item in iterator:

tests/test_create.py

@@ -782,3 +782,10 @@ def test_drop(fresh_db):
     assert ["t"] == fresh_db.table_names()
     assert None is fresh_db["t"].drop()
     assert [] == fresh_db.table_names()
+
+
+def test_drop_view(fresh_db):
+    fresh_db.create_view("foo_view", "select 1")
+    assert ["foo_view"] == fresh_db.view_names()
+    assert None is fresh_db["foo_view"].drop()
+    assert [] == fresh_db.view_names()

tests/test_introspect.py

@@ -1,4 +1,4 @@
-from sqlite_utils.db import Index
+from sqlite_utils.db import Index, View
 import pytest
 
 
@@ -6,6 +6,11 @@ def test_table_names(existing_db):
     assert ["foo"] == existing_db.table_names()
 
 
+def test_view_names(fresh_db):
+    fresh_db.create_view("foo_view", "select 1")
+    assert ["foo_view"] == fresh_db.view_names()
+
+
 def test_table_names_fts4(existing_db):
     existing_db["woo"].insert({"title": "Hello"}).enable_fts(
         ["title"], fts_version="FTS4"
@@ -36,6 +41,16 @@ def test_tables(existing_db):
     assert "foo" == existing_db.tables[0].name
 
 
+def test_views(fresh_db):
+    fresh_db.create_view("foo_view", "select 1")
+    assert 1 == len(fresh_db.views)
+    view = fresh_db.views[0]
+    assert isinstance(view, View)
+    assert "foo_view" == view.name
+    assert "<View foo_view (1)>" == repr(view)
+    assert {"1": str} == view.columns_dict
+
+
 def test_count(existing_db):
     assert 3 == existing_db["foo"].count