db.table() only returns tables, added db.view(), refs #657

Author: simonw

docs/python-api.rst

@@ -251,21 +251,36 @@ In this example ``next()`` is used to retrieve the first result in the iterator
 Accessing tables
 ================
 
-Tables are accessed using the indexing operator, like so:
+Tables are accessed using the ``db.table()`` method, like so:
+
+.. code-block:: python
+
+    table = db.table("my_table")
+
+Using this factory function allows you to set :ref:`python_api_table_configuration`. Additional keyword arguments to ``db.table()`` will be used if a further method call causes the table to be created.
+
+The ``db.table()`` method  will always return a :ref:`reference_db_table` instance, or raise a ``sqlite_utils.db.NoTable`` exception if the table name is actually a SQL view.
+
+You can also access tables or views using dictionary-style syntax, like this:
 
 .. code-block:: python
 
     table = db["my_table"]
 
-If the table does not yet exist, it will be created the first time you attempt to insert or upsert data into it.
+If a table accessed using either of these methods does not yet exist, it will be created the first time you attempt to insert or upsert data into it.
+
+.. _python_api_view:
 
-You can also access tables using the ``.table()`` method like so:
+Accessing views
+===============
+
+SQL views can be accessed using the ``db.view()`` method, like so:
 
 .. code-block:: python
 
-    table = db.table("my_table")
+    view = db.view("my_view")
 
-Using this factory function allows you to set :ref:`python_api_table_configuration`.
+This will return a  :ref:`reference_db_view` instance, or raise a ``sqlite_utils.db.NoView`` exception if the view does not exist.
 
 .. _python_api_tables:
 

sqlite_utils/cli.py

@@ -186,6 +186,8 @@ def tables(
     if schema:
         headers.append("schema")
 
+    method = db.view if views else db.table
+
     def _iter():
         if views:
             items = db.view_names()
@@ -194,15 +196,15 @@ def _iter():
         for name in items:
             row = [name]
             if counts:
-                row.append(db[name].count)
+                row.append(method(name).count)
             if columns:
-                cols = [c.name for c in db[name].columns]
+                cols = [c.name for c in method(name).columns]
                 if csv:
                     row.append("\n".join(cols))
                 else:
                     row.append(cols)
             if schema:
-                row.append(db[name].schema)
+                row.append(method(name).schema)
             yield row
 
     if table or fmt:
@@ -1693,7 +1695,7 @@ def create_view(path, view, select, ignore, replace, load_extension):
         if ignore:
             return
         elif replace:
-            db[view].drop()
+            db.view(view).drop()
         else:
             raise click.ClickException(
                 'View "{}" already exists. Use --replace to delete and replace it.'.format(

sqlite_utils/db.py

@@ -247,6 +247,10 @@ class NoTable(Exception):
     "Specified table does not exist"
 
 
+class NoView(Exception):
+    "Specified view does not exist"
+
+
 class BadPrimaryKey(Exception):
     "Table does not have a single obvious primary key"
 
@@ -419,6 +423,8 @@ def __getitem__(self, table_name: str) -> Union["Table", "View"]:
 
         :param table_name: The name of the table
         """
+        if table_name in self.view_names():
+            return self.view(table_name)
         return self.table(table_name)
 
     def __repr__(self) -> str:
@@ -541,7 +547,7 @@ def executescript(self, sql: str) -> sqlite3.Cursor:
             self._tracer(sql, None)
         return self.conn.executescript(sql)
 
-    def table(self, table_name: str, **kwargs) -> Union["Table", "View"]:
+    def table(self, table_name: str, **kwargs) -> "Table":
         """
         Return a table object, optionally configured with default options.
 
@@ -550,10 +556,19 @@ def table(self, table_name: str, **kwargs) -> Union["Table", "View"]:
         :param table_name: Name of the table
         """
         if table_name in self.view_names():
-            return View(self, table_name, **kwargs)
-        else:
-            kwargs.setdefault("strict", self.strict)
-            return Table(self, table_name, **kwargs)
+            raise NoTable("Table {} is actually a view".format(table_name))
+        kwargs.setdefault("strict", self.strict)
+        return Table(self, table_name, **kwargs)
+
+    def view(self, view_name: str) -> "View":
+        """
+        Return a view object.
+
+        :param view_name: Name of the view
+        """
+        if view_name not in self.view_names():
+            raise NoView("View {} does not exist".format(view_name))
+        return View(self, view_name)
 
     def quote(self, value: str) -> str:
         """
@@ -637,12 +652,12 @@ def view_names(self) -> List[str]:
     @property
     def tables(self) -> List["Table"]:
         "List of Table objects in this database."
-        return cast(List["Table"], [self[name] for name in self.table_names()])
+        return [self.table(name) for name in self.table_names()]
 
     @property
     def views(self) -> List["View"]:
         "List of View objects in this database."
-        return cast(List["View"], [self[name] for name in self.view_names()])
+        return [self.view(name) for name in self.view_names()]
 
     @property
     def triggers(self) -> List[Trigger]:
@@ -808,7 +823,7 @@ def resolve_foreign_keys(
             or a tuple of (column, other_table, other_column), or a tuple of
             (table, column, other_table, other_column)
         """
-        table = cast(Table, self[name])
+        table = self.table(name)
         if all(isinstance(fk, ForeignKey) for fk in foreign_keys):
             return cast(List[ForeignKey], foreign_keys)
         if all(isinstance(fk, str) for fk in foreign_keys):
@@ -1039,11 +1054,11 @@ def create_table(
         # Transform table to match the new definition if table already exists:
         if self[name].exists():
             if ignore:
-                return cast(Table, self[name])
+                return self.table(name)
             elif replace:
                 self[name].drop()
         if transform and self[name].exists():
-            table = cast(Table, self[name])
+            table = self.table(name)
             should_transform = False
             # First add missing columns and figure out columns to drop
             existing_columns = table.columns_dict
@@ -1109,7 +1124,7 @@ def create_table(
             strict=strict,
         )
         self.execute(sql)
-        created_table = self.table(
+        return self.table(
             name,
             pk=pk,
             foreign_keys=foreign_keys,
@@ -1119,7 +1134,6 @@ def create_table(
             hash_id=hash_id,
             hash_id_columns=hash_id_columns,
         )
-        return cast(Table, created_table)
 
     def rename_table(self, name: str, new_name: str):
         """
@@ -1196,12 +1210,9 @@ def add_foreign_keys(self, foreign_keys: Iterable[Tuple[str, str, str, str]]):
 
         # Verify that all tables and columns exist
         for table, column, other_table, other_column in foreign_keys:
-            if not self[table].exists():
+            if not self.table(table).exists():
                 raise AlterError("No such table: {}".format(table))
-            table_obj = self[table]
-            if not isinstance(table_obj, Table):
-                raise AlterError("Must be a table, not a view: {}".format(table))
-            table_obj = cast(Table, table_obj)
+            table_obj = self.table(table)
             if column not in table_obj.columns_dict:
                 raise AlterError("No such column: {} in {}".format(column, table))
             if not self[other_table].exists():
@@ -1231,7 +1242,7 @@ def add_foreign_keys(self, foreign_keys: Iterable[Tuple[str, str, str, str]]):
             by_table.setdefault(fk[0], []).append(fk)
 
         for table, fks in by_table.items():
-            cast(Table, self[table]).transform(add_foreign_keys=fks)
+            self.table(table).transform(add_foreign_keys=fks)
 
         self.vacuum()
 
@@ -3655,7 +3666,7 @@ def m2m(
           already exists.
         """
         if isinstance(other_table, str):
-            other_table = cast(Table, self.db.table(other_table, pk=pk))
+            other_table = self.db.table(other_table, pk=pk)
         our_id = self.last_pk
         if lookup is not None:
             assert record_or_iterable is None, "Provide lookup= or record, not both"

tests/test_enable_counts.py

@@ -129,7 +129,7 @@ def test_uses_counts_after_enable_counts(counts_db_path):
     db = Database(counts_db_path)
     logged = []
     with db.tracer(lambda sql, parameters: logged.append((sql, parameters))):
-        assert db["foo"].count == 1
+        assert db.table("foo").count == 1
         assert logged == [
             ("select name from sqlite_master where type = 'view'", None),
             ("select count(*) from [foo]", []),
@@ -138,7 +138,7 @@ def test_uses_counts_after_enable_counts(counts_db_path):
         assert not db.use_counts_table
         db.enable_counts()
         assert db.use_counts_table
-        assert db["foo"].count == 1
+        assert db.table("foo").count == 1
     assert logged == [
         (
             "CREATE TABLE IF NOT EXISTS [_counts](\n   [table] TEXT PRIMARY KEY,\n   count INTEGER DEFAULT 0\n);",

tests/test_tracer.py

@@ -6,20 +6,21 @@ def test_tracer():
     db = Database(
         memory=True, tracer=lambda sql, params: collected.append((sql, params))
     )
-    db["dogs"].insert({"name": "Cleopaws"})
-    db["dogs"].enable_fts(["name"])
-    db["dogs"].search("Cleopaws")
+    dogs = db.table("dogs")
+    dogs.insert({"name": "Cleopaws"})
+    dogs.enable_fts(["name"])
+    dogs.search("Cleopaws")
     assert collected == [
         ("PRAGMA recursive_triggers=on;", None),
         ("select name from sqlite_master where type = 'view'", None),
         ("select name from sqlite_master where type = 'table'", None),
         ("select name from sqlite_master where type = 'view'", None),
+        ("select name from sqlite_master where type = 'view'", None),
         ("select name from sqlite_master where type = 'table'", None),
         ("select name from sqlite_master where type = 'view'", None),
         ("CREATE TABLE [dogs] (\n   [name] TEXT\n);\n        ", None),
         ("select name from sqlite_master where type = 'view'", None),
         ("INSERT INTO [dogs] ([name]) VALUES (?)", ["Cleopaws"]),
-        ("select name from sqlite_master where type = 'view'", None),
         (
             "CREATE VIRTUAL TABLE [dogs_fts] USING FTS5 (\n    [name],\n    content=[dogs]\n)",
             None,
@@ -28,7 +29,6 @@ def test_tracer():
             "INSERT INTO [dogs_fts] (rowid, [name])\n    SELECT rowid, [name] FROM [dogs];",
             None,
         ),
-        ("select name from sqlite_master where type = 'view'", None),
     ]
 
 
@@ -40,60 +40,58 @@ def tracer(sql, params):
 
     db = Database(memory=True)
 
-    db["dogs"].insert({"name": "Cleopaws"})
-    db["dogs"].enable_fts(["name"])
+    dogs = db.table("dogs")
+
+    dogs.insert({"name": "Cleopaws"})
+    dogs.enable_fts(["name"])
 
     assert len(collected) == 0
 
     with db.tracer(tracer):
-        list(db["dogs"].search("Cleopaws"))
+        list(dogs.search("Cleopaws"))
 
     assert len(collected) == 5
     assert collected == [
-        ("select name from sqlite_master where type = 'view'", None),
         (
-            (
-                "SELECT name FROM sqlite_master\n"
-                "    WHERE rootpage = 0\n"
-                "    AND (\n"
-                "        sql LIKE :like\n"
-                "        OR sql LIKE :like2\n"
-                "        OR (\n"
-                "            tbl_name = :table\n"
-                "            AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
-                "        )\n"
-                "    )",
-                {
-                    "like": "%VIRTUAL TABLE%USING FTS%content=[dogs]%",
-                    "like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
-                    "table": "dogs",
-                },
-            )
+            "SELECT name FROM sqlite_master\n"
+            "    WHERE rootpage = 0\n"
+            "    AND (\n"
+            "        sql LIKE :like\n"
+            "        OR sql LIKE :like2\n"
+            "        OR (\n"
+            "            tbl_name = :table\n"
+            "            AND sql LIKE '%VIRTUAL TABLE%USING FTS%'\n"
+            "        )\n"
+            "    )",
+            {
+                "like": "%VIRTUAL TABLE%USING FTS%content=[dogs]%",
+                "like2": '%VIRTUAL TABLE%USING FTS%content="dogs"%',
+                "table": "dogs",
+            },
         ),
         ("select name from sqlite_master where type = 'view'", None),
+        ("select name from sqlite_master where type = 'view'", None),
         ("select sql from sqlite_master where name = ?", ("dogs_fts",)),
         (
-            (
-                "with original as (\n"
-                "    select\n"
-                "        rowid,\n"
-                "        *\n"
-                "    from [dogs]\n"
-                ")\n"
-                "select\n"
-                "    [original].*\n"
-                "from\n"
-                "    [original]\n"
-                "    join [dogs_fts] on [original].rowid = [dogs_fts].rowid\n"
-                "where\n"
-                "    [dogs_fts] match :query\n"
-                "order by\n"
-                "    [dogs_fts].rank"
-            ),
+            "with original as (\n"
+            "    select\n"
+            "        rowid,\n"
+            "        *\n"
+            "    from [dogs]\n"
+            ")\n"
+            "select\n"
+            "    [original].*\n"
+            "from\n"
+            "    [original]\n"
+            "    join [dogs_fts] on [original].rowid = [dogs_fts].rowid\n"
+            "where\n"
+            "    [dogs_fts] match :query\n"
+            "order by\n"
+            "    [dogs_fts].rank",
             {"query": "Cleopaws"},
         ),
     ]
 
     # Outside the with block collected should not be appended to
-    db["dogs"].insert({"name": "Cleopaws"})
+    dogs.insert({"name": "Cleopaws"})
     assert len(collected) == 5