Table(
"some_table",
metadata,
- Column("id", Integer, Sequence("some_seq", optional=True), primary_key=True),
+ Column(
+ "id", Integer, Sequence("some_seq", start=1, optional=True), primary_key=True
+ ),
)
The above :class:`.Sequence` is only used for DDL and INSERT statements if the
--- /dev/null
+.. change::
+ :tags: schema, mssql
+ :tickets: 7211
+
+ The :class:`.Sequence` construct restores itself to the DDL behavior it
+ had prior to the 1.4 series, where creating a :class:`.Sequence` with
+ no additional arguments will emit a simple ``CREATE SEQUENCE`` instruction
+ **without** any additional parameters for "start value". For most backends,
+ this is how things worked previously in any case; **however**, for
+ MS SQL Server, the default value on this database is
+ ``-2**63``; to prevent this generally impractical default
+ from taking effect on SQL Server, the :paramref:`.Sequence.start` parameter
+ should be provided. As usage of :class:`.Sequence` is unusual
+ for SQL Server which for many years has standardized on ``IDENTITY``,
+ it is hoped that this change has minimal impact.
+
+ .. seealso::
+
+ :ref:`change_7211`
:ref:`postgresql_psycopg`
+
.. _ticket_8054:
Dialect support for oracledb
:ticket:`8567`
+.. _change_7211:
+
+The ``Sequence`` construct reverts to not having any explicit default "start" value; impacts MS SQL Server
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Prior to SQLAlchemy 1.4, the :class:`.Sequence` construct would emit only
+simple ``CREATE SEQUENCE`` DDL, if no additional arguments were specified::
+
+ >>> # SQLAlchemy 1.3 (and 2.0)
+ >>> from sqlalchemy import Sequence
+ >>> from sqlalchemy.schema import CreateSequence
+ >>> print(CreateSequence(Sequence("my_seq")))
+ CREATE SEQUENCE my_seq
+
+However, as :class:`.Sequence` support was added for MS SQL Server, where the
+default start value is inconveniently set to ``-2**63``,
+version 1.4 decided to default the DDL to emit a start value of 1, if
+:paramref:`.Sequence.start` were not otherwise provided::
+
+ >>> # SQLAlchemy 1.4 (only)
+ >>> from sqlalchemy import Sequence
+ >>> from sqlalchemy.schema import CreateSequence
+ >>> print(CreateSequence(Sequence("my_seq")))
+ CREATE SEQUENCE my_seq START WITH 1
+
+This change has introduced other complexities, including that when
+the :paramref:`.Sequence.min_value` parameter is included, this default of
+``1`` should in fact default to what :paramref:`.Sequence.min_value`
+states, else a min_value that's below the start_value may be seen as
+contradictory. As looking at this issue started to become a bit of a
+rabbit hole of other various edge cases, we decided to instead revert this
+change and restore the original behavior of :class:`.Sequence` which is
+to have no opinion, and just emit CREATE SEQUENCE, allowing the database
+itself to make its decisions on how the various parameters of ``SEQUENCE``
+should interact with each other.
+
+Therefore, to ensure that the start value is 1 on all backends,
+**the start value of 1 may be indicated explicitly**, as below::
+
+ >>> # All SQLAlchemy versions
+ >>> from sqlalchemy import Sequence
+ >>> from sqlalchemy.schema import CreateSequence
+ >>> print(CreateSequence(Sequence("my_seq", start=1)))
+ CREATE SEQUENCE my_seq START WITH 1
+
+Beyond all of that, for autogeneration of integer primary keys on modern
+backends including PostgreSQL, Oracle, SQL Server, the :class:`.Identity`
+construct should be preferred, which also works the same way in 1.4 and 2.0
+with no changes in behavior.
+
+
+:ticket:`7211`
+
+
.. _change_6980:
"with_variant()" clones the original TypeEngine rather than changing the type
passing it directly to a SQL execution method::
with my_engine.connect() as conn:
- seq = Sequence("some_sequence")
+ seq = Sequence("some_sequence", start=1)
nextid = conn.execute(seq)
In order to embed the "next value" function of a :class:`.Sequence`
method, which will render at statement compilation time a SQL function that is
appropriate for the target backend::
- >>> my_seq = Sequence("some_sequence")
+ >>> my_seq = Sequence("some_sequence", start=1)
>>> stmt = select(my_seq.next_value())
>>> print(stmt.compile(dialect=postgresql.dialect()))
SELECT nextval('some_sequence') AS next_value_1
* The :class:`.Sequence` will inherit the :paramref:`_schema.MetaData.schema`
parameter specified to the target :class:`_schema.MetaData`, which
- affects the production of CREATE / DROP DDL, if any.
-
+ affects the production of CREATE / DROP DDL as well as how the
+ :meth:`.Sequence.next_value` function is rendered in SQL statements.
* The :meth:`_schema.MetaData.create_all` and :meth:`_schema.MetaData.drop_all`
methods will emit CREATE / DROP for this :class:`.Sequence`,
:class:`_schema.Column` as the **Python side default generator**::
Column(
- "cart_id", Integer, Sequence("cart_id_seq", metadata=metadata_obj), primary_key=True
+ "cart_id",
+ Integer,
+ Sequence("cart_id_seq", metadata=metadata_obj, start=1),
+ primary_key=True,
)
In the above case, the :class:`.Sequence` will automatically be subject
:class:`_schema.Column` both as the Python-side default generator as well as
the server-side default generator::
- cart_id_seq = Sequence("cart_id_seq", metadata=metadata_obj)
+ cart_id_seq = Sequence("cart_id_seq", metadata=metadata_obj, start=1)
table = Table(
"cartitems",
metadata_obj,
class CartItem(Base):
__tablename__ = "cartitems"
- cart_id_seq = Sequence("cart_id_seq", metadata=Base.metadata)
+ cart_id_seq = Sequence("cart_id_seq", metadata=Base.metadata, start=1)
cart_id = Column(
Integer, cart_id_seq, server_default=cart_id_seq.next_value(), primary_key=True
)
class MyOracleModel(Base):
__tablename__ = "my_table"
- id = mapped_column(Integer, Sequence("my_sequence"), primary_key=True)
+ id = mapped_column(Integer, Sequence("my_sequence", start=1), primary_key=True)
data = mapped_column(String(50))
The INSERT for a model as above on Oracle looks like:
SEQUENCE support
----------------
-The :class:`.Sequence` object now creates "real" sequences, i.e.,
-``CREATE SEQUENCE``. To provide compatibility with other dialects,
-:class:`.Sequence` defaults to a start value of 1, even though the
-T-SQL defaults is -9223372036854775808.
+The :class:`.Sequence` object creates "real" sequences, i.e.,
+``CREATE SEQUENCE``::
-.. versionadded:: 1.4.0
+ >>> from sqlalchemy import Sequence
+ >>> from sqlalchemy.schema import CreateSequence
+ >>> from sqlalchemy.dialects import mssql
+ >>> print(CreateSequence(Sequence("my_seq", start=1)).compile(dialect=mssql.dialect()))
+ CREATE SEQUENCE my_seq START WITH 1
+
+For integer primary key generation, SQL Server's ``IDENTITY`` construct should
+generally be preferred vs. sequence.
+
+..tip::
+
+ The default start value for T-SQL is ``-2**63`` instead of 1 as
+ in most other SQL databases. Users should explicitly set the
+ :paramref:`.Sequence.start` to 1 if that's the expected default::
+
+ seq = Sequence("my_sequence", start=1)
+
+.. versionadded:: 1.4 added SQL Server support for :class:`.Sequence`
+
+.. versionchanged:: 2.0 The SQL Server dialect will no longer implicitly
+ render "START WITH 1" for ``CREATE SEQUENCE``, which was the behavior
+ first implemented in version 1.4.
MAX on VARCHAR / NVARCHAR
-------------------------
supports_sequences = True
sequences_optional = True
- # T-SQL's actual default is -9223372036854775808
+ # This is actually used for autoincrement, where itentity is used that
+ # starts with 1.
+ # for sequences T-SQL's actual default is -9223372036854775808
default_sequence_base = 1
supports_native_boolean = False
from ...testing.provision import drop_db
from ...testing.provision import get_temp_table_name
from ...testing.provision import log
+from ...testing.provision import normalize_sequence
from ...testing.provision import run_reap_dbs
from ...testing.provision import temp_table_keyword_args
)
)
)
+
+
+@normalize_sequence.for_db("mssql")
+def normalize_sequence(cfg, sequence):
+ if sequence.start is None:
+ sequence.start = 1
+ return sequence
to a Column construct::
t = Table('mytable', metadata,
- Column('id', Integer, Sequence('id_seq'), primary_key=True),
+ Column('id', Integer, Sequence('id_seq', start=1), primary_key=True),
Column(...), ...
)
This step is also required when using table reflection, i.e. autoload_with=engine::
t = Table('mytable', metadata,
- Column('id', Integer, Sequence('id_seq'), primary_key=True),
+ Column('id', Integer, Sequence('id_seq', start=1), primary_key=True),
autoload_with=engine
)
To specify a specific named sequence to be used for primary key generation,
use the :func:`~sqlalchemy.schema.Sequence` construct::
- Table('sometable', metadata,
- Column('id', Integer, Sequence('some_id_seq'), primary_key=True)
+ Table(
+ "sometable",
+ metadata,
+ Column(
+ "id", Integer, Sequence("some_id_seq", start=1), primary_key=True
)
+ )
When SQLAlchemy issues a single INSERT statement, to fulfill the contract of
having the "last insert identifier" available, a RETURNING clause is added to
include_set_input_sizes: Optional[Set[Any]] = None
exclude_set_input_sizes: Optional[Set[Any]] = None
- # the first value we'd get for an autoincrement
- # column.
+ # the first value we'd get for an autoincrement column.
default_sequence_base = 1
# most DBAPIs happy with this for execute().
if prefix:
text += prefix
- if create.element.start is None:
- create.element.start = self.dialect.default_sequence_base
options = self.get_identity_options(create.element)
if options:
text += " " + options
some_table = Table(
'some_table', metadata,
- Column('id', Integer, Sequence('some_table_seq'),
+ Column('id', Integer, Sequence('some_table_seq', start=1),
primary_key=True)
)
:param start: the starting index of the sequence. This value is
used when the CREATE SEQUENCE command is emitted to the database
- as the value of the "START WITH" clause. If ``None``, the
+ as the value of the "START WITH" clause. If ``None``, the
clause is omitted, which on most platforms indicates a starting
value of 1.
+
+ .. versionchanged:: 2.0 The :paramref:`.Sequence.start` parameter
+ is required in order to have DDL emit "START WITH". This is a
+ reversal of a change made in version 1.4 which would implicitly
+ render "START WITH 1" if the :paramref:`.Sequence.start` were
+ not included. See :ref:`change_7211` for more detail.
+
:param increment: the increment value of the sequence. This
value is used when the CREATE SEQUENCE command is emitted to
the database as the value of the "INCREMENT BY" clause. If ``None``,
raise NotImplementedError(
f"backend does not include an upsert implementation: {cfg.db.url}"
)
+
+
+@register.init
+def normalize_sequence(cfg, sequence):
+ """Normalize sequence parameters for dialect that don't start with 1
+ by default.
+
+ The default implementation does nothing
+ """
+ return sequence
from ..assertions import eq_
from ..assertions import is_true
from ..config import requirements
+from ..provision import normalize_sequence
from ..schema import Column
from ..schema import Table
from ... import inspect
Column(
"id",
Integer,
- Sequence("tab_id_seq"),
+ normalize_sequence(config, Sequence("tab_id_seq")),
primary_key=True,
),
Column("data", String(50)),
Column(
"id",
Integer,
- Sequence("tab_id_seq", data_type=Integer, optional=True),
+ normalize_sequence(
+ config,
+ Sequence("tab_id_seq", data_type=Integer, optional=True),
+ ),
primary_key=True,
),
Column("data", String(50)),
Column(
"id",
Integer,
- Sequence("noret_id_seq"),
+ normalize_sequence(config, Sequence("noret_id_seq")),
primary_key=True,
),
Column("data", String(50)),
Column(
"id",
Integer,
- Sequence("noret_sch_id_seq", schema=config.test_schema),
+ normalize_sequence(
+ config,
+ Sequence(
+ "noret_sch_id_seq", schema=config.test_schema
+ ),
+ ),
primary_key=True,
),
Column("data", String(50)),
Column(
"id",
Integer,
- Sequence("noret_sch_id_seq", schema="alt_schema"),
+ normalize_sequence(
+ config, Sequence("noret_sch_id_seq", schema="alt_schema")
+ ),
primary_key=True,
),
Column("data", String(50)),
@testing.requires.schemas
def test_nextval_direct_schema_translate(self, connection):
- seq = Sequence("noret_sch_id_seq", schema="alt_schema")
+ seq = normalize_sequence(
+ config, Sequence("noret_sch_id_seq", schema="alt_schema")
+ )
connection = connection.execution_options(
schema_translate_map={"alt_schema": config.test_schema}
)
table = Table(
"x",
MetaData(),
- Column("y", Integer, Sequence("y_seq")),
+ Column(
+ "y", Integer, normalize_sequence(config, Sequence("y_seq"))
+ ),
Column("q", Integer),
)
seq_nextval = connection.dialect.statement_compiler(
statement=None, dialect=connection.dialect
- ).visit_sequence(Sequence("y_seq"))
+ ).visit_sequence(normalize_sequence(config, Sequence("y_seq")))
self.assert_compile(
stmt,
"INSERT INTO x (y, q) VALUES (%s, 5)" % (seq_nextval,),
@classmethod
def define_tables(cls, metadata):
- Sequence("user_id_seq", metadata=metadata)
- Sequence(
- "other_seq", metadata=metadata, nomaxvalue=True, nominvalue=True
+ normalize_sequence(config, Sequence("user_id_seq", metadata=metadata))
+ normalize_sequence(
+ config,
+ Sequence(
+ "other_seq",
+ metadata=metadata,
+ nomaxvalue=True,
+ nominvalue=True,
+ ),
)
if testing.requires.schemas.enabled:
- Sequence(
- "user_id_seq", schema=config.test_schema, metadata=metadata
+ normalize_sequence(
+ config,
+ Sequence(
+ "user_id_seq", schema=config.test_schema, metadata=metadata
+ ),
)
- Sequence(
- "schema_seq", schema=config.test_schema, metadata=metadata
+ normalize_sequence(
+ config,
+ Sequence(
+ "schema_seq", schema=config.test_schema, metadata=metadata
+ ),
)
Table(
"user_id_table",
def test_has_sequence_cache(self, connection, metadata):
insp = inspect(connection)
eq_(insp.has_sequence("user_id_seq"), True)
- ss = Sequence("new_seq", metadata=metadata)
+ ss = normalize_sequence(config, Sequence("new_seq", metadata=metadata))
eq_(insp.has_sequence("new_seq"), False)
ss.create(connection)
try:
from sqlalchemy import Sequence
from sqlalchemy import String
from sqlalchemy import Table
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
+from sqlalchemy.testing.provision import normalize_sequence
class SequenceTest(fixtures.TablesTest):
Column(
"id", Integer, default=Sequence("int_seq", data_type=Integer())
),
+ Column(
+ "id_provision",
+ Integer,
+ default=normalize_sequence(
+ config, Sequence("id_provision", data_type=Integer())
+ ),
+ ),
+ Column(
+ "id_start",
+ Integer,
+ default=Sequence("id_start", data_type=Integer(), start=42),
+ ),
Column("txt", String(50)),
)
def test_int_seq(self, connection):
t = self.tables.int_seq_t
connection.execute(t.insert().values({"txt": "int_seq test"}))
- result = connection.scalar(select(t.c.id))
- eq_(result, 1)
+ result = connection.execute(select(t)).first()
+ eq_(result.id, -(2**31))
+ eq_(result.id_provision, 1)
+ eq_(result.id_start, 42)
def test_bigint_seq(self, connection):
t = self.tables.bigint_seq_t
ddl.CreateSequence(
Sequence("my_seq", nomaxvalue=True, nominvalue=True)
),
- "CREATE SEQUENCE my_seq START WITH 1 NOMINVALUE NOMAXVALUE",
+ "CREATE SEQUENCE my_seq NOMINVALUE NOMAXVALUE",
dialect=oracle.OracleDialect(),
)
(SmallInteger, "AS SMALLINT "),
(BigInteger, "AS BIGINT "),
)
- def test_create_index_concurrently(self, type_, text):
+ def test_compile_type(self, type_, text):
s = Sequence("s1", data_type=type_)
self.assert_compile(
schema.CreateSequence(s),
- "CREATE SEQUENCE s1 %sSTART WITH 1" % text,
+ f"CREATE SEQUENCE s1 {text}".strip(),
dialect=postgresql.dialect(),
)
from sqlalchemy.schema import ForeignKeyConstraint
from sqlalchemy.schema import Sequence
from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import config
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
@testing.fixture
def produce_subject(self):
- return Sequence("my_seq")
+ return normalize_sequence(config, Sequence("my_seq"))
@testing.fixture
def produce_table_integrated_subject(self, metadata, produce_subject):
from sqlalchemy.testing import is_not
from sqlalchemy.testing import is_true
from sqlalchemy.testing.assertsql import CompiledSQL
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import gc_collect
Column(
"pk",
Integer,
- Sequence("testtable_pk_seq"),
+ normalize_sequence(config, Sequence("testtable_pk_seq")),
primary_key=True,
),
)
Column(
"x",
Integer,
- Sequence("t_id_seq"),
+ normalize_sequence(config, Sequence("t_id_seq")),
primary_key=True,
),
implicit_returning=False,
from sqlalchemy.testing import mock
from sqlalchemy.testing import not_in
from sqlalchemy.testing import skip
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
Column(
"user_id",
sa.Integer,
- sa.Sequence("user_id_seq", optional=True),
+ normalize_sequence(
+ config, sa.Sequence("user_id_seq", optional=True)
+ ),
primary_key=True,
),
Column("user_name", sa.String(40)),
Column(
"address_id",
sa.Integer,
- sa.Sequence("address_id_seq", optional=True),
+ normalize_sequence(
+ config, sa.Sequence("address_id_seq", optional=True)
+ ),
primary_key=True,
),
Column("user_id", sa.Integer, sa.ForeignKey("users.user_id")),
Column(
"order_id",
sa.Integer,
- sa.Sequence("order_id_seq", optional=True),
+ normalize_sequence(
+ config, sa.Sequence("order_id_seq", optional=True)
+ ),
primary_key=True,
),
Column("user_id", sa.Integer, sa.ForeignKey("users.user_id")),
Column(
"item_id",
sa.INT,
- sa.Sequence("items_id_seq", optional=True),
+ normalize_sequence(
+ config, sa.Sequence("items_id_seq", optional=True)
+ ),
primary_key=True,
),
Column("order_id", sa.INT, sa.ForeignKey("orders")),
Column(
"id",
sa.Integer,
- sa.Sequence(cname + "_id_seq"),
+ normalize_sequence(config, sa.Sequence(cname + "_id_seq")),
primary_key=True,
),
Column(cname, Integer),
from sqlalchemy.orm import Session
from sqlalchemy.orm import sessionmaker
from sqlalchemy.testing import async_test
+from sqlalchemy.testing import config
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import expect_deprecated
from sqlalchemy.testing.assertions import is_false
+from sqlalchemy.testing.provision import normalize_sequence
from .test_engine_py3k import AsyncFixture as _AsyncFixture
from ...orm import _fixtures
async def test_sequence_execute(
self, async_session: AsyncSession, metadata, use_scalar
):
- seq = Sequence("some_sequence", metadata=metadata)
+ seq = normalize_sequence(
+ config, Sequence("some_sequence", metadata=metadata)
+ )
sync_connection = (await async_session.connection()).sync_connection
from sqlalchemy.orm.interfaces import MANYTOONE
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import AssertsExecutionResults
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.fixtures import ComparableEntity
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
Column(
"person_id",
Integer,
- Sequence("person_id_seq", optional=True),
+ normalize_sequence(
+ config, Sequence("person_id_seq", optional=True)
+ ),
primary_key=True,
),
Column(
from sqlalchemy import Table
from sqlalchemy.orm import class_mapper
from sqlalchemy.orm import relationship
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.provision import normalize_sequence
class InheritTest(fixtures.MappedTest):
Column(
"principal_id",
Integer,
- Sequence("principal_id_seq", optional=False),
+ normalize_sequence(
+ config, Sequence("principal_id_seq", optional=False)
+ ),
primary_key=True,
),
Column("name", String(50), nullable=False),
Column(
"id",
Integer,
- Sequence("foo_id_seq", optional=True),
+ normalize_sequence(
+ config, Sequence("foo_id_seq", optional=True)
+ ),
primary_key=True,
),
Column("data", String(20)),
Column(
"id",
Integer,
- Sequence("foo_seq", optional=True),
+ normalize_sequence(config, Sequence("foo_seq", optional=True)),
primary_key=True,
),
Column("data", String(20)),
from sqlalchemy import Integer
from sqlalchemy import testing
from sqlalchemy.orm import relationship
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from test.orm import _fixtures
Table(
"foo",
metadata,
- Column("id", Integer, sa.Sequence("foo_id_seq"), primary_key=True),
+ Column(
+ "id",
+ Integer,
+ normalize_sequence(config, sa.Sequence("foo_id_seq")),
+ primary_key=True,
+ ),
Column("bar", Integer),
Column("range", Integer),
)
from sqlalchemy.testing import mock
from sqlalchemy.testing import pickleable
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import gc_collect
def test_sequence_execute(
self, connection, metadata, add_do_orm_execute_event, use_scalar
):
- seq = Sequence("some_sequence", metadata=metadata)
+ seq = normalize_sequence(
+ config, Sequence("some_sequence", metadata=metadata)
+ )
metadata.create_all(connection)
sess = Session(connection)
from sqlalchemy.orm.persistence import _sort_states
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_true
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.assertsql import Conditional
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.util import OrderedDict
Column(
"secondary_id",
Integer,
- sa.Sequence("sec_id_seq"),
+ normalize_sequence(config, sa.Sequence("sec_id_seq")),
unique=True,
)
)
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.assertsql import Conditional
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from test.orm import _fixtures
@testing.requires.insert_returning
def test_b(self, base, run_test):
- seq = Sequence("x_seq")
+ seq = normalize_sequence(config, Sequence("x_seq"))
class A(base):
__tablename__ = "a"
self.assert_compile(
schema.CreateSequence(s1),
- "CREATE SEQUENCE __[SCHEMA__none].s1 START WITH 1",
+ "CREATE SEQUENCE __[SCHEMA__none].s1",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
schema.CreateSequence(s2),
- "CREATE SEQUENCE __[SCHEMA_foo].s2 START WITH 1",
+ "CREATE SEQUENCE __[SCHEMA_foo].s2",
schema_translate_map=schema_translate_map,
)
self.assert_compile(
schema.CreateSequence(s3),
- "CREATE SEQUENCE __[SCHEMA_bar].s3 START WITH 1",
+ "CREATE SEQUENCE __[SCHEMA_bar].s3",
schema_translate_map=schema_translate_map,
)
from sqlalchemy.sql import text
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import expect_deprecated
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.types import TypeDecorator
Column(
"id",
Integer,
- Sequence("ai_id_seq", optional=True),
+ normalize_sequence(
+ config, Sequence("ai_id_seq", optional=True)
+ ),
primary_key=True,
),
Column("int1", Integer),
# and autoincrement=False. Using a ForeignKey
# would have the same effect
- some_seq = Sequence("some_seq")
+ some_seq = normalize_sequence(config, Sequence("some_seq"))
dataset_no_autoinc = Table(
"x",
self._run_test(default=literal_column("1", type_=self.MyInteger))
def test_sequence(self):
- self._run_test(Sequence("foo_seq"))
+ self._run_test(normalize_sequence(config, Sequence("foo_seq")))
def test_text_clause_default_no_type(self):
self._run_test(default=text("1"))
from sqlalchemy.sql.selectable import SelectStatementGrouping
from sqlalchemy.testing import assertions
from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_true
from sqlalchemy.testing import mock
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
Column(
"id",
Integer,
- Sequence("ai_id_seq", optional=True),
+ normalize_sequence(
+ config, Sequence("ai_id_seq", optional=True)
+ ),
primary_key=True,
),
Column("int1", Integer),
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing.assertions import expect_warnings
from sqlalchemy.testing.engines import all_dialects
+from sqlalchemy.testing.provision import normalize_sequence
table1 = table(
Column(
"id",
Integer,
- Sequence("t1idseq", optional=True),
+ normalize_sequence(config, Sequence("t1idseq", optional=True)),
primary_key=True,
),
Column("value", Integer),
Column(
"id",
Integer,
- Sequence("t2idseq", optional=True),
+ normalize_sequence(config, Sequence("t2idseq", optional=True)),
primary_key=True,
),
Column("value", Integer, default=7),
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
+from sqlalchemy.testing.provision import normalize_sequence
class ORMExpr:
)
def test_insert_literal_binds_sequence_notimplemented(self):
- table = Table("x", MetaData(), Column("y", Integer, Sequence("y_seq")))
+ table = Table(
+ "x",
+ MetaData(),
+ Column(
+ "y", Integer, normalize_sequence(config, Sequence("y_seq"))
+ ),
+ )
dialect = default.DefaultDialect()
dialect.supports_sequences = True
t1 = Table(
"t",
m,
- Column("id", Integer, Sequence("id_seq"), primary_key=True),
+ Column(
+ "id",
+ Integer,
+ normalize_sequence(config, Sequence("id_seq")),
+ primary_key=True,
+ ),
Column("data", String),
)
t1 = Table(
"t",
m,
- Column("id", Integer, Sequence("id_seq"), primary_key=True),
+ Column(
+ "id",
+ Integer,
+ normalize_sequence(config, Sequence("id_seq")),
+ primary_key=True,
+ ),
Column("data", String),
)
"t",
m,
Column("id", Integer, primary_key=True),
- Column("counter", Sequence("counter_seq")),
+ Column(
+ "counter", normalize_sequence(config, Sequence("counter_seq"))
+ ),
Column("data", String),
)
t1 = Table(
"t",
m,
- Column("id", Integer, Sequence("id_seq"), primary_key=True),
+ Column(
+ "id",
+ Integer,
+ normalize_sequence(config, Sequence("id_seq")),
+ primary_key=True,
+ ),
Column("data", String),
)
from sqlalchemy import VARCHAR
from sqlalchemy.engine import cursor as _cursor
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import mock
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
Column(
"id",
Integer,
- Sequence("t4_id_seq", optional=True),
+ normalize_sequence(
+ config, Sequence("t4_id_seq", optional=True)
+ ),
primary_key=True,
),
Column("foo", String(30), primary_key=True),
Column(
"id",
Integer,
- Sequence("t4_id_seq"),
+ normalize_sequence(config, Sequence("t4_id_seq")),
primary_key=True,
),
Column("foo", String(30)),
Column(
"id",
Integer,
- Sequence("t_id_seq"),
+ normalize_sequence(config, Sequence("t_id_seq")),
primary_key=True,
),
Column("data", String(50)),
self._test(
connection,
t.insert().values(
- id=func.next_value(Sequence("t_id_seq")), data="data", x=5
+ id=func.next_value(
+ normalize_sequence(config, Sequence("t_id_seq"))
+ ),
+ data="data",
+ x=5,
),
(testing.db.dialect.default_sequence_base, "data", 5),
)
@classmethod
def define_tables(cls, metadata):
- seq = Sequence("tid_seq")
+ seq = provision.normalize_sequence(config, Sequence("tid_seq"))
Table(
"returning_tbl",
metadata,
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import AssertsExecutionResults
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import in_
from sqlalchemy.testing import is_not
from sqlalchemy.testing import ne_
from sqlalchemy.testing.assertions import expect_raises_message
+from sqlalchemy.testing.provision import normalize_sequence
metadata = MetaData()
Column(
"person_id",
Integer,
- Sequence("person_id_seq", optional=True),
+ normalize_sequence(
+ config, Sequence("person_id_seq", optional=True)
+ ),
primary_key=True,
),
Column("name", String(50)),
from sqlalchemy.schema import DropSequence
from sqlalchemy.sql import select
from sqlalchemy.testing import assert_raises_message
+from sqlalchemy.testing import config
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_false
from sqlalchemy.testing.assertsql import AllOf
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.assertsql import EachOf
+from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
__dialect__ = "default"
__backend__ = True
- def test_create_drop_ddl(self):
- self.assert_compile(
- CreateSequence(Sequence("foo_seq")),
- "CREATE SEQUENCE foo_seq START WITH 1",
- )
-
- self.assert_compile(
- CreateSequence(Sequence("foo_seq", start=5)),
- "CREATE SEQUENCE foo_seq START WITH 5",
- )
-
- self.assert_compile(
- CreateSequence(Sequence("foo_seq", increment=2)),
- "CREATE SEQUENCE foo_seq INCREMENT BY 2 START WITH 1",
- )
-
- self.assert_compile(
- CreateSequence(Sequence("foo_seq", increment=2, start=5)),
- "CREATE SEQUENCE foo_seq INCREMENT BY 2 START WITH 5",
- )
-
- self.assert_compile(
- CreateSequence(
- Sequence("foo_seq", increment=2, start=0, minvalue=0)
- ),
- "CREATE SEQUENCE foo_seq INCREMENT BY 2 START WITH 0 MINVALUE 0",
- )
-
- self.assert_compile(
- CreateSequence(
- Sequence("foo_seq", increment=2, start=1, maxvalue=5)
- ),
- "CREATE SEQUENCE foo_seq INCREMENT BY 2 START WITH 1 MAXVALUE 5",
- )
-
- self.assert_compile(
- CreateSequence(
- Sequence("foo_seq", increment=2, start=1, nomaxvalue=True)
- ),
- "CREATE SEQUENCE foo_seq INCREMENT BY 2 START WITH 1 NO MAXVALUE",
- )
-
- self.assert_compile(
- CreateSequence(
- Sequence("foo_seq", increment=2, start=0, nominvalue=True)
- ),
- "CREATE SEQUENCE foo_seq INCREMENT BY 2 START WITH 0 NO MINVALUE",
- )
-
- self.assert_compile(
- CreateSequence(
- Sequence("foo_seq", start=1, maxvalue=10, cycle=True)
- ),
- "CREATE SEQUENCE foo_seq START WITH 1 MAXVALUE 10 CYCLE",
- )
-
- self.assert_compile(
- CreateSequence(Sequence("foo_seq", cache=1000, order=True)),
- "CREATE SEQUENCE foo_seq START WITH 1 CACHE 1000 ORDER",
- )
-
+ @testing.combinations(
+ (Sequence("foo_seq"), ""),
+ (Sequence("foo_seq", start=5), "START WITH 5"),
+ (Sequence("foo_seq", increment=2), "INCREMENT BY 2"),
+ (
+ Sequence("foo_seq", increment=2, start=5),
+ "INCREMENT BY 2 START WITH 5",
+ ),
+ (
+ Sequence("foo_seq", increment=2, start=0, minvalue=0),
+ "INCREMENT BY 2 START WITH 0 MINVALUE 0",
+ ),
+ (
+ Sequence("foo_seq", increment=2, start=1, maxvalue=5),
+ "INCREMENT BY 2 START WITH 1 MAXVALUE 5",
+ ),
+ (
+ Sequence("foo_seq", increment=2, start=1, nomaxvalue=True),
+ "INCREMENT BY 2 START WITH 1 NO MAXVALUE",
+ ),
+ (
+ Sequence("foo_seq", increment=2, start=0, nominvalue=True),
+ "INCREMENT BY 2 START WITH 0 NO MINVALUE",
+ ),
+ (
+ Sequence("foo_seq", start=1, maxvalue=10, cycle=True),
+ "START WITH 1 MAXVALUE 10 CYCLE",
+ ),
+ (
+ Sequence("foo_seq", cache=1000, order=True),
+ "CACHE 1000 ORDER",
+ ),
+ (Sequence("foo_seq", order=True), "ORDER"),
+ (Sequence("foo_seq", minvalue=42), "MINVALUE 42"),
+ (Sequence("foo_seq", minvalue=-42), "MINVALUE -42"),
+ (
+ Sequence("foo_seq", minvalue=42, increment=2),
+ "INCREMENT BY 2 MINVALUE 42",
+ ),
+ (
+ Sequence("foo_seq", minvalue=-42, increment=2),
+ "INCREMENT BY 2 MINVALUE -42",
+ ),
+ (
+ Sequence("foo_seq", minvalue=42, increment=-2),
+ "INCREMENT BY -2 MINVALUE 42",
+ ),
+ (
+ Sequence("foo_seq", minvalue=-42, increment=-2),
+ "INCREMENT BY -2 MINVALUE -42",
+ ),
+ (Sequence("foo_seq", maxvalue=99), "MAXVALUE 99"),
+ (Sequence("foo_seq", maxvalue=-99), "MAXVALUE -99"),
+ (
+ Sequence("foo_seq", maxvalue=99, increment=2),
+ "INCREMENT BY 2 MAXVALUE 99",
+ ),
+ (
+ Sequence("foo_seq", maxvalue=99, increment=-2),
+ "INCREMENT BY -2 MAXVALUE 99",
+ ),
+ (
+ Sequence("foo_seq", maxvalue=-99, increment=-2),
+ "INCREMENT BY -2 MAXVALUE -99",
+ ),
+ (
+ Sequence("foo_seq", minvalue=42, maxvalue=99),
+ "MINVALUE 42 MAXVALUE 99",
+ ),
+ (
+ Sequence("foo_seq", minvalue=42, maxvalue=99, increment=2),
+ "INCREMENT BY 2 MINVALUE 42 MAXVALUE 99",
+ ),
+ (
+ Sequence("foo_seq", minvalue=-42, maxvalue=-9, increment=2),
+ "INCREMENT BY 2 MINVALUE -42 MAXVALUE -9",
+ ),
+ (
+ Sequence("foo_seq", minvalue=42, maxvalue=99, increment=-2),
+ "INCREMENT BY -2 MINVALUE 42 MAXVALUE 99",
+ ),
+ (
+ Sequence("foo_seq", minvalue=-42, maxvalue=-9, increment=-2),
+ "INCREMENT BY -2 MINVALUE -42 MAXVALUE -9",
+ ),
+ )
+ def test_create_ddl(self, sequence, sql):
+ before = sequence.start
self.assert_compile(
- CreateSequence(Sequence("foo_seq", order=True)),
- "CREATE SEQUENCE foo_seq START WITH 1 ORDER",
+ CreateSequence(sequence),
+ ("CREATE SEQUENCE foo_seq " + sql).strip(),
)
+ eq_(sequence.start, before)
+ def test_drop_ddl(self):
self.assert_compile(
CreateSequence(Sequence("foo_seq"), if_not_exists=True),
- "CREATE SEQUENCE IF NOT EXISTS foo_seq START WITH 1",
+ "CREATE SEQUENCE IF NOT EXISTS foo_seq",
)
self.assert_compile(
@classmethod
def setup_test_class(cls):
- cls.seq = Sequence("my_sequence")
+ cls.seq = normalize_sequence(config, Sequence("my_sequence"))
cls.seq.create(testing.db)
@classmethod
assert ret >= testing.db.dialect.default_sequence_base
def test_execute(self, connection):
- s = Sequence("my_sequence")
+ s = normalize_sequence(config, Sequence("my_sequence"))
self._assert_seq_result(connection.scalar(s))
def test_execute_deprecated(self, connection):
- s = Sequence("my_sequence", optional=True)
+ s = normalize_sequence(config, Sequence("my_sequence", optional=True))
with expect_deprecated(
r"Using the .execute\(\) method to invoke a "
"""test dialect executes a Sequence, returns nextval, whether
or not "optional" is set"""
- s = Sequence("my_sequence", optional=True)
+ s = normalize_sequence(config, Sequence("my_sequence", optional=True))
self._assert_seq_result(connection.scalar(s))
def test_execute_next_value(self, connection):
"""test func.next_value().execute()/.scalar() works
with connectionless execution."""
- s = Sequence("my_sequence")
+ s = normalize_sequence(config, Sequence("my_sequence"))
self._assert_seq_result(connection.scalar(s.next_value()))
def test_execute_optional_next_value(self, connection):
"""test func.next_value().execute()/.scalar() works
with connectionless execution."""
- s = Sequence("my_sequence", optional=True)
+ s = normalize_sequence(config, Sequence("my_sequence", optional=True))
self._assert_seq_result(connection.scalar(s.next_value()))
def test_func_embedded_select(self, connection):
"""test can use next_value() in select column expr"""
- s = Sequence("my_sequence")
+ s = normalize_sequence(config, Sequence("my_sequence"))
self._assert_seq_result(connection.scalar(select(s.next_value())))
@testing.requires.sequences_in_other_clauses
t1 = Table("t", metadata, Column("x", Integer))
t1.create(testing.db)
connection.execute(t1.insert(), [{"x": 1}, {"x": 300}, {"x": 301}])
- s = Sequence("my_sequence")
+ s = normalize_sequence(config, Sequence("my_sequence"))
eq_(
list(
connection.execute(t1.select().where(t1.c.x > s.next_value()))
Column("x", Integer),
)
t1.create(testing.db)
- s = Sequence("my_sequence")
+ s = normalize_sequence(config, Sequence("my_sequence"))
connection.execute(t1.insert().values(x=s.next_value()))
self._assert_seq_result(connection.scalar(t1.select()))
Column("x", Integer, primary_key=True),
implicit_returning=False,
)
- s = Sequence("my_sequence_here", metadata=metadata)
+ s = normalize_sequence(
+ config, Sequence("my_sequence_here", metadata=metadata)
+ )
conn = connection
t1.create(conn)
t1 = Table(
"t",
metadata,
- Column("x", Integer, Sequence("my_seq"), primary_key=True),
+ Column(
+ "x",
+ Integer,
+ normalize_sequence(config, Sequence("my_seq")),
+ primary_key=True,
+ ),
Column("data", String(50)),
implicit_returning=_implicit_returning,
)
t1 = Table(
"t",
metadata,
- Column("x", Integer, Sequence("my_seq"), primary_key=True),
+ Column(
+ "x",
+ Integer,
+ normalize_sequence(config, Sequence("my_seq")),
+ primary_key=True,
+ ),
Column("data", String(50)),
implicit_returning=_implicit_returning,
)
"""test inserted_primary_key contains the result when
pk_col=next_value(), when implicit returning is used."""
- s = Sequence("my_sequence")
+ s = normalize_sequence(config, Sequence("my_sequence"))
t1 = Table(
"t",
metadata,
(Sequence("foo_seq", increment=5),),
)
def test_start_increment(self, seq):
+ seq = normalize_sequence(config, seq)
seq.create(testing.db)
try:
with testing.db.connect() as conn:
"""test next_value() used on non-sequence platform
raises NotImplementedError."""
- s = Sequence("my_seq")
+ s = normalize_sequence(config, Sequence("my_seq"))
d = sqlite.dialect()
assert_raises_message(
NotImplementedError,
)
def test_checkfirst_sequence(self, connection):
- s = Sequence("my_sequence")
+ s = normalize_sequence(config, Sequence("my_sequence"))
s.create(connection, checkfirst=False)
assert self._has_sequence(connection, "my_sequence")
s.create(connection, checkfirst=True)
def test_checkfirst_table(self, connection):
m = MetaData()
- s = Sequence("my_sequence")
+ s = normalize_sequence(config, Sequence("my_sequence"))
t = Table("t", m, Column("c", Integer, s, primary_key=True))
t.create(connection, checkfirst=False)
assert self._has_sequence(connection, "my_sequence")
@testing.provide_metadata
def test_table_overrides_metadata_create(self, connection):
metadata = self.metadata
- Sequence("s1", metadata=metadata)
- s2 = Sequence("s2", metadata=metadata)
- s3 = Sequence("s3")
+ normalize_sequence(config, Sequence("s1", metadata=metadata))
+ s2 = normalize_sequence(config, Sequence("s2", metadata=metadata))
+ s3 = normalize_sequence(config, Sequence("s3"))
t = Table("t", metadata, Column("c", Integer, s3, primary_key=True))
assert s3.metadata is metadata
Integer,
autoincrement=True,
primary_key=True,
- default=Sequence(
- "my_sequence", metadata=self.metadata
+ default=normalize_sequence(
+ config, Sequence("my_sequence", metadata=self.metadata)
).next_value(),
),
)
@testing.provide_metadata
def test_shared_sequence(self, connection):
# test case for #6071
- common_seq = Sequence("common_sequence", metadata=self.metadata)
+ common_seq = normalize_sequence(
+ config, Sequence("common_sequence", metadata=self.metadata)
+ )
Table(
"table_1",
self.metadata,
is_false(testing.db.dialect.has_table(connection, "table_2"))
def test_next_value_type(self):
- seq = Sequence("my_sequence", data_type=BigInteger)
+ seq = normalize_sequence(
+ config, Sequence("my_sequence", data_type=BigInteger)
+ )
assert isinstance(seq.next_value().type, BigInteger)
Column(
"cart_id",
Integer,
- Sequence("cart_id_seq"),
+ normalize_sequence(config, Sequence("cart_id_seq")),
primary_key=True,
autoincrement=False,
),
Column(
"obj_id",
Integer,
- Sequence("obj_id_seq"),
+ normalize_sequence(config, Sequence("obj_id_seq")),
),
Column("name", String(128)),
Column(
"id",
Integer,
- Sequence("Manager_id_seq", optional=True),
+ normalize_sequence(
+ config, Sequence("Manager_id_seq", optional=True)
+ ),
primary_key=True,
),
implicit_returning=implicit_returning,
def define_tables(cls, metadata):
m = metadata
- s = Sequence("t_seq", metadata=m)
+ s = normalize_sequence(config, Sequence("t_seq", metadata=m))
Table(
"t_seq_test",
m,
Column("data", String(50)),
)
- s2 = Sequence("t_seq_2", metadata=m)
+ s2 = normalize_sequence(config, Sequence("t_seq_2", metadata=m))
Table(
"t_seq_test_2",
m,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
