Building SQLAlchemy engine with ATTACHed schema

sa_sqlite_attached.py

# coding: utf-8                                                                 

import sqlite3
from types import ModuleType


def sqlite_attached(schema_map):
    mod = ModuleType('', 'sqlite3_attach_%08x' %(id(schema_map)))
    for k, v in vars(sqlite3).items():
        if k not in ['connect']:
            setattr(mod, k, v)
    def connect_with_attach(*args, **kwargs):
        con = sqlite3.connect(*args, **kwargs)
        for schema, filename in schema_map.items():
            con.execute("ATTACH '?1' AS ?2", (filename, schema))
        return con
    mod.connect = connect_with_attach
    return mod
                                                                                

def demo():
    from sqlalchemy import create_engine
    from sqlalchemy.sql import column, select, table
    db_map = {
        'db1': '/tmp/database1.db',
        'db2': '/tmp/database2.db',
    }
    e = create_engine('sqlite://', module=sqlite_attached(db_map))
    e.execute('DROP TABLE IF EXISTS db1.t1')
    e.execute('CREATE TABLE db1.t1 (id integer, msg varchar)')
    e.execute("INSERT INTO db1.t1 VALUES (1, 'one'), (2, 'two')")
    e.execute('DROP TABLE IF EXISTS db2.t2')
    e.execute('CREATE TABLE db2.t2 (id integer, msg varchar)')
    e.execute("INSERT INTO db2.t2 VALUES (1, 'mono'), (2, 'di')")
    t1 = table('t1', column('id'), column('msg')); t1.schema='db1'
    t2 = table('t2', column('id'), column('msg')); t2.schema='db2'
    t1_t2 = t1.join(t2, onclause=(t1.c.id==t2.c.id))
    for row in e.execute(select([t1])):
        print(row)
    for row in e.execute(select([t2])):
        print(row)
    exp = select([t1.c.id, t1.c.msg, t2.c.msg], from_obj=t1_t2)
    for row in e.execute(exp):
        print(row)


if __name__ == '__main__':
    demo()