db.attach(alias, filepath) method, closes #113

Will also be useful for #236

Author: simonw

docs/python-api.rst

@@ -45,6 +45,27 @@ Connections use ``PRAGMA recursive_triggers=on`` by default. If you don't want t
 
     db = Database(memory=True, recursive_triggers=False)
 
+.. _python_api_attach:
+
+Attaching additional databases
+------------------------------
+
+SQLite supports cross-database SQL queries, which can join data from tables in more than one database file. You can attach an additional database using the ``.attach()`` method, providing an alias to use for that database and the path to the SQLite file on disk.
+
+.. code-block:: python
+
+    db = Database("first.db")
+    db.attach("second", "second.db")
+    # Now you can run queries like this one:
+    cursor = db.execute("""
+    select * from table_in_first
+        union all
+    select * from second.table_in_second
+    """)
+    print(cursor.fetchall())
+
+You can reference tables in the attached database using the alias value you passed to ``db.attach(alias, filepath)`` as a prefix, for example the ``second.table_in_second`` reference in the SQL query above.
+
 .. _python_api_tracing:
 
 Tracing queries

sqlite_utils/db.py

@@ -219,6 +219,14 @@ def register(fn):
     def register_fts4_bm25(self):
         self.register_function(rank_bm25, deterministic=True)
 
+    def attach(self, alias, filepath):
+        attach_sql = """
+            ATTACH DATABASE '{}' AS [{}];
+        """.format(
+            str(pathlib.Path(filepath).resolve()), alias
+        ).strip()
+        self.execute(attach_sql)
+
     def execute(self, sql, parameters=None):
         if self._tracer:
             self._tracer(sql, parameters)

tests/test_attach.py

@@ -0,0 +1,16 @@
+from sqlite_utils import Database
+
+
+def test_attach(tmpdir):
+    foo_path = str(tmpdir / "foo.db")
+    bar_path = str(tmpdir / "bar.db")
+    db = Database(foo_path)
+    with db.conn:
+        db["foo"].insert({"id": 1, "text": "foo"})
+    db2 = Database(bar_path)
+    with db2.conn:
+        db2["bar"].insert({"id": 1, "text": "bar"})
+    db.attach("bar", bar_path)
+    assert db.execute(
+        "select * from foo union all select * from bar.bar"
+    ).fetchall() == [(1, "foo"), (1, "bar")]