--- /dev/null
+.. change:
+ :tags: bug, mysql
+ :tickets: 4096
+ :versions: 1.2.0b3
+
+ Fixed issue where CURRENT_TIMESTAMP would not reflect correctly
+ in the MariaDB 10.2 series due to a syntax change, where the function
+ is now represented as ``current_timestamp()``.
+
+.. change:
+ :tags: bug, mysql
+ :tickets: 4098
+ :versions: 1.2.0b3
+
+ MariaDB 10.2 now supports CHECK constraints (warning: use version 10.2.9
+ or greater due to upstream issues noted in :ticket:`4097`). Reflection
+ now takes these CHECK constraints into account when they are present in
+ the ``SHOW CREATE TABLE`` output.
def _is_mariadb(self):
return 'MariaDB' in self.server_version_info
+ @property
+ def _is_mariadb_102(self):
+ return self._is_mariadb and \
+ self._mariadb_normalized_version_info > (10, 2)
+
@property
def _mariadb_normalized_version_info(self):
if len(self.server_version_info) > 5:
fkeys = []
- for spec in parsed_state.constraints:
- # only FOREIGN KEYs
+ for spec in parsed_state.fk_constraints:
ref_name = spec['table'][-1]
ref_schema = len(spec['table']) > 1 and \
spec['table'][-2] or schema
fkeys.append(fkey_d)
return fkeys
+ @reflection.cache
+ def get_check_constraints(
+ self, connection, table_name, schema=None, **kw):
+
+ parsed_state = self._parsed_state_or_create(
+ connection, table_name, schema, **kw)
+
+ return [
+ {"name": spec['name'], "sqltext": spec['sqltext']}
+ for spec in parsed_state.ck_constraints
+ ]
+
@reflection.cache
def get_table_comment(self, connection, table_name, schema=None, **kw):
parsed_state = self._parsed_state_or_create(
self.table_options = {}
self.table_name = None
self.keys = []
- self.constraints = []
+ self.fk_constraints = []
+ self.ck_constraints = []
@log.class_logger
util.warn("Unknown schema content: %r" % line)
elif type_ == 'key':
state.keys.append(spec)
- elif type_ == 'constraint':
- state.constraints.append(spec)
+ elif type_ == 'fk_constraint':
+ state.fk_constraints.append(spec)
+ elif type_ == 'ck_constraint':
+ state.ck_constraints.append(spec)
else:
pass
return state
spec['columns'] = self._parse_keyexprs(spec['columns'])
return 'key', spec
- # CONSTRAINT
- m = self._re_constraint.match(line)
+ # FOREIGN KEY CONSTRAINT
+ m = self._re_fk_constraint.match(line)
if m:
spec = m.groupdict()
spec['table'] = \
for c in self._parse_keyexprs(spec['local'])]
spec['foreign'] = [c[0]
for c in self._parse_keyexprs(spec['foreign'])]
- return 'constraint', spec
+ return 'fk_constraint', spec
+
+ # CHECK constraint
+ m = self._re_ck_constraint.match(line)
+ if m:
+ spec = m.groupdict()
+ return 'ck_constraint', spec
# PARTITION and SUBPARTITION
m = self._re_partition.match(line)
r"(?: +COLLATE +(?P<collate>[\w_]+))?"
r"(?: +(?P<notnull>(?:NOT )?NULL))?"
r"(?: +DEFAULT +(?P<default>"
- r"(?:NULL|'(?:''|[^'])*'|\w+"
- r"(?: +ON UPDATE \w+)?)"
+ r"(?:NULL|'(?:''|[^'])*'|[\w\(\)]+"
+ r"(?: +ON UPDATE [\w\(\)]+)?)"
r"))?"
r"(?: +(?P<autoincr>AUTO_INCREMENT))?"
r"(?: +COMMENT +'(?P<comment>(?:''|[^'])*)')?"
# unique constraints come back as KEYs
kw = quotes.copy()
kw['on'] = 'RESTRICT|CASCADE|SET NULL|NOACTION'
- self._re_constraint = _re_compile(
+ self._re_fk_constraint = _re_compile(
r' '
r'CONSTRAINT +'
r'%(iq)s(?P<name>(?:%(esc_fq)s|[^%(fq)s])+)%(fq)s +'
% kw
)
+ # CONSTRAINT `CONSTRAINT_1` CHECK (`x` > 5)'
+ # testing on MariaDB 10.2 shows that the CHECK constraint
+ # is returned on a line by itself, so to match without worrying
+ # about parenthesis in the expresion we go to the end of the line
+ self._re_ck_constraint = _re_compile(
+ r' '
+ r'CONSTRAINT +'
+ r'%(iq)s(?P<name>(?:%(esc_fq)s|[^%(fq)s])+)%(fq)s +'
+ r'CHECK +'
+ r'\((?P<sqltext>.+)\),?'
+ % kw
+ )
+
# PARTITION
#
# punt!
from sqlalchemy.testing import fixtures, AssertsExecutionResults
from sqlalchemy import testing
from sqlalchemy.testing import assert_raises_message, expect_warnings
+import re
class TypeReflectionTest(fixtures.TestBase):
assert reflected.c.c5.default is None
assert reflected.c.c5.server_default is None
assert reflected.c.c6.default is None
- eq_(
- str(reflected.c.c6.server_default.arg).upper(),
- "CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"
+ assert re.match(
+ r"CURRENT_TIMESTAMP(\(\))? ON UPDATE CURRENT_TIMESTAMP(\(\))?",
+ str(reflected.c.c6.server_default.arg).upper()
)
reflected.create()
try:
assert reflected.c.c5.default is None
assert reflected.c.c5.server_default is None
assert reflected.c.c6.default is None
- eq_(
- str(reflected.c.c6.server_default.arg).upper(),
- "CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"
+ assert re.match(
+ r"CURRENT_TIMESTAMP(\(\))? ON UPDATE CURRENT_TIMESTAMP(\(\))?",
+ str(reflected.c.c6.server_default.arg).upper()
)
def test_reflection_with_table_options(self):
for d in inspect(testing.db).get_columns("nn_t%d" % idx)
)
+ if testing.db.dialect._is_mariadb_102:
+ current_timestamp = "current_timestamp()"
+ else:
+ current_timestamp = "CURRENT_TIMESTAMP"
+
eq_(
reflected,
[
{'name': 'z', 'nullable': True, 'default': None},
{'name': 'q', 'nullable': True, 'default': None},
{'name': 'p', 'nullable': True,
- 'default': 'CURRENT_TIMESTAMP'},
+ 'default': current_timestamp},
{'name': 'r', 'nullable': False,
- 'default': "CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"},
+ 'default':
+ "%(current_timestamp)s ON UPDATE %(current_timestamp)s" %
+ {"current_timestamp": current_timestamp}},
{'name': 's', 'nullable': False,
- 'default': 'CURRENT_TIMESTAMP'},
+ 'default': current_timestamp},
{'name': 't', 'nullable': False,
- 'default': "CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"},
+ 'default':
+ "%(current_timestamp)s ON UPDATE %(current_timestamp)s" %
+ {"current_timestamp": current_timestamp}},
{'name': 'u', 'nullable': False,
- 'default': 'CURRENT_TIMESTAMP'},
+ 'default': current_timestamp},
]
)
class RawReflectionTest(fixtures.TestBase):
+ __backend__ = True
+
def setup(self):
dialect = mysql.dialect()
self.parser = _reflection.MySQLTableDefinitionParser(
" KEY (`id`) USING BTREE COMMENT 'prefix''text''suffix'")
def test_fk_reflection(self):
- regex = self.parser._re_constraint
+ regex = self.parser._re_fk_constraint
m = regex.match(' CONSTRAINT `addresses_user_id_fkey` '
'FOREIGN KEY (`user_id`) '
ComparesTables, engines, AssertsCompiledSQL,
fixtures, skip)
from sqlalchemy.testing.schema import Table, Column
-from sqlalchemy.testing import eq_, is_true, assert_raises, \
+from sqlalchemy.testing import eq_, eq_regex, is_true, assert_raises, \
assert_raises_message
from sqlalchemy import testing
from sqlalchemy.util import ue
const for const in
t2.constraints if isinstance(const, sa.CheckConstraint)][0]
- eq_(ck.sqltext.text, "q > 10")
+ eq_regex(ck.sqltext.text, r".?q.? > 10")
eq_(ck.name, "ck1")
@testing.provide_metadata
def enforces_check_constraints(self):
"""Target database must also enforce check constraints."""
- def mysql_not_mariadb_102(config):
- return against(config, "mysql") and (
- not config.db.dialect._is_mariadb or
- config.db.dialect._mariadb_normalized_version_info < (10, 2)
- )
-
return self.check_constraints + fails_on(
- mysql_not_mariadb_102,
+ self._mysql_not_mariadb_102,
"check constraints don't enforce on MySQL, MariaDB<10.2"
)
@property
def check_constraint_reflection(self):
- return fails_on_everything_except(
- "postgresql",
- "sqlite",
- "oracle"
- )
+ return only_if(
+ ["postgresql", "sqlite", "oracle",
+ self._mariadb_102]
+ )
@property
def temp_table_names(self):
return only_if(check)
+ def _mariadb_102(self, config):
+ return against(config, "mysql") and \
+ config.db.dialect._is_mariadb and \
+ config.db.dialect._mariadb_normalized_version_info > (10, 2)
+
+ def _mysql_not_mariadb_102(self, config):
+ return against(config, "mysql") and (
+ not config.db.dialect._is_mariadb or
+ config.db.dialect._mariadb_normalized_version_info < (10, 2)
+ )
+
def _has_mysql_on_windows(self, config):
return against(config, 'mysql') and \
config.db.dialect._detect_casing(config.db) == 1
@testing.requires.enforces_check_constraints
def test_check_constraint(self):
assert_raises(
- (exc.IntegrityError, exc.ProgrammingError, exc.OperationalError),
+ (
+ exc.IntegrityError, exc.ProgrammingError,
+ exc.OperationalError,
+ # PyMySQL raising InternalError until
+ # https://github.com/PyMySQL/PyMySQL/issues/607 is resolved
+ exc.InternalError),
testing.db.execute,
"insert into non_native_enum_table "
"(id, someenum) values(1, 'four')")
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
