--- /dev/null
+.. change::
+ :tags: bug, schema, mysql, mariadb, oracle, postgresql
+ :tickets: 6338
+
+ Ensure that the MySQL and MariaDB dialect ignore the
+ :class:`_sql.Identity` construct while rendering the ``AUTO_INCREMENT``
+ keyword in a create table.
+
+ The Oracle and PostgreSQL compiler was updated to not render
+ :class:`_sql.Identity` if the database version does not support it
+ (Oracle < 12 and PostgreSQL < 10). Previously it was rendered regardless
+ of the database version.
if (
column.table is not None
and column is column.table._autoincrement_column
- and column.server_default is None
+ and (
+ column.server_default is None
+ or isinstance(column.server_default, sa_schema.Identity)
+ )
and not (
self.dialect.supports_sequences
and isinstance(column.default, sa_schema.Sequence)
supports_default_values = False
supports_default_metavalue = True
supports_empty_insert = False
+ supports_identity_columns = True
statement_compiler = OracleCompiler
ddl_compiler = OracleDDLCompiler
self.colspecs.pop(sqltypes.Interval)
self.use_ansi = False
+ self.supports_identity_columns = self.server_version_info >= (12,)
+
def _get_effective_compat_server_version_info(self, connection):
# dialect does not need compat levels below 12.2, so don't query
# in those cases
if isinstance(impl_type, sqltypes.TypeDecorator):
impl_type = impl_type.impl
+ has_identity = (
+ column.identity is not None
+ and self.dialect.supports_identity_columns
+ )
+
if (
column.primary_key
and column is column.table._autoincrement_column
self.dialect.supports_smallserial
or not isinstance(impl_type, sqltypes.SmallInteger)
)
- and column.identity is None
+ and not has_identity
and (
column.default is None
or (
if column.computed is not None:
colspec += " " + self.process(column.computed)
- if column.identity is not None:
+ if has_identity:
colspec += " " + self.process(column.identity)
- if not column.nullable and not column.identity:
+ if not column.nullable and not has_identity:
colspec += " NOT NULL"
- elif column.nullable and column.identity:
+ elif column.nullable and has_identity:
colspec += " NULL"
return colspec
supports_empty_insert = False
supports_multivalues_insert = True
+ supports_identity_columns = True
+
default_paramstyle = "pyformat"
ischema_names = ischema_names
colspecs = colspecs
9,
2,
)
+ self.supports_identity_columns = self.server_version_info >= (10,)
def on_connect(self):
if self.isolation_level is not None:
supports_sequences = False
sequences_optional = False
preexecute_autoincrement_sequences = False
+ supports_identity_columns = False
postfetch_lastrowid = True
implicit_returning = False
full_returning = False
supports_statement_cache = True
+ supports_identity_columns = True
+
supports_sequences = True
sequences_optional = True
preexecute_autoincrement_sequences = False
if column.computed is not None:
colspec += " " + self.process(column.computed)
- if column.identity is not None:
+ if (
+ column.identity is not None
+ and self.dialect.supports_identity_columns
+ ):
colspec += " " + self.process(column.identity)
- if not column.nullable and not column.identity:
+ if not column.nullable and (
+ not column.identity or not self.dialect.supports_identity_columns
+ ):
colspec += " NOT NULL"
return colspec
or ties. basically this is "not mssql"
"""
return exclusions.closed()
+
+ @property
+ def autoincrement_without_sequence(self):
+ """If autoincrement=True on a column does not require an explicit
+ sequence. This should be false only for oracle.
+ """
+ return exclusions.open()
assert_raises((DatabaseError, ProgrammingError), fn)
+class IdentityAutoincrementTest(fixtures.TablesTest):
+ __backend__ = True
+ __requires__ = ("autoincrement_without_sequence",)
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "tbl",
+ metadata,
+ Column(
+ "id",
+ Integer,
+ Identity(),
+ primary_key=True,
+ autoincrement=True,
+ ),
+ Column("desc", String(100)),
+ )
+
+ def test_autoincrement_with_identity(self, connection):
+ res = connection.execute(self.tables.tbl.insert(), {"desc": "row"})
+ res = connection.execute(self.tables.tbl.select()).first()
+ eq_(res, (1, "row"))
+
+
class ExistsTest(fixtures.TablesTest):
__backend__ = True
"CREATE TABLE t (y INTEGER GENERATED %s AS IDENTITY)" % text,
)
+ def test_column_identity_not_supported(self):
+ m = MetaData()
+ t = Table("t", m, Column("y", Integer, Identity(always=None)))
+ dd = oracle.OracleDialect()
+ dd.supports_identity_columns = False
+ self.assert_compile(
+ schema.CreateTable(t),
+ "CREATE TABLE t (y INTEGER NOT NULL)",
+ dialect=dd,
+ )
+
class SequenceTest(fixtures.TestBase, AssertsCompiledSQL):
def test_basic(self):
dialect=postgresql.dialect(),
)
- def test_column_identity(self):
+ @testing.combinations(True, False)
+ def test_column_identity(self, pk):
# all other tests are in test_identity_column.py
m = MetaData()
t = Table(
"t",
m,
- Column("y", Integer, Identity(always=True, start=4, increment=7)),
+ Column(
+ "y",
+ Integer,
+ Identity(always=True, start=4, increment=7),
+ primary_key=pk,
+ ),
)
self.assert_compile(
schema.CreateTable(t),
"CREATE TABLE t (y INTEGER GENERATED ALWAYS AS IDENTITY "
- "(INCREMENT BY 7 START WITH 4))",
+ "(INCREMENT BY 7 START WITH 4)%s)"
+ % (", PRIMARY KEY (y)" if pk else ""),
+ )
+
+ @testing.combinations(True, False)
+ def test_column_identity_no_support(self, pk):
+ m = MetaData()
+ t = Table(
+ "t",
+ m,
+ Column(
+ "y",
+ Integer,
+ Identity(always=True, start=4, increment=7),
+ primary_key=pk,
+ ),
+ )
+ dd = PGDialect()
+ dd.supports_identity_columns = False
+ self.assert_compile(
+ schema.CreateTable(t),
+ "CREATE TABLE t (y %s%s)"
+ % (
+ "SERIAL NOT NULL" if pk else "INTEGER NOT NULL",
+ ", PRIMARY KEY (y)" if pk else "",
+ ),
+ dialect=dd,
)
def test_column_identity_null(self):
@property
def fetch_offset_with_options(self):
return skip_if("mssql")
+
+ @property
+ def autoincrement_without_sequence(self):
+ return skip_if("oracle")
+import re
+
from sqlalchemy import Column
from sqlalchemy import Identity
from sqlalchemy import Integer
from sqlalchemy import Sequence
from sqlalchemy import Table
from sqlalchemy import testing
+from sqlalchemy.engine import URL
from sqlalchemy.exc import ArgumentError
from sqlalchemy.schema import CreateTable
from sqlalchemy.testing import assert_raises_message
)
def test_create_ddl(self, identity_args, text):
- if getattr(self, "__dialect__", None) != "default" and testing.against(
- "oracle"
- ):
+ if getattr(
+ self, "__dialect__", None
+ ) != "default_enhanced" and testing.against("oracle"):
text = text.replace("NO MINVALUE", "NOMINVALUE")
text = text.replace("NO MAXVALUE", "NOMAXVALUE")
text = text.replace("NO CYCLE", "NOCYCLE")
is_(t.c.c.nullable, False)
nullable = ""
- if getattr(self, "__dialect__", None) != "default" and testing.against(
- "postgresql"
- ):
+ if getattr(
+ self, "__dialect__", None
+ ) != "default_enhanced" and testing.against("postgresql"):
nullable = " NULL"
self.assert_compile(
class DefaultDialectIdentityDDL(_IdentityDDLFixture, fixtures.TestBase):
# this uses the default dialect
- __dialect__ = "default"
+ __dialect__ = "default_enhanced"
class NotSupportingIdentityDDL(testing.AssertsCompiledSQL, fixtures.TestBase):
- # a dialect that doesn't render IDENTITY
- __dialect__ = "sqlite"
+ def get_dialect(self, dialect):
+ dd = URL.create(dialect).get_dialect()()
+ if dialect in {"oracle", "postgresql"}:
+ dd.supports_identity_columns = False
+ return dd
+
+ @testing.combinations("sqlite", "mysql", "mariadb", "postgresql", "oracle")
+ def test_identity_is_ignored(self, dialect):
- @testing.skip_if(testing.requires.identity_columns)
- def test_identity_is_ignored(self):
t = Table(
"foo_table",
MetaData(),
Column("foo", Integer(), Identity("always", start=3)),
)
+ t_exp = Table(
+ "foo_table",
+ MetaData(),
+ Column("foo", Integer(), nullable=False),
+ )
+ dialect = self.get_dialect(dialect)
+ exp = CreateTable(t_exp).compile(dialect=dialect).string
+ self.assert_compile(
+ CreateTable(t), re.sub(r"[\n\t]", "", exp), dialect=dialect
+ )
+
+ @testing.combinations(
+ "sqlite",
+ "mysql",
+ "mariadb",
+ "postgresql",
+ "oracle",
+ argnames="dialect",
+ )
+ @testing.combinations(True, "auto", argnames="autoincrement")
+ def test_identity_is_ignored_in_pk(self, dialect, autoincrement):
+ t = Table(
+ "foo_table",
+ MetaData(),
+ Column(
+ "foo",
+ Integer(),
+ Identity("always", start=3),
+ primary_key=True,
+ autoincrement=autoincrement,
+ ),
+ )
+ t_exp = Table(
+ "foo_table",
+ MetaData(),
+ Column(
+ "foo", Integer(), primary_key=True, autoincrement=autoincrement
+ ),
+ )
+ dialect = self.get_dialect(dialect)
+ exp = CreateTable(t_exp).compile(dialect=dialect).string
self.assert_compile(
- CreateTable(t), "CREATE TABLE foo_table (foo INTEGER NOT NULL)"
+ CreateTable(t), re.sub(r"[\n\t]", "", exp), dialect=dialect
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
