From: Javid Khan Date: Tue, 21 Jul 2026 05:53:51 +0000 (-0400) Subject: escape single quotes in postgresql nextval() identifier rendering X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=dbf2f9583b6767a14a5db3dc40777b6bf0c2c0df;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git escape single quotes in postgresql nextval() identifier rendering 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. Fixes: #13429 Closes: #13430 Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/13430 Pull-request-sha: 0c19bcbd85c69b492542c43a693600595cbd0916 Change-Id: I6198d491a8ead7429fb5ac70e0502a6644f06810 --- diff --git a/doc/build/changelog/unreleased_20/13429.rst b/doc/build/changelog/unreleased_20/13429.rst new file mode 100644 index 0000000000..8576c53d37 --- /dev/null +++ b/doc/build/changelog/unreleased_20/13429.rst @@ -0,0 +1,9 @@ +.. 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. diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py index 88654295d6..7816e0c910 100644 --- a/lib/sqlalchemy/dialects/postgresql/base.py +++ b/lib/sqlalchemy/dialects/postgresql/base.py @@ -2371,7 +2371,9 @@ class PGCompiler(compiler.SQLCompiler): 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 = "" @@ -3244,9 +3246,43 @@ class PGTypeCompiler(compiler.GenericTypeCompiler): 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( @@ -3448,8 +3484,8 @@ class PGExecutionContext(default.DefaultExecutionContext): 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_, ) @@ -3486,13 +3522,12 @@ class PGExecutionContext(default.DefaultExecutionContext): 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) diff --git a/test/dialect/postgresql/test_compiler.py b/test/dialect/postgresql/test_compiler.py index b2da47cd08..215e0e3a51 100644 --- a/test/dialect/postgresql/test_compiler.py +++ b/test/dialect/postgresql/test_compiler.py @@ -61,6 +61,7 @@ from sqlalchemy.dialects.postgresql import Range 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 @@ -125,6 +126,53 @@ class SequenceTest(fixtures.TestBase, AssertsCompiledSQL): 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() diff --git a/test/dialect/postgresql/test_dialect.py b/test/dialect/postgresql/test_dialect.py index 2efac75c38..cfcc40f36f 100644 --- a/test/dialect/postgresql/test_dialect.py +++ b/test/dialect/postgresql/test_dialect.py @@ -15,6 +15,7 @@ from sqlalchemy import event 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 @@ -40,6 +41,7 @@ from sqlalchemy.dialects.postgresql.psycopg2 import ( 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 @@ -1358,13 +1360,26 @@ $$ LANGUAGE plpgsql; ).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")