.. changelog::
:version: 1.1.0b1
+ .. change::
+ :tags: change, sqlite
+ :pullreq: github:198
+
+ Added support to the SQLite dialect for the
+ :meth:`.Inspector.get_schema_names` method to work with SQLite;
+ pull request courtesy Brian Van Klaveren. Also repaired support
+ for creation of indexes with schemas as well as reflection of
+ foreign key constraints in schema-bound tables.
+
+ .. seealso::
+
+ :ref:`change_sqlite_schemas`
+
.. change::
:tags: change, mssql
:tickets: 3434
some issues may be moved to later milestones in order to allow
for a timely release.
- Document last updated: September 19, 2015
+ Document last updated: September 28, 2015
Introduction
============
Dialect Improvements and Changes - SQLite
=============================================
+.. _change_sqlite_schemas:
+
+Improved Support for Remote Schemas
+------------------------------------
+
+The SQLite dialect now implements :meth:`.Inspector.get_schema_names`
+and additionally has improved support for tables and indexes that are
+created and reflected from a remote schema, which in SQLite is a
+database that is assigned a name via the ``ATTACH`` statement; previously,
+the ``CREATE INDEX`` DDL didn't work correctly for a schema-bound table
+and the :meth:`.Inspector.get_foreign_keys` method will now indicate the
+given schema in the results. Cross-schema foreign keys aren't supported.
+
Dialect Improvements and Changes - SQL Server
=============================================
return preparer.format_table(table, use_schema=False)
- def visit_create_index(self, create):
+ def visit_create_index(self, create, include_schema=False,
+ include_table_schema=True):
index = create.element
-
- text = super(SQLiteDDLCompiler, self).visit_create_index(
- create, include_table_schema=False)
+ self._verify_index_table(index)
+ preparer = self.preparer
+ text = "CREATE "
+ if index.unique:
+ text += "UNIQUE "
+ text += "INDEX %s ON %s (%s)" \
+ % (
+ self._prepared_index_name(index,
+ include_schema=True),
+ preparer.format_table(index.table,
+ use_schema=False),
+ ', '.join(
+ self.sql_compiler.process(
+ expr, include_table=False, literal_binds=True) for
+ expr in index.expressions)
+ )
whereclause = index.dialect_options["sqlite"]["where"]
if whereclause is not None:
s = "PRAGMA database_list"
dl = connection.execute(s)
- return [db[1] for db in dl]
+ return [db[1] for db in dl if db[1] != "temp"]
@reflection.cache
def get_table_names(self, connection, schema=None, **kw):
fk = fks[numerical_id] = {
'name': None,
'constrained_columns': [],
- 'referred_schema': None,
+ 'referred_schema': schema,
'referred_table': rtbl,
'referred_columns': [],
}
from sqlalchemy import text
from sqlalchemy.util import compat
from . import config, engines
-
+import os
FOLLOWER_IDENT = None
db_opts = {}
_update_db_opts(db_url, db_opts)
eng = engines.testing_engine(db_url, db_opts)
+ _post_configure_engine(db_url, eng, follower_ident)
eng.connect().close()
cfg = config.Config.register(eng, db_opts, options, file_config)
if follower_ident:
pass
+@register.init
+def _post_configure_engine(url, engine):
+ pass
+
+
@register.init
def _follower_url_from_main(url, ident):
url = sa_url.make_url(url)
return sa_url.make_url("sqlite:///%s.db" % ident)
+@_post_configure_engine.for_db("sqlite")
+def _sqlite_post_configure_engine(url, engine, follower_ident):
+ from sqlalchemy import event
+
+ @event.listens_for(engine, "connect")
+ def connect(dbapi_connection, connection_record):
+ # use file DBs in all cases, memory acts kind of strangely
+ # as an attached
+ if not follower_ident:
+ dbapi_connection.execute(
+ 'ATTACH DATABASE "test_schema.db" AS test_schema')
+ else:
+ dbapi_connection.execute(
+ 'ATTACH DATABASE "%s_test_schema.db" AS test_schema'
+ % follower_ident)
+
+
@_create_db.for_db("postgresql")
def _pg_create_db(cfg, eng, ident):
with eng.connect().execution_options(
@_drop_db.for_db("sqlite")
def _sqlite_drop_db(cfg, eng, ident):
- pass
- #os.remove("%s.db" % ident)
+ if ident:
+ os.remove("%s_test_schema.db" % ident)
+ else:
+ os.remove("%s.db" % ident)
@_drop_db.for_db("mysql")
assert e.pool.__class__ is pool.NullPool
-
-class AttachedMemoryDBTest(fixtures.TestBase):
+class AttachedDBTest(fixtures.TestBase):
__only_on__ = 'sqlite'
- dbname = None
-
- def setUp(self):
- self.conn = conn = testing.db.connect()
- if self.dbname is None:
- dbname = ':memory:'
- else:
- dbname = self.dbname
- conn.execute('ATTACH DATABASE "%s" AS test_schema' % dbname)
- self.metadata = MetaData()
-
- def tearDown(self):
- self.metadata.drop_all(self.conn)
- self.conn.execute('DETACH DATABASE test_schema')
- if self.dbname:
- os.remove(self.dbname)
-
def _fixture(self):
meta = self.metadata
+ self.conn = testing.db.connect()
ct = Table(
'created', meta,
Column('id', Integer),
meta.create_all(self.conn)
return ct
+ def setup(self):
+ self.conn = testing.db.connect()
+ self.metadata = MetaData()
+
+ def teardown(self):
+ self.metadata.drop_all(self.conn)
+ self.conn.close()
+
def test_no_tables(self):
insp = inspect(self.conn)
eq_(insp.get_table_names("test_schema"), [])
insp = inspect(self.conn)
eq_(insp.get_schema_names(), ["main", "test_schema"])
+ # implicitly creates a "temp" schema
+ self.conn.execute("select * from sqlite_temp_master")
+
+ # we're not including it
+ insp = inspect(self.conn)
+ eq_(insp.get_schema_names(), ["main", "test_schema"])
+
def test_reflect_system_table(self):
meta = MetaData(self.conn)
alt_master = Table(
eq_(row['name'], 'foo')
-class AttachedFileDBTest(AttachedMemoryDBTest):
- dbname = 'attached_db.db'
-
-
class SQLTest(fixtures.TestBase, AssertsCompiledSQL):
"""Tests SQLite-dialect specific compilation."""
named 'test_schema'."""
return skip_if([
- "sqlite",
"firebird"
], "no schema support")