--- /dev/null
+.. change::
+ :tags: bug, postgresql
+ :tickets: 13429
+
+ Fixed bug in the PostgreSQL dialect where a single quote in a sequence,
+ table, or schema name, such as one supplied via a ``schema_translate_map``
+ or an explicit :class:`.Sequence`, could result in a malformed
+ ``nextval()`` statement. The quote is now properly escaped. Pull request
+ courtesy dxbjavid.
return f"power{self.function_argspec(fn)}"
def visit_sequence(self, seq, **kw):
- return "nextval('%s')" % self.preparer.format_sequence(seq)
+ return "nextval(%s)" % self.preparer._format_sequence_string_literal(
+ seq
+ )
def limit_clause(self, select, **kw):
text = ""
return "JSONPATH"
+class _CompilerSequence:
+ """Minimal stand-in for :class:`.Sequence`.
+
+ Used for the implicit sequence behind a SERIAL column, where no
+ :class:`.Sequence` object exists but a name still has to be rendered
+ through :meth:`.IdentifierPreparer.format_sequence`.
+
+ """
+
+ __slots__ = ("name", "schema")
+
+ # the schema handed to us has already been resolved against the
+ # schema translate map, so don't let format_sequence() translate it
+ # a second time
+ _use_schema_map = False
+
+ def __init__(self, name, schema=None):
+ self.name = name
+ self.schema = schema
+
+
class PGIdentifierPreparer(compiler.IdentifierPreparer):
reserved_words = RESERVED_WORDS
+ def _format_sequence_string_literal(self, sequence):
+ """Render a sequence name as a SQL string literal.
+
+ ``nextval()`` and friends take the sequence as a string rather than
+ as an identifier, so the quoted identifier produced by
+ :meth:`.format_sequence` has to be escaped a second time for the
+ enclosing literal.
+
+ """
+ return "'%s'" % self.format_sequence(
+ sequence, use_schema=True
+ ).replace("'", "''")
+
def _unquote_identifier(self, value):
if value[0] == self.initial_quote:
value = value[1:-1].replace(
def fire_sequence(self, seq, type_):
return self._execute_scalar(
(
- "select nextval('%s')"
- % self.identifier_preparer.format_sequence(seq)
+ "select nextval(%s)"
+ % self.identifier_preparer._format_sequence_string_literal(seq)
),
type_,
)
else:
effective_schema = None
- if effective_schema is not None:
- exc = 'select nextval(\'"%s"."%s"\')' % (
- effective_schema,
- seq_name,
+ exc = (
+ "select nextval(%s)"
+ % self.identifier_preparer._format_sequence_string_literal(
+ _CompilerSequence(seq_name, effective_schema)
)
- else:
- exc = "select nextval('\"%s\"')" % (seq_name,)
+ )
return self._execute_scalar(exc, column.type)
from sqlalchemy.dialects.postgresql import REGCONFIG
from sqlalchemy.dialects.postgresql import TSQUERY
from sqlalchemy.dialects.postgresql import TSRANGE
+from sqlalchemy.dialects.postgresql.base import _CompilerSequence
from sqlalchemy.dialects.postgresql.base import PGDialect
from sqlalchemy.dialects.postgresql.psycopg2 import PGDialect_psycopg2
from sqlalchemy.dialects.postgresql.ranges import MultiRange
dialect=postgresql.dialect(),
)
+ def test_nextval_quote_escaping(self):
+ """test for #13429"""
+
+ # a single quote in a sequence or schema name must be escaped so it
+ # can't terminate the nextval() string literal early
+ self.assert_compile(
+ Sequence("some'seq").next_value(),
+ "nextval('\"some''seq\"')",
+ dialect=postgresql.dialect(),
+ )
+ self.assert_compile(
+ Sequence("some'seq", schema="my'schema").next_value(),
+ "nextval('\"my''schema\".\"some''seq\"')",
+ dialect=postgresql.dialect(),
+ )
+
+ @testing.combinations(
+ ("my_seq", None, "'my_seq'"),
+ ("my_seq", "my_schema", "'my_schema.my_seq'"),
+ ("My_Seq", "Some_Schema", '\'"Some_Schema"."My_Seq"\''),
+ ("some'seq", "my'schema", "'\"my''schema\".\"some''seq\"'"),
+ )
+ def test_format_sequence_string_literal(self, name, schema_, expected):
+ """test for #13429"""
+
+ preparer = postgresql.dialect().identifier_preparer
+ eq_(
+ preparer._format_sequence_string_literal(
+ _CompilerSequence(name, schema_)
+ ),
+ expected,
+ )
+
+ def test_compiler_sequence_ignores_schema_translate(self):
+ """test for #13429"""
+
+ # the schema is resolved by the execution context before it reaches
+ # _CompilerSequence, so it must not be translated a second time
+ preparer = postgresql.dialect().identifier_preparer
+ preparer = preparer._with_schema_translate({"bar": "foo"})
+ eq_(
+ preparer._format_sequence_string_literal(
+ _CompilerSequence("my_seq", "bar")
+ ),
+ "'bar.my_seq'",
+ )
+
class CompileTest(fixtures.TestBase, AssertsCompiledSQL):
__dialect__ = postgresql.dialect()
from sqlalchemy import exc
from sqlalchemy import extract
from sqlalchemy import func
+from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy import literal
from sqlalchemy import MetaData
EXECUTEMANY_VALUES_PLUS_BATCH,
)
from sqlalchemy.engine import url
+from sqlalchemy.schema import CreateSequence
from sqlalchemy.sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
from sqlalchemy.testing import config
from sqlalchemy.testing import engines
).scalar()
eq_(r, exp)
- @testing.provide_metadata
- def test_checksfor_sequence(self, connection):
- meta1 = self.metadata
- seq = Sequence("fooseq")
- t = Table("mytable", meta1, Column("col1", Integer, seq))
+ @testing.combinations(
+ ("fooseq", None),
+ ("foo'seq", None),
+ ("fooseq", config.test_schema),
+ ("foo'seq", config.test_schema),
+ argnames="seqname,schema",
+ )
+ def test_checksfor_sequence(self, connection, metadata, seqname, schema):
+ """an old sequence test updated with new test names for #13429"""
+ meta1 = metadata
+ seq = Sequence(seqname, schema=schema)
+ t = Table(
+ "mytable", meta1, Column("col1", Integer, seq), schema=schema
+ )
seq.drop(connection)
- connection.execute(text("CREATE SEQUENCE fooseq"))
+ assert seqname not in inspect(connection).get_sequence_names(
+ schema=schema
+ )
+ connection.execute(CreateSequence(seq))
+ assert seqname in inspect(connection).get_sequence_names(schema=schema)
t.create(connection, checkfirst=True)
@testing.combinations(True, False, argnames="implicit_returning")
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
