symbol when appropriate in conjunction with the ``VARBINARY``
data type. Pull request courtesy Sheila Allen.
+ .. change::
+ :tags: feature, sql
+ :pullreq: bitbucket:80
+
+ Implemented reflection of CHECK constraints for SQLite and Postgresql.
+ This is available via the new inspector method
+ :meth:`.Inspector.get_check_constraints` as well as when reflecting
+ :class:`.Table` objects in the form of :class:`.CheckConstraint`
+ objects present in the constraints collection. Pull request courtesy
+ Alex Grönholm.
+
.. change::
:tags: change, orm
:tickets: 3394
:ticket:`2729`
+Check constraints now reflect
+-----------------------------
+
+The Postgresql dialect now supports reflection of CHECK constraints
+both within the method :meth:`.Inspector.get_check_constraints` as well
+as within :class:`.Table` reflection within the :attr:`.Table.constraints`
+collection.
+
Support for PyGreSQL
--------------------
:ticket:`3629`
+Check constraints now reflect
+-----------------------------
+
+The SQLite dialect now supports reflection of CHECK constraints
+both within the method :meth:`.Inspector.get_check_constraints` as well
+as within :class:`.Table` reflection within the :attr:`.Table.constraints`
+collection.
+
Dialect Improvements and Changes - SQL Server
=============================================
for name, uc in uniques.items()
]
+ @reflection.cache
+ def get_check_constraints(
+ self, connection, table_name, schema=None, **kw):
+ table_oid = self.get_table_oid(connection, table_name, schema,
+ info_cache=kw.get('info_cache'))
+
+ CHECK_SQL = """
+ SELECT
+ cons.conname as name,
+ cons.consrc as src
+ FROM
+ pg_catalog.pg_constraint cons
+ WHERE
+ cons.conrelid = :table_oid AND
+ cons.contype = 'c'
+ """
+
+ c = connection.execute(sql.text(CHECK_SQL), table_oid=table_oid)
+
+ return [
+ {'name': name,
+ 'sqltext': src[1:-1]}
+ for name, src in c.fetchall()
+ ]
+
def _load_enums(self, connection, schema=None):
schema = schema or self.default_schema_name
if not self.supports_native_enum:
# the PRIMARY KEY may have an entry.
return unique_constraints
+ @reflection.cache
+ def get_check_constraints(self, connection, table_name,
+ schema=None, **kw):
+ table_data = self._get_table_sql(
+ connection, table_name, schema=schema, **kw)
+ if not table_data:
+ return []
+
+ CHECK_PATTERN = (
+ '(?:CONSTRAINT (\w+) +)?'
+ 'CHECK *\( *(.+) *\),? *'
+ )
+ check_constraints = []
+ # NOTE: we aren't using re.S here because we actually are
+ # taking advantage of each CHECK constraint being all on one
+ # line in the table definition in order to delineate. This
+ # necessarily makes assumptions as to how the CREATE TABLE
+ # was emitted.
+ for match in re.finditer(CHECK_PATTERN, table_data, re.I):
+ check_constraints.append({
+ 'sqltext': match.group(2),
+ 'name': match.group(1)
+ })
+
+ return check_constraints
+
@reflection.cache
def get_indexes(self, connection, table_name, schema=None, **kw):
pragma_indexes = self._get_table_pragma(
raise NotImplementedError()
+ def get_check_constraints(
+ self, connection, table_name, schema=None, **kw):
+ """Return information about check constraints in `table_name`.
+
+ Given a string `table_name` and an optional string `schema`, return
+ check constraint information as a list of dicts with these keys:
+
+ name
+ the check constraint's name
+
+ sqltext
+ the check constraint's SQL expression
+
+ \**kw
+ other options passed to the dialect's get_check_constraints()
+ method.
+
+ .. versionadded:: 1.1.0
+
+ """
+
+ raise NotImplementedError()
+
def normalize_name(self, name):
"""convert the given name to lowercase if it is detected as
case insensitive.
return self.dialect.get_unique_constraints(
self.bind, table_name, schema, info_cache=self.info_cache, **kw)
+ def get_check_constraints(self, table_name, schema=None, **kw):
+ """Return information about check constraints in `table_name`.
+
+ Given a string `table_name` and an optional string `schema`, return
+ check constraint information as a list of dicts with these keys:
+
+ name
+ the check constraint's name
+
+ sqltext
+ the check constraint's SQL expression
+
+ :param table_name: string name of the table. For special quoting,
+ use :class:`.quoted_name`.
+
+ :param schema: string schema name; if omitted, uses the default schema
+ of the database connection. For special quoting,
+ use :class:`.quoted_name`.
+
+ .. versionadded:: 1.1.0
+
+ """
+
+ return self.dialect.get_check_constraints(
+ self.bind, table_name, schema, info_cache=self.info_cache, **kw)
+
def reflecttable(self, table, include_columns, exclude_columns=()):
"""Given a Table object, load its internal constructs based on
introspection.
table_name, schema, table, cols_by_orig_name,
include_columns, exclude_columns, reflection_options)
+ self._reflect_check_constraints(
+ table_name, schema, table, cols_by_orig_name,
+ include_columns, exclude_columns, reflection_options)
+
def _reflect_column(
self, table, col_d, include_columns,
exclude_columns, cols_by_orig_name):
constrained_cols.append(constrained_col)
table.append_constraint(
sa_schema.UniqueConstraint(*constrained_cols, name=conname))
+
+ def _reflect_check_constraints(
+ self, table_name, schema, table, cols_by_orig_name,
+ include_columns, exclude_columns, reflection_options):
+ try:
+ constraints = self.get_check_constraints(table_name, schema)
+ except NotImplementedError:
+ # optional dialect feature
+ return
+
+ for const_d in constraints:
+ table.append_constraint(
+ sa_schema.CheckConstraint(**const_d))
# coding: utf-8
from sqlalchemy.engine import reflection
+from sqlalchemy.sql.schema import CheckConstraint
from sqlalchemy.testing.assertions import eq_, assert_raises, \
AssertsExecutionResults
from sqlalchemy.testing import fixtures
assert indexes['ix_a'].unique
self.assert_('ix_a' not in constraints)
+ @testing.provide_metadata
+ def test_reflect_check_constraint(self):
+ meta = self.metadata
+
+ cc_table = Table(
+ 'pgsql_cc', meta,
+ Column('a', Integer()),
+ CheckConstraint('a > 1 AND a < 5', name='cc1'),
+ CheckConstraint('a = 1 OR (a > 2 AND a < 5)', name='cc2'))
+
+ cc_table.create()
+
+ reflected = Table('pgsql_cc', MetaData(testing.db), autoload=True)
+
+ check_constraints = dict((uc.name, uc.sqltext.text)
+ for uc in reflected.constraints
+ if isinstance(uc, CheckConstraint))
+
+ eq_(check_constraints, {
+ u'cc1': u'(a > 1) AND (a < 5)',
+ u'cc2': u'(a = 1) OR ((a > 2) AND (a < 5))'
+ })
+
class CustomTypeReflectionTest(fixtures.TestBase):
# will contain an "autoindex"
conn.execute("create table o (foo varchar(20) primary key)")
+ conn.execute(
+ "CREATE TABLE cp ("
+ "q INTEGER check (q > 1 AND q < 6),\n"
+ "CONSTRAINT cq CHECK (q == 1 OR (q > 2 AND q < 5))\n"
+ ")"
+ )
+
@classmethod
def teardown_class(cls):
with testing.db.begin() as conn:
{'constrained_columns': [], 'name': None}
)
+ def test_check_constraint(self):
+ inspector = Inspector(testing.db)
+ eq_(
+ inspector.get_check_constraints("cp"),
+ [{'sqltext': 'q > 1 AND q < 6', 'name': None},
+ {'sqltext': 'q == 1 OR (q > 2 AND q < 5)', 'name': 'cq'}]
+ )
+
class SavepointTest(fixtures.TablesTest):
assert set([t2.c.name, t2.c.id]) == set(r2.columns)
assert set([t2.c.name]) == set(r3.columns)
+ @testing.requires.check_constraint_reflection
+ @testing.provide_metadata
+ def test_check_constraint_reflection(self):
+ m1 = self.metadata
+ Table(
+ 'x', m1,
+ Column('q', Integer),
+ sa.CheckConstraint('q > 10', name="ck1")
+ )
+ m1.create_all()
+ m2 = MetaData(testing.db)
+ t2 = Table('x', m2, autoload=True)
+
+ ck = [
+ const for const in
+ t2.constraints if isinstance(const, sa.CheckConstraint)][0]
+
+ eq_(ck.sqltext.text, "q > 10")
+ eq_(ck.name, "ck1")
+
@testing.provide_metadata
def test_index_reflection_cols_busted(self):
t = Table('x', self.metadata,
"sqlite"
)
+ @property
+ def check_constraint_reflection(self):
+ return fails_on_everything_except(
+ "postgresql",
+ "sqlite"
+ )
+
@property
def temp_table_names(self):
"""target dialect supports listing of temporary table names"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
