--- /dev/null
+.. change::
+ :tags: bug, sqlite
+ :tickets: 12425
+
+ Expanded the rules for when to apply parenthesis to a server default in DDL
+ to suit the general case of a default string that contains non-word
+ characters such as spaces or operators and is not a string literal.
+
+.. change::
+ :tags: bug, mysql
+ :tickets: 12425
+
+ Fixed issue in MySQL server default reflection where a default that has
+ spaces would not be correctly reflected. Additionally, expanded the rules
+ for when to apply parenthesis to a server default in DDL to suit the
+ general case of a default string that contains non-word characters such as
+ spaces or operators and is not a string literal.
+
CREATE TABLE a (
id INTEGER NOT NULL,
data VARCHAR NOT NULL,
- create_date DATETIME DEFAULT (CURRENT_TIMESTAMP) NOT NULL,
+ create_date DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
PRIMARY KEY (id)
)
...
colspec.append("AUTO_INCREMENT")
else:
default = self.get_column_default_string(column)
+
if default is not None:
if (
- isinstance(
- column.server_default.arg, functions.FunctionElement
- )
- and self.dialect._support_default_function
+ self.dialect._support_default_function
+ and not re.match(r"^\s*[\'\"\(]", default)
+ and "ON UPDATE" not in default
+ and re.match(r".*\W.*", default)
):
colspec.append(f"DEFAULT ({default})")
else:
r"(?: +COLLATE +(?P<collate>[\w_]+))?"
r"(?: +(?P<notnull>(?:NOT )?NULL))?"
r"(?: +DEFAULT +(?P<default>"
- r"(?:NULL|'(?:''|[^'])*'|[\-\w\.\(\)]+"
+ r"(?:NULL|'(?:''|[^'])*'|\(.+?\)|[\-\w\.\(\)]+"
r"(?: +ON UPDATE [\-\w\.\(\)]+)?)"
r"))?"
r"(?: +(?:GENERATED ALWAYS)? ?AS +(?P<generated>\("
from ...engine import reflection
from ...engine.reflection import ReflectionDefaults
from ...sql import coercions
-from ...sql import ColumnElement
from ...sql import compiler
from ...sql import elements
from ...sql import roles
colspec = self.preparer.format_column(column) + " " + coltype
default = self.get_column_default_string(column)
if default is not None:
- if isinstance(column.server_default.arg, ColumnElement):
- default = "(" + default + ")"
- colspec += " DEFAULT " + default
+
+ if not re.match(r"""^\s*[\'\"\(]""", default) and re.match(
+ r".*\W.*", default
+ ):
+ colspec += f" DEFAULT ({default})"
+ else:
+ colspec += f" DEFAULT {default}"
if not column.nullable:
colspec += " NOT NULL"
)
-def eq_regex(a, b, msg=None):
- assert re.match(b, a), msg or "%r !~ %r" % (a, b)
+def eq_regex(a, b, msg=None, flags=0):
+ assert re.match(b, a, flags), msg or "%r !~ %r" % (a, b)
def eq_(a, b, msg=None):
"""
return self.precision_numerics_many_significant_digits
+ @property
+ def server_defaults(self):
+ """Target backend supports server side defaults for columns"""
+
+ return exclusions.closed()
+
+ @property
+ def expression_server_defaults(self):
+ """Target backend supports server side defaults with SQL expressions
+ for columns"""
+
+ return exclusions.closed()
+
@property
def implicit_decimal_binds(self):
"""target backend will return a selected Decimal as a Decimal, not
from .. import config
from .. import engines
from .. import eq_
+from .. import eq_regex
from .. import expect_raises
from .. import expect_raises_message
from .. import expect_warnings
from ..provision import temp_table_keyword_args
from ..schema import Column
from ..schema import Table
+from ... import Boolean
+from ... import DateTime
from ... import event
from ... import ForeignKey
from ... import func
eq_(opts, expected)
# eq_(dict((k, opts[k]) for k in opts if opts[k]), expected)
+ @testing.combinations(
+ (Integer, sa.text("10"), r"'?10'?"),
+ (Integer, "10", r"'?10'?"),
+ (Boolean, sa.true(), r"1|true"),
+ (
+ Integer,
+ sa.text("3 + 5"),
+ r"3\+5",
+ testing.requires.expression_server_defaults,
+ ),
+ (
+ Integer,
+ sa.text("(3 * 5)"),
+ r"3\*5",
+ testing.requires.expression_server_defaults,
+ ),
+ (DateTime, func.now(), r"current_timestamp|now|getdate"),
+ (
+ Integer,
+ sa.literal_column("3") + sa.literal_column("5"),
+ r"3\+5",
+ testing.requires.expression_server_defaults,
+ ),
+ argnames="datatype, default, expected_reg",
+ )
+ @testing.requires.server_defaults
+ def test_server_defaults(
+ self, metadata, connection, datatype, default, expected_reg
+ ):
+ t = Table(
+ "t",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("thecol", datatype, server_default=default),
+ )
+ t.create(connection)
+
+ reflected = inspect(connection).get_columns("t")[1]["default"]
+ reflected_sanitized = re.sub(r"[\(\) \']", "", reflected)
+ eq_regex(reflected_sanitized, expected_reg, flags=re.IGNORECASE)
+
class NormalizedNameTest(fixtures.TablesTest):
__requires__ = ("denormalized_names",)
self.assert_compile(
schema.CreateTable(tbl),
"CREATE TABLE testtbl ("
- "time DATETIME DEFAULT (CURRENT_TIMESTAMP), "
+ "time DATETIME DEFAULT CURRENT_TIMESTAMP, "
"name VARCHAR(255) DEFAULT 'some str', "
"description VARCHAR(255) DEFAULT (lower('hi')), "
"data JSON DEFAULT (json_object()))",
from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import Computed
+from sqlalchemy import DateTime
from sqlalchemy import exc
from sqlalchemy import false
from sqlalchemy import ForeignKey
+from sqlalchemy import func
from sqlalchemy import Integer
+from sqlalchemy import literal_column
from sqlalchemy import MetaData
from sqlalchemy import or_
from sqlalchemy import schema
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import Table
+from sqlalchemy import testing
+from sqlalchemy import text
from sqlalchemy import true
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import combinations
)
+class ServerDefaultCreateTest(fixtures.TestBase):
+ @testing.combinations(
+ (Integer, text("10")),
+ (Integer, text("'10'")),
+ (Integer, "10"),
+ (Boolean, true()),
+ (Integer, text("3+5"), testing.requires.mysql_expression_defaults),
+ (Integer, text("3 + 5"), testing.requires.mysql_expression_defaults),
+ (Integer, text("(3 * 5)"), testing.requires.mysql_expression_defaults),
+ (DateTime, func.now()),
+ (
+ Integer,
+ literal_column("3") + literal_column("5"),
+ testing.requires.mysql_expression_defaults,
+ ),
+ argnames="datatype, default",
+ )
+ def test_create_server_defaults(
+ self, connection, metadata, datatype, default
+ ):
+ t = Table(
+ "t",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("thecol", datatype, server_default=default),
+ )
+ t.create(connection)
+
+
class MatchTest(fixtures.TablesTest):
__only_on__ = "mysql", "mariadb"
__backend__ = True
")",
)
- def test_column_defaults_ddl(self):
+ @testing.combinations(
+ (
+ Boolean(create_constraint=True),
+ sql.false(),
+ "BOOLEAN DEFAULT 0, CHECK (x IN (0, 1))",
+ ),
+ (
+ String(),
+ func.sqlite_version(),
+ "VARCHAR DEFAULT (sqlite_version())",
+ ),
+ (Integer(), func.abs(-5) + 17, "INTEGER DEFAULT (abs(-5) + 17)"),
+ (
+ # test #12425
+ String(),
+ func.now(),
+ "VARCHAR DEFAULT CURRENT_TIMESTAMP",
+ ),
+ (
+ # test #12425
+ String(),
+ func.datetime(func.now(), "localtime"),
+ "VARCHAR DEFAULT (datetime(CURRENT_TIMESTAMP, 'localtime'))",
+ ),
+ (
+ # test #12425
+ String(),
+ text("datetime(CURRENT_TIMESTAMP, 'localtime')"),
+ "VARCHAR DEFAULT (datetime(CURRENT_TIMESTAMP, 'localtime'))",
+ ),
+ (
+ # default with leading spaces that should not be
+ # parenthesized
+ String,
+ text(" 'some default'"),
+ "VARCHAR DEFAULT 'some default'",
+ ),
+ (String, text("'some default'"), "VARCHAR DEFAULT 'some default'"),
+ argnames="datatype,default,expected",
+ )
+ def test_column_defaults_ddl(self, datatype, default, expected):
t = Table(
"t",
MetaData(),
Column(
"x",
- Boolean(create_constraint=True),
- server_default=sql.false(),
+ datatype,
+ server_default=default,
),
)
self.assert_compile(
CreateTable(t),
- "CREATE TABLE t (x BOOLEAN DEFAULT (0), CHECK (x IN (0, 1)))",
- )
-
- t = Table(
- "t",
- MetaData(),
- Column("x", String(), server_default=func.sqlite_version()),
- )
- self.assert_compile(
- CreateTable(t),
- "CREATE TABLE t (x VARCHAR DEFAULT (sqlite_version()))",
- )
-
- t = Table(
- "t",
- MetaData(),
- Column("x", Integer(), server_default=func.abs(-5) + 17),
- )
- self.assert_compile(
- CreateTable(t), "CREATE TABLE t (x INTEGER DEFAULT (abs(-5) + 17))"
+ f"CREATE TABLE t (x {expected})",
)
def test_create_partial_index(self):
-"""Requirements specific to SQLAlchemy's own unit tests.
-
-
-"""
+"""Requirements specific to SQLAlchemy's own unit tests."""
from sqlalchemy import exc
from sqlalchemy.sql import sqltypes
]
)
+ @property
+ def server_defaults(self):
+ """Target backend supports server side defaults for columns"""
+
+ return exclusions.open()
+
+ @property
+ def expression_server_defaults(self):
+ return skip_if(
+ lambda config: against(config, "mysql", "mariadb")
+ and not self._mysql_expression_defaults(config)
+ )
+
@property
def qmark_paramstyle(self):
return only_on(["sqlite", "+pyodbc"])
# 2. they dont enforce check constraints
return not self._mysql_check_constraints_exist(config)
+ def _mysql_expression_defaults(self, config):
+ return (against(config, ["mysql", "mariadb"])) and (
+ config.db.dialect._support_default_function
+ )
+
+ @property
+ def mysql_expression_defaults(self):
+ return only_if(self._mysql_expression_defaults)
+
def _mysql_not_mariadb_102(self, config):
return (against(config, ["mysql", "mariadb"])) and (
not config.db.dialect._is_mariadb
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
