Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# dialects/postgresql/base.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
9r"""
10.. dialect:: postgresql
11 :name: PostgreSQL
12 :normal_support: 9.6+
13 :best_effort: 9+
15.. _postgresql_sequences:
17Sequences/SERIAL/IDENTITY
18-------------------------
20PostgreSQL supports sequences, and SQLAlchemy uses these as the default means
21of creating new primary key values for integer-based primary key columns. When
22creating tables, SQLAlchemy will issue the ``SERIAL`` datatype for
23integer-based primary key columns, which generates a sequence and server side
24default corresponding to the column.
26To specify a specific named sequence to be used for primary key generation,
27use the :func:`~sqlalchemy.schema.Sequence` construct::
29 Table(
30 "sometable",
31 metadata,
32 Column(
33 "id", Integer, Sequence("some_id_seq", start=1), primary_key=True
34 ),
35 )
37When SQLAlchemy issues a single INSERT statement, to fulfill the contract of
38having the "last insert identifier" available, a RETURNING clause is added to
39the INSERT statement which specifies the primary key columns should be
40returned after the statement completes. The RETURNING functionality only takes
41place if PostgreSQL 8.2 or later is in use. As a fallback approach, the
42sequence, whether specified explicitly or implicitly via ``SERIAL``, is
43executed independently beforehand, the returned value to be used in the
44subsequent insert. Note that when an
45:func:`~sqlalchemy.sql.expression.insert()` construct is executed using
46"executemany" semantics, the "last inserted identifier" functionality does not
47apply; no RETURNING clause is emitted nor is the sequence pre-executed in this
48case.
51PostgreSQL 10 and above IDENTITY columns
52^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
54PostgreSQL 10 and above have a new IDENTITY feature that supersedes the use
55of SERIAL. The :class:`_schema.Identity` construct in a
56:class:`_schema.Column` can be used to control its behavior::
58 from sqlalchemy import Table, Column, MetaData, Integer, Computed
60 metadata = MetaData()
62 data = Table(
63 "data",
64 metadata,
65 Column(
66 "id", Integer, Identity(start=42, cycle=True), primary_key=True
67 ),
68 Column("data", String),
69 )
71The CREATE TABLE for the above :class:`_schema.Table` object would be:
73.. sourcecode:: sql
75 CREATE TABLE data (
76 id INTEGER GENERATED BY DEFAULT AS IDENTITY (START WITH 42 CYCLE),
77 data VARCHAR,
78 PRIMARY KEY (id)
79 )
81.. versionchanged:: 1.4 Added :class:`_schema.Identity` construct
82 in a :class:`_schema.Column` to specify the option of an autoincrementing
83 column.
85.. note::
87 Previous versions of SQLAlchemy did not have built-in support for rendering
88 of IDENTITY, and could use the following compilation hook to replace
89 occurrences of SERIAL with IDENTITY::
91 from sqlalchemy.schema import CreateColumn
92 from sqlalchemy.ext.compiler import compiles
95 @compiles(CreateColumn, "postgresql")
96 def use_identity(element, compiler, **kw):
97 text = compiler.visit_create_column(element, **kw)
98 text = text.replace("SERIAL", "INT GENERATED BY DEFAULT AS IDENTITY")
99 return text
101 Using the above, a table such as::
103 t = Table(
104 "t", m, Column("id", Integer, primary_key=True), Column("data", String)
105 )
107 Will generate on the backing database as:
109 .. sourcecode:: sql
111 CREATE TABLE t (
112 id INT GENERATED BY DEFAULT AS IDENTITY,
113 data VARCHAR,
114 PRIMARY KEY (id)
115 )
117.. _postgresql_monotonic_functions:
119PostgreSQL 18 and above UUID with uuidv7 as a server default
120^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
122PostgreSQL 18's ``uuidv7`` SQL function is available as any other
123SQL function using the :data:`_sql.func` namespace::
125 >>> from sqlalchemy import select, func
126 >>> print(select(func.uuidv7()))
127 SELECT uuidv7() AS uuidv7_1
129When using ``func.uuidv7()`` as a default on a :class:`.Column` using either
130Core or ORM, an extra directive ``monotonic=True`` may be passed which
131indicates this function produces monotonically increasing values; this in turn
132allows Core and ORM to use a more efficient batched form of INSERT for large
133insert operations::
135 import uuid
138 class MyClass(Base):
139 __tablename__ = "my_table"
141 id: Mapped[uuid.UUID] = mapped_column(
142 server_default=func.uuidv7(monotonic=True)
143 )
145With the above mapping, the ORM will be able to efficiently batch rows when
146running bulk insert operations using the :ref:`engine_insertmanyvalues`
147feature.
149.. versionadded:: 2.1
150 Added ``monotonic=True`` to allow functions like PostgreSQL's
151 ``uuidv7()`` to work with batched "insertmanyvalues"
153.. seealso::
155 :ref:`engine_insertmanyvalues_monotonic_functions`
158.. _postgresql_ss_cursors:
160Server Side Cursors
161-------------------
163Server-side cursor support is available for the psycopg2, asyncpg
164dialects and may also be available in others.
166Server side cursors are enabled on a per-statement basis by using the
167:paramref:`.Connection.execution_options.stream_results` connection execution
168option::
170 with engine.connect() as conn:
171 result = conn.execution_options(stream_results=True).execute(
172 text("select * from table")
173 )
175Note that some kinds of SQL statements may not be supported with
176server side cursors; generally, only SQL statements that return rows should be
177used with this option.
179.. deprecated:: 1.4 The dialect-level server_side_cursors flag is deprecated
180 and will be removed in a future release. Please use the
181 :paramref:`_engine.Connection.stream_results` execution option for
182 unbuffered cursor support.
184.. seealso::
186 :ref:`engine_stream_results`
188.. _postgresql_isolation_level:
190Transaction Isolation Level
191---------------------------
193Most SQLAlchemy dialects support setting of transaction isolation level
194using the :paramref:`_sa.create_engine.isolation_level` parameter
195at the :func:`_sa.create_engine` level, and at the :class:`_engine.Connection`
196level via the :paramref:`.Connection.execution_options.isolation_level`
197parameter.
199For PostgreSQL dialects, this feature works either by making use of the
200DBAPI-specific features, such as psycopg2's isolation level flags which will
201embed the isolation level setting inline with the ``"BEGIN"`` statement, or for
202DBAPIs with no direct support by emitting ``SET SESSION CHARACTERISTICS AS
203TRANSACTION ISOLATION LEVEL <level>`` ahead of the ``"BEGIN"`` statement
204emitted by the DBAPI. For the special AUTOCOMMIT isolation level,
205DBAPI-specific techniques are used which is typically an ``.autocommit``
206flag on the DBAPI connection object.
208To set isolation level using :func:`_sa.create_engine`::
210 engine = create_engine(
211 "postgresql+pg8000://scott:tiger@localhost/test",
212 isolation_level="REPEATABLE READ",
213 )
215To set using per-connection execution options::
217 with engine.connect() as conn:
218 conn = conn.execution_options(isolation_level="REPEATABLE READ")
219 with conn.begin():
220 ... # work with transaction
222There are also more options for isolation level configurations, such as
223"sub-engine" objects linked to a main :class:`_engine.Engine` which each apply
224different isolation level settings. See the discussion at
225:ref:`dbapi_autocommit` for background.
227Valid values for ``isolation_level`` on most PostgreSQL dialects include:
229* ``READ COMMITTED``
230* ``READ UNCOMMITTED``
231* ``REPEATABLE READ``
232* ``SERIALIZABLE``
233* ``AUTOCOMMIT``
235.. seealso::
237 :ref:`dbapi_autocommit`
239 :ref:`postgresql_readonly_deferrable`
241 :ref:`psycopg2_isolation_level`
243 :ref:`pg8000_isolation_level`
245.. _postgresql_readonly_deferrable:
247Setting READ ONLY / DEFERRABLE
248------------------------------
250Most PostgreSQL dialects support setting the "READ ONLY" and "DEFERRABLE"
251characteristics of the transaction, which is in addition to the isolation level
252setting. These two attributes can be established either in conjunction with or
253independently of the isolation level by passing the ``postgresql_readonly`` and
254``postgresql_deferrable`` flags with
255:meth:`_engine.Connection.execution_options`. The example below illustrates
256passing the ``"SERIALIZABLE"`` isolation level at the same time as setting
257"READ ONLY" and "DEFERRABLE"::
259 with engine.connect() as conn:
260 conn = conn.execution_options(
261 isolation_level="SERIALIZABLE",
262 postgresql_readonly=True,
263 postgresql_deferrable=True,
264 )
265 with conn.begin():
266 ... # work with transaction
268Note that some DBAPIs such as asyncpg only support "readonly" with
269SERIALIZABLE isolation.
271.. versionadded:: 1.4
272 Added support for the ``postgresql_readonly``
273 and ``postgresql_deferrable`` execution options.
275.. _postgresql_reset_on_return:
277Temporary Table / Resource Reset for Connection Pooling
278-------------------------------------------------------
280The :class:`.QueuePool` connection pool implementation used
281by the SQLAlchemy :class:`.Engine` object includes
282:ref:`reset on return <pool_reset_on_return>` behavior that will invoke
283the DBAPI ``.rollback()`` method when connections are returned to the pool.
284While this rollback will clear out the immediate state used by the previous
285transaction, it does not cover a wider range of session-level state, including
286temporary tables as well as other server state such as prepared statement
287handles and statement caches. The PostgreSQL database includes a variety
288of commands which may be used to reset this state, including
289``DISCARD``, ``RESET``, ``DEALLOCATE``, and ``UNLISTEN``.
292To install
293one or more of these commands as the means of performing reset-on-return,
294the :meth:`.PoolEvents.reset` event hook may be used, as demonstrated
295in the example below. The implementation
296will end transactions in progress as well as discard temporary tables
297using the ``CLOSE``, ``RESET`` and ``DISCARD`` commands; see the PostgreSQL
298documentation for background on what each of these statements do.
300The :paramref:`_sa.create_engine.pool_reset_on_return` parameter
301is set to ``None`` so that the custom scheme can replace the default behavior
302completely. The custom hook implementation calls ``.rollback()`` in any case,
303as it's usually important that the DBAPI's own tracking of commit/rollback
304will remain consistent with the state of the transaction::
307 from sqlalchemy import create_engine
308 from sqlalchemy import event
310 postgresql_engine = create_engine(
311 "postgresql+psycopg2://scott:tiger@hostname/dbname",
312 # disable default reset-on-return scheme
313 pool_reset_on_return=None,
314 )
317 @event.listens_for(postgresql_engine, "reset")
318 def _reset_postgresql(dbapi_connection, connection_record, reset_state):
319 if not reset_state.terminate_only:
320 dbapi_connection.execute("CLOSE ALL")
321 dbapi_connection.execute("RESET ALL")
322 dbapi_connection.execute("DISCARD TEMP")
324 # so that the DBAPI itself knows that the connection has been
325 # reset
326 dbapi_connection.rollback()
328.. versionchanged:: 2.0.0b3 Added additional state arguments to
329 the :meth:`.PoolEvents.reset` event and additionally ensured the event
330 is invoked for all "reset" occurrences, so that it's appropriate
331 as a place for custom "reset" handlers. Previous schemes which
332 use the :meth:`.PoolEvents.checkin` handler remain usable as well.
334.. seealso::
336 :ref:`pool_reset_on_return` - in the :ref:`pooling_toplevel` documentation
338.. _postgresql_alternate_search_path:
340Setting Alternate Search Paths on Connect
341------------------------------------------
343The PostgreSQL ``search_path`` variable refers to the list of schema names
344that will be implicitly referenced when a particular table or other
345object is referenced in a SQL statement. As detailed in the next section
346:ref:`postgresql_schema_reflection`, SQLAlchemy is generally organized around
347the concept of keeping this variable at its default value of ``public``,
348however, in order to have it set to any arbitrary name or names when connections
349are used automatically, the "SET SESSION search_path" command may be invoked
350for all connections in a pool using the following event handler, as discussed
351at :ref:`schema_set_default_connections`::
353 from sqlalchemy import event
354 from sqlalchemy import create_engine
356 engine = create_engine("postgresql+psycopg2://scott:tiger@host/dbname")
359 @event.listens_for(engine, "connect", insert=True)
360 def set_search_path(dbapi_connection, connection_record):
361 existing_autocommit = dbapi_connection.autocommit
362 dbapi_connection.autocommit = True
363 cursor = dbapi_connection.cursor()
364 cursor.execute("SET SESSION search_path='%s'" % schema_name)
365 cursor.close()
366 dbapi_connection.autocommit = existing_autocommit
368The reason the recipe is complicated by use of the ``.autocommit`` DBAPI
369attribute is so that when the ``SET SESSION search_path`` directive is invoked,
370it is invoked outside of the scope of any transaction and therefore will not
371be reverted when the DBAPI connection has a rollback.
373.. seealso::
375 :ref:`schema_set_default_connections` - in the :ref:`metadata_toplevel` documentation
377.. _postgresql_schema_reflection:
379Remote-Schema Table Introspection and PostgreSQL search_path
380------------------------------------------------------------
382.. admonition:: Section Best Practices Summarized
384 keep the ``search_path`` variable set to its default of ``public``, without
385 any other schema names. Ensure the username used to connect **does not**
386 match remote schemas, or ensure the ``"$user"`` token is **removed** from
387 ``search_path``. For other schema names, name these explicitly
388 within :class:`_schema.Table` definitions. Alternatively, the
389 ``postgresql_ignore_search_path`` option will cause all reflected
390 :class:`_schema.Table` objects to have a :attr:`_schema.Table.schema`
391 attribute set up.
393The PostgreSQL dialect can reflect tables from any schema, as outlined in
394:ref:`metadata_reflection_schemas`.
396In all cases, the first thing SQLAlchemy does when reflecting tables is
397to **determine the default schema for the current database connection**.
398It does this using the PostgreSQL ``current_schema()``
399function, illustated below using a PostgreSQL client session (i.e. using
400the ``psql`` tool):
402.. sourcecode:: sql
404 test=> select current_schema();
405 current_schema
406 ----------------
407 public
408 (1 row)
410Above we see that on a plain install of PostgreSQL, the default schema name
411is the name ``public``.
413However, if your database username **matches the name of a schema**, PostgreSQL's
414default is to then **use that name as the default schema**. Below, we log in
415using the username ``scott``. When we create a schema named ``scott``, **it
416implicitly changes the default schema**:
418.. sourcecode:: sql
420 test=> select current_schema();
421 current_schema
422 ----------------
423 public
424 (1 row)
426 test=> create schema scott;
427 CREATE SCHEMA
428 test=> select current_schema();
429 current_schema
430 ----------------
431 scott
432 (1 row)
434The behavior of ``current_schema()`` is derived from the
435`PostgreSQL search path
436<https://www.postgresql.org/docs/current/static/ddl-schemas.html#DDL-SCHEMAS-PATH>`_
437variable ``search_path``, which in modern PostgreSQL versions defaults to this:
439.. sourcecode:: sql
441 test=> show search_path;
442 search_path
443 -----------------
444 "$user", public
445 (1 row)
447Where above, the ``"$user"`` variable will inject the current username as the
448default schema, if one exists. Otherwise, ``public`` is used.
450When a :class:`_schema.Table` object is reflected, if it is present in the
451schema indicated by the ``current_schema()`` function, **the schema name assigned
452to the ".schema" attribute of the Table is the Python "None" value**. Otherwise, the
453".schema" attribute will be assigned the string name of that schema.
455With regards to tables which these :class:`_schema.Table`
456objects refer to via foreign key constraint, a decision must be made as to how
457the ``.schema`` is represented in those remote tables, in the case where that
458remote schema name is also a member of the current ``search_path``.
460By default, the PostgreSQL dialect mimics the behavior encouraged by
461PostgreSQL's own ``pg_get_constraintdef()`` builtin procedure. This function
462returns a sample definition for a particular foreign key constraint,
463omitting the referenced schema name from that definition when the name is
464also in the PostgreSQL schema search path. The interaction below
465illustrates this behavior:
467.. sourcecode:: sql
469 test=> CREATE TABLE test_schema.referred(id INTEGER PRIMARY KEY);
470 CREATE TABLE
471 test=> CREATE TABLE referring(
472 test(> id INTEGER PRIMARY KEY,
473 test(> referred_id INTEGER REFERENCES test_schema.referred(id));
474 CREATE TABLE
475 test=> SET search_path TO public, test_schema;
476 test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
477 test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n
478 test-> ON n.oid = c.relnamespace
479 test-> JOIN pg_catalog.pg_constraint r ON c.oid = r.conrelid
480 test-> WHERE c.relname='referring' AND r.contype = 'f'
481 test-> ;
482 pg_get_constraintdef
483 ---------------------------------------------------
484 FOREIGN KEY (referred_id) REFERENCES referred(id)
485 (1 row)
487Above, we created a table ``referred`` as a member of the remote schema
488``test_schema``, however when we added ``test_schema`` to the
489PG ``search_path`` and then asked ``pg_get_constraintdef()`` for the
490``FOREIGN KEY`` syntax, ``test_schema`` was not included in the output of
491the function.
493On the other hand, if we set the search path back to the typical default
494of ``public``:
496.. sourcecode:: sql
498 test=> SET search_path TO public;
499 SET
501The same query against ``pg_get_constraintdef()`` now returns the fully
502schema-qualified name for us:
504.. sourcecode:: sql
506 test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
507 test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n
508 test-> ON n.oid = c.relnamespace
509 test-> JOIN pg_catalog.pg_constraint r ON c.oid = r.conrelid
510 test-> WHERE c.relname='referring' AND r.contype = 'f';
511 pg_get_constraintdef
512 ---------------------------------------------------------------
513 FOREIGN KEY (referred_id) REFERENCES test_schema.referred(id)
514 (1 row)
516SQLAlchemy will by default use the return value of ``pg_get_constraintdef()``
517in order to determine the remote schema name. That is, if our ``search_path``
518were set to include ``test_schema``, and we invoked a table
519reflection process as follows::
521 >>> from sqlalchemy import Table, MetaData, create_engine, text
522 >>> engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
523 >>> with engine.connect() as conn:
524 ... conn.execute(text("SET search_path TO test_schema, public"))
525 ... metadata_obj = MetaData()
526 ... referring = Table("referring", metadata_obj, autoload_with=conn)
527 <sqlalchemy.engine.result.CursorResult object at 0x101612ed0>
529The above process would deliver to the :attr:`_schema.MetaData.tables`
530collection
531``referred`` table named **without** the schema::
533 >>> metadata_obj.tables["referred"].schema is None
534 True
536To alter the behavior of reflection such that the referred schema is
537maintained regardless of the ``search_path`` setting, use the
538``postgresql_ignore_search_path`` option, which can be specified as a
539dialect-specific argument to both :class:`_schema.Table` as well as
540:meth:`_schema.MetaData.reflect`::
542 >>> with engine.connect() as conn:
543 ... conn.execute(text("SET search_path TO test_schema, public"))
544 ... metadata_obj = MetaData()
545 ... referring = Table(
546 ... "referring",
547 ... metadata_obj,
548 ... autoload_with=conn,
549 ... postgresql_ignore_search_path=True,
550 ... )
551 <sqlalchemy.engine.result.CursorResult object at 0x1016126d0>
553We will now have ``test_schema.referred`` stored as schema-qualified::
555 >>> metadata_obj.tables["test_schema.referred"].schema
556 'test_schema'
558.. sidebar:: Best Practices for PostgreSQL Schema reflection
560 The description of PostgreSQL schema reflection behavior is complex, and
561 is the product of many years of dealing with widely varied use cases and
562 user preferences. But in fact, there's no need to understand any of it if
563 you just stick to the simplest use pattern: leave the ``search_path`` set
564 to its default of ``public`` only, never refer to the name ``public`` as
565 an explicit schema name otherwise, and refer to all other schema names
566 explicitly when building up a :class:`_schema.Table` object. The options
567 described here are only for those users who can't, or prefer not to, stay
568 within these guidelines.
570.. seealso::
572 :ref:`reflection_schema_qualified_interaction` - discussion of the issue
573 from a backend-agnostic perspective
575 `The Schema Search Path
576 <https://www.postgresql.org/docs/current/static/ddl-schemas.html#DDL-SCHEMAS-PATH>`_
577 - on the PostgreSQL website.
579INSERT/UPDATE...RETURNING
580-------------------------
582The dialect supports PG 8.2's ``INSERT..RETURNING``, ``UPDATE..RETURNING`` and
583``DELETE..RETURNING`` syntaxes. ``INSERT..RETURNING`` is used by default
584for single-row INSERT statements in order to fetch newly generated
585primary key identifiers. To specify an explicit ``RETURNING`` clause,
586use the :meth:`._UpdateBase.returning` method on a per-statement basis::
588 # INSERT..RETURNING
589 result = (
590 table.insert().returning(table.c.col1, table.c.col2).values(name="foo")
591 )
592 print(result.fetchall())
594 # UPDATE..RETURNING
595 result = (
596 table.update()
597 .returning(table.c.col1, table.c.col2)
598 .where(table.c.name == "foo")
599 .values(name="bar")
600 )
601 print(result.fetchall())
603 # DELETE..RETURNING
604 result = (
605 table.delete()
606 .returning(table.c.col1, table.c.col2)
607 .where(table.c.name == "foo")
608 )
609 print(result.fetchall())
611.. _postgresql_insert_on_conflict:
613INSERT...ON CONFLICT (Upsert)
614------------------------------
616Starting with version 9.5, PostgreSQL allows "upserts" (update or insert) of
617rows into a table via the ``ON CONFLICT`` clause of the ``INSERT`` statement. A
618candidate row will only be inserted if that row does not violate any unique
619constraints. In the case of a unique constraint violation, a secondary action
620can occur which can be either "DO UPDATE", indicating that the data in the
621target row should be updated, or "DO NOTHING", which indicates to silently skip
622this row.
624Conflicts are determined using existing unique constraints and indexes. These
625constraints may be identified either using their name as stated in DDL,
626or they may be inferred by stating the columns and conditions that comprise
627the indexes.
629SQLAlchemy provides ``ON CONFLICT`` support via the PostgreSQL-specific
630:func:`_postgresql.insert()` function, which provides
631the generative methods :meth:`_postgresql.Insert.on_conflict_do_update`
632and :meth:`~.postgresql.Insert.on_conflict_do_nothing`:
634.. sourcecode:: pycon+sql
636 >>> from sqlalchemy.dialects.postgresql import insert
637 >>> insert_stmt = insert(my_table).values(
638 ... id="some_existing_id", data="inserted value"
639 ... )
640 >>> do_nothing_stmt = insert_stmt.on_conflict_do_nothing(index_elements=["id"])
641 >>> print(do_nothing_stmt)
642 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
643 ON CONFLICT (id) DO NOTHING
644 {stop}
646 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
647 ... constraint="pk_my_table", set_=dict(data="updated value")
648 ... )
649 >>> print(do_update_stmt)
650 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
651 ON CONFLICT ON CONSTRAINT pk_my_table DO UPDATE SET data = %(param_1)s
653.. seealso::
655 `INSERT .. ON CONFLICT
656 <https://www.postgresql.org/docs/current/static/sql-insert.html#SQL-ON-CONFLICT>`_
657 - in the PostgreSQL documentation.
659Specifying the Target
660^^^^^^^^^^^^^^^^^^^^^
662Both methods supply the "target" of the conflict using either the
663named constraint or by column inference:
665* The :paramref:`_postgresql.Insert.on_conflict_do_update.index_elements` argument
666 specifies a sequence containing string column names, :class:`_schema.Column`
667 objects, and/or SQL expression elements, which would identify a unique
668 index:
670 .. sourcecode:: pycon+sql
672 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
673 ... index_elements=["id"], set_=dict(data="updated value")
674 ... )
675 >>> print(do_update_stmt)
676 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
677 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
678 {stop}
680 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
681 ... index_elements=[my_table.c.id], set_=dict(data="updated value")
682 ... )
683 >>> print(do_update_stmt)
684 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
685 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
687* When using :paramref:`_postgresql.Insert.on_conflict_do_update.index_elements` to
688 infer an index, a partial index can be inferred by also specifying the
689 use the :paramref:`_postgresql.Insert.on_conflict_do_update.index_where` parameter:
691 .. sourcecode:: pycon+sql
693 >>> stmt = insert(my_table).values(user_email="a@b.com", data="inserted data")
694 >>> stmt = stmt.on_conflict_do_update(
695 ... index_elements=[my_table.c.user_email],
696 ... index_where=my_table.c.user_email.like("%@gmail.com"),
697 ... set_=dict(data=stmt.excluded.data),
698 ... )
699 >>> print(stmt)
700 {printsql}INSERT INTO my_table (data, user_email)
701 VALUES (%(data)s, %(user_email)s) ON CONFLICT (user_email)
702 WHERE user_email LIKE %(user_email_1)s DO UPDATE SET data = excluded.data
704* The :paramref:`_postgresql.Insert.on_conflict_do_update.constraint` argument is
705 used to specify an index directly rather than inferring it. This can be
706 the name of a UNIQUE constraint, a PRIMARY KEY constraint, or an INDEX:
708 .. sourcecode:: pycon+sql
710 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
711 ... constraint="my_table_idx_1", set_=dict(data="updated value")
712 ... )
713 >>> print(do_update_stmt)
714 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
715 ON CONFLICT ON CONSTRAINT my_table_idx_1 DO UPDATE SET data = %(param_1)s
716 {stop}
718 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
719 ... constraint="my_table_pk", set_=dict(data="updated value")
720 ... )
721 >>> print(do_update_stmt)
722 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
723 ON CONFLICT ON CONSTRAINT my_table_pk DO UPDATE SET data = %(param_1)s
724 {stop}
726* The :paramref:`_postgresql.Insert.on_conflict_do_update.constraint` argument may
727 also refer to a SQLAlchemy construct representing a constraint,
728 e.g. :class:`.UniqueConstraint`, :class:`.PrimaryKeyConstraint`,
729 :class:`.Index`, or :class:`.ExcludeConstraint`. In this use,
730 if the constraint has a name, it is used directly. Otherwise, if the
731 constraint is unnamed, then inference will be used, where the expressions
732 and optional WHERE clause of the constraint will be spelled out in the
733 construct. This use is especially convenient
734 to refer to the named or unnamed primary key of a :class:`_schema.Table`
735 using the
736 :attr:`_schema.Table.primary_key` attribute:
738 .. sourcecode:: pycon+sql
740 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
741 ... constraint=my_table.primary_key, set_=dict(data="updated value")
742 ... )
743 >>> print(do_update_stmt)
744 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
745 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
747The SET Clause
748^^^^^^^^^^^^^^^
750``ON CONFLICT...DO UPDATE`` is used to perform an update of the already
751existing row, using any combination of new values as well as values
752from the proposed insertion. These values are specified using the
753:paramref:`_postgresql.Insert.on_conflict_do_update.set_` parameter. This
754parameter accepts a dictionary which consists of direct values
755for UPDATE:
757.. sourcecode:: pycon+sql
759 >>> stmt = insert(my_table).values(id="some_id", data="inserted value")
760 >>> do_update_stmt = stmt.on_conflict_do_update(
761 ... index_elements=["id"], set_=dict(data="updated value")
762 ... )
763 >>> print(do_update_stmt)
764 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
765 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
767.. warning::
769 The :meth:`_expression.Insert.on_conflict_do_update`
770 method does **not** take into
771 account Python-side default UPDATE values or generation functions, e.g.
772 those specified using :paramref:`_schema.Column.onupdate`.
773 These values will not be exercised for an ON CONFLICT style of UPDATE,
774 unless they are manually specified in the
775 :paramref:`_postgresql.Insert.on_conflict_do_update.set_` dictionary.
777Updating using the Excluded INSERT Values
778^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
780In order to refer to the proposed insertion row, the special alias
781:attr:`~.postgresql.Insert.excluded` is available as an attribute on
782the :class:`_postgresql.Insert` object; this object is a
783:class:`_expression.ColumnCollection`
784which alias contains all columns of the target
785table:
787.. sourcecode:: pycon+sql
789 >>> stmt = insert(my_table).values(
790 ... id="some_id", data="inserted value", author="jlh"
791 ... )
792 >>> do_update_stmt = stmt.on_conflict_do_update(
793 ... index_elements=["id"],
794 ... set_=dict(data="updated value", author=stmt.excluded.author),
795 ... )
796 >>> print(do_update_stmt)
797 {printsql}INSERT INTO my_table (id, data, author)
798 VALUES (%(id)s, %(data)s, %(author)s)
799 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author
801Additional WHERE Criteria
802^^^^^^^^^^^^^^^^^^^^^^^^^
804The :meth:`_expression.Insert.on_conflict_do_update` method also accepts
805a WHERE clause using the :paramref:`_postgresql.Insert.on_conflict_do_update.where`
806parameter, which will limit those rows which receive an UPDATE:
808.. sourcecode:: pycon+sql
810 >>> stmt = insert(my_table).values(
811 ... id="some_id", data="inserted value", author="jlh"
812 ... )
813 >>> on_update_stmt = stmt.on_conflict_do_update(
814 ... index_elements=["id"],
815 ... set_=dict(data="updated value", author=stmt.excluded.author),
816 ... where=(my_table.c.status == 2),
817 ... )
818 >>> print(on_update_stmt)
819 {printsql}INSERT INTO my_table (id, data, author)
820 VALUES (%(id)s, %(data)s, %(author)s)
821 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author
822 WHERE my_table.status = %(status_1)s
824Skipping Rows with DO NOTHING
825^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
827``ON CONFLICT`` may be used to skip inserting a row entirely
828if any conflict with a unique or exclusion constraint occurs; below
829this is illustrated using the
830:meth:`~.postgresql.Insert.on_conflict_do_nothing` method:
832.. sourcecode:: pycon+sql
834 >>> stmt = insert(my_table).values(id="some_id", data="inserted value")
835 >>> stmt = stmt.on_conflict_do_nothing(index_elements=["id"])
836 >>> print(stmt)
837 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
838 ON CONFLICT (id) DO NOTHING
840If ``DO NOTHING`` is used without specifying any columns or constraint,
841it has the effect of skipping the INSERT for any unique or exclusion
842constraint violation which occurs:
844.. sourcecode:: pycon+sql
846 >>> stmt = insert(my_table).values(id="some_id", data="inserted value")
847 >>> stmt = stmt.on_conflict_do_nothing()
848 >>> print(stmt)
849 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
850 ON CONFLICT DO NOTHING
852.. _postgresql_match:
854Full Text Search
855----------------
857PostgreSQL's full text search system is available through the use of the
858:data:`.func` namespace, combined with the use of custom operators
859via the :meth:`.Operators.bool_op` method. For simple cases with some
860degree of cross-backend compatibility, the :meth:`.Operators.match` operator
861may also be used.
863.. _postgresql_simple_match:
865Simple plain text matching with ``match()``
866^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
868The :meth:`.Operators.match` operator provides for cross-compatible simple
869text matching. For the PostgreSQL backend, it's hardcoded to generate
870an expression using the ``@@`` operator in conjunction with the
871``plainto_tsquery()`` PostgreSQL function.
873On the PostgreSQL dialect, an expression like the following::
875 select(sometable.c.text.match("search string"))
877would emit to the database:
879.. sourcecode:: sql
881 SELECT text @@ plainto_tsquery('search string') FROM table
883Above, passing a plain string to :meth:`.Operators.match` will automatically
884make use of ``plainto_tsquery()`` to specify the type of tsquery. This
885establishes basic database cross-compatibility for :meth:`.Operators.match`
886with other backends.
888.. versionchanged:: 2.0 The default tsquery generation function used by the
889 PostgreSQL dialect with :meth:`.Operators.match` is ``plainto_tsquery()``.
891 To render exactly what was rendered in 1.4, use the following form::
893 from sqlalchemy import func
895 select(sometable.c.text.bool_op("@@")(func.to_tsquery("search string")))
897 Which would emit:
899 .. sourcecode:: sql
901 SELECT text @@ to_tsquery('search string') FROM table
903Using PostgreSQL full text functions and operators directly
904^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
906Text search operations beyond the simple use of :meth:`.Operators.match`
907may make use of the :data:`.func` namespace to generate PostgreSQL full-text
908functions, in combination with :meth:`.Operators.bool_op` to generate
909any boolean operator.
911For example, the query::
913 select(func.to_tsquery("cat").bool_op("@>")(func.to_tsquery("cat & rat")))
915would generate:
917.. sourcecode:: sql
919 SELECT to_tsquery('cat') @> to_tsquery('cat & rat')
922The :class:`_postgresql.TSVECTOR` type can provide for explicit CAST::
924 from sqlalchemy.dialects.postgresql import TSVECTOR
925 from sqlalchemy import select, cast
927 select(cast("some text", TSVECTOR))
929produces a statement equivalent to:
931.. sourcecode:: sql
933 SELECT CAST('some text' AS TSVECTOR) AS anon_1
935The ``func`` namespace is augmented by the PostgreSQL dialect to set up
936correct argument and return types for most full text search functions.
937These functions are used automatically by the :attr:`_sql.func` namespace
938assuming the ``sqlalchemy.dialects.postgresql`` package has been imported,
939or :func:`_sa.create_engine` has been invoked using a ``postgresql``
940dialect. These functions are documented at:
942* :class:`_postgresql.to_tsvector`
943* :class:`_postgresql.to_tsquery`
944* :class:`_postgresql.plainto_tsquery`
945* :class:`_postgresql.phraseto_tsquery`
946* :class:`_postgresql.websearch_to_tsquery`
947* :class:`_postgresql.ts_headline`
949Specifying the "regconfig" with ``match()`` or custom operators
950^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
952PostgreSQL's ``plainto_tsquery()`` function accepts an optional
953"regconfig" argument that is used to instruct PostgreSQL to use a
954particular pre-computed GIN or GiST index in order to perform the search.
955When using :meth:`.Operators.match`, this additional parameter may be
956specified using the ``postgresql_regconfig`` parameter, such as::
958 select(mytable.c.id).where(
959 mytable.c.title.match("somestring", postgresql_regconfig="english")
960 )
962Which would emit:
964.. sourcecode:: sql
966 SELECT mytable.id FROM mytable
967 WHERE mytable.title @@ plainto_tsquery('english', 'somestring')
969When using other PostgreSQL search functions with :data:`.func`, the
970"regconfig" parameter may be passed directly as the initial argument::
972 select(mytable.c.id).where(
973 func.to_tsvector("english", mytable.c.title).bool_op("@@")(
974 func.to_tsquery("english", "somestring")
975 )
976 )
978produces a statement equivalent to:
980.. sourcecode:: sql
982 SELECT mytable.id FROM mytable
983 WHERE to_tsvector('english', mytable.title) @@
984 to_tsquery('english', 'somestring')
986It is recommended that you use the ``EXPLAIN ANALYZE...`` tool from
987PostgreSQL to ensure that you are generating queries with SQLAlchemy that
988take full advantage of any indexes you may have created for full text search.
990.. seealso::
992 `Full Text Search <https://www.postgresql.org/docs/current/textsearch-controls.html>`_ - in the PostgreSQL documentation
995FROM ONLY ...
996-------------
998The dialect supports PostgreSQL's ONLY keyword for targeting only a particular
999table in an inheritance hierarchy. This can be used to produce the
1000``SELECT ... FROM ONLY``, ``UPDATE ONLY ...``, and ``DELETE FROM ONLY ...``
1001syntaxes. It uses SQLAlchemy's hints mechanism::
1003 # SELECT ... FROM ONLY ...
1004 result = table.select().with_hint(table, "ONLY", "postgresql")
1005 print(result.fetchall())
1007 # UPDATE ONLY ...
1008 table.update(values=dict(foo="bar")).with_hint(
1009 "ONLY", dialect_name="postgresql"
1010 )
1012 # DELETE FROM ONLY ...
1013 table.delete().with_hint("ONLY", dialect_name="postgresql")
1015.. _postgresql_indexes:
1017PostgreSQL-Specific Index Options
1018---------------------------------
1020Several extensions to the :class:`.Index` construct are available, specific
1021to the PostgreSQL dialect.
1023.. _postgresql_covering_indexes:
1025Covering Indexes
1026^^^^^^^^^^^^^^^^
1028A covering index includes additional columns that are not part of the index key
1029but are stored in the index, allowing PostgreSQL to satisfy queries using only
1030the index without accessing the table (an "index-only scan"). This is
1031indicated on the index using the ``INCLUDE`` clause. The
1032``postgresql_include`` option for :class:`.Index` (as well as
1033:class:`.UniqueConstraint`) renders ``INCLUDE(colname)`` for the given string
1034names::
1036 Index("my_index", table.c.x, postgresql_include=["y"])
1038would render the index as ``CREATE INDEX my_index ON table (x) INCLUDE (y)``
1040Note that this feature requires PostgreSQL 11 or later.
1042.. seealso::
1044 :ref:`postgresql_constraint_options_include` - the same feature implemented
1045 for :class:`.UniqueConstraint`
1047.. versionadded:: 1.4
1048 Support for covering indexes with :class:`.Index`.
1050.. _postgresql_partial_indexes:
1052Partial Indexes
1053^^^^^^^^^^^^^^^
1055Partial indexes add criterion to the index definition so that the index is
1056applied to a subset of rows. These can be specified on :class:`.Index`
1057using the ``postgresql_where`` keyword argument::
1059 Index("my_index", my_table.c.id, postgresql_where=my_table.c.value > 10)
1061.. _postgresql_operator_classes:
1063Operator Classes
1064^^^^^^^^^^^^^^^^
1066PostgreSQL allows the specification of an *operator class* for each column of
1067an index (see
1068https://www.postgresql.org/docs/current/interactive/indexes-opclass.html).
1069The :class:`.Index` construct allows these to be specified via the
1070``postgresql_ops`` keyword argument::
1072 Index(
1073 "my_index",
1074 my_table.c.id,
1075 my_table.c.data,
1076 postgresql_ops={"data": "text_pattern_ops", "id": "int4_ops"},
1077 )
1079Note that the keys in the ``postgresql_ops`` dictionaries are the
1080"key" name of the :class:`_schema.Column`, i.e. the name used to access it from
1081the ``.c`` collection of :class:`_schema.Table`, which can be configured to be
1082different than the actual name of the column as expressed in the database.
1084If ``postgresql_ops`` is to be used against a complex SQL expression such
1085as a function call, then to apply to the column it must be given a label
1086that is identified in the dictionary by name, e.g.::
1088 Index(
1089 "my_index",
1090 my_table.c.id,
1091 func.lower(my_table.c.data).label("data_lower"),
1092 postgresql_ops={"data_lower": "text_pattern_ops", "id": "int4_ops"},
1093 )
1095Operator classes are also supported by the
1096:class:`_postgresql.ExcludeConstraint` construct using the
1097:paramref:`_postgresql.ExcludeConstraint.ops` parameter. See that parameter for
1098details.
1100Index Types
1101^^^^^^^^^^^
1103PostgreSQL provides several index types: B-Tree, Hash, GiST, and GIN, as well
1104as the ability for users to create their own (see
1105https://www.postgresql.org/docs/current/static/indexes-types.html). These can be
1106specified on :class:`.Index` using the ``postgresql_using`` keyword argument::
1108 Index("my_index", my_table.c.data, postgresql_using="gin")
1110The value passed to the keyword argument will be simply passed through to the
1111underlying CREATE INDEX command, so it *must* be a valid index type for your
1112version of PostgreSQL.
1114.. _postgresql_index_storage:
1116Index Storage Parameters
1117^^^^^^^^^^^^^^^^^^^^^^^^
1119PostgreSQL allows storage parameters to be set on indexes. The storage
1120parameters available depend on the index method used by the index. Storage
1121parameters can be specified on :class:`.Index` using the ``postgresql_with``
1122keyword argument::
1124 Index("my_index", my_table.c.data, postgresql_with={"fillfactor": 50})
1126PostgreSQL allows to define the tablespace in which to create the index.
1127The tablespace can be specified on :class:`.Index` using the
1128``postgresql_tablespace`` keyword argument::
1130 Index("my_index", my_table.c.data, postgresql_tablespace="my_tablespace")
1132Note that the same option is available on :class:`_schema.Table` as well.
1134.. _postgresql_index_concurrently:
1136Indexes with CONCURRENTLY
1137^^^^^^^^^^^^^^^^^^^^^^^^^
1139The PostgreSQL index option CONCURRENTLY is supported by passing the
1140flag ``postgresql_concurrently`` to the :class:`.Index` construct::
1142 tbl = Table("testtbl", m, Column("data", Integer))
1144 idx1 = Index("test_idx1", tbl.c.data, postgresql_concurrently=True)
1146The above index construct will render DDL for CREATE INDEX, assuming
1147PostgreSQL 8.2 or higher is detected or for a connection-less dialect, as:
1149.. sourcecode:: sql
1151 CREATE INDEX CONCURRENTLY test_idx1 ON testtbl (data)
1153For DROP INDEX, assuming PostgreSQL 9.2 or higher is detected or for
1154a connection-less dialect, it will emit:
1156.. sourcecode:: sql
1158 DROP INDEX CONCURRENTLY test_idx1
1160When using CONCURRENTLY, the PostgreSQL database requires that the statement
1161be invoked outside of a transaction block. The Python DBAPI enforces that
1162even for a single statement, a transaction is present, so to use this
1163construct, the DBAPI's "autocommit" mode must be used::
1165 metadata = MetaData()
1166 table = Table("foo", metadata, Column("id", String))
1167 index = Index("foo_idx", table.c.id, postgresql_concurrently=True)
1169 with engine.connect() as conn:
1170 with conn.execution_options(isolation_level="AUTOCOMMIT"):
1171 table.create(conn)
1173.. seealso::
1175 :ref:`postgresql_isolation_level`
1177.. _postgresql_index_reflection:
1179PostgreSQL Index Reflection
1180---------------------------
1182Notes on Index reflection
1185Implicit UNIQUE index for UNIQUE CONSTRAINT
1186^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
1188The PostgreSQL database creates a UNIQUE INDEX implicitly whenever the
1189UNIQUE CONSTRAINT construct is used. When inspecting a table using
1190:class:`_reflection.Inspector`, the :meth:`_reflection.Inspector.get_indexes`
1191and the :meth:`_reflection.Inspector.get_unique_constraints`
1192will report on these
1193two constructs distinctly; in the case of the index, the key
1194``duplicates_constraint`` will be present in the index entry if it is
1195detected as mirroring a constraint. When performing reflection using
1196``Table(..., autoload_with=engine)``, the UNIQUE INDEX is **not** returned
1197in :attr:`_schema.Table.indexes` when it is detected as mirroring a
1198:class:`.UniqueConstraint` in the :attr:`_schema.Table.constraints` collection.
1200.. _postgresql_invalid_index_reflection:
1202Invalid indexes
1203^^^^^^^^^^^^^^^
1205Indexes marked "invalid" by PostgreSQL (i.e. an index that exists in the
1206catalog, but is ignored by the query planner because it may be incomplete,
1207while still being maintained on writes) remain present in reflection. The
1208:meth:`_reflection.Inspector.get_indexes` and
1209:meth:`_reflection.Inspector.get_multi_indexes` methods include
1210``postgresql_invalid=True`` in the index's ``dialect_options`` dictionary when
1211``pg_index.indisvalid`` is false. Valid indexes omit this flag.
1213When reflecting a :class:`_schema.Table`, the flag, if present, is stored
1214separately from DDL options on each reflected :class:`.Index`::
1216 table = Table("my_table", MetaData(), autoload_with=engine)
1217 for index in table.indexes:
1218 if index.reflect_only_elements["postgresql"].get("invalid"):
1219 print(index.name)
1221The :attr:`.Index.reflect_only_elements` mapping is read-only. These values are
1222not included in :attr:`.Index.dialect_kwargs` and do not affect DDL.
1224An invalid index may be left by a failed ``CREATE INDEX CONCURRENTLY``, may
1225still be building, or may be a partitioned index awaiting attachment of its
1226partition indexes. This flag reports the state at reflection time; it does not
1227request an index rebuild.
1229.. versionadded:: 2.1
1231Special Reflection Options
1232--------------------------
1234The :class:`_reflection.Inspector`
1235used for the PostgreSQL backend is an instance
1236of :class:`.PGInspector`, which offers additional methods::
1238 from sqlalchemy import create_engine, inspect
1240 engine = create_engine("postgresql+psycopg2://localhost/test")
1241 insp = inspect(engine) # will be a PGInspector
1243 print(insp.get_enums())
1245.. autoclass:: PGInspector
1246 :members:
1248.. _postgresql_table_options:
1250PostgreSQL Table Options
1251------------------------
1253Several options for CREATE TABLE are supported directly by the PostgreSQL
1254dialect in conjunction with the :class:`_schema.Table` construct, detailed
1255in the following sections.
1257.. seealso::
1259 `PostgreSQL CREATE TABLE options
1260 <https://www.postgresql.org/docs/current/static/sql-createtable.html>`_ -
1261 in the PostgreSQL documentation.
1263``INHERITS``
1264^^^^^^^^^^^^
1266Specifies one or more parent tables from which this table inherits columns and
1267constraints, enabling table inheritance hierarchies in PostgreSQL.
1269::
1271 Table("some_table", metadata, ..., postgresql_inherits="some_supertable")
1273 Table("some_table", metadata, ..., postgresql_inherits=("t1", "t2", ...))
1275For schema-qualified parent table names, use :class:`.quoted_name` with
1276``quote=False`` to prevent the dotted name from being quoted as a single
1277identifier::
1279 from sqlalchemy.sql import quoted_name
1281 Table(
1282 "some_table",
1283 metadata,
1284 ...,
1285 postgresql_inherits=quoted_name(
1286 "my_schema.some_supertable", quote=False
1287 ),
1288 )
1290SQLAlchemy does not automatically copy the columns from the inherited tables
1291mentioned in the ``postgresql_inherits`` argument into the new
1292:class:`_schema.Table`. To populate the new table columns reflection may be
1293used, or a function similar to the following one::
1295 def get_parent_columns(tbl: sa.Table) -> list[sa.Column]:
1296 return [
1297 sa.Column(
1298 c.name,
1299 c.type,
1300 key=c.key,
1301 nullable=c.nullable,
1302 # Set system=true to omit from the CREATE TABLE statement
1303 system=True,
1304 )
1305 for c in tbl.columns
1306 ]
1309 documents = Table(
1310 "some_table",
1311 metadata,
1312 *get_parent_columns(some_supertable),
1313 # ... # add other columns here normally if needed
1314 postgresql_inherits="some_supertable",
1315 )
1317``ON COMMIT``
1318^^^^^^^^^^^^^
1320Controls the behavior of temporary tables at transaction commit, with options
1321to preserve rows, delete rows, or drop the table.
1323::
1325 Table("some_table", metadata, ..., postgresql_on_commit="PRESERVE ROWS")
1327``PARTITION BY``
1328^^^^^^^^^^^^^^^^
1330Declares the table as a partitioned table using the specified partitioning
1331strategy (RANGE, LIST, or HASH) on the given column(s).
1333::
1335 Table(
1336 "some_table",
1337 metadata,
1338 ...,
1339 postgresql_partition_by="LIST (part_column)",
1340 )
1342``TABLESPACE``
1343^^^^^^^^^^^^^^
1345Specifies the tablespace where the table will be stored, allowing control over
1346the physical location of table data on disk.
1348::
1350 Table("some_table", metadata, ..., postgresql_tablespace="some_tablespace")
1352The above option is also available on the :class:`.Index` construct.
1354``USING``
1355^^^^^^^^^
1357Specifies the table access method to use for storing table data, such as
1358``heap`` (the default) or other custom access methods.
1360::
1362 Table("some_table", metadata, ..., postgresql_using="heap")
1364.. versionadded:: 2.0.26
1366.. _postgresql_table_options_with:
1368``WITH``
1369^^^^^^^^
1371::
1373 Table("some_table", metadata, ..., postgresql_with={"fillfactor": 100})
1375The ``postgresql_with`` parameter accepts a dictionary of storage parameters
1376that will be applied to the table using the ``WITH`` clause in the
1377``CREATE TABLE`` statement. Storage parameters control various aspects of
1378table behavior such as the fill factor for pages, autovacuum settings, and
1379toast table parameters.
1381When the table is created, the parameters are rendered as a comma-separated
1382list within the ``WITH`` clause. For example::
1384 Table(
1385 "mytable",
1386 metadata,
1387 Column("id", Integer, primary_key=True),
1388 postgresql_with={
1389 "fillfactor": 70,
1390 "autovacuum_enabled": False,
1391 "toast.vacuum_truncate": True,
1392 },
1393 )
1395This will generate DDL similar to:
1397.. sourcecode:: sql
1399 CREATE TABLE mytable (
1400 id INTEGER NOT NULL,
1401 PRIMARY KEY (id)
1402 ) WITH (fillfactor = 70, autovacuum_enabled = false, toast.vacuum_truncate = true)
1404The values in the dictionary can be integers, strings, boolean values, or
1405``None``. Boolean values are rendered as lowercase ``true``/``false``.
1406A ``None`` value renders the parameter name only, without an ``=`` value.
1407Parameter names can include dots to specify parameters for associated objects
1408like toast tables (e.g., ``toast.vacuum_truncate``).
1410.. seealso::
1412 `PostgreSQL Storage Parameters
1413 <https://www.postgresql.org/docs/current/sql-createtable.html#SQL-CREATETABLE-STORAGE-PARAMETERS>`_ -
1414 documentation on available storage parameters.
1416.. versionadded:: 2.1
1418``WITH OIDS``
1419^^^^^^^^^^^^^
1421Enables the legacy OID (object identifier) system column for the table, which
1422assigns a unique identifier to each row.
1424::
1426 Table("some_table", metadata, ..., postgresql_with_oids=True)
1428.. note:: Support for tables "with OIDs" was removed in postgresql 12.
1430``WITHOUT OIDS``
1431^^^^^^^^^^^^^^^^
1433Explicitly disables the OID system column for the table (the default behavior
1434in modern PostgreSQL versions).
1436::
1438 Table("some_table", metadata, ..., postgresql_with_oids=False)
1440.. _postgresql_view_options:
1442PostgreSQL View Options
1443-----------------------
1445The following sections indicate options which are supported by the PostgreSQL
1446dialect in conjunction with :class:`.CreateView`.
1448``WITH``
1449^^^^^^^^
1451Specifies view-level options for :class:`.CreateView` using the ``WITH``
1452clause::
1454 from sqlalchemy import CreateView, select
1456 CreateView(
1457 select(my_table),
1458 "my_view",
1459 postgresql_with={"security_invoker": True},
1460 )
1462This generates:
1464.. sourcecode:: sql
1466 CREATE VIEW my_view WITH (security_invoker = true) AS SELECT ...
1468The same formatting rules that apply to :ref:`postgresql_table_options_with`
1469apply here: booleans render as lowercase ``true``/``false``, ``None``
1470renders as the parameter name only, and other values render as-is.
1472.. versionadded:: 2.1
1474.. _postgresql_constraint_options:
1476PostgreSQL Constraint Options
1477-----------------------------
1479The following sections indicate options which are supported by the PostgreSQL
1480dialect in conjunction with selected constraint constructs.
1483``NOT VALID``
1484^^^^^^^^^^^^^
1486Allows a constraint to be added without validating existing rows, improving
1487performance when adding constraints to large tables. This option applies
1488towards CHECK and FOREIGN KEY constraints when the constraint is being added
1489to an existing table via ALTER TABLE, and has the effect that existing rows
1490are not scanned during the ALTER operation against the constraint being added.
1492When using a SQL migration tool such as `Alembic <https://alembic.sqlalchemy.org>`_
1493that renders ALTER TABLE constructs, the ``postgresql_not_valid`` argument
1494may be specified as an additional keyword argument within the operation
1495that creates the constraint, as in the following Alembic example::
1497 def update():
1498 op.create_foreign_key(
1499 "fk_user_address",
1500 "address",
1501 "user",
1502 ["user_id"],
1503 ["id"],
1504 postgresql_not_valid=True,
1505 )
1507The keyword is ultimately accepted directly by the
1508:class:`_schema.CheckConstraint`, :class:`_schema.ForeignKeyConstraint`
1509and :class:`_schema.ForeignKey` constructs; when using a tool like
1510Alembic, dialect-specific keyword arguments are passed through to
1511these constructs from the migration operation directives::
1513 CheckConstraint("some_field IS NOT NULL", postgresql_not_valid=True)
1515 ForeignKeyConstraint(
1516 ["some_id"], ["some_table.some_id"], postgresql_not_valid=True
1517 )
1519.. versionadded:: 1.4.32
1521.. seealso::
1523 `PostgreSQL ALTER TABLE options
1524 <https://www.postgresql.org/docs/current/static/sql-altertable.html>`_ -
1525 in the PostgreSQL documentation.
1527.. _postgresql_constraint_options_include:
1529``INCLUDE``
1530^^^^^^^^^^^
1532This keyword is applicable to both a ``UNIQUE`` constraint as well as an
1533``INDEX``. The ``postgresql_include`` option available for
1534:class:`.UniqueConstraint` as well as :class:`.Index` creates a covering index
1535by including additional columns in the underlying index without making them
1536part of the key constraint. This option adds one or more columns as a "payload"
1537to the index created automatically by PostgreSQL for the constraint. For
1538example, the following table definition::
1540 Table(
1541 "mytable",
1542 metadata,
1543 Column("id", Integer, nullable=False),
1544 Column("value", Integer, nullable=False),
1545 UniqueConstraint("id", postgresql_include=["value"]),
1546 )
1548would produce the DDL statement
1550.. sourcecode:: sql
1552 CREATE TABLE mytable (
1553 id INTEGER NOT NULL,
1554 value INTEGER NOT NULL,
1555 UNIQUE (id) INCLUDE (value)
1556 )
1558Note that this feature requires PostgreSQL 11 or later.
1560.. versionadded:: 2.0.41
1561 Added support for ``postgresql_include`` to :class:`.UniqueConstraint`,
1562 to complement the existing feature in :class:`.Index`.
1564.. seealso::
1566 :ref:`postgresql_covering_indexes` - background on ``postgresql_include``
1567 for the :class:`.Index` construct.
1570Column list with foreign key ``ON DELETE SET`` actions
1571^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
1573Allows selective column updates when a foreign key action is triggered, limiting
1574which columns are set to NULL or DEFAULT upon deletion of a referenced row.
1575This applies to :class:`.ForeignKey` and :class:`.ForeignKeyConstraint`, the
1576:paramref:`.ForeignKey.ondelete` parameter will accept on the PostgreSQL
1577backend only a string list of column names inside parenthesis, following the
1578``SET NULL`` or ``SET DEFAULT`` phrases, which will limit the set of columns
1579that are subject to the action::
1581 fktable = Table(
1582 "fktable",
1583 metadata,
1584 Column("tid", Integer),
1585 Column("id", Integer),
1586 Column("fk_id_del_set_null", Integer),
1587 ForeignKeyConstraint(
1588 columns=["tid", "fk_id_del_set_null"],
1589 refcolumns=[pktable.c.tid, pktable.c.id],
1590 ondelete="SET NULL (fk_id_del_set_null)",
1591 ),
1592 )
1594.. versionadded:: 2.0.40
1596.. _postgresql_computed_column_notes:
1598Computed Columns (GENERATED ALWAYS AS)
1599---------------------------------------
1601SQLAlchemy's support for the "GENERATED ALWAYS AS" SQL instruction, which
1602establishes a dynamic, automatically populated value for a column, is available
1603using the :ref:`computed_ddl` feature of SQLAlchemy DDL. E.g.::
1605 from sqlalchemy import Table, Column, MetaData, Integer, Computed
1607 metadata_obj = MetaData()
1609 square = Table(
1610 "square",
1611 metadata_obj,
1612 Column("id", Integer, primary_key=True),
1613 Column("side", Integer),
1614 Column("area", Integer, Computed("side * side")),
1615 Column("perimeter", Integer, Computed("4 * side", persisted=True)),
1616 )
1618There are two general varieties of the "computed" column, ``VIRTUAL`` and ``STORED``.
1619A ``STORED`` computed column computes and persists its value at INSERT/UPDATE time,
1620while a ``VIRTUAL`` computed column computes its value on access without persisting it.
1621This preference is indicated using the :paramref:`.Computed.persisted` parameter,
1622which defaults to ``None`` to use the database default behavior.
1624For PostgreSQL, prior to version 18 only the ``STORED`` variant was supported,
1625requiring the ``STORED`` keyword to be emitted explicitly. PostgreSQL 18 added
1626support for ``VIRTUAL`` columns and made ``VIRTUAL`` the default behavior.
1628To accommodate this change, SQLAlchemy's behavior when
1629:paramref:`.Computed.persisted` is not specified depends on the PostgreSQL
1630version: on PostgreSQL 18 and later, no keyword is rendered, allowing the
1631database to use its default of ``VIRTUAL``; on PostgreSQL 17 and earlier,
1632``STORED`` is rendered and a warning is emitted. To ensure consistent
1633``STORED`` behavior across all PostgreSQL versions, explicitly set
1634``persisted=True``.
1636.. versionchanged:: 2.1
1638 PostgreSQL 18+ now defaults to ``VIRTUAL`` when :paramref:`.Computed.persisted`
1639 is not specified. A warning is emitted for older versions of PostgreSQL
1640 when this parameter is not indicated.
1645``NULLS NOT DISTINCT``
1646^^^^^^^^^^^^^^^^^^^^^^
1648By default, two ``null`` values are not considered equal for unique constraints
1649and indexes. Therefore, seemingly duplicate rows may be stored if one of the
1650values in the constraint is ``null``. This default behavior is implementation
1651defined, so other SQL dialects may behave differently than PostgreSQL.
1653The ``NULLS NOT DISTINCT`` clause can be used to change this behavior, treating
1654null values as equal and preventing unintended duplicate rows. The opposite
1655``NULLS DISTINCT`` clause can also be used to make PostgreSQL's default behavior
1656explict.
1658The ``postgresql_nulls_not_distinct`` parameter can be set to ``True`` to
1659add the ``NULLS NOT DISTINCT`` clause, or ``False`` to add ``NULLS DISTINCT``.
1660Not setting it, or passing ``None``, will not add a clause and keep the default
1661behavior.
1663This feature requires PostgreSQL 15 or later.
1665.. versionadded:: 2.0.16
1668.. _postgresql_table_valued_overview:
1670Table values, Table and Column valued functions, Row and Tuple objects
1671-----------------------------------------------------------------------
1673PostgreSQL makes great use of modern SQL forms such as table-valued functions,
1674tables and rows as values. These constructs are commonly used as part
1675of PostgreSQL's support for complex datatypes such as JSON, ARRAY, and other
1676datatypes. SQLAlchemy's SQL expression language has native support for
1677most table-valued and row-valued forms.
1679.. _postgresql_table_valued:
1681Table-Valued Functions
1682^^^^^^^^^^^^^^^^^^^^^^^
1684Many PostgreSQL built-in functions are intended to be used in the FROM clause
1685of a SELECT statement, and are capable of returning table rows or sets of table
1686rows. A large portion of PostgreSQL's JSON functions for example such as
1687``json_array_elements()``, ``json_object_keys()``, ``json_each_text()``,
1688``json_each()``, ``json_to_record()``, ``json_populate_recordset()`` use such
1689forms. These classes of SQL function calling forms in SQLAlchemy are available
1690using the :meth:`_functions.FunctionElement.table_valued` method in conjunction
1691with :class:`_functions.Function` objects generated from the :data:`_sql.func`
1692namespace.
1694Examples from PostgreSQL's reference documentation follow below:
1696* ``json_each()``:
1698 .. sourcecode:: pycon+sql
1700 >>> from sqlalchemy import select, func
1701 >>> stmt = select(
1702 ... func.json_each('{"a":"foo", "b":"bar"}').table_valued("key", "value")
1703 ... )
1704 >>> print(stmt)
1705 {printsql}SELECT anon_1.key, anon_1.value
1706 FROM json_each(:json_each_1) AS anon_1
1708* ``json_populate_record()``:
1710 .. sourcecode:: pycon+sql
1712 >>> from sqlalchemy import select, func, literal_column
1713 >>> stmt = select(
1714 ... func.json_populate_record(
1715 ... literal_column("null::myrowtype"), '{"a":1,"b":2}'
1716 ... ).table_valued("a", "b", name="x")
1717 ... )
1718 >>> print(stmt)
1719 {printsql}SELECT x.a, x.b
1720 FROM json_populate_record(null::myrowtype, :json_populate_record_1) AS x
1722* ``json_to_record()`` - this form uses a PostgreSQL specific form of derived
1723 columns in the alias, where we may make use of :func:`_sql.column` elements with
1724 types to produce them. The :meth:`_functions.FunctionElement.table_valued`
1725 method produces a :class:`_sql.TableValuedAlias` construct, and the method
1726 :meth:`_sql.TableValuedAlias.render_derived` method sets up the derived
1727 columns specification:
1729 .. sourcecode:: pycon+sql
1731 >>> from sqlalchemy import select, func, column, Integer, Text
1732 >>> stmt = select(
1733 ... func.json_to_record('{"a":1,"b":[1,2,3],"c":"bar"}')
1734 ... .table_valued(
1735 ... column("a", Integer),
1736 ... column("b", Text),
1737 ... column("d", Text),
1738 ... )
1739 ... .render_derived(name="x", with_types=True)
1740 ... )
1741 >>> print(stmt)
1742 {printsql}SELECT x.a, x.b, x.d
1743 FROM json_to_record(:json_to_record_1) AS x(a INTEGER, b TEXT, d TEXT)
1745* ``WITH ORDINALITY`` - part of the SQL standard, ``WITH ORDINALITY`` adds an
1746 ordinal counter to the output of a function and is accepted by a limited set
1747 of PostgreSQL functions including ``unnest()`` and ``generate_series()``. The
1748 :meth:`_functions.FunctionElement.table_valued` method accepts a keyword
1749 parameter ``with_ordinality`` for this purpose, which accepts the string name
1750 that will be applied to the "ordinality" column:
1752 .. sourcecode:: pycon+sql
1754 >>> from sqlalchemy import select, func
1755 >>> stmt = select(
1756 ... func.generate_series(4, 1, -1)
1757 ... .table_valued("value", with_ordinality="ordinality")
1758 ... .render_derived()
1759 ... )
1760 >>> print(stmt)
1761 {printsql}SELECT anon_1.value, anon_1.ordinality
1762 FROM generate_series(:generate_series_1, :generate_series_2, :generate_series_3)
1763 WITH ORDINALITY AS anon_1(value, ordinality)
1765.. versionadded:: 1.4.0b2
1767.. seealso::
1769 :ref:`tutorial_functions_table_valued` - in the :ref:`unified_tutorial`
1771.. _postgresql_column_valued:
1773Column Valued Functions
1774^^^^^^^^^^^^^^^^^^^^^^^
1776Similar to the table valued function, a column valued function is present
1777in the FROM clause, but delivers itself to the columns clause as a single
1778scalar value. PostgreSQL functions such as ``json_array_elements()``,
1779``unnest()`` and ``generate_series()`` may use this form. Column valued functions are available using the
1780:meth:`_functions.FunctionElement.column_valued` method of :class:`_functions.FunctionElement`:
1782* ``json_array_elements()``:
1784 .. sourcecode:: pycon+sql
1786 >>> from sqlalchemy import select, func
1787 >>> stmt = select(
1788 ... func.json_array_elements('["one", "two"]').column_valued("x")
1789 ... )
1790 >>> print(stmt)
1791 {printsql}SELECT x
1792 FROM json_array_elements(:json_array_elements_1) AS x
1794* ``unnest()`` - in order to generate a PostgreSQL ARRAY literal, the
1795 :func:`_postgresql.array` construct may be used:
1797 .. sourcecode:: pycon+sql
1799 >>> from sqlalchemy.dialects.postgresql import array
1800 >>> from sqlalchemy import select, func
1801 >>> stmt = select(func.unnest(array([1, 2])).column_valued())
1802 >>> print(stmt)
1803 {printsql}SELECT anon_1
1804 FROM unnest(ARRAY[%(param_1)s, %(param_2)s]) AS anon_1
1806 The function can of course be used against an existing table-bound column
1807 that's of type :class:`_types.ARRAY`:
1809 .. sourcecode:: pycon+sql
1811 >>> from sqlalchemy import table, column, ARRAY, Integer
1812 >>> from sqlalchemy import select, func
1813 >>> t = table("t", column("value", ARRAY(Integer)))
1814 >>> stmt = select(func.unnest(t.c.value).column_valued("unnested_value"))
1815 >>> print(stmt)
1816 {printsql}SELECT unnested_value
1817 FROM unnest(t.value) AS unnested_value
1819.. seealso::
1821 :ref:`tutorial_functions_column_valued` - in the :ref:`unified_tutorial`
1824Row Types
1825^^^^^^^^^
1827Built-in support for rendering a ``ROW`` may be approximated using
1828``func.ROW`` with the :attr:`_sa.func` namespace, or by using the
1829:func:`_sql.tuple_` construct:
1831.. sourcecode:: pycon+sql
1833 >>> from sqlalchemy import table, column, func, tuple_
1834 >>> t = table("t", column("id"), column("fk"))
1835 >>> stmt = (
1836 ... t.select()
1837 ... .where(tuple_(t.c.id, t.c.fk) > (1, 2))
1838 ... .where(func.ROW(t.c.id, t.c.fk) < func.ROW(3, 7))
1839 ... )
1840 >>> print(stmt)
1841 {printsql}SELECT t.id, t.fk
1842 FROM t
1843 WHERE (t.id, t.fk) > (:param_1, :param_2) AND ROW(t.id, t.fk) < ROW(:ROW_1, :ROW_2)
1845.. seealso::
1847 `PostgreSQL Row Constructors
1848 <https://www.postgresql.org/docs/current/sql-expressions.html#SQL-SYNTAX-ROW-CONSTRUCTORS>`_
1850 `PostgreSQL Row Constructor Comparison
1851 <https://www.postgresql.org/docs/current/functions-comparisons.html#ROW-WISE-COMPARISON>`_
1853Table Types passed to Functions
1854^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
1856PostgreSQL supports passing a table as an argument to a function, which is
1857known as a "record" type. SQLAlchemy :class:`_sql.FromClause` objects
1858such as :class:`_schema.Table` support this special form using the
1859:meth:`_sql.FromClause.table_valued` method, which is comparable to the
1860:meth:`_functions.FunctionElement.table_valued` method except that the collection
1861of columns is already established by that of the :class:`_sql.FromClause`
1862itself:
1864.. sourcecode:: pycon+sql
1866 >>> from sqlalchemy import table, column, func, select
1867 >>> a = table("a", column("id"), column("x"), column("y"))
1868 >>> stmt = select(func.row_to_json(a.table_valued()))
1869 >>> print(stmt)
1870 {printsql}SELECT row_to_json(a) AS row_to_json_1
1871 FROM a
1873.. versionadded:: 1.4.0b2
1876.. _postgresql_collation:
1878Schema-Qualified Collations
1879----------------------------
1881PostgreSQL supports collations that are qualified by a schema name, such as
1882``CREATE COLLATION my_schema.my_collation (...)``. To refer to such a
1883collation, use the :paramref:`.String.collation_schema` parameter (or
1884:paramref:`_postgresql.DOMAIN.collation_schema` for a :class:`_postgresql.DOMAIN`)
1885in conjunction with :paramref:`.String.collation`, rather than attempting to
1886embed the schema name inside the ``collation`` string itself::
1888 Column(
1889 "data",
1890 String(collation="my_collation", collation_schema="my_schema"),
1891 )
1893The above renders DDL similar to:
1895.. sourcecode:: sql
1897 data VARCHAR COLLATE "my_schema"."my_collation"
1899.. versionadded:: 2.1
1901.. seealso::
1903 :paramref:`.String.collation`
1905 :paramref:`.String.collation_schema`
1907""" # noqa: E501
1909from __future__ import annotations
1911from collections import defaultdict
1912from functools import lru_cache
1913import re
1914from typing import Any
1915from typing import cast
1916from typing import Dict
1917from typing import List
1918from typing import Optional
1919from typing import Tuple
1920from typing import TYPE_CHECKING
1921from typing import TypedDict
1922from typing import Union
1924from . import arraylib as _array
1925from . import json as _json
1926from . import pg_catalog
1927from . import ranges as _ranges
1928from .ext import _regconfig_fn
1929from .ext import aggregate_order_by
1930from .hstore import HSTORE
1931from .named_types import CreateDomainType as CreateDomainType # noqa: F401
1932from .named_types import CreateEnumType as CreateEnumType # noqa: F401
1933from .named_types import DOMAIN as DOMAIN # noqa: F401
1934from .named_types import DropDomainType as DropDomainType # noqa: F401
1935from .named_types import DropEnumType as DropEnumType # noqa: F401
1936from .named_types import ENUM as ENUM # noqa: F401
1937from .named_types import NamedType as NamedType # noqa: F401
1938from .types import _DECIMAL_TYPES # noqa: F401
1939from .types import _FLOAT_TYPES # noqa: F401
1940from .types import _INT_TYPES # noqa: F401
1941from .types import BIT as BIT
1942from .types import BYTEA as BYTEA
1943from .types import CIDR as CIDR
1944from .types import CITEXT as CITEXT
1945from .types import INET as INET
1946from .types import INTERVAL as INTERVAL
1947from .types import MACADDR as MACADDR
1948from .types import MACADDR8 as MACADDR8
1949from .types import MONEY as MONEY
1950from .types import OID as OID
1951from .types import PGBit as PGBit # noqa: F401
1952from .types import PGCidr as PGCidr # noqa: F401
1953from .types import PGInet as PGInet # noqa: F401
1954from .types import PGInterval as PGInterval # noqa: F401
1955from .types import PGMacAddr as PGMacAddr # noqa: F401
1956from .types import PGMacAddr8 as PGMacAddr8 # noqa: F401
1957from .types import PGUuid as PGUuid
1958from .types import REGCLASS as REGCLASS
1959from .types import REGCONFIG as REGCONFIG # noqa: F401
1960from .types import TIME as TIME
1961from .types import TIMESTAMP as TIMESTAMP
1962from .types import TSVECTOR as TSVECTOR
1963from ... import exc
1964from ... import schema
1965from ... import select
1966from ... import sql
1967from ... import util
1968from ...engine import characteristics
1969from ...engine import default
1970from ...engine import interfaces
1971from ...engine import ObjectKind
1972from ...engine import ObjectScope
1973from ...engine import reflection
1974from ...engine import URL
1975from ...engine.reflection import ReflectionDefaults
1976from ...sql import bindparam
1977from ...sql import coercions
1978from ...sql import compiler
1979from ...sql import elements
1980from ...sql import expression
1981from ...sql import functions
1982from ...sql import roles
1983from ...sql import sqltypes
1984from ...sql import util as sql_util
1985from ...sql.base import DialectKWArgConst
1986from ...sql.compiler import InsertmanyvaluesSentinelOpts
1987from ...sql.visitors import InternalTraversal
1988from ...types import BIGINT
1989from ...types import BOOLEAN
1990from ...types import CHAR
1991from ...types import DATE
1992from ...types import DOUBLE_PRECISION
1993from ...types import FLOAT
1994from ...types import INTEGER
1995from ...types import NUMERIC
1996from ...types import REAL
1997from ...types import SMALLINT
1998from ...types import TEXT
1999from ...types import UUID as UUID
2000from ...types import VARCHAR
2002IDX_USING = re.compile(r"^(?:btree|hash|gist|gin|[\w_]+)$", re.I)
2004RESERVED_WORDS = {
2005 "all",
2006 "analyse",
2007 "analyze",
2008 "and",
2009 "any",
2010 "array",
2011 "as",
2012 "asc",
2013 "asymmetric",
2014 "both",
2015 "case",
2016 "cast",
2017 "check",
2018 "collate",
2019 "column",
2020 "constraint",
2021 "create",
2022 "current_catalog",
2023 "current_date",
2024 "current_role",
2025 "current_time",
2026 "current_timestamp",
2027 "current_user",
2028 "default",
2029 "deferrable",
2030 "desc",
2031 "distinct",
2032 "do",
2033 "else",
2034 "end",
2035 "except",
2036 "false",
2037 "fetch",
2038 "for",
2039 "foreign",
2040 "from",
2041 "grant",
2042 "group",
2043 "having",
2044 "in",
2045 "initially",
2046 "intersect",
2047 "into",
2048 "leading",
2049 "limit",
2050 "localtime",
2051 "localtimestamp",
2052 "new",
2053 "not",
2054 "null",
2055 "of",
2056 "off",
2057 "offset",
2058 "old",
2059 "on",
2060 "only",
2061 "or",
2062 "order",
2063 "placing",
2064 "primary",
2065 "references",
2066 "returning",
2067 "select",
2068 "session_user",
2069 "some",
2070 "symmetric",
2071 "table",
2072 "then",
2073 "to",
2074 "trailing",
2075 "true",
2076 "union",
2077 "unique",
2078 "user",
2079 "using",
2080 "variadic",
2081 "when",
2082 "where",
2083 "window",
2084 "with",
2085 "authorization",
2086 "between",
2087 "binary",
2088 "cross",
2089 "current_schema",
2090 "freeze",
2091 "full",
2092 "ilike",
2093 "inner",
2094 "is",
2095 "isnull",
2096 "join",
2097 "left",
2098 "like",
2099 "natural",
2100 "notnull",
2101 "outer",
2102 "over",
2103 "overlaps",
2104 "right",
2105 "similar",
2106 "verbose",
2107}
2110colspecs = {
2111 sqltypes.ARRAY: _array.ARRAY,
2112 sqltypes.Interval: INTERVAL,
2113 sqltypes.Enum: ENUM,
2114 sqltypes.JSON.JSONPathType: _json.JSONPATH,
2115 sqltypes.JSON: _json.JSON,
2116 sqltypes.Uuid: PGUuid,
2117}
2120ischema_names = {
2121 "_array": _array.ARRAY,
2122 "hstore": HSTORE,
2123 "json": _json.JSON,
2124 "jsonb": _json.JSONB,
2125 "int4range": _ranges.INT4RANGE,
2126 "int8range": _ranges.INT8RANGE,
2127 "numrange": _ranges.NUMRANGE,
2128 "daterange": _ranges.DATERANGE,
2129 "tsrange": _ranges.TSRANGE,
2130 "tstzrange": _ranges.TSTZRANGE,
2131 "int4multirange": _ranges.INT4MULTIRANGE,
2132 "int8multirange": _ranges.INT8MULTIRANGE,
2133 "nummultirange": _ranges.NUMMULTIRANGE,
2134 "datemultirange": _ranges.DATEMULTIRANGE,
2135 "tsmultirange": _ranges.TSMULTIRANGE,
2136 "tstzmultirange": _ranges.TSTZMULTIRANGE,
2137 "integer": INTEGER,
2138 "bigint": BIGINT,
2139 "smallint": SMALLINT,
2140 "character varying": VARCHAR,
2141 "character": CHAR,
2142 '"char"': sqltypes.String,
2143 "name": sqltypes.String,
2144 "text": TEXT,
2145 "numeric": NUMERIC,
2146 "float": FLOAT,
2147 "real": REAL,
2148 "inet": INET,
2149 "cidr": CIDR,
2150 "citext": CITEXT,
2151 "uuid": UUID,
2152 "bit": BIT,
2153 "bit varying": BIT,
2154 "macaddr": MACADDR,
2155 "macaddr8": MACADDR8,
2156 "money": MONEY,
2157 "oid": OID,
2158 "regclass": REGCLASS,
2159 "double precision": DOUBLE_PRECISION,
2160 "timestamp": TIMESTAMP,
2161 "timestamp with time zone": TIMESTAMP,
2162 "timestamp without time zone": TIMESTAMP,
2163 "time with time zone": TIME,
2164 "time without time zone": TIME,
2165 "date": DATE,
2166 "time": TIME,
2167 "bytea": BYTEA,
2168 "boolean": BOOLEAN,
2169 "interval": INTERVAL,
2170 "tsvector": TSVECTOR,
2171}
2174class PGCompiler(compiler.SQLCompiler):
2175 def visit_to_tsvector_func(self, element, **kw):
2176 return self._assert_pg_ts_ext(element, **kw)
2178 def visit_to_tsquery_func(self, element, **kw):
2179 return self._assert_pg_ts_ext(element, **kw)
2181 def visit_plainto_tsquery_func(self, element, **kw):
2182 return self._assert_pg_ts_ext(element, **kw)
2184 def visit_phraseto_tsquery_func(self, element, **kw):
2185 return self._assert_pg_ts_ext(element, **kw)
2187 def visit_websearch_to_tsquery_func(self, element, **kw):
2188 return self._assert_pg_ts_ext(element, **kw)
2190 def visit_ts_headline_func(self, element, **kw):
2191 return self._assert_pg_ts_ext(element, **kw)
2193 def _assert_pg_ts_ext(self, element, **kw):
2194 if not isinstance(element, _regconfig_fn):
2195 # other options here include trying to rewrite the function
2196 # with the correct types. however, that means we have to
2197 # "un-SQL-ize" the first argument, which can't work in a
2198 # generalized way. Also, parent compiler class has already added
2199 # the incorrect return type to the result map. So let's just
2200 # make sure the function we want is used up front.
2202 raise exc.CompileError(
2203 f'Can\'t compile "{element.name}()" full text search '
2204 f"function construct that does not originate from the "
2205 f'"sqlalchemy.dialects.postgresql" package. '
2206 f'Please ensure "import sqlalchemy.dialects.postgresql" is '
2207 f"called before constructing "
2208 f'"sqlalchemy.func.{element.name}()" to ensure registration '
2209 f"of the correct argument and return types."
2210 )
2212 return f"{element.name}{self.function_argspec(element, **kw)}"
2214 def render_bind_cast(self, type_, dbapi_type, sqltext):
2215 if dbapi_type._type_affinity is sqltypes.String and dbapi_type.length:
2216 # use VARCHAR with no length for VARCHAR cast.
2217 # see #9511
2218 dbapi_type = sqltypes.STRINGTYPE
2219 return f"""{sqltext}::{
2220 self.dialect.type_compiler_instance.process(
2221 dbapi_type, identifier_preparer=self.preparer
2222 )
2223 }"""
2225 def visit_array(self, element, **kw):
2226 if not element.clauses and not element.type.item_type._isnull:
2227 return "ARRAY[]::%s" % element.type.compile(self.dialect)
2228 return "ARRAY[%s]" % self.visit_clauselist(element, **kw)
2230 def visit_slice(self, element, **kw):
2231 return "%s:%s" % (
2232 self.process(element.start, **kw),
2233 self.process(element.stop, **kw),
2234 )
2236 def visit_bitwise_xor_op_binary(self, binary, operator, **kw):
2237 return self._generate_generic_binary(binary, " # ", **kw)
2239 def visit_json_getitem_op_binary(
2240 self, binary, operator, _cast_applied=False, **kw
2241 ):
2242 if (
2243 not _cast_applied
2244 and binary.type._type_affinity is not sqltypes.JSON
2245 ):
2246 kw["_cast_applied"] = True
2247 return self.process(sql.cast(binary, binary.type), **kw)
2249 kw["eager_grouping"] = True
2251 if (
2252 not _cast_applied
2253 and isinstance(binary.left.type, _json.JSONB)
2254 and self.dialect._supports_jsonb_subscripting
2255 ):
2256 left = binary.left
2257 if isinstance(left, (functions.FunctionElement, elements.Cast)):
2258 left = elements.Grouping(left)
2260 # for pg14+JSONB use subscript notation: col['key'] instead
2261 # of col -> 'key'
2262 return "%s[%s]" % (
2263 self.process(left, **kw),
2264 self.process(binary.right, **kw),
2265 )
2266 else:
2267 # Fall back to arrow notation for older versions or when cast
2268 # is applied
2269 return self._generate_generic_binary(
2270 binary, " -> " if not _cast_applied else " ->> ", **kw
2271 )
2273 def visit_json_path_getitem_op_binary(
2274 self, binary, operator, _cast_applied=False, **kw
2275 ):
2276 if (
2277 not _cast_applied
2278 and binary.type._type_affinity is not sqltypes.JSON
2279 ):
2280 kw["_cast_applied"] = True
2281 return self.process(sql.cast(binary, binary.type), **kw)
2283 kw["eager_grouping"] = True
2284 return self._generate_generic_binary(
2285 binary, " #> " if not _cast_applied else " #>> ", **kw
2286 )
2288 def visit_hstore_getitem_op_binary(self, binary, operator, **kw):
2289 kw["eager_grouping"] = True
2291 if self.dialect._supports_jsonb_subscripting:
2292 # use subscript notation: col['key'] instead of col -> 'key'
2293 # For function calls, wrap in parentheses: (func())[key]
2294 left_str = self.process(binary.left, **kw)
2295 if isinstance(binary.left, sql.functions.FunctionElement):
2296 left_str = f"({left_str})"
2297 return "%s[%s]" % (
2298 left_str,
2299 self.process(binary.right, **kw),
2300 )
2301 else:
2302 # Fall back to arrow notation for older versions
2303 return self._generate_generic_binary(binary, " -> ", **kw)
2305 def visit_getitem_binary(self, binary, operator, **kw):
2306 return "%s[%s]" % (
2307 self.process(binary.left, **kw),
2308 self.process(binary.right, **kw),
2309 )
2311 def visit_aggregate_order_by(self, element, **kw):
2312 return "%s ORDER BY %s" % (
2313 self.process(element.target, **kw),
2314 self.process(element.order_by, **kw),
2315 )
2317 def visit_match_op_binary(self, binary, operator, **kw):
2318 if "postgresql_regconfig" in binary.modifiers:
2319 regconfig = self.render_literal_value(
2320 binary.modifiers["postgresql_regconfig"], sqltypes.STRINGTYPE
2321 )
2322 if regconfig:
2323 return "%s @@ plainto_tsquery(%s, %s)" % (
2324 self.process(binary.left, **kw),
2325 regconfig,
2326 self.process(binary.right, **kw),
2327 )
2328 return "%s @@ plainto_tsquery(%s)" % (
2329 self.process(binary.left, **kw),
2330 self.process(binary.right, **kw),
2331 )
2333 def visit_ilike_case_insensitive_operand(self, element, **kw):
2334 return element.element._compiler_dispatch(self, **kw)
2336 def visit_ilike_op_binary(self, binary, operator, **kw):
2337 escape = binary.modifiers.get("escape", None)
2339 return "%s ILIKE %s" % (
2340 self.process(binary.left, **kw),
2341 self.process(binary.right, **kw),
2342 ) + (
2343 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
2344 if escape is not None
2345 else ""
2346 )
2348 def visit_not_ilike_op_binary(self, binary, operator, **kw):
2349 escape = binary.modifiers.get("escape", None)
2350 return "%s NOT ILIKE %s" % (
2351 self.process(binary.left, **kw),
2352 self.process(binary.right, **kw),
2353 ) + (
2354 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
2355 if escape is not None
2356 else ""
2357 )
2359 def _regexp_match(self, base_op, binary, operator, kw):
2360 flags = binary.modifiers["flags"]
2361 if flags is None:
2362 return self._generate_generic_binary(
2363 binary, " %s " % base_op, **kw
2364 )
2365 if flags == "i":
2366 return self._generate_generic_binary(
2367 binary, " %s* " % base_op, **kw
2368 )
2369 return "%s %s CONCAT('(?', %s, ')', %s)" % (
2370 self.process(binary.left, **kw),
2371 base_op,
2372 self.render_literal_value(flags, sqltypes.STRINGTYPE),
2373 self.process(binary.right, **kw),
2374 )
2376 def visit_regexp_match_op_binary(self, binary, operator, **kw):
2377 return self._regexp_match("~", binary, operator, kw)
2379 def visit_not_regexp_match_op_binary(self, binary, operator, **kw):
2380 return self._regexp_match("!~", binary, operator, kw)
2382 def visit_regexp_replace_op_binary(self, binary, operator, **kw):
2383 string = self.process(binary.left, **kw)
2384 pattern_replace = self.process(binary.right, **kw)
2385 flags = binary.modifiers["flags"]
2386 if flags is None:
2387 return "REGEXP_REPLACE(%s, %s)" % (
2388 string,
2389 pattern_replace,
2390 )
2391 else:
2392 return "REGEXP_REPLACE(%s, %s, %s)" % (
2393 string,
2394 pattern_replace,
2395 self.render_literal_value(flags, sqltypes.STRINGTYPE),
2396 )
2398 def visit_empty_set_expr(self, element_types, **kw):
2399 # cast the empty set to the type we are comparing against. if
2400 # we are comparing against the null type, pick an arbitrary
2401 # datatype for the empty set
2402 return "SELECT %s WHERE 1!=1" % (
2403 ", ".join(
2404 "CAST(NULL AS %s)"
2405 % self.dialect.type_compiler_instance.process(
2406 INTEGER() if type_._isnull else type_
2407 )
2408 for type_ in element_types or [INTEGER()]
2409 ),
2410 )
2412 def render_literal_value(self, value, type_):
2413 value = super().render_literal_value(value, type_)
2415 if self.dialect._backslash_escapes:
2416 value = value.replace("\\", "\\\\")
2417 return value
2419 def visit_aggregate_strings_func(self, fn, **kw):
2420 return super().visit_aggregate_strings_func(
2421 fn, use_function_name="string_agg", **kw
2422 )
2424 def visit_pow_func(self, fn, **kw):
2425 return f"power{self.function_argspec(fn)}"
2427 def visit_sequence(self, seq, **kw):
2428 return "nextval(%s)" % self.preparer._format_sequence_string_literal(
2429 seq
2430 )
2432 def limit_clause(self, select, **kw):
2433 text = ""
2434 if select._limit_clause is not None:
2435 text += " \n LIMIT " + self.process(select._limit_clause, **kw)
2436 if select._offset_clause is not None:
2437 if select._limit_clause is None:
2438 text += "\n LIMIT ALL"
2439 text += " OFFSET " + self.process(select._offset_clause, **kw)
2440 return text
2442 def format_from_hint_text(self, sqltext, table, hint, iscrud):
2443 if hint.upper() != "ONLY":
2444 raise exc.CompileError("Unrecognized hint: %r" % hint)
2445 return "ONLY " + sqltext
2447 def get_select_precolumns(self, select, **kw):
2448 # Do not call super().get_select_precolumns because
2449 # it will warn/raise when distinct on is present
2450 if select._distinct or select._distinct_on:
2451 if select._distinct_on:
2452 return (
2453 "DISTINCT ON ("
2454 + ", ".join(
2455 [
2456 self.process(col, **kw)
2457 for col in select._distinct_on
2458 ]
2459 )
2460 + ") "
2461 )
2462 else:
2463 return "DISTINCT "
2464 else:
2465 return ""
2467 def visit_postgresql_distinct_on(self, element, **kw):
2468 if self.stack[-1]["selectable"]._distinct_on:
2469 raise exc.CompileError(
2470 "Cannot mix ``select.ext(distinct_on(...))`` and "
2471 "``select.distinct(...)``"
2472 )
2474 if element._distinct_on:
2475 cols = ", ".join(
2476 self.process(col, **kw) for col in element._distinct_on
2477 )
2478 return f"ON ({cols})"
2479 else:
2480 return None
2482 def for_update_clause(self, select, **kw):
2483 if select._for_update_arg.read:
2484 if select._for_update_arg.key_share:
2485 tmp = " FOR KEY SHARE"
2486 else:
2487 tmp = " FOR SHARE"
2488 elif select._for_update_arg.key_share:
2489 tmp = " FOR NO KEY UPDATE"
2490 else:
2491 tmp = " FOR UPDATE"
2493 if select._for_update_arg.of:
2494 tables = util.OrderedSet()
2495 for c in select._for_update_arg.of:
2496 tables.update(sql_util.surface_selectables_only(c))
2498 of_kw = dict(kw)
2499 of_kw.update(ashint=True, use_schema=False)
2500 tmp += " OF " + ", ".join(
2501 self.process(table, **of_kw) for table in tables
2502 )
2504 if select._for_update_arg.nowait:
2505 tmp += " NOWAIT"
2506 if select._for_update_arg.skip_locked:
2507 tmp += " SKIP LOCKED"
2509 return tmp
2511 def visit_substring_func(self, func, **kw):
2512 s = self.process(func.clauses.clauses[0], **kw)
2513 start = self.process(func.clauses.clauses[1], **kw)
2514 if len(func.clauses.clauses) > 2:
2515 length = self.process(func.clauses.clauses[2], **kw)
2516 return "SUBSTRING(%s FROM %s FOR %s)" % (s, start, length)
2517 else:
2518 return "SUBSTRING(%s FROM %s)" % (s, start)
2520 def _on_conflict_target(self, clause, **kw):
2521 if clause.constraint_target is not None:
2522 # target may be a name of an Index, UniqueConstraint or
2523 # ExcludeConstraint. While there is a separate
2524 # "max_identifier_length" for indexes, PostgreSQL uses the same
2525 # length for all objects so we can use
2526 # truncate_and_render_constraint_name
2527 target_text = (
2528 "ON CONSTRAINT %s"
2529 % self.preparer.truncate_and_render_constraint_name(
2530 clause.constraint_target
2531 )
2532 )
2533 elif clause.inferred_target_elements is not None:
2534 target_text = "(%s)" % ", ".join(
2535 (
2536 self.preparer.quote(c)
2537 if isinstance(c, str)
2538 else self.process(c, include_table=False, use_schema=False)
2539 )
2540 for c in clause.inferred_target_elements
2541 )
2542 if clause.inferred_target_whereclause is not None:
2543 whereclause_kw = dict(kw)
2544 whereclause_kw.update(include_table=False, use_schema=False)
2545 target_text += " WHERE %s" % self.process(
2546 clause.inferred_target_whereclause,
2547 **whereclause_kw,
2548 )
2549 else:
2550 target_text = ""
2552 return target_text
2554 def visit_on_conflict_do_nothing(self, on_conflict, **kw):
2555 target_text = self._on_conflict_target(on_conflict, **kw)
2557 if target_text:
2558 return "ON CONFLICT %s DO NOTHING" % target_text
2559 else:
2560 return "ON CONFLICT DO NOTHING"
2562 def visit_on_conflict_do_update(self, on_conflict, **kw):
2563 clause = on_conflict
2565 target_text = self._on_conflict_target(on_conflict, **kw)
2567 action_set_ops = []
2569 set_parameters = dict(clause.update_values_to_set)
2570 # create a list of column assignment clauses as tuples
2572 insert_statement = self.stack[-1]["selectable"]
2573 cols = insert_statement.table.c
2574 set_kw = dict(kw)
2575 set_kw.update(use_schema=False)
2576 for c in cols:
2577 col_key = c.key
2579 if col_key in set_parameters:
2580 value = set_parameters.pop(col_key)
2581 elif c in set_parameters:
2582 value = set_parameters.pop(c)
2583 else:
2584 continue
2586 assert not coercions._is_literal(value)
2587 if (
2588 isinstance(value, elements.BindParameter)
2589 and value.type._isnull
2590 ):
2591 value = value._with_binary_element_type(c.type)
2593 value_text = self.process(
2594 value.self_group(), is_upsert_set=True, **set_kw
2595 )
2597 key_text = self.preparer.quote(c.name)
2598 action_set_ops.append("%s = %s" % (key_text, value_text))
2600 # check for names that don't match columns
2601 if set_parameters:
2602 util.warn(
2603 "Additional column names not matching "
2604 "any column keys in table '%s': %s"
2605 % (
2606 self.current_executable.table.name,
2607 (", ".join("'%s'" % c for c in set_parameters)),
2608 )
2609 )
2610 for k, v in set_parameters.items():
2611 key_text = (
2612 self.preparer.quote(k)
2613 if isinstance(k, str)
2614 else self.process(k, use_schema=False)
2615 )
2616 value_text = self.process(
2617 coercions.expect(roles.ExpressionElementRole, v),
2618 is_upsert_set=True,
2619 **set_kw,
2620 )
2621 action_set_ops.append("%s = %s" % (key_text, value_text))
2623 action_text = ", ".join(action_set_ops)
2624 if clause.update_whereclause is not None:
2625 where_kw = dict(kw)
2626 where_kw.update(include_table=True, use_schema=False)
2627 action_text += " WHERE %s" % self.process(
2628 clause.update_whereclause, **where_kw
2629 )
2631 return "ON CONFLICT %s DO UPDATE SET %s" % (target_text, action_text)
2633 def update_from_clause(
2634 self, update_stmt, from_table, extra_froms, from_hints, **kw
2635 ):
2636 kw["asfrom"] = True
2637 return "FROM " + ", ".join(
2638 t._compiler_dispatch(self, fromhints=from_hints, **kw)
2639 for t in extra_froms
2640 )
2642 def delete_extra_from_clause(
2643 self, delete_stmt, from_table, extra_froms, from_hints, **kw
2644 ):
2645 """Render the DELETE .. USING clause specific to PostgreSQL."""
2646 kw["asfrom"] = True
2647 return "USING " + ", ".join(
2648 t._compiler_dispatch(self, fromhints=from_hints, **kw)
2649 for t in extra_froms
2650 )
2652 def fetch_clause(self, select, **kw):
2653 # pg requires parens for non literal clauses. It's also required for
2654 # bind parameters if a ::type casts is used by the driver (asyncpg),
2655 # so it's easiest to just always add it
2656 text = ""
2657 if select._offset_clause is not None:
2658 text += "\n OFFSET (%s) ROWS" % self.process(
2659 select._offset_clause, **kw
2660 )
2661 if select._fetch_clause is not None:
2662 text += "\n FETCH FIRST (%s)%s ROWS %s" % (
2663 self.process(select._fetch_clause, **kw),
2664 " PERCENT" if select._fetch_clause_options["percent"] else "",
2665 (
2666 "WITH TIES"
2667 if select._fetch_clause_options["with_ties"]
2668 else "ONLY"
2669 ),
2670 )
2671 return text
2674class PGDDLCompiler(compiler.DDLCompiler):
2675 def get_column_specification(self, column, **kwargs):
2676 colspec = self.preparer.format_column(column)
2677 impl_type = column.type.dialect_impl(self.dialect)
2678 if isinstance(impl_type, sqltypes.TypeDecorator):
2679 impl_type = impl_type.impl
2681 has_identity = (
2682 column.identity is not None
2683 and self.dialect.supports_identity_columns
2684 )
2686 if (
2687 column.primary_key
2688 and column is column.table._autoincrement_column
2689 and (
2690 self.dialect.supports_smallserial
2691 or not isinstance(impl_type, sqltypes.SmallInteger)
2692 )
2693 and not has_identity
2694 and (
2695 column.default is None
2696 or (
2697 isinstance(column.default, schema.Sequence)
2698 and column.default.optional
2699 )
2700 )
2701 ):
2702 if isinstance(impl_type, sqltypes.BigInteger):
2703 colspec += " BIGSERIAL"
2704 elif isinstance(impl_type, sqltypes.SmallInteger):
2705 colspec += " SMALLSERIAL"
2706 else:
2707 colspec += " SERIAL"
2708 else:
2709 colspec += " " + self.dialect.type_compiler_instance.process(
2710 column.type,
2711 type_expression=column,
2712 identifier_preparer=self.preparer,
2713 )
2714 default = self.get_column_default_string(column)
2715 if default is not None:
2716 colspec += " DEFAULT " + default
2718 if column.computed is not None:
2719 colspec += " " + self.process(column.computed)
2720 if has_identity:
2721 colspec += " " + self.process(column.identity)
2723 if not column.nullable and not has_identity:
2724 colspec += " NOT NULL"
2725 elif column.nullable and has_identity:
2726 colspec += " NULL"
2727 return colspec
2729 def _define_constraint_validity(self, constraint):
2730 not_valid = constraint.dialect_options["postgresql"]["not_valid"]
2731 return " NOT VALID" if not_valid else ""
2733 def _define_include(self, obj):
2734 includeclause = obj.dialect_options["postgresql"]["include"]
2735 if not includeclause:
2736 return ""
2737 inclusions = [
2738 obj.table.c[col] if isinstance(col, str) else col
2739 for col in includeclause
2740 ]
2741 return " INCLUDE (%s)" % ", ".join(
2742 [self.preparer.quote(c.name) for c in inclusions]
2743 )
2745 def visit_check_constraint(self, constraint, **kw):
2746 if constraint._type_bound:
2747 typ = list(constraint.columns)[0].type
2748 if (
2749 isinstance(typ, sqltypes.ARRAY)
2750 and isinstance(typ.item_type, sqltypes.Enum)
2751 and not typ.item_type.native_enum
2752 ):
2753 raise exc.CompileError(
2754 "PostgreSQL dialect cannot produce the CHECK constraint "
2755 "for ARRAY of non-native ENUM; please specify "
2756 "create_constraint=False on this Enum datatype."
2757 )
2759 text = super().visit_check_constraint(constraint)
2760 text += self._define_constraint_validity(constraint)
2761 return text
2763 def visit_foreign_key_constraint(self, constraint, **kw):
2764 text = super().visit_foreign_key_constraint(constraint)
2765 text += self._define_constraint_validity(constraint)
2766 return text
2768 def visit_primary_key_constraint(self, constraint, **kw):
2769 text = self.define_constraint_preamble(constraint, **kw)
2770 text += self.define_primary_key_body(constraint, **kw)
2771 text += self._define_include(constraint)
2772 text += self.define_constraint_deferrability(constraint)
2773 return text
2775 def visit_unique_constraint(self, constraint, **kw):
2776 if len(constraint) == 0:
2777 return ""
2778 text = self.define_constraint_preamble(constraint, **kw)
2779 text += self.define_unique_body(constraint, **kw)
2780 text += self._define_include(constraint)
2781 text += self.define_constraint_deferrability(constraint)
2782 return text
2784 @util.memoized_property
2785 def _fk_ondelete_pattern(self):
2786 # a keyword phrase, optionally followed by the column list accepted
2787 # by SET NULL / SET DEFAULT. column names may be double-quoted, as
2788 # rendered by pg_get_constraintdef() when reflected
2789 ident = r'(?:[\w$]+|"(?:[^"]|"")+")'
2790 return re.compile(
2791 r"^\w+(?:\s+\w+)*"
2792 rf"(?:\s*\(\s*{ident}(?:\s*,\s*{ident})*\s*\))?$"
2793 )
2795 def define_constraint_ondelete_cascade(self, constraint):
2796 return " ON DELETE %s" % self.preparer.validate_sql_phrase(
2797 constraint.ondelete, self._fk_ondelete_pattern
2798 )
2800 def visit_create_enum_type(self, create, **kw):
2801 type_ = create.element
2803 return "CREATE TYPE %s AS ENUM (%s)" % (
2804 self.preparer.format_type(type_),
2805 ", ".join(
2806 self.sql_compiler.process(sql.literal(e), literal_binds=True)
2807 for e in type_.enums
2808 ),
2809 )
2811 def visit_drop_enum_type(self, drop, **kw):
2812 type_ = drop.element
2814 return "DROP TYPE %s" % (self.preparer.format_type(type_))
2816 def visit_create_domain_type(self, create, **kw):
2817 domain: DOMAIN = create.element
2819 options = []
2820 if domain.collation is not None:
2821 collation = self.preparer.format_collation(
2822 domain.collation, domain.collation_schema
2823 )
2824 options.append(f"COLLATE {collation}")
2825 if domain.default is not None:
2826 default = self.render_default_string(domain.default)
2827 options.append(f"DEFAULT {default}")
2828 if domain.constraint_name is not None:
2829 name = self.preparer.truncate_and_render_constraint_name(
2830 domain.constraint_name
2831 )
2832 options.append(f"CONSTRAINT {name}")
2833 if domain.not_null:
2834 options.append("NOT NULL")
2835 if domain.check is not None:
2836 check = self.sql_compiler.process(
2837 domain.check, include_table=False, literal_binds=True
2838 )
2839 options.append(f"CHECK ({check})")
2841 return (
2842 f"CREATE DOMAIN {self.preparer.format_type(domain)} AS "
2843 f"{self.type_compiler.process(domain.data_type)} "
2844 f"{' '.join(options)}"
2845 )
2847 def visit_drop_domain_type(self, drop, **kw):
2848 domain = drop.element
2849 return f"DROP DOMAIN {self.preparer.format_type(domain)}"
2851 def _prepare_withclause_opts(self, withclause):
2852 with_opts = []
2853 for param, value in withclause.items():
2854 if value is None:
2855 with_opts.append(param)
2856 elif isinstance(value, bool):
2857 with_opts.append(
2858 "%s = %s" % (param, "true" if value else "false")
2859 )
2860 else:
2861 with_opts.append("%s = %s" % (param, value))
2862 return ", ".join(with_opts)
2864 def create_table_select_suffixes(self, element, type_, **kw):
2865 if type_ == "view":
2866 withclause = element.dialect_options["postgresql"]["with"]
2867 if withclause:
2868 return "WITH (%s)" % self._prepare_withclause_opts(withclause)
2869 return ""
2871 def visit_create_index(self, create, **kw):
2872 preparer = self.preparer
2873 index = create.element
2874 self._verify_index_table(index)
2875 text = "CREATE "
2876 if index.unique:
2877 text += "UNIQUE "
2879 text += "INDEX "
2881 if self.dialect._supports_create_index_concurrently:
2882 concurrently = index.dialect_options["postgresql"]["concurrently"]
2883 if concurrently:
2884 text += "CONCURRENTLY "
2886 if create.if_not_exists:
2887 text += "IF NOT EXISTS "
2889 text += "%s ON %s " % (
2890 self._prepared_index_name(index, include_schema=False),
2891 preparer.format_table(index.table),
2892 )
2894 using = index.dialect_options["postgresql"]["using"]
2895 if using:
2896 text += (
2897 "USING %s "
2898 % self.preparer.validate_sql_phrase(using, IDX_USING).lower()
2899 )
2901 ops = index.dialect_options["postgresql"]["ops"]
2902 text += "(%s)" % (
2903 ", ".join(
2904 [
2905 self.sql_compiler.process(
2906 (
2907 expr.self_group()
2908 if not isinstance(expr, expression.ColumnClause)
2909 else expr
2910 ),
2911 include_table=False,
2912 literal_binds=True,
2913 )
2914 + (
2915 (" " + ops[expr.key])
2916 if hasattr(expr, "key") and expr.key in ops
2917 else ""
2918 )
2919 for expr in index.expressions
2920 ]
2921 )
2922 )
2924 text += self._define_include(index)
2926 nulls_not_distinct = index.dialect_options["postgresql"][
2927 "nulls_not_distinct"
2928 ]
2929 if nulls_not_distinct is True:
2930 text += " NULLS NOT DISTINCT"
2931 elif nulls_not_distinct is False:
2932 text += " NULLS DISTINCT"
2934 withclause = index.dialect_options["postgresql"]["with"]
2935 if withclause:
2936 text += " WITH (%s)" % self._prepare_withclause_opts(withclause)
2938 tablespace_name = index.dialect_options["postgresql"]["tablespace"]
2939 if tablespace_name:
2940 text += " TABLESPACE %s" % preparer.quote(tablespace_name)
2942 whereclause = index.dialect_options["postgresql"]["where"]
2943 if whereclause is not None:
2944 whereclause = coercions.expect(
2945 roles.DDLExpressionRole, whereclause
2946 )
2948 where_compiled = self.sql_compiler.process(
2949 whereclause, include_table=False, literal_binds=True
2950 )
2951 text += " WHERE " + where_compiled
2953 return text
2955 def define_unique_constraint_distinct(self, constraint, **kw):
2956 nulls_not_distinct = constraint.dialect_options["postgresql"][
2957 "nulls_not_distinct"
2958 ]
2959 if nulls_not_distinct is True:
2960 nulls_not_distinct_param = "NULLS NOT DISTINCT "
2961 elif nulls_not_distinct is False:
2962 nulls_not_distinct_param = "NULLS DISTINCT "
2963 else:
2964 nulls_not_distinct_param = ""
2965 return nulls_not_distinct_param
2967 def visit_drop_index(self, drop, **kw):
2968 index = drop.element
2970 text = "\nDROP INDEX "
2972 if self.dialect._supports_drop_index_concurrently:
2973 concurrently = index.dialect_options["postgresql"]["concurrently"]
2974 if concurrently:
2975 text += "CONCURRENTLY "
2977 if drop.if_exists:
2978 text += "IF EXISTS "
2980 text += self._prepared_index_name(index, include_schema=True)
2981 return text
2983 def visit_exclude_constraint(self, constraint, **kw):
2984 text = ""
2985 if constraint.name is not None:
2986 text += "CONSTRAINT %s " % self.preparer.format_constraint(
2987 constraint
2988 )
2989 elements = []
2990 kw["include_table"] = False
2991 kw["literal_binds"] = True
2992 for expr, name, op in constraint._render_exprs:
2993 exclude_element = self.sql_compiler.process(expr, **kw) + (
2994 (" " + constraint.ops[expr.key])
2995 if hasattr(expr, "key") and expr.key in constraint.ops
2996 else ""
2997 )
2999 elements.append("%s WITH %s" % (exclude_element, op))
3000 text += "EXCLUDE USING %s (%s)" % (
3001 self.preparer.validate_sql_phrase(
3002 constraint.using, IDX_USING
3003 ).lower(),
3004 ", ".join(elements),
3005 )
3006 if constraint.where is not None:
3007 text += " WHERE (%s)" % self.sql_compiler.process(
3008 constraint.where, literal_binds=True
3009 )
3010 text += self.define_constraint_deferrability(constraint)
3011 return text
3013 def post_create_table(self, table):
3014 table_opts = []
3015 pg_opts = table.dialect_options["postgresql"]
3017 inherits = pg_opts.get("inherits")
3018 if inherits is not None:
3019 if not isinstance(inherits, (list, tuple)):
3020 inherits = (inherits,)
3021 table_opts.append(
3022 "\n INHERITS ("
3023 + ", ".join(self.preparer.quote(name) for name in inherits)
3024 + ")"
3025 )
3027 if pg_opts["partition_by"]:
3028 table_opts.append("\n PARTITION BY %s" % pg_opts["partition_by"])
3030 if pg_opts["using"]:
3031 table_opts.append("\n USING %s" % pg_opts["using"])
3033 if pg_opts["with"]:
3034 table_opts.append(
3035 "\n WITH (%s)" % self._prepare_withclause_opts(pg_opts["with"])
3036 )
3038 if pg_opts["with_oids"] is True:
3039 table_opts.append("\n WITH OIDS")
3040 elif pg_opts["with_oids"] is False:
3041 table_opts.append("\n WITHOUT OIDS")
3043 if pg_opts["on_commit"]:
3044 on_commit_options = pg_opts["on_commit"].replace("_", " ").upper()
3045 table_opts.append("\n ON COMMIT %s" % on_commit_options)
3047 if pg_opts["tablespace"]:
3048 tablespace_name = pg_opts["tablespace"]
3049 table_opts.append(
3050 "\n TABLESPACE %s" % self.preparer.quote(tablespace_name)
3051 )
3053 return "".join(table_opts)
3055 def visit_computed_column(self, generated, **kw):
3056 if self.dialect.supports_virtual_generated_columns:
3057 return super().visit_computed_column(generated, **kw)
3058 if generated.persisted is False:
3059 raise exc.CompileError(
3060 "PostrgreSQL computed columns do not support 'virtual' "
3061 "persistence; set the 'persisted' flag to None or True for "
3062 "PostgreSQL support."
3063 )
3064 elif generated.persisted is None:
3065 util.warn(
3066 f"Computed column {generated.column} is being created as "
3067 "'STORED' since the current PostgreSQL version does not "
3068 "support VIRTUAL columns. On PostgreSQL 18+, when "
3069 "'persisted' is not "
3070 "specified, no keyword will be rendered and VIRTUAL will be "
3071 "used by default. Set 'persisted=True' to ensure STORED "
3072 "behavior across all PostgreSQL versions."
3073 )
3075 return "GENERATED ALWAYS AS (%s) STORED" % self.sql_compiler.process(
3076 generated.sqltext, include_table=False, literal_binds=True
3077 )
3079 def visit_create_sequence(self, create, **kw):
3080 prefix = None
3081 if create.element.data_type is not None:
3082 prefix = " AS %s" % self.type_compiler.process(
3083 create.element.data_type
3084 )
3086 return super().visit_create_sequence(create, prefix=prefix, **kw)
3088 def _can_comment_on_constraint(self, ddl_instance):
3089 constraint = ddl_instance.element
3090 if constraint.name is None:
3091 raise exc.CompileError(
3092 f"Can't emit COMMENT ON for constraint {constraint!r}: "
3093 "it has no name"
3094 )
3095 if constraint.table is None:
3096 raise exc.CompileError(
3097 f"Can't emit COMMENT ON for constraint {constraint!r}: "
3098 "it has no associated table"
3099 )
3101 def visit_set_constraint_comment(self, create, **kw):
3102 self._can_comment_on_constraint(create)
3103 return "COMMENT ON CONSTRAINT %s ON %s IS %s" % (
3104 self.preparer.format_constraint(create.element),
3105 self.preparer.format_table(create.element.table),
3106 self.sql_compiler.render_literal_value(
3107 create.element.comment, sqltypes.String()
3108 ),
3109 )
3111 def visit_drop_constraint_comment(self, drop, **kw):
3112 self._can_comment_on_constraint(drop)
3113 return "COMMENT ON CONSTRAINT %s ON %s IS NULL" % (
3114 self.preparer.format_constraint(drop.element),
3115 self.preparer.format_table(drop.element.table),
3116 )
3119class PGTypeCompiler(compiler.GenericTypeCompiler):
3120 def visit_TSVECTOR(self, type_, **kw):
3121 return "TSVECTOR"
3123 def visit_TSQUERY(self, type_, **kw):
3124 return "TSQUERY"
3126 def visit_INET(self, type_, **kw):
3127 return "INET"
3129 def visit_CIDR(self, type_, **kw):
3130 return "CIDR"
3132 def visit_CITEXT(self, type_, **kw):
3133 return "CITEXT"
3135 def visit_MACADDR(self, type_, **kw):
3136 return "MACADDR"
3138 def visit_MACADDR8(self, type_, **kw):
3139 return "MACADDR8"
3141 def visit_MONEY(self, type_, **kw):
3142 return "MONEY"
3144 def visit_OID(self, type_, **kw):
3145 return "OID"
3147 def visit_REGCONFIG(self, type_, **kw):
3148 return "REGCONFIG"
3150 def visit_REGCLASS(self, type_, **kw):
3151 return "REGCLASS"
3153 def visit_FLOAT(self, type_, **kw):
3154 if not type_.precision:
3155 return "FLOAT"
3156 else:
3157 return "FLOAT(%(precision)s)" % {"precision": type_.precision}
3159 def visit_double(self, type_, **kw):
3160 return self.visit_DOUBLE_PRECISION(type, **kw)
3162 def visit_BIGINT(self, type_, **kw):
3163 return "BIGINT"
3165 def visit_HSTORE(self, type_, **kw):
3166 return "HSTORE"
3168 def visit_JSON(self, type_, **kw):
3169 return "JSON"
3171 def visit_JSONB(self, type_, **kw):
3172 return "JSONB"
3174 def visit_INT4MULTIRANGE(self, type_, **kw):
3175 return "INT4MULTIRANGE"
3177 def visit_INT8MULTIRANGE(self, type_, **kw):
3178 return "INT8MULTIRANGE"
3180 def visit_NUMMULTIRANGE(self, type_, **kw):
3181 return "NUMMULTIRANGE"
3183 def visit_DATEMULTIRANGE(self, type_, **kw):
3184 return "DATEMULTIRANGE"
3186 def visit_TSMULTIRANGE(self, type_, **kw):
3187 return "TSMULTIRANGE"
3189 def visit_TSTZMULTIRANGE(self, type_, **kw):
3190 return "TSTZMULTIRANGE"
3192 def visit_INT4RANGE(self, type_, **kw):
3193 return "INT4RANGE"
3195 def visit_INT8RANGE(self, type_, **kw):
3196 return "INT8RANGE"
3198 def visit_NUMRANGE(self, type_, **kw):
3199 return "NUMRANGE"
3201 def visit_DATERANGE(self, type_, **kw):
3202 return "DATERANGE"
3204 def visit_TSRANGE(self, type_, **kw):
3205 return "TSRANGE"
3207 def visit_TSTZRANGE(self, type_, **kw):
3208 return "TSTZRANGE"
3210 def visit_json_int_index(self, type_, **kw):
3211 return "INT"
3213 def visit_json_str_index(self, type_, **kw):
3214 return "TEXT"
3216 def visit_datetime(self, type_, **kw):
3217 return self.visit_TIMESTAMP(type_, **kw)
3219 def visit_enum(self, type_, **kw):
3220 if not type_.native_enum or not self.dialect.supports_native_enum:
3221 return super().visit_enum(type_, **kw)
3222 else:
3223 return self.visit_ENUM(type_, **kw)
3225 def visit_ENUM(self, type_, identifier_preparer=None, **kw):
3226 if identifier_preparer is None:
3227 identifier_preparer = self.dialect.identifier_preparer
3228 return identifier_preparer.format_type(type_)
3230 def visit_DOMAIN(self, type_, identifier_preparer=None, **kw):
3231 if identifier_preparer is None:
3232 identifier_preparer = self.dialect.identifier_preparer
3233 return identifier_preparer.format_type(type_)
3235 def visit_TIMESTAMP(self, type_, **kw):
3236 return "TIMESTAMP%s %s" % (
3237 (
3238 "(%d)" % type_.precision
3239 if getattr(type_, "precision", None) is not None
3240 else ""
3241 ),
3242 (type_.timezone and "WITH" or "WITHOUT") + " TIME ZONE",
3243 )
3245 def visit_TIME(self, type_, **kw):
3246 return "TIME%s %s" % (
3247 (
3248 "(%d)" % type_.precision
3249 if getattr(type_, "precision", None) is not None
3250 else ""
3251 ),
3252 (type_.timezone and "WITH" or "WITHOUT") + " TIME ZONE",
3253 )
3255 def visit_INTERVAL(self, type_, **kw):
3256 text = "INTERVAL"
3257 if type_.fields is not None:
3258 text += " " + type_.fields
3259 if type_.precision is not None:
3260 text += " (%d)" % type_.precision
3261 return text
3263 def visit_BIT(self, type_, **kw):
3264 if type_.varying:
3265 compiled = "BIT VARYING"
3266 if type_.length is not None:
3267 compiled += "(%d)" % type_.length
3268 else:
3269 compiled = "BIT(%d)" % type_.length
3270 return compiled
3272 def visit_uuid(self, type_, **kw):
3273 if type_.native_uuid:
3274 return self.visit_UUID(type_, **kw)
3275 else:
3276 return super().visit_uuid(type_, **kw)
3278 def visit_UUID(self, type_, **kw):
3279 return "UUID"
3281 def visit_large_binary(self, type_, **kw):
3282 return self.visit_BYTEA(type_, **kw)
3284 def visit_BYTEA(self, type_, **kw):
3285 return "BYTEA"
3287 def visit_ARRAY(self, type_, **kw):
3288 inner = self.process(type_.item_type, **kw)
3289 return re.sub(
3290 r"((?: COLLATE.*)?)$",
3291 (
3292 r"%s\1"
3293 % (
3294 "[]"
3295 * (type_.dimensions if type_.dimensions is not None else 1)
3296 )
3297 ),
3298 inner,
3299 count=1,
3300 )
3302 def visit_json_path(self, type_, **kw):
3303 return self.visit_JSONPATH(type_, **kw)
3305 def visit_JSONPATH(self, type_, **kw):
3306 return "JSONPATH"
3309class _CompilerSequence:
3310 """Minimal stand-in for :class:`.Sequence`.
3312 Used for the implicit sequence behind a SERIAL column, where no
3313 :class:`.Sequence` object exists but a name still has to be rendered
3314 through :meth:`.IdentifierPreparer.format_sequence`.
3316 """
3318 __slots__ = ("name", "schema")
3320 # the schema handed to us has already been resolved against the
3321 # schema translate map, so don't let format_sequence() translate it
3322 # a second time
3323 _use_schema_map = False
3325 def __init__(self, name, schema=None):
3326 self.name = name
3327 self.schema = schema
3330class PGIdentifierPreparer(compiler.IdentifierPreparer):
3331 reserved_words = RESERVED_WORDS
3333 def _format_sequence_string_literal(self, sequence):
3334 """Render a sequence name as a SQL string literal.
3336 ``nextval()`` and friends take the sequence as a string rather than
3337 as an identifier, so the quoted identifier produced by
3338 :meth:`.format_sequence` has to be escaped a second time for the
3339 enclosing literal.
3341 """
3342 return "'%s'" % self.format_sequence(
3343 sequence, use_schema=True
3344 ).replace("'", "''")
3346 def _unquote_identifier(self, value):
3347 if value[0] == self.initial_quote:
3348 value = value[1:-1].replace(
3349 self.escape_to_quote, self.escape_quote
3350 )
3351 return value
3353 def format_type(self, type_, use_schema=True):
3354 if not type_.name:
3355 raise exc.CompileError(
3356 f"PostgreSQL {type_.__class__.__name__} type requires a name."
3357 )
3359 name = self.quote(type_.name)
3360 effective_schema = self.schema_for_object(type_)
3362 # a built-in type with the same name will obscure this type, so raise
3363 # for that case. this applies really to any visible type with the same
3364 # name in any other visible schema that would not be appropriate for
3365 # us to check against, so this is not a robust check, but
3366 # at least do something for an obvious built-in name conflict
3367 if (
3368 effective_schema is None
3369 and type_.name in self.dialect.ischema_names
3370 ):
3371 raise exc.CompileError(
3372 f"{type_!r} has name "
3373 f"'{type_.name}' that matches an existing type, and "
3374 "requires an explicit schema name in order to be rendered "
3375 "in DDL."
3376 )
3378 if (
3379 not self.omit_schema
3380 and use_schema
3381 and effective_schema is not None
3382 ):
3383 name = f"{self.quote_schema(effective_schema)}.{name}"
3384 return name
3387class ReflectedNamedType(TypedDict):
3388 """Represents a reflected named type."""
3390 name: str
3391 """Name of the type."""
3392 schema: str
3393 """The schema of the type."""
3394 visible: bool
3395 """Indicates if this type is in the current search path."""
3398class ReflectedDomainConstraint(TypedDict):
3399 """Represents a reflect check constraint of a domain."""
3401 name: str
3402 """Name of the constraint."""
3403 check: str
3404 """The check constraint text."""
3407class ReflectedDomain(ReflectedNamedType):
3408 """Represents a reflected enum."""
3410 type: str
3411 """The string name of the underlying data type of the domain."""
3412 nullable: bool
3413 """Indicates if the domain allows null or not."""
3414 default: Optional[str]
3415 """The string representation of the default value of this domain
3416 or ``None`` if none present.
3417 """
3418 constraints: List[ReflectedDomainConstraint]
3419 """The constraints defined in the domain, if any.
3420 The constraint are in order of evaluation by postgresql.
3421 """
3422 collation: Optional[str]
3423 """The collation for the domain."""
3424 collation_schema: Optional[str]
3425 """The schema of the collation for the domain, if the collation is not
3426 visible on the current search_path."""
3429class ReflectedEnum(ReflectedNamedType):
3430 """Represents a reflected enum."""
3432 labels: List[str]
3433 """The labels that compose the enum."""
3436class PGInspector(reflection.Inspector):
3437 dialect: PGDialect
3439 def get_table_oid(
3440 self, table_name: str, schema: Optional[str] = None
3441 ) -> int:
3442 """Return the OID for the given table name.
3444 :param table_name: string name of the table. For special quoting,
3445 use :class:`.quoted_name`.
3447 :param schema: string schema name; if omitted, uses the default schema
3448 of the database connection. For special quoting,
3449 use :class:`.quoted_name`.
3451 """
3453 with self._operation_context() as conn:
3454 return self.dialect.get_table_oid(
3455 conn, table_name, schema, info_cache=self.info_cache
3456 )
3458 def get_domains(
3459 self, schema: Optional[str] = None
3460 ) -> List[ReflectedDomain]:
3461 """Return a list of DOMAIN objects.
3463 Each member is a dictionary containing these fields:
3465 * name - name of the domain
3466 * schema - the schema name for the domain.
3467 * visible - boolean, whether or not this domain is visible
3468 in the default search path.
3469 * type - the type defined by this domain.
3470 * nullable - Indicates if this domain can be ``NULL``.
3471 * default - The default value of the domain or ``None`` if the
3472 domain has no default.
3473 * constraints - A list of dict with the constraint defined by this
3474 domain. Each element contains two keys: ``name`` of the
3475 constraint and ``check`` with the constraint text.
3477 :param schema: schema name. If None, the default schema
3478 (typically 'public') is used. May also be set to ``'*'`` to
3479 indicate load domains for all schemas.
3481 .. versionadded:: 2.0
3483 """
3484 with self._operation_context() as conn:
3485 return self.dialect._load_domains(
3486 conn, schema, info_cache=self.info_cache
3487 )
3489 def get_enums(self, schema: Optional[str] = None) -> List[ReflectedEnum]:
3490 """Return a list of ENUM objects.
3492 Each member is a dictionary containing these fields:
3494 * name - name of the enum
3495 * schema - the schema name for the enum.
3496 * visible - boolean, whether or not this enum is visible
3497 in the default search path.
3498 * labels - a list of string labels that apply to the enum.
3500 :param schema: schema name. If None, the default schema
3501 (typically 'public') is used. May also be set to ``'*'`` to
3502 indicate load enums for all schemas.
3504 """
3505 with self._operation_context() as conn:
3506 return self.dialect._load_enums(
3507 conn, schema, info_cache=self.info_cache
3508 )
3510 def get_foreign_table_names(
3511 self, schema: Optional[str] = None
3512 ) -> List[str]:
3513 """Return a list of FOREIGN TABLE names.
3515 Behavior is similar to that of
3516 :meth:`_reflection.Inspector.get_table_names`,
3517 except that the list is limited to those tables that report a
3518 ``relkind`` value of ``f``.
3520 """
3521 with self._operation_context() as conn:
3522 return self.dialect._get_foreign_table_names(
3523 conn, schema, info_cache=self.info_cache
3524 )
3526 def has_type(
3527 self, type_name: str, schema: Optional[str] = None, **kw: Any
3528 ) -> bool:
3529 """Return if the database has the specified type in the provided
3530 schema.
3532 :param type_name: the type to check.
3533 :param schema: schema name. If None, the default schema
3534 (typically 'public') is used. May also be set to ``'*'`` to
3535 check in all schemas.
3537 .. versionadded:: 2.0
3539 """
3540 with self._operation_context() as conn:
3541 return self.dialect.has_type(
3542 conn, type_name, schema, info_cache=self.info_cache
3543 )
3546class PGExecutionContext(default.DefaultExecutionContext):
3547 def fire_sequence(self, seq, type_):
3548 return self._execute_scalar(
3549 (
3550 "select nextval(%s)"
3551 % self.identifier_preparer._format_sequence_string_literal(seq)
3552 ),
3553 type_,
3554 )
3556 def get_insert_default(self, column):
3557 if column.primary_key and column is column.table._autoincrement_column:
3558 if column.server_default and column.server_default.has_argument:
3559 # pre-execute passive defaults on primary key columns
3560 return self._execute_scalar(
3561 "select %s" % column.server_default.arg, column.type
3562 )
3564 elif column.default is None or (
3565 column.default.is_sequence and column.default.optional
3566 ):
3567 # execute the sequence associated with a SERIAL primary
3568 # key column. for non-primary-key SERIAL, the ID just
3569 # generates server side.
3571 try:
3572 seq_name = column._postgresql_seq_name
3573 except AttributeError:
3574 tab = column.table.name
3575 col = column.name
3576 tab = tab[0 : 29 + max(0, (29 - len(col)))]
3577 col = col[0 : 29 + max(0, (29 - len(tab)))]
3578 name = "%s_%s_seq" % (tab, col)
3579 column._postgresql_seq_name = seq_name = name
3581 if column.table is not None:
3582 effective_schema = self.connection.schema_for_object(
3583 column.table
3584 )
3585 else:
3586 effective_schema = None
3588 exc = (
3589 "select nextval(%s)"
3590 % self.identifier_preparer._format_sequence_string_literal(
3591 _CompilerSequence(seq_name, effective_schema)
3592 )
3593 )
3595 return self._execute_scalar(exc, column.type)
3597 return super().get_insert_default(column)
3600class PGReadOnlyConnectionCharacteristic(
3601 characteristics.ConnectionCharacteristic
3602):
3603 transactional = True
3605 def reset_characteristic(self, dialect, dbapi_conn):
3606 dialect.set_readonly(dbapi_conn, False)
3608 def set_characteristic(self, dialect, dbapi_conn, value):
3609 dialect.set_readonly(dbapi_conn, value)
3611 def get_characteristic(self, dialect, dbapi_conn):
3612 return dialect.get_readonly(dbapi_conn)
3615class PGDeferrableConnectionCharacteristic(
3616 characteristics.ConnectionCharacteristic
3617):
3618 transactional = True
3620 def reset_characteristic(self, dialect, dbapi_conn):
3621 dialect.set_deferrable(dbapi_conn, False)
3623 def set_characteristic(self, dialect, dbapi_conn, value):
3624 dialect.set_deferrable(dbapi_conn, value)
3626 def get_characteristic(self, dialect, dbapi_conn):
3627 return dialect.get_deferrable(dbapi_conn)
3630class PGDialect(default._BackendsMultiReflection, default.DefaultDialect):
3631 name = "postgresql"
3632 supports_statement_cache = True
3633 supports_alter = True
3634 max_identifier_length = 63
3635 supports_sane_rowcount = True
3637 bind_typing = interfaces.BindTyping.RENDER_CASTS
3639 supports_native_enum = True
3640 supports_native_boolean = True
3641 supports_native_uuid = True
3642 supports_smallserial = True
3643 supports_virtual_generated_columns = True
3645 supports_sequences = True
3646 sequences_optional = True
3647 preexecute_autoincrement_sequences = True
3648 postfetch_lastrowid = False
3649 use_insertmanyvalues = True
3651 returns_native_bytes = True
3653 insertmanyvalues_implicit_sentinel = (
3654 InsertmanyvaluesSentinelOpts.ANY_AUTOINCREMENT
3655 | InsertmanyvaluesSentinelOpts.USE_INSERT_FROM_SELECT
3656 | InsertmanyvaluesSentinelOpts.RENDER_SELECT_COL_CASTS
3657 )
3659 supports_comments = True
3660 supports_constraint_comments = True
3661 supports_default_values = True
3663 supports_default_metavalue = True
3665 supports_empty_insert = False
3666 supports_multivalues_insert = True
3668 supports_identity_columns = True
3670 default_paramstyle = "pyformat"
3671 ischema_names = ischema_names
3672 colspecs = colspecs
3674 statement_compiler = PGCompiler
3675 ddl_compiler = PGDDLCompiler
3676 type_compiler_cls = PGTypeCompiler
3677 preparer = PGIdentifierPreparer
3678 execution_ctx_cls = PGExecutionContext
3679 inspector = PGInspector
3681 update_returning = True
3682 delete_returning = True
3683 insert_returning = True
3684 update_returning_multifrom = True
3685 delete_returning_multifrom = True
3687 connection_characteristics = (
3688 default.DefaultDialect.connection_characteristics
3689 )
3690 connection_characteristics = connection_characteristics.union(
3691 {
3692 "postgresql_readonly": PGReadOnlyConnectionCharacteristic(),
3693 "postgresql_deferrable": PGDeferrableConnectionCharacteristic(),
3694 }
3695 )
3697 construct_arguments = [
3698 (
3699 schema.Index,
3700 {
3701 "using": False,
3702 "include": None,
3703 "where": None,
3704 "ops": {},
3705 "concurrently": False,
3706 "with": {},
3707 "tablespace": None,
3708 "nulls_not_distinct": None,
3709 "invalid": DialectKWArgConst.REFLECTED_ONLY,
3710 },
3711 ),
3712 (
3713 schema.Table,
3714 {
3715 "ignore_search_path": False,
3716 "tablespace": None,
3717 "partition_by": None,
3718 "with_oids": None,
3719 "with": None,
3720 "on_commit": None,
3721 "inherits": None,
3722 "using": None,
3723 },
3724 ),
3725 (
3726 schema.CheckConstraint,
3727 {
3728 "not_valid": False,
3729 },
3730 ),
3731 (
3732 schema.ForeignKeyConstraint,
3733 {
3734 "not_valid": False,
3735 },
3736 ),
3737 (
3738 schema.PrimaryKeyConstraint,
3739 {"include": None},
3740 ),
3741 (
3742 schema.UniqueConstraint,
3743 {
3744 "include": None,
3745 "nulls_not_distinct": None,
3746 },
3747 ),
3748 (
3749 schema.CreateView,
3750 {
3751 "with": None,
3752 },
3753 ),
3754 ]
3756 reflection_options = ("postgresql_ignore_search_path",)
3758 _backslash_escapes = False
3759 _supports_create_index_concurrently = True
3760 _supports_drop_index_concurrently = True
3761 _supports_jsonb_subscripting = True
3762 _pg_am_btree_oid = -1
3764 def __init__(
3765 self,
3766 native_inet_types=None,
3767 json_serializer=None,
3768 json_deserializer=None,
3769 **kwargs,
3770 ):
3771 default.DefaultDialect.__init__(self, **kwargs)
3773 self._native_inet_types = native_inet_types
3774 self._json_deserializer = json_deserializer
3775 self._json_serializer = json_serializer
3777 def initialize(self, connection):
3778 super().initialize(connection)
3780 # https://www.postgresql.org/docs/9.3/static/release-9-2.html#AEN116689
3781 self.supports_smallserial = self.server_version_info >= (9, 2)
3783 self._set_backslash_escapes(connection)
3785 self._supports_drop_index_concurrently = self.server_version_info >= (
3786 9,
3787 2,
3788 )
3789 self.supports_identity_columns = self.server_version_info >= (10,)
3791 self._supports_jsonb_subscripting = self.server_version_info >= (14,)
3793 self.supports_virtual_generated_columns = self.server_version_info >= (
3794 18,
3795 )
3797 def get_isolation_level_values(self, dbapi_conn):
3798 # note the generic dialect doesn't have AUTOCOMMIT, however
3799 # all postgresql dialects should include AUTOCOMMIT.
3800 return (
3801 "SERIALIZABLE",
3802 "READ UNCOMMITTED",
3803 "READ COMMITTED",
3804 "REPEATABLE READ",
3805 )
3807 def set_isolation_level(self, dbapi_connection, level):
3808 cursor = dbapi_connection.cursor()
3809 cursor.execute(
3810 "SET SESSION CHARACTERISTICS AS TRANSACTION "
3811 f"ISOLATION LEVEL {level}"
3812 )
3813 cursor.execute("COMMIT")
3814 cursor.close()
3816 def get_isolation_level(self, dbapi_connection):
3817 cursor = dbapi_connection.cursor()
3818 cursor.execute("show transaction isolation level")
3819 val = cursor.fetchone()[0]
3820 cursor.close()
3821 return val.upper()
3823 def set_readonly(self, connection, value):
3824 raise NotImplementedError()
3826 def get_readonly(self, connection):
3827 raise NotImplementedError()
3829 def set_deferrable(self, connection, value):
3830 raise NotImplementedError()
3832 def get_deferrable(self, connection):
3833 raise NotImplementedError()
3835 def _split_multihost_from_url(self, url: URL) -> Union[
3836 Tuple[None, None],
3837 Tuple[Tuple[Optional[str], ...], Tuple[Optional[int], ...]],
3838 ]:
3839 hosts: Optional[Tuple[Optional[str], ...]] = None
3840 ports_str: Union[str, Tuple[Optional[str], ...], None] = None
3842 integrated_multihost = False
3844 if "host" in url.query:
3845 if isinstance(url.query["host"], (list, tuple)):
3846 integrated_multihost = True
3847 hosts, ports_str = zip(
3848 *[
3849 token.split(":") if ":" in token else (token, None)
3850 for token in url.query["host"]
3851 ]
3852 )
3854 elif isinstance(url.query["host"], str):
3855 hosts = tuple(url.query["host"].split(","))
3857 if (
3858 "port" not in url.query
3859 and len(hosts) == 1
3860 and ":" in hosts[0]
3861 ):
3862 # internet host is alphanumeric plus dots or hyphens.
3863 # this is essentially rfc1123, which refers to rfc952.
3864 # https://stackoverflow.com/questions/3523028/
3865 # valid-characters-of-a-hostname
3866 host_port_match = re.match(
3867 r"^([a-zA-Z0-9\-\.]*)(?:\:(\d*))?$", hosts[0]
3868 )
3869 if host_port_match:
3870 integrated_multihost = True
3871 h, p = host_port_match.group(1, 2)
3872 if TYPE_CHECKING:
3873 assert isinstance(h, str)
3874 assert isinstance(p, str)
3875 hosts = (h,)
3876 ports_str = cast(
3877 "Tuple[Optional[str], ...]", (p,) if p else (None,)
3878 )
3880 if "port" in url.query:
3881 if integrated_multihost:
3882 raise exc.ArgumentError(
3883 "Can't mix 'multihost' formats together; use "
3884 '"host=h1,h2,h3&port=p1,p2,p3" or '
3885 '"host=h1:p1&host=h2:p2&host=h3:p3" separately'
3886 )
3887 if isinstance(url.query["port"], (list, tuple)):
3888 ports_str = url.query["port"]
3889 elif isinstance(url.query["port"], str):
3890 ports_str = tuple(url.query["port"].split(","))
3892 ports: Optional[Tuple[Optional[int], ...]] = None
3894 if ports_str:
3895 try:
3896 ports = tuple(int(x) if x else None for x in ports_str)
3897 except ValueError:
3898 raise exc.ArgumentError(
3899 f"Received non-integer port arguments: {ports_str}"
3900 ) from None
3902 if ports and (
3903 (not hosts and len(ports) > 1)
3904 or (
3905 hosts
3906 and ports
3907 and len(hosts) != len(ports)
3908 and (len(hosts) > 1 or len(ports) > 1)
3909 )
3910 ):
3911 raise exc.ArgumentError("number of hosts and ports don't match")
3913 if hosts is not None:
3914 if ports is None:
3915 ports = tuple(None for _ in hosts)
3917 return hosts, ports # type: ignore
3919 def do_begin_twophase(self, connection, xid):
3920 self.do_begin(connection.connection)
3922 def do_prepare_twophase(self, connection, xid):
3923 connection.execute(
3924 sql.text("PREPARE TRANSACTION :xid").bindparams(
3925 sql.bindparam("xid", xid, literal_execute=True)
3926 )
3927 )
3929 def do_rollback_twophase(
3930 self, connection, xid, is_prepared=True, recover=False
3931 ):
3932 if is_prepared:
3933 if recover:
3934 # FIXME: ugly hack to get out of transaction
3935 # context when committing recoverable transactions
3936 # Must find out a way how to make the dbapi not
3937 # open a transaction.
3938 connection.exec_driver_sql("ROLLBACK")
3939 connection.execute(
3940 sql.text("ROLLBACK PREPARED :xid").bindparams(
3941 sql.bindparam("xid", xid, literal_execute=True)
3942 )
3943 )
3944 connection.exec_driver_sql("BEGIN")
3945 self.do_rollback(connection.connection)
3946 else:
3947 self.do_rollback(connection.connection)
3949 def do_commit_twophase(
3950 self, connection, xid, is_prepared=True, recover=False
3951 ):
3952 if is_prepared:
3953 if recover:
3954 connection.exec_driver_sql("ROLLBACK")
3955 connection.execute(
3956 sql.text("COMMIT PREPARED :xid").bindparams(
3957 sql.bindparam("xid", xid, literal_execute=True)
3958 )
3959 )
3960 connection.exec_driver_sql("BEGIN")
3961 self.do_rollback(connection.connection)
3962 else:
3963 self.do_commit(connection.connection)
3965 def do_recover_twophase(self, connection):
3966 return connection.scalars(
3967 sql.text("SELECT gid FROM pg_prepared_xacts")
3968 ).all()
3970 def _get_default_schema_name(self, connection):
3971 return connection.exec_driver_sql("select current_schema()").scalar()
3973 @reflection.cache
3974 def has_schema(self, connection, schema, **kw):
3975 query = select(pg_catalog.pg_namespace.c.nspname).where(
3976 pg_catalog.pg_namespace.c.nspname == schema
3977 )
3978 return bool(connection.scalar(query))
3980 def _pg_class_filter_scope_schema(
3981 self, query, schema, scope, pg_class_table=None
3982 ):
3983 if pg_class_table is None:
3984 pg_class_table = pg_catalog.pg_class
3985 query = query.join(
3986 pg_catalog.pg_namespace,
3987 pg_catalog.pg_namespace.c.oid == pg_class_table.c.relnamespace,
3988 )
3990 if scope is ObjectScope.DEFAULT:
3991 query = query.where(pg_class_table.c.relpersistence != "t")
3992 elif scope is ObjectScope.TEMPORARY:
3993 query = query.where(pg_class_table.c.relpersistence == "t")
3995 if schema is None:
3996 query = query.where(
3997 pg_catalog.pg_table_is_visible(pg_class_table.c.oid),
3998 # ignore pg_catalog schema
3999 pg_catalog.pg_namespace.c.nspname != "pg_catalog",
4000 )
4001 else:
4002 query = query.where(pg_catalog.pg_namespace.c.nspname == schema)
4003 return query
4005 def _pg_class_relkind_condition(self, relkinds, pg_class_table=None):
4006 if pg_class_table is None:
4007 pg_class_table = pg_catalog.pg_class
4008 # uses the any form instead of in otherwise postgresql complaings
4009 # that 'IN could not convert type character to "char"'
4010 return pg_class_table.c.relkind == sql.any_(_array.array(relkinds))
4012 @lru_cache()
4013 def _has_multi_table_query(self, schema):
4014 query = select(pg_catalog.pg_class.c.relname).where(
4015 pg_catalog.pg_class.c.relname.in_(
4016 bindparam("table_names", expanding=True)
4017 ),
4018 self._pg_class_relkind_condition(
4019 pg_catalog.RELKINDS_ALL_TABLE_LIKE
4020 ),
4021 )
4022 return self._pg_class_filter_scope_schema(
4023 query, schema, scope=ObjectScope.ANY
4024 )
4026 @reflection.cache
4027 def has_table(self, connection, table_name, schema=None, **kw):
4028 # NOTE: it's not worth calling into the multi table since the query
4029 # is compatible also with this single case.
4030 self._ensure_has_table_connection(connection)
4031 query = self._has_multi_table_query(schema)
4032 return bool(connection.scalar(query, {"table_names": [table_name]}))
4034 def has_multi_table(self, connection, table_names, schema=None, **kw):
4035 query = self._has_multi_table_query(schema)
4036 params = {"table_names": table_names}
4037 existing = set(connection.scalars(query, params).all())
4038 retval = {(schema, table): table in existing for table in table_names}
4039 return retval.items()
4041 @reflection.cache
4042 def has_sequence(self, connection, sequence_name, schema=None, **kw):
4043 query = select(pg_catalog.pg_class.c.relname).where(
4044 pg_catalog.pg_class.c.relkind == "S",
4045 pg_catalog.pg_class.c.relname == sequence_name,
4046 )
4047 query = self._pg_class_filter_scope_schema(
4048 query, schema, scope=ObjectScope.ANY
4049 )
4050 return bool(connection.scalar(query))
4052 @reflection.cache
4053 def has_type(self, connection, type_name, schema=None, **kw):
4054 query = (
4055 select(pg_catalog.pg_type.c.typname)
4056 .join(
4057 pg_catalog.pg_namespace,
4058 pg_catalog.pg_namespace.c.oid
4059 == pg_catalog.pg_type.c.typnamespace,
4060 )
4061 .where(pg_catalog.pg_type.c.typname == type_name)
4062 )
4063 if schema is None:
4064 query = query.where(
4065 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid),
4066 # ignore pg_catalog schema
4067 pg_catalog.pg_namespace.c.nspname != "pg_catalog",
4068 )
4069 elif schema != "*":
4070 query = query.where(pg_catalog.pg_namespace.c.nspname == schema)
4072 return bool(connection.scalar(query))
4074 def _get_server_version_info(self, connection):
4075 v = connection.exec_driver_sql("select pg_catalog.version()").scalar()
4076 m = re.match(
4077 r".*(?:PostgreSQL|EnterpriseDB) "
4078 r"(\d+)\.?(\d+)?(?:\.(\d+))?(?:\.\d+)?(?:devel|beta)?",
4079 v,
4080 )
4081 if not m:
4082 raise AssertionError(
4083 "Could not determine version from string '%s'" % v
4084 )
4085 return tuple([int(x) for x in m.group(1, 2, 3) if x is not None])
4087 @reflection.cache
4088 def get_table_oid(self, connection, table_name, schema=None, **kw):
4089 """Fetch the oid for schema.table_name."""
4090 query = select(pg_catalog.pg_class.c.oid).where(
4091 pg_catalog.pg_class.c.relname == table_name,
4092 self._pg_class_relkind_condition(
4093 pg_catalog.RELKINDS_ALL_TABLE_LIKE
4094 ),
4095 )
4096 query = self._pg_class_filter_scope_schema(
4097 query, schema, scope=ObjectScope.ANY
4098 )
4099 table_oid = connection.scalar(query)
4100 if table_oid is None:
4101 raise exc.NoSuchTableError(
4102 f"{schema}.{table_name}" if schema else table_name
4103 )
4104 return table_oid
4106 @reflection.cache
4107 def get_schema_names(self, connection, **kw):
4108 query = (
4109 select(pg_catalog.pg_namespace.c.nspname)
4110 .where(
4111 ~pg_catalog.pg_namespace.c.nspname.startswith(
4112 "pg_", autoescape=True
4113 )
4114 )
4115 .order_by(pg_catalog.pg_namespace.c.nspname)
4116 )
4117 return connection.scalars(query).all()
4119 def _get_relnames_for_relkinds(self, connection, schema, relkinds, scope):
4120 query = select(pg_catalog.pg_class.c.relname).where(
4121 self._pg_class_relkind_condition(relkinds)
4122 )
4123 query = self._pg_class_filter_scope_schema(query, schema, scope=scope)
4124 return connection.scalars(query).all()
4126 @reflection.cache
4127 def get_table_names(self, connection, schema=None, **kw):
4128 return self._get_relnames_for_relkinds(
4129 connection,
4130 schema,
4131 pg_catalog.RELKINDS_TABLE_NO_FOREIGN,
4132 scope=ObjectScope.DEFAULT,
4133 )
4135 @reflection.cache
4136 def get_temp_table_names(self, connection, **kw):
4137 return self._get_relnames_for_relkinds(
4138 connection,
4139 schema=None,
4140 relkinds=pg_catalog.RELKINDS_TABLE_NO_FOREIGN,
4141 scope=ObjectScope.TEMPORARY,
4142 )
4144 @reflection.cache
4145 def _get_foreign_table_names(self, connection, schema=None, **kw):
4146 return self._get_relnames_for_relkinds(
4147 connection, schema, relkinds=("f",), scope=ObjectScope.ANY
4148 )
4150 @reflection.cache
4151 def get_view_names(self, connection, schema=None, **kw):
4152 return self._get_relnames_for_relkinds(
4153 connection,
4154 schema,
4155 pg_catalog.RELKINDS_VIEW,
4156 scope=ObjectScope.DEFAULT,
4157 )
4159 @reflection.cache
4160 def get_materialized_view_names(self, connection, schema=None, **kw):
4161 return self._get_relnames_for_relkinds(
4162 connection,
4163 schema,
4164 pg_catalog.RELKINDS_MAT_VIEW,
4165 scope=ObjectScope.DEFAULT,
4166 )
4168 @reflection.cache
4169 def get_temp_view_names(self, connection, schema=None, **kw):
4170 return self._get_relnames_for_relkinds(
4171 connection,
4172 schema,
4173 # NOTE: do not include temp materialzied views (that do not
4174 # seem to be a thing at least up to version 14)
4175 pg_catalog.RELKINDS_VIEW,
4176 scope=ObjectScope.TEMPORARY,
4177 )
4179 @reflection.cache
4180 def get_sequence_names(self, connection, schema=None, **kw):
4181 return self._get_relnames_for_relkinds(
4182 connection, schema, relkinds=("S",), scope=ObjectScope.ANY
4183 )
4185 @reflection.cache
4186 def get_view_definition(self, connection, view_name, schema=None, **kw):
4187 query = (
4188 select(pg_catalog.pg_get_viewdef(pg_catalog.pg_class.c.oid))
4189 .select_from(pg_catalog.pg_class)
4190 .where(
4191 pg_catalog.pg_class.c.relname == view_name,
4192 self._pg_class_relkind_condition(
4193 pg_catalog.RELKINDS_VIEW + pg_catalog.RELKINDS_MAT_VIEW
4194 ),
4195 )
4196 )
4197 query = self._pg_class_filter_scope_schema(
4198 query, schema, scope=ObjectScope.ANY
4199 )
4200 res = connection.scalar(query)
4201 if res is None:
4202 raise exc.NoSuchTableError(
4203 f"{schema}.{view_name}" if schema else view_name
4204 )
4205 else:
4206 return res
4208 def _prepare_filter_names(self, filter_names):
4209 if filter_names:
4210 return True, {"filter_names": filter_names}
4211 else:
4212 return False, {}
4214 def _kind_to_relkinds(self, kind: ObjectKind) -> Tuple[str, ...]:
4215 if kind is ObjectKind.ANY:
4216 return pg_catalog.RELKINDS_ALL_TABLE_LIKE
4217 relkinds = ()
4218 if ObjectKind.TABLE in kind:
4219 relkinds += pg_catalog.RELKINDS_TABLE
4220 if ObjectKind.VIEW in kind:
4221 relkinds += pg_catalog.RELKINDS_VIEW
4222 if ObjectKind.MATERIALIZED_VIEW in kind:
4223 relkinds += pg_catalog.RELKINDS_MAT_VIEW
4224 return relkinds
4226 @lru_cache()
4227 def _table_options_query(self, schema, has_filter_names, scope, kind):
4228 inherits_sq = (
4229 select(
4230 pg_catalog.pg_inherits.c.inhrelid,
4231 sql.func.array_agg(
4232 aggregate_order_by(
4233 pg_catalog.pg_class.c.relname,
4234 pg_catalog.pg_inherits.c.inhseqno,
4235 )
4236 ).label("parent_table_names"),
4237 )
4238 .select_from(pg_catalog.pg_inherits)
4239 .join(
4240 pg_catalog.pg_class,
4241 pg_catalog.pg_inherits.c.inhparent
4242 == pg_catalog.pg_class.c.oid,
4243 )
4244 .group_by(pg_catalog.pg_inherits.c.inhrelid)
4245 .subquery("inherits")
4246 )
4248 if self.server_version_info < (12,):
4249 # this is not in the pg_catalog.pg_class since it was
4250 # removed in PostgreSQL version 12
4251 has_oids = sql.column("relhasoids", BOOLEAN)
4252 else:
4253 has_oids = sql.null().label("relhasoids")
4255 relkinds = self._kind_to_relkinds(kind)
4256 query = (
4257 select(
4258 pg_catalog.pg_class.c.oid,
4259 pg_catalog.pg_class.c.relname,
4260 pg_catalog.pg_class.c.reloptions,
4261 has_oids,
4262 sql.case(
4263 (
4264 sql.and_(
4265 pg_catalog.pg_am.c.amname.is_not(None),
4266 pg_catalog.pg_am.c.amname
4267 != sql.func.current_setting(
4268 "default_table_access_method"
4269 ),
4270 ),
4271 pg_catalog.pg_am.c.amname,
4272 ),
4273 else_=sql.null(),
4274 ).label("access_method_name"),
4275 pg_catalog.pg_tablespace.c.spcname.label("tablespace_name"),
4276 inherits_sq.c.parent_table_names,
4277 )
4278 .select_from(pg_catalog.pg_class)
4279 .outerjoin(
4280 # NOTE: on postgresql < 12, this could be avoided
4281 # since relam is always 0 so nothing is joined.
4282 pg_catalog.pg_am,
4283 pg_catalog.pg_class.c.relam == pg_catalog.pg_am.c.oid,
4284 )
4285 .outerjoin(
4286 inherits_sq,
4287 pg_catalog.pg_class.c.oid == inherits_sq.c.inhrelid,
4288 )
4289 .outerjoin(
4290 pg_catalog.pg_tablespace,
4291 pg_catalog.pg_tablespace.c.oid
4292 == pg_catalog.pg_class.c.reltablespace,
4293 )
4294 .where(self._pg_class_relkind_condition(relkinds))
4295 )
4296 query = self._pg_class_filter_scope_schema(query, schema, scope=scope)
4297 if has_filter_names:
4298 query = query.where(
4299 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
4300 )
4301 return query
4303 def get_multi_table_options(
4304 self, connection, schema, filter_names, scope, kind, **kw
4305 ):
4306 has_filter_names, params = self._prepare_filter_names(filter_names)
4307 query = self._table_options_query(
4308 schema, has_filter_names, scope, kind
4309 )
4310 rows = connection.execute(query, params).mappings()
4311 table_options = {}
4313 for row in rows:
4314 current: dict[str, Any] = {}
4315 if row["access_method_name"] is not None:
4316 current["postgresql_using"] = row["access_method_name"]
4318 if row["parent_table_names"]:
4319 current["postgresql_inherits"] = tuple(
4320 row["parent_table_names"]
4321 )
4323 if row["reloptions"]:
4324 current["postgresql_with"] = dict(
4325 option.split("=", 1) for option in row["reloptions"]
4326 )
4328 if row["relhasoids"]:
4329 current["postgresql_with_oids"] = True
4331 if row["tablespace_name"] is not None:
4332 current["postgresql_tablespace"] = row["tablespace_name"]
4334 table_options[(schema, row["relname"])] = current
4336 return table_options.items()
4338 @lru_cache()
4339 def _columns_query(self, schema, has_filter_names, scope, kind):
4340 # NOTE: the query with the default and identity options scalar
4341 # subquery is faster than trying to use outer joins for them
4342 generated = (
4343 pg_catalog.pg_attribute.c.attgenerated.label("generated")
4344 if self.server_version_info >= (12,)
4345 else sql.null().label("generated")
4346 )
4347 if self.server_version_info >= (10,):
4348 # join lateral performs worse (~2x slower) than a scalar_subquery
4349 # also the subquery can be run only if the column is an identity
4350 identity = sql.case(
4351 ( # attidentity != '' is required or it will reflect also
4352 # serial columns as identity.
4353 pg_catalog.pg_attribute.c.attidentity != "",
4354 select(
4355 sql.func.json_build_object(
4356 "always",
4357 pg_catalog.pg_attribute.c.attidentity == "a",
4358 "start",
4359 pg_catalog.pg_sequence.c.seqstart,
4360 "increment",
4361 pg_catalog.pg_sequence.c.seqincrement,
4362 "minvalue",
4363 pg_catalog.pg_sequence.c.seqmin,
4364 "maxvalue",
4365 pg_catalog.pg_sequence.c.seqmax,
4366 "cache",
4367 pg_catalog.pg_sequence.c.seqcache,
4368 "cycle",
4369 pg_catalog.pg_sequence.c.seqcycle,
4370 type_=sqltypes.JSON(),
4371 )
4372 )
4373 .select_from(pg_catalog.pg_sequence)
4374 .where(
4375 # not needed but pg seems to like it
4376 pg_catalog.pg_attribute.c.attidentity != "",
4377 pg_catalog.pg_sequence.c.seqrelid
4378 == sql.cast(
4379 sql.cast(
4380 pg_catalog.pg_get_serial_sequence(
4381 sql.cast(
4382 sql.cast(
4383 pg_catalog.pg_attribute.c.attrelid,
4384 REGCLASS,
4385 ),
4386 TEXT,
4387 ),
4388 pg_catalog.pg_attribute.c.attname,
4389 ),
4390 REGCLASS,
4391 ),
4392 OID,
4393 ),
4394 )
4395 .correlate(pg_catalog.pg_attribute)
4396 .scalar_subquery(),
4397 ),
4398 else_=sql.null(),
4399 ).label("identity_options")
4400 else:
4401 identity = sql.null().label("identity_options")
4403 # join lateral performs the same as scalar_subquery here, also
4404 # the subquery can be run only if the column has a default
4405 default = sql.case(
4406 (
4407 pg_catalog.pg_attribute.c.atthasdef,
4408 select(
4409 pg_catalog.pg_get_expr(
4410 pg_catalog.pg_attrdef.c.adbin,
4411 pg_catalog.pg_attrdef.c.adrelid,
4412 )
4413 )
4414 .select_from(pg_catalog.pg_attrdef)
4415 .where(
4416 # not needed but pg seems to like it
4417 pg_catalog.pg_attribute.c.atthasdef,
4418 pg_catalog.pg_attrdef.c.adrelid
4419 == pg_catalog.pg_attribute.c.attrelid,
4420 pg_catalog.pg_attrdef.c.adnum
4421 == pg_catalog.pg_attribute.c.attnum,
4422 )
4423 .correlate(pg_catalog.pg_attribute)
4424 .scalar_subquery(),
4425 ),
4426 else_=sql.null(),
4427 ).label("default")
4429 # get the name and schema of the collate when it's different from
4430 # the default one; the schema is only included when the collation
4431 # is not visible on the current search_path
4432 collate = sql.case(
4433 (
4434 sql.and_(
4435 pg_catalog.pg_attribute.c.attcollation != 0,
4436 select(pg_catalog.pg_type.c.typcollation)
4437 .where(
4438 pg_catalog.pg_type.c.oid
4439 == pg_catalog.pg_attribute.c.atttypid,
4440 )
4441 .correlate(pg_catalog.pg_attribute)
4442 .scalar_subquery()
4443 != pg_catalog.pg_attribute.c.attcollation,
4444 ),
4445 select(
4446 sql.func.json_build_object(
4447 "name",
4448 pg_catalog.pg_collation.c.collname,
4449 "schema",
4450 sql.case(
4451 (
4452 pg_catalog.pg_collation_is_visible(
4453 pg_catalog.pg_collation.c.oid
4454 ),
4455 sql.null(),
4456 ),
4457 else_=pg_catalog.pg_namespace.c.nspname,
4458 ),
4459 type_=sqltypes.JSON(),
4460 )
4461 )
4462 .select_from(pg_catalog.pg_collation)
4463 .join(
4464 pg_catalog.pg_namespace,
4465 pg_catalog.pg_namespace.c.oid
4466 == pg_catalog.pg_collation.c.collnamespace,
4467 )
4468 .where(
4469 pg_catalog.pg_collation.c.oid
4470 == pg_catalog.pg_attribute.c.attcollation
4471 )
4472 .correlate(pg_catalog.pg_attribute)
4473 .scalar_subquery(),
4474 ),
4475 else_=sql.null(),
4476 ).label("collation")
4478 relkinds = self._kind_to_relkinds(kind)
4479 query = (
4480 select(
4481 pg_catalog.pg_attribute.c.attname.label("name"),
4482 pg_catalog.format_type(
4483 pg_catalog.pg_attribute.c.atttypid,
4484 pg_catalog.pg_attribute.c.atttypmod,
4485 ).label("format_type"),
4486 default,
4487 pg_catalog.pg_attribute.c.attnotnull.label("not_null"),
4488 pg_catalog.pg_class.c.relname.label("table_name"),
4489 pg_catalog.pg_description.c.description.label("comment"),
4490 generated,
4491 identity,
4492 collate,
4493 )
4494 .select_from(pg_catalog.pg_class)
4495 # NOTE: postgresql support table with no user column, meaning
4496 # there is no row with pg_attribute.attnum > 0. use a left outer
4497 # join to avoid filtering these tables.
4498 .outerjoin(
4499 pg_catalog.pg_attribute,
4500 sql.and_(
4501 pg_catalog.pg_class.c.oid
4502 == pg_catalog.pg_attribute.c.attrelid,
4503 pg_catalog.pg_attribute.c.attnum > 0,
4504 ~pg_catalog.pg_attribute.c.attisdropped,
4505 ),
4506 )
4507 .outerjoin(
4508 pg_catalog.pg_description,
4509 sql.and_(
4510 pg_catalog.pg_description.c.objoid
4511 == pg_catalog.pg_attribute.c.attrelid,
4512 pg_catalog.pg_description.c.objsubid
4513 == pg_catalog.pg_attribute.c.attnum,
4514 ),
4515 )
4516 .where(self._pg_class_relkind_condition(relkinds))
4517 .order_by(
4518 pg_catalog.pg_class.c.relname, pg_catalog.pg_attribute.c.attnum
4519 )
4520 )
4521 query = self._pg_class_filter_scope_schema(query, schema, scope=scope)
4522 if has_filter_names:
4523 query = query.where(
4524 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
4525 )
4526 return query
4528 def get_multi_columns(
4529 self, connection, schema, filter_names, scope, kind, **kw
4530 ):
4531 has_filter_names, params = self._prepare_filter_names(filter_names)
4532 query = self._columns_query(schema, has_filter_names, scope, kind)
4533 rows = connection.execute(query, params).mappings()
4535 named_type_loader = _NamedTypeLoader(self, connection, kw)
4536 columns = self._get_columns_info(rows, named_type_loader, schema)
4538 return columns.items()
4540 _format_type_args_pattern = re.compile(r"\((.*)\)")
4541 _format_type_args_delim = re.compile(r"\s*,\s*")
4542 _format_array_spec_pattern = re.compile(r"((?:\[\])*)$")
4544 def _reflect_type(
4545 self,
4546 format_type: Optional[str],
4547 named_type_loader: _NamedTypeLoader,
4548 type_description: str,
4549 collation: Optional[str],
4550 collation_schema: Optional[str] = None,
4551 ) -> sqltypes.TypeEngine[Any]:
4552 """
4553 Attempts to reconstruct a column type defined in ischema_names based
4554 on the information available in the format_type.
4556 If the `format_type` cannot be associated with a known `ischema_names`,
4557 it is treated as a reference to a known PostgreSQL named `ENUM` or
4558 `DOMAIN` type.
4559 """
4560 type_description = type_description or "unknown type"
4561 if format_type is None:
4562 util.warn(
4563 "PostgreSQL format_type() returned NULL for %s"
4564 % type_description
4565 )
4566 return sqltypes.NULLTYPE
4568 attype_args_match = self._format_type_args_pattern.search(format_type)
4569 if attype_args_match and attype_args_match.group(1):
4570 attype_args = self._format_type_args_delim.split(
4571 attype_args_match.group(1)
4572 )
4573 else:
4574 attype_args = ()
4576 match_array_dim = self._format_array_spec_pattern.search(format_type)
4577 # Each "[]" in array specs corresponds to an array dimension
4578 array_dim = len(match_array_dim.group(1) or "") // 2
4580 # Remove all parameters and array specs from format_type to obtain an
4581 # ischema_name candidate
4582 attype = self._format_type_args_pattern.sub("", format_type)
4583 attype = self._format_array_spec_pattern.sub("", attype)
4585 schema_type = self.ischema_names.get(attype.lower(), None)
4586 args, kwargs = (), {}
4588 if attype == "numeric":
4589 if len(attype_args) == 2:
4590 precision, scale = map(int, attype_args)
4591 args = (precision, scale)
4593 elif attype == "double precision":
4594 args = (53,)
4596 elif attype == "integer":
4597 args = ()
4599 elif attype in ("timestamp with time zone", "time with time zone"):
4600 kwargs["timezone"] = True
4601 if len(attype_args) == 1:
4602 kwargs["precision"] = int(attype_args[0])
4604 elif attype in (
4605 "timestamp without time zone",
4606 "time without time zone",
4607 "time",
4608 ):
4609 kwargs["timezone"] = False
4610 if len(attype_args) == 1:
4611 kwargs["precision"] = int(attype_args[0])
4613 elif attype == "bit varying":
4614 kwargs["varying"] = True
4615 if len(attype_args) == 1:
4616 charlen = int(attype_args[0])
4617 args = (charlen,)
4619 # a domain or enum can start with interval, so be mindful of that.
4620 elif attype == "interval" or attype.startswith("interval "):
4621 schema_type = INTERVAL
4623 field_match = re.match(r"interval (.+)", attype)
4624 if field_match:
4625 kwargs["fields"] = field_match.group(1)
4627 if len(attype_args) == 1:
4628 kwargs["precision"] = int(attype_args[0])
4630 else:
4631 enum_or_domain_key = tuple(util.quoted_token_parser(attype))
4633 if (
4634 schema_type is None
4635 and enum_or_domain_key in named_type_loader.enums
4636 ):
4637 schema_type = ENUM
4638 enum = named_type_loader.enums[enum_or_domain_key]
4640 kwargs["name"] = enum["name"]
4642 if not enum["visible"]:
4643 kwargs["schema"] = enum["schema"]
4644 args = tuple(enum["labels"])
4645 elif (
4646 schema_type is None
4647 and enum_or_domain_key in named_type_loader.domains
4648 ):
4649 schema_type = DOMAIN
4650 domain = named_type_loader.domains[enum_or_domain_key]
4652 data_type = self._reflect_type(
4653 domain["type"],
4654 named_type_loader,
4655 type_description="DOMAIN '%s'" % domain["name"],
4656 collation=domain["collation"],
4657 collation_schema=domain["collation_schema"],
4658 )
4659 args = (domain["name"], data_type)
4661 kwargs["collation"] = domain["collation"]
4662 kwargs["collation_schema"] = domain["collation_schema"]
4663 kwargs["default"] = domain["default"]
4664 kwargs["not_null"] = not domain["nullable"]
4665 kwargs["create_type"] = False
4667 if domain["constraints"]:
4668 # We only support a single constraint
4669 check_constraint = domain["constraints"][0]
4671 kwargs["constraint_name"] = check_constraint["name"]
4672 kwargs["check"] = check_constraint["check"]
4674 if not domain["visible"]:
4675 kwargs["schema"] = domain["schema"]
4677 else:
4678 try:
4679 charlen = int(attype_args[0])
4680 args = (charlen, *attype_args[1:])
4681 except (ValueError, IndexError):
4682 args = attype_args
4684 if not schema_type:
4685 util.warn(
4686 "Did not recognize type '%s' of %s"
4687 % (attype, type_description)
4688 )
4689 return sqltypes.NULLTYPE
4691 if collation is not None:
4692 kwargs["collation"] = collation
4693 if collation_schema is not None:
4694 kwargs["collation_schema"] = collation_schema
4696 data_type = schema_type(*args, **kwargs)
4697 if array_dim >= 1:
4698 # postgres does not preserve dimensionality or size of array types.
4699 data_type = _array.ARRAY(data_type)
4701 return data_type
4703 def _get_columns_info(self, rows, named_type_loader, schema):
4704 columns = defaultdict(list)
4705 for row_dict in rows:
4706 # ensure that each table has an entry, even if it has no columns
4707 if row_dict["name"] is None:
4708 columns[(schema, row_dict["table_name"])] = (
4709 ReflectionDefaults.columns()
4710 )
4711 continue
4712 table_cols = columns[(schema, row_dict["table_name"])]
4714 collation_info = row_dict["collation"]
4715 if collation_info is not None:
4716 collation = collation_info["name"]
4717 collation_schema = collation_info["schema"]
4718 else:
4719 collation = collation_schema = None
4721 coltype = self._reflect_type(
4722 row_dict["format_type"],
4723 named_type_loader,
4724 type_description="column '%s'" % row_dict["name"],
4725 collation=collation,
4726 collation_schema=collation_schema,
4727 )
4729 default = row_dict["default"]
4730 name = row_dict["name"]
4731 generated = row_dict["generated"]
4732 nullable = not row_dict["not_null"]
4734 if isinstance(coltype, DOMAIN):
4735 if not default:
4736 # domain can override the default value but
4737 # can't set it to None
4738 if coltype.default is not None:
4739 default = coltype.default
4741 nullable = nullable and not coltype.not_null
4743 identity = row_dict["identity_options"]
4745 # If a zero byte or blank string depending on driver (is also
4746 # absent for older PG versions), then not a generated column.
4747 # Otherwise, s = stored, v = virtual.
4748 if generated not in (None, "", b"\x00"):
4749 computed = dict(
4750 sqltext=default, persisted=generated in ("s", b"s")
4751 )
4752 default = None
4753 else:
4754 computed = None
4756 # adjust the default value
4757 autoincrement = False
4758 if default is not None:
4759 match = re.search(r"""(nextval\(')([^']+)('.*$)""", default)
4760 if match is not None:
4761 if issubclass(coltype._type_affinity, sqltypes.Integer):
4762 autoincrement = True
4763 # the default is related to a Sequence
4764 if "." not in match.group(2) and schema is not None:
4765 # unconditionally quote the schema name. this could
4766 # later be enhanced to obey quoting rules /
4767 # "quote schema"
4768 default = (
4769 match.group(1)
4770 + ('"%s"' % schema)
4771 + "."
4772 + match.group(2)
4773 + match.group(3)
4774 )
4776 column_info = {
4777 "name": name,
4778 "type": coltype,
4779 "nullable": nullable,
4780 "default": default,
4781 "autoincrement": autoincrement or identity is not None,
4782 "comment": row_dict["comment"],
4783 }
4784 if computed is not None:
4785 column_info["computed"] = computed
4786 if identity is not None:
4787 column_info["identity"] = identity
4789 table_cols.append(column_info)
4791 return columns
4793 @lru_cache()
4794 def _table_oids_query(self, schema, has_filter_names, scope, kind):
4795 relkinds = self._kind_to_relkinds(kind)
4796 oid_q = select(
4797 pg_catalog.pg_class.c.oid, pg_catalog.pg_class.c.relname
4798 ).where(self._pg_class_relkind_condition(relkinds))
4799 oid_q = self._pg_class_filter_scope_schema(oid_q, schema, scope=scope)
4801 if has_filter_names:
4802 oid_q = oid_q.where(
4803 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
4804 )
4805 return oid_q
4807 @reflection.flexi_cache(
4808 ("schema", InternalTraversal.dp_string),
4809 ("filter_names", InternalTraversal.dp_string_list),
4810 ("kind", InternalTraversal.dp_plain_obj),
4811 ("scope", InternalTraversal.dp_plain_obj),
4812 )
4813 def _get_table_oids(
4814 self, connection, schema, filter_names, scope, kind, **kw
4815 ):
4816 has_filter_names, params = self._prepare_filter_names(filter_names)
4817 oid_q = self._table_oids_query(schema, has_filter_names, scope, kind)
4818 result = connection.execute(oid_q, params)
4819 return result.all()
4821 @util.memoized_property
4822 def _constraint_query(self):
4823 if self.server_version_info >= (11, 0):
4824 indnkeyatts = pg_catalog.pg_index.c.indnkeyatts
4825 else:
4826 indnkeyatts = pg_catalog.pg_index.c.indnatts.label("indnkeyatts")
4828 if self.server_version_info >= (15,):
4829 indnullsnotdistinct = pg_catalog.pg_index.c.indnullsnotdistinct
4830 else:
4831 indnullsnotdistinct = sql.false().label("indnullsnotdistinct")
4833 con_sq = (
4834 select(
4835 pg_catalog.pg_constraint.c.conrelid,
4836 pg_catalog.pg_constraint.c.conname,
4837 sql.func.unnest(pg_catalog.pg_index.c.indkey).label("attnum"),
4838 sql.func.generate_subscripts(
4839 pg_catalog.pg_index.c.indkey, 1
4840 ).label("ord"),
4841 indnkeyatts,
4842 indnullsnotdistinct,
4843 pg_catalog.pg_description.c.description,
4844 )
4845 .join(
4846 pg_catalog.pg_index,
4847 pg_catalog.pg_constraint.c.conindid
4848 == pg_catalog.pg_index.c.indexrelid,
4849 )
4850 .outerjoin(
4851 pg_catalog.pg_description,
4852 sql.and_(
4853 pg_catalog.pg_description.c.objoid
4854 == pg_catalog.pg_constraint.c.oid,
4855 pg_catalog.pg_description.c.classoid
4856 == sql.func.cast("pg_catalog.pg_constraint", REGCLASS),
4857 ),
4858 )
4859 .where(
4860 pg_catalog.pg_constraint.c.contype == bindparam("contype"),
4861 pg_catalog.pg_constraint.c.conrelid.in_(bindparam("oids")),
4862 # NOTE: filtering also on pg_index.indrelid for oids does
4863 # not seem to have a performance effect, but it may be an
4864 # option if perf problems are reported
4865 )
4866 .subquery("con")
4867 )
4869 attr_sq = (
4870 select(
4871 con_sq.c.conrelid,
4872 con_sq.c.conname,
4873 con_sq.c.description,
4874 con_sq.c.ord,
4875 con_sq.c.indnkeyatts,
4876 con_sq.c.indnullsnotdistinct,
4877 pg_catalog.pg_attribute.c.attname,
4878 )
4879 .select_from(pg_catalog.pg_attribute)
4880 .join(
4881 con_sq,
4882 sql.and_(
4883 pg_catalog.pg_attribute.c.attnum == con_sq.c.attnum,
4884 pg_catalog.pg_attribute.c.attrelid == con_sq.c.conrelid,
4885 ),
4886 )
4887 .where(
4888 # NOTE: restate the condition here, since pg15 otherwise
4889 # seems to get confused on pscopg2 sometimes, doing
4890 # a sequential scan of pg_attribute.
4891 # The condition in the con_sq subquery is not actually needed
4892 # in pg15, but it may be needed in older versions. Keeping it
4893 # does not seems to have any impact in any case.
4894 con_sq.c.conrelid.in_(bindparam("oids"))
4895 )
4896 .subquery("attr")
4897 )
4899 return (
4900 select(
4901 attr_sq.c.conrelid,
4902 sql.func.array_agg(
4903 # NOTE: cast since some postgresql derivatives may
4904 # not support array_agg on the name type
4905 aggregate_order_by(
4906 attr_sq.c.attname.cast(TEXT), attr_sq.c.ord
4907 )
4908 ).label("cols"),
4909 attr_sq.c.conname,
4910 sql.func.min(attr_sq.c.description).label("description"),
4911 sql.func.min(attr_sq.c.indnkeyatts).label("indnkeyatts"),
4912 sql.func.bool_and(attr_sq.c.indnullsnotdistinct).label(
4913 "indnullsnotdistinct"
4914 ),
4915 )
4916 .group_by(attr_sq.c.conrelid, attr_sq.c.conname)
4917 .order_by(attr_sq.c.conrelid, attr_sq.c.conname)
4918 )
4920 def _reflect_constraint(
4921 self, connection, contype, schema, filter_names, scope, kind, **kw
4922 ):
4923 # used to reflect primary and unique constraint
4924 table_oids = self._get_table_oids(
4925 connection, schema, filter_names, scope, kind, **kw
4926 )
4927 batches = list(table_oids)
4928 is_unique = contype == "u"
4930 while batches:
4931 batch = batches[0:3000]
4932 batches[0:3000] = []
4934 result = connection.execute(
4935 self._constraint_query,
4936 {"oids": [r[0] for r in batch], "contype": contype},
4937 ).mappings()
4939 result_by_oid = defaultdict(list)
4940 for row_dict in result:
4941 result_by_oid[row_dict["conrelid"]].append(row_dict)
4943 for oid, tablename in batch:
4944 for_oid = result_by_oid.get(oid, ())
4945 if for_oid:
4946 for row in for_oid:
4947 # See note in get_multi_indexes
4948 all_cols = row["cols"]
4949 indnkeyatts = row["indnkeyatts"]
4950 if len(all_cols) > indnkeyatts:
4951 inc_cols = all_cols[indnkeyatts:]
4952 cst_cols = all_cols[:indnkeyatts]
4953 else:
4954 inc_cols = []
4955 cst_cols = all_cols
4957 opts = {}
4958 if self.server_version_info >= (11,):
4959 opts["postgresql_include"] = inc_cols
4960 if is_unique:
4961 opts["postgresql_nulls_not_distinct"] = row[
4962 "indnullsnotdistinct"
4963 ]
4964 yield (
4965 tablename,
4966 cst_cols,
4967 row["conname"],
4968 row["description"],
4969 opts,
4970 )
4971 else:
4972 yield tablename, None, None, None, None
4974 def get_multi_pk_constraint(
4975 self, connection, schema, filter_names, scope, kind, **kw
4976 ):
4977 result = self._reflect_constraint(
4978 connection, "p", schema, filter_names, scope, kind, **kw
4979 )
4981 # only a single pk can be present for each table. Return an entry
4982 # even if a table has no primary key
4983 default = ReflectionDefaults.pk_constraint
4985 def pk_constraint(pk_name, cols, comment, opts):
4986 info = {
4987 "constrained_columns": cols,
4988 "name": pk_name,
4989 "comment": comment,
4990 }
4991 if opts:
4992 info["dialect_options"] = opts
4993 return info
4995 return (
4996 (
4997 (schema, table_name),
4998 (
4999 pk_constraint(pk_name, cols, comment, opts)
5000 if pk_name is not None
5001 else default()
5002 ),
5003 )
5004 for table_name, cols, pk_name, comment, opts in result
5005 )
5007 @lru_cache()
5008 def _foreing_key_query(self, schema, has_filter_names, scope, kind):
5009 pg_class_ref = pg_catalog.pg_class.alias("cls_ref")
5010 pg_namespace_ref = pg_catalog.pg_namespace.alias("nsp_ref")
5011 relkinds = self._kind_to_relkinds(kind)
5012 query = (
5013 select(
5014 pg_catalog.pg_class.c.relname,
5015 pg_catalog.pg_constraint.c.conname,
5016 # NOTE: avoid calling pg_get_constraintdef when not needed
5017 # to speed up the query
5018 sql.case(
5019 (
5020 pg_catalog.pg_constraint.c.oid.is_not(None),
5021 pg_catalog.pg_get_constraintdef(
5022 pg_catalog.pg_constraint.c.oid, True
5023 ),
5024 ),
5025 else_=None,
5026 ),
5027 pg_namespace_ref.c.nspname,
5028 pg_catalog.pg_description.c.description,
5029 )
5030 .select_from(pg_catalog.pg_class)
5031 .outerjoin(
5032 pg_catalog.pg_constraint,
5033 sql.and_(
5034 pg_catalog.pg_class.c.oid
5035 == pg_catalog.pg_constraint.c.conrelid,
5036 pg_catalog.pg_constraint.c.contype == "f",
5037 ),
5038 )
5039 .outerjoin(
5040 pg_class_ref,
5041 pg_class_ref.c.oid == pg_catalog.pg_constraint.c.confrelid,
5042 )
5043 .outerjoin(
5044 pg_namespace_ref,
5045 pg_class_ref.c.relnamespace == pg_namespace_ref.c.oid,
5046 )
5047 .outerjoin(
5048 pg_catalog.pg_description,
5049 sql.and_(
5050 pg_catalog.pg_description.c.objoid
5051 == pg_catalog.pg_constraint.c.oid,
5052 pg_catalog.pg_description.c.classoid
5053 == sql.func.cast("pg_catalog.pg_constraint", REGCLASS),
5054 ),
5055 )
5056 .order_by(
5057 pg_catalog.pg_class.c.relname,
5058 pg_catalog.pg_constraint.c.conname,
5059 )
5060 .where(self._pg_class_relkind_condition(relkinds))
5061 )
5062 query = self._pg_class_filter_scope_schema(query, schema, scope)
5063 if has_filter_names:
5064 query = query.where(
5065 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
5066 )
5067 return query
5069 @util.memoized_property
5070 def _fk_regex_pattern(self):
5071 # optionally quoted token
5072 qtoken = r'(?:"(?:[^"]|"")+"|[\w]+?)'
5074 # https://www.postgresql.org/docs/current/static/sql-createtable.html
5075 return re.compile(
5076 rf"FOREIGN KEY \((.*?)(?:, PERIOD ({qtoken}))?\) "
5077 rf"REFERENCES (?:({qtoken})\.)?({qtoken})"
5078 rf"\(((?:{qtoken}(?: *, *)?)+)(?:, PERIOD ({qtoken}))?\)"
5079 r"[\s]?(MATCH (FULL|PARTIAL|SIMPLE)+)?"
5080 r"[\s]?(?:ON (UPDATE|DELETE) "
5081 r"(CASCADE|RESTRICT|NO ACTION|"
5082 r"SET (?:NULL|DEFAULT)(?:\s\(.+\))?)+)?"
5083 r"[\s]?(?:ON (UPDATE|DELETE) "
5084 r"(CASCADE|RESTRICT|NO ACTION|"
5085 r"SET (?:NULL|DEFAULT)(?:\s\(.+\))?)+)?"
5086 r"[\s]?(DEFERRABLE|NOT DEFERRABLE)?"
5087 r"[\s]?(INITIALLY (DEFERRED|IMMEDIATE)+)?"
5088 )
5090 def _parse_fk(self, condef):
5091 FK_REGEX = self._fk_regex_pattern
5092 m = re.search(FK_REGEX, condef).groups()
5094 (
5095 constrained_columns,
5096 constrained_period_column,
5097 referred_schema,
5098 referred_table,
5099 referred_columns,
5100 referred_period_column,
5101 _,
5102 match,
5103 upddelkey1,
5104 upddelval1,
5105 upddelkey2,
5106 upddelval2,
5107 deferrable,
5108 _,
5109 initially,
5110 ) = m
5112 # TODO: the PERIOD designation of a temporal foreign key
5113 # (PostgreSQL 18+) is not yet reflected
5114 if constrained_period_column:
5115 constrained_columns += f", {constrained_period_column}"
5116 if referred_period_column:
5117 referred_columns += f", {referred_period_column}"
5119 onupdate = (
5120 upddelval1
5121 if upddelkey1 == "UPDATE"
5122 else upddelval2 if upddelkey2 == "UPDATE" else None
5123 )
5124 ondelete = (
5125 upddelval1
5126 if upddelkey1 == "DELETE"
5127 else upddelval2 if upddelkey2 == "DELETE" else None
5128 )
5130 return (
5131 constrained_columns,
5132 referred_schema,
5133 referred_table,
5134 referred_columns,
5135 match,
5136 onupdate,
5137 ondelete,
5138 deferrable,
5139 initially,
5140 )
5142 def get_multi_foreign_keys(
5143 self,
5144 connection,
5145 schema,
5146 filter_names,
5147 scope,
5148 kind,
5149 postgresql_ignore_search_path=False,
5150 **kw,
5151 ):
5152 preparer = self.identifier_preparer
5154 has_filter_names, params = self._prepare_filter_names(filter_names)
5155 query = self._foreing_key_query(schema, has_filter_names, scope, kind)
5156 result = connection.execute(query, params)
5158 fkeys = defaultdict(list)
5159 default = ReflectionDefaults.foreign_keys
5160 for table_name, conname, condef, conschema, comment in result:
5161 # ensure that each table has an entry, even if it has
5162 # no foreign keys
5163 if conname is None:
5164 fkeys[(schema, table_name)] = default()
5165 continue
5166 table_fks = fkeys[(schema, table_name)]
5168 (
5169 constrained_columns,
5170 referred_schema,
5171 referred_table,
5172 referred_columns,
5173 match,
5174 onupdate,
5175 ondelete,
5176 deferrable,
5177 initially,
5178 ) = self._parse_fk(condef)
5180 if deferrable is not None:
5181 deferrable = True if deferrable == "DEFERRABLE" else False
5182 constrained_columns = [
5183 preparer._unquote_identifier(x)
5184 for x in re.split(r"\s*,\s*", constrained_columns)
5185 ]
5187 if postgresql_ignore_search_path:
5188 # when ignoring search path, we use the actual schema
5189 # provided it isn't the "default" schema
5190 if conschema != self.default_schema_name:
5191 referred_schema = conschema
5192 else:
5193 referred_schema = schema
5194 elif referred_schema:
5195 # referred_schema is the schema that we regexp'ed from
5196 # pg_get_constraintdef(). If the schema is in the search
5197 # path, pg_get_constraintdef() will give us None.
5198 referred_schema = preparer._unquote_identifier(referred_schema)
5199 elif schema is not None and schema == conschema:
5200 # If the actual schema matches the schema of the table
5201 # we're reflecting, then we will use that.
5202 referred_schema = schema
5204 referred_table = preparer._unquote_identifier(referred_table)
5205 referred_columns = [
5206 preparer._unquote_identifier(x)
5207 for x in re.split(r"\s*,\s", referred_columns)
5208 ]
5209 options = {
5210 k: v
5211 for k, v in [
5212 ("onupdate", onupdate),
5213 ("ondelete", ondelete),
5214 ("initially", initially),
5215 ("deferrable", deferrable),
5216 ("match", match),
5217 ]
5218 if v is not None and v != "NO ACTION"
5219 }
5220 fkey_d = {
5221 "name": conname,
5222 "constrained_columns": constrained_columns,
5223 "referred_schema": referred_schema,
5224 "referred_table": referred_table,
5225 "referred_columns": referred_columns,
5226 "options": options,
5227 "comment": comment,
5228 }
5229 table_fks.append(fkey_d)
5230 return fkeys.items()
5232 @util.memoized_property
5233 def _index_query(self):
5234 # NOTE: pg_index is used as from two times to improve performance,
5235 # since extraing all the index information from `idx_sq` to avoid
5236 # the second pg_index use leads to a worse performing query in
5237 # particular when querying for a single table (as of pg 17)
5238 # NOTE: repeating oids clause improve query performance
5240 # subquery to get the columns
5241 idx_sq = (
5242 select(
5243 pg_catalog.pg_index.c.indexrelid,
5244 pg_catalog.pg_index.c.indrelid,
5245 sql.func.unnest(pg_catalog.pg_index.c.indkey).label("attnum"),
5246 sql.func.unnest(pg_catalog.pg_index.c.indclass).label(
5247 "att_opclass"
5248 ),
5249 sql.func.generate_subscripts(
5250 pg_catalog.pg_index.c.indkey, 1
5251 ).label("ord"),
5252 )
5253 .where(
5254 ~pg_catalog.pg_index.c.indisprimary,
5255 pg_catalog.pg_index.c.indrelid.in_(bindparam("oids")),
5256 )
5257 .subquery("idx")
5258 )
5260 attr_sq = (
5261 select(
5262 idx_sq.c.indexrelid,
5263 idx_sq.c.indrelid,
5264 idx_sq.c.ord,
5265 # NOTE: always using pg_get_indexdef is too slow so just
5266 # invoke when the element is an expression
5267 sql.case(
5268 (
5269 idx_sq.c.attnum == 0,
5270 pg_catalog.pg_get_indexdef(
5271 idx_sq.c.indexrelid, idx_sq.c.ord + 1, True
5272 ),
5273 ),
5274 # NOTE: need to cast this since attname is of type "name"
5275 # that's limited to 63 bytes, while pg_get_indexdef
5276 # returns "text" so its output may get cut
5277 else_=pg_catalog.pg_attribute.c.attname.cast(TEXT),
5278 ).label("element"),
5279 (idx_sq.c.attnum == 0).label("is_expr"),
5280 # since it's converted to array cast it to bigint (oid are
5281 # "unsigned four-byte integer") to make it easier for
5282 # dialects to interpret
5283 idx_sq.c.att_opclass.cast(BIGINT),
5284 )
5285 .select_from(idx_sq)
5286 .outerjoin(
5287 # do not remove rows where idx_sq.c.attnum is 0
5288 pg_catalog.pg_attribute,
5289 sql.and_(
5290 pg_catalog.pg_attribute.c.attnum == idx_sq.c.attnum,
5291 pg_catalog.pg_attribute.c.attrelid == idx_sq.c.indrelid,
5292 ),
5293 )
5294 .where(idx_sq.c.indrelid.in_(bindparam("oids")))
5295 .subquery("idx_attr")
5296 )
5298 cols_sq = (
5299 select(
5300 attr_sq.c.indexrelid,
5301 sql.func.min(attr_sq.c.indrelid),
5302 sql.func.array_agg(
5303 aggregate_order_by(attr_sq.c.element, attr_sq.c.ord)
5304 ).label("elements"),
5305 sql.func.array_agg(
5306 aggregate_order_by(attr_sq.c.is_expr, attr_sq.c.ord)
5307 ).label("elements_is_expr"),
5308 sql.func.array_agg(
5309 aggregate_order_by(attr_sq.c.att_opclass, attr_sq.c.ord)
5310 ).label("elements_opclass"),
5311 )
5312 .group_by(attr_sq.c.indexrelid)
5313 .subquery("idx_cols")
5314 )
5316 if self.server_version_info >= (11, 0):
5317 indnkeyatts = pg_catalog.pg_index.c.indnkeyatts
5318 else:
5319 indnkeyatts = pg_catalog.pg_index.c.indnatts.label("indnkeyatts")
5321 if self.server_version_info >= (15,):
5322 nulls_not_distinct = pg_catalog.pg_index.c.indnullsnotdistinct
5323 else:
5324 nulls_not_distinct = sql.false().label("indnullsnotdistinct")
5326 return (
5327 select(
5328 pg_catalog.pg_index.c.indrelid,
5329 pg_catalog.pg_class.c.relname,
5330 pg_catalog.pg_index.c.indisunique,
5331 pg_catalog.pg_index.c.indisvalid,
5332 pg_catalog.pg_constraint.c.conrelid.is_not(None).label(
5333 "has_constraint"
5334 ),
5335 pg_catalog.pg_index.c.indoption,
5336 pg_catalog.pg_class.c.reloptions,
5337 # will get the value using the pg_am cached dict
5338 pg_catalog.pg_class.c.relam,
5339 # NOTE: pg_get_expr is very fast so this case has almost no
5340 # performance impact
5341 sql.case(
5342 (
5343 pg_catalog.pg_index.c.indpred.is_not(None),
5344 pg_catalog.pg_get_expr(
5345 pg_catalog.pg_index.c.indpred,
5346 pg_catalog.pg_index.c.indrelid,
5347 ),
5348 ),
5349 else_=None,
5350 ).label("filter_definition"),
5351 indnkeyatts,
5352 nulls_not_distinct,
5353 cols_sq.c.elements,
5354 cols_sq.c.elements_is_expr,
5355 # will get the value using the pg_opclass cached dict
5356 cols_sq.c.elements_opclass,
5357 )
5358 .select_from(pg_catalog.pg_index)
5359 .where(
5360 pg_catalog.pg_index.c.indrelid.in_(bindparam("oids")),
5361 ~pg_catalog.pg_index.c.indisprimary,
5362 )
5363 .join(
5364 pg_catalog.pg_class,
5365 pg_catalog.pg_index.c.indexrelid == pg_catalog.pg_class.c.oid,
5366 )
5367 .outerjoin(
5368 cols_sq,
5369 pg_catalog.pg_index.c.indexrelid == cols_sq.c.indexrelid,
5370 )
5371 .outerjoin(
5372 pg_catalog.pg_constraint,
5373 sql.and_(
5374 pg_catalog.pg_index.c.indrelid
5375 == pg_catalog.pg_constraint.c.conrelid,
5376 pg_catalog.pg_index.c.indexrelid
5377 == pg_catalog.pg_constraint.c.conindid,
5378 pg_catalog.pg_constraint.c.contype
5379 == sql.any_(_array.array(("p", "u", "x"))),
5380 ),
5381 )
5382 .order_by(
5383 pg_catalog.pg_index.c.indrelid, pg_catalog.pg_class.c.relname
5384 )
5385 )
5387 def get_multi_indexes(
5388 self, connection, schema, filter_names, scope, kind, **kw
5389 ):
5390 table_oids = self._get_table_oids(
5391 connection, schema, filter_names, scope, kind, **kw
5392 )
5394 pg_am_btree_oid = self._load_pg_am_btree_oid(connection)
5395 # lazy load only if needed, the assumption is that most indexes
5396 # will use btree so it may not be needed at all
5397 pg_am_dict = None
5398 pg_opclass_dict = self._load_pg_opclass_notdefault_dict(
5399 connection, **kw
5400 )
5402 indexes = defaultdict(list)
5403 default = ReflectionDefaults.indexes
5405 batches = list(table_oids)
5407 while batches:
5408 batch = batches[0:3000]
5409 batches[0:3000] = []
5411 result = connection.execute(
5412 self._index_query, {"oids": [r[0] for r in batch]}
5413 ).mappings()
5415 result_by_oid = defaultdict(list)
5416 for row_dict in result:
5417 result_by_oid[row_dict["indrelid"]].append(row_dict)
5419 for oid, table_name in batch:
5420 if oid not in result_by_oid:
5421 # ensure that each table has an entry, even if reflection
5422 # is skipped because not supported
5423 indexes[(schema, table_name)] = default()
5424 continue
5426 for row in result_by_oid[oid]:
5427 index_name = row["relname"]
5429 table_indexes = indexes[(schema, table_name)]
5431 all_elements = row["elements"]
5432 all_elements_is_expr = row["elements_is_expr"]
5433 all_elements_opclass = row["elements_opclass"]
5434 indnkeyatts = row["indnkeyatts"]
5435 # "The number of key columns in the index, not counting any
5436 # included columns, which are merely stored and do not
5437 # participate in the index semantics"
5438 if len(all_elements) > indnkeyatts:
5439 # this is a "covering index" which has INCLUDE columns
5440 # as well as regular index columns
5441 inc_cols = all_elements[indnkeyatts:]
5442 idx_elements = all_elements[:indnkeyatts]
5443 idx_elements_is_expr = all_elements_is_expr[
5444 :indnkeyatts
5445 ]
5446 # postgresql does not support expression on included
5447 # columns as of v14: "ERROR: expressions are not
5448 # supported in included columns".
5449 assert all(
5450 not is_expr
5451 for is_expr in all_elements_is_expr[indnkeyatts:]
5452 )
5453 idx_elements_opclass = all_elements_opclass[
5454 :indnkeyatts
5455 ]
5456 else:
5457 idx_elements = all_elements
5458 idx_elements_is_expr = all_elements_is_expr
5459 inc_cols = []
5460 idx_elements_opclass = all_elements_opclass
5462 index = {"name": index_name, "unique": row["indisunique"]}
5463 if any(idx_elements_is_expr):
5464 index["column_names"] = [
5465 None if is_expr else expr
5466 for expr, is_expr in zip(
5467 idx_elements, idx_elements_is_expr
5468 )
5469 ]
5470 index["expressions"] = idx_elements
5471 else:
5472 index["column_names"] = idx_elements
5474 dialect_options = {}
5476 postgresql_ops = {}
5477 for name, opclass in zip(
5478 idx_elements, idx_elements_opclass
5479 ):
5480 # is not in the dict if the opclass is the default one
5481 opclass_name = pg_opclass_dict.get(opclass)
5482 if opclass_name is not None:
5483 postgresql_ops[name] = opclass_name
5485 if postgresql_ops:
5486 dialect_options["postgresql_ops"] = postgresql_ops
5488 sorting = {}
5489 for col_index, col_flags in enumerate(row["indoption"]):
5490 col_sorting = ()
5491 # try to set flags only if they differ from PG
5492 # defaults...
5493 if col_flags & 0x01:
5494 col_sorting += ("desc",)
5495 if not (col_flags & 0x02):
5496 col_sorting += ("nulls_last",)
5497 else:
5498 if col_flags & 0x02:
5499 col_sorting += ("nulls_first",)
5500 if col_sorting:
5501 sorting[idx_elements[col_index]] = col_sorting
5502 if sorting:
5503 index["column_sorting"] = sorting
5504 if row["has_constraint"]:
5505 index["duplicates_constraint"] = index_name
5507 if row["reloptions"]:
5508 dialect_options["postgresql_with"] = dict(
5509 [
5510 option.split("=", 1)
5511 for option in row["reloptions"]
5512 ]
5513 )
5514 # it *might* be nice to include that this is 'btree' in the
5515 # reflection info. But we don't want an Index object
5516 # to have a ``postgresql_using`` in it that is just the
5517 # default, so for the moment leaving this out.
5518 if row["relam"] != pg_am_btree_oid:
5519 if pg_am_dict is None:
5520 pg_am_dict = self._load_pg_am_dict(
5521 connection, **kw
5522 )
5523 dialect_options["postgresql_using"] = pg_am_dict[
5524 row["relam"]
5525 ]
5526 if row["filter_definition"]:
5527 dialect_options["postgresql_where"] = row[
5528 "filter_definition"
5529 ]
5530 if self.server_version_info >= (11,):
5531 dialect_options["postgresql_include"] = inc_cols
5532 if row["indnullsnotdistinct"]:
5533 # the default is False, so ignore it.
5534 dialect_options["postgresql_nulls_not_distinct"] = row[
5535 "indnullsnotdistinct"
5536 ]
5538 if not row["indisvalid"]:
5539 dialect_options["postgresql_invalid"] = True
5541 if dialect_options:
5542 index["dialect_options"] = dialect_options
5544 table_indexes.append(index)
5545 return indexes.items()
5547 def get_multi_unique_constraints(
5548 self,
5549 connection,
5550 schema,
5551 filter_names,
5552 scope,
5553 kind,
5554 **kw,
5555 ):
5556 result = self._reflect_constraint(
5557 connection, "u", schema, filter_names, scope, kind, **kw
5558 )
5560 # each table can have multiple unique constraints
5561 uniques = defaultdict(list)
5562 default = ReflectionDefaults.unique_constraints
5563 for table_name, cols, con_name, comment, options in result:
5564 # ensure a list is created for each table. leave it empty if
5565 # the table has no unique constraint
5566 if con_name is None:
5567 uniques[(schema, table_name)] = default()
5568 continue
5570 uc_dict = {
5571 "column_names": cols,
5572 "name": con_name,
5573 "comment": comment,
5574 }
5575 if options:
5576 uc_dict["dialect_options"] = options
5578 uniques[(schema, table_name)].append(uc_dict)
5579 return uniques.items()
5581 @lru_cache()
5582 def _comment_query(self, schema, has_filter_names, scope, kind):
5583 relkinds = self._kind_to_relkinds(kind)
5584 query = (
5585 select(
5586 pg_catalog.pg_class.c.relname,
5587 pg_catalog.pg_description.c.description,
5588 )
5589 .select_from(pg_catalog.pg_class)
5590 .outerjoin(
5591 pg_catalog.pg_description,
5592 sql.and_(
5593 pg_catalog.pg_class.c.oid
5594 == pg_catalog.pg_description.c.objoid,
5595 pg_catalog.pg_description.c.objsubid == 0,
5596 pg_catalog.pg_description.c.classoid
5597 == sql.func.cast("pg_catalog.pg_class", REGCLASS),
5598 ),
5599 )
5600 .where(self._pg_class_relkind_condition(relkinds))
5601 )
5602 query = self._pg_class_filter_scope_schema(query, schema, scope)
5603 if has_filter_names:
5604 query = query.where(
5605 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
5606 )
5607 return query
5609 def get_multi_table_comment(
5610 self, connection, schema, filter_names, scope, kind, **kw
5611 ):
5612 has_filter_names, params = self._prepare_filter_names(filter_names)
5613 query = self._comment_query(schema, has_filter_names, scope, kind)
5614 result = connection.execute(query, params)
5616 default = ReflectionDefaults.table_comment
5617 return (
5618 (
5619 (schema, table),
5620 {"text": comment} if comment is not None else default(),
5621 )
5622 for table, comment in result
5623 )
5625 @lru_cache()
5626 def _check_constraint_query(self, schema, has_filter_names, scope, kind):
5627 relkinds = self._kind_to_relkinds(kind)
5628 query = (
5629 select(
5630 pg_catalog.pg_class.c.relname,
5631 pg_catalog.pg_constraint.c.conname,
5632 # NOTE: avoid calling pg_get_constraintdef when not needed
5633 # to speed up the query
5634 sql.case(
5635 (
5636 pg_catalog.pg_constraint.c.oid.is_not(None),
5637 pg_catalog.pg_get_constraintdef(
5638 pg_catalog.pg_constraint.c.oid, True
5639 ),
5640 ),
5641 else_=None,
5642 ),
5643 pg_catalog.pg_description.c.description,
5644 )
5645 .select_from(pg_catalog.pg_class)
5646 .outerjoin(
5647 pg_catalog.pg_constraint,
5648 sql.and_(
5649 pg_catalog.pg_class.c.oid
5650 == pg_catalog.pg_constraint.c.conrelid,
5651 pg_catalog.pg_constraint.c.contype == "c",
5652 ),
5653 )
5654 .outerjoin(
5655 pg_catalog.pg_description,
5656 sql.and_(
5657 pg_catalog.pg_description.c.objoid
5658 == pg_catalog.pg_constraint.c.oid,
5659 pg_catalog.pg_description.c.classoid
5660 == sql.func.cast("pg_catalog.pg_constraint", REGCLASS),
5661 ),
5662 )
5663 .order_by(
5664 pg_catalog.pg_class.c.relname,
5665 pg_catalog.pg_constraint.c.conname,
5666 )
5667 .where(self._pg_class_relkind_condition(relkinds))
5668 )
5669 query = self._pg_class_filter_scope_schema(query, schema, scope)
5670 if has_filter_names:
5671 query = query.where(
5672 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
5673 )
5674 return query
5676 def get_multi_check_constraints(
5677 self, connection, schema, filter_names, scope, kind, **kw
5678 ):
5679 has_filter_names, params = self._prepare_filter_names(filter_names)
5680 query = self._check_constraint_query(
5681 schema, has_filter_names, scope, kind
5682 )
5683 result = connection.execute(query, params)
5685 check_constraints = defaultdict(list)
5686 default = ReflectionDefaults.check_constraints
5687 for table_name, check_name, src, comment in result:
5688 # only two cases for check_name and src: both null or both defined
5689 if check_name is None and src is None:
5690 check_constraints[(schema, table_name)] = default()
5691 continue
5692 # samples:
5693 # "CHECK (((a > 1) AND (a < 5)))"
5694 # "CHECK (((a = 1) OR ((a > 2) AND (a < 5))))"
5695 # "CHECK (((a > 1) AND (a < 5))) NOT VALID"
5696 # "CHECK (some_boolean_function(a))"
5697 # "CHECK (((a\n < 1)\n OR\n (a\n >= 5))\n)"
5698 # "CHECK (a NOT NULL) NO INHERIT"
5699 # "CHECK (a NOT NULL) NO INHERIT NOT VALID"
5701 m = re.match(
5702 r"^CHECK *\((.+)\)( NO INHERIT)?( NOT VALID)?$",
5703 src,
5704 flags=re.DOTALL,
5705 )
5706 if not m:
5707 util.warn("Could not parse CHECK constraint text: %r" % src)
5708 sqltext = ""
5709 else:
5710 sqltext = util.strip_outer_parens(m.group(1))
5711 entry = {
5712 "name": check_name,
5713 "sqltext": sqltext,
5714 "comment": comment,
5715 }
5716 if m:
5717 do = {}
5718 if " NOT VALID" in m.groups():
5719 do["not_valid"] = True
5720 if " NO INHERIT" in m.groups():
5721 do["no_inherit"] = True
5722 if do:
5723 entry["dialect_options"] = do
5725 check_constraints[(schema, table_name)].append(entry)
5726 return check_constraints.items()
5728 def _pg_type_filter_schema(self, query, schema):
5729 if schema is None:
5730 query = query.where(
5731 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid),
5732 # ignore pg_catalog schema
5733 pg_catalog.pg_namespace.c.nspname != "pg_catalog",
5734 )
5735 elif schema != "*":
5736 query = query.where(pg_catalog.pg_namespace.c.nspname == schema)
5737 return query
5739 @lru_cache()
5740 def _enum_query(self, schema):
5741 lbl_agg_sq = (
5742 select(
5743 pg_catalog.pg_enum.c.enumtypid,
5744 sql.func.array_agg(
5745 aggregate_order_by(
5746 # NOTE: cast since some postgresql derivatives may
5747 # not support array_agg on the name type
5748 pg_catalog.pg_enum.c.enumlabel.cast(TEXT),
5749 pg_catalog.pg_enum.c.enumsortorder,
5750 )
5751 ).label("labels"),
5752 )
5753 .group_by(pg_catalog.pg_enum.c.enumtypid)
5754 .subquery("lbl_agg")
5755 )
5757 query = (
5758 select(
5759 pg_catalog.pg_type.c.typname.label("name"),
5760 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid).label(
5761 "visible"
5762 ),
5763 pg_catalog.pg_namespace.c.nspname.label("schema"),
5764 lbl_agg_sq.c.labels.label("labels"),
5765 )
5766 .join(
5767 pg_catalog.pg_namespace,
5768 pg_catalog.pg_namespace.c.oid
5769 == pg_catalog.pg_type.c.typnamespace,
5770 )
5771 .outerjoin(
5772 lbl_agg_sq, pg_catalog.pg_type.c.oid == lbl_agg_sq.c.enumtypid
5773 )
5774 .where(pg_catalog.pg_type.c.typtype == "e")
5775 .order_by(
5776 pg_catalog.pg_namespace.c.nspname, pg_catalog.pg_type.c.typname
5777 )
5778 )
5780 return self._pg_type_filter_schema(query, schema)
5782 @reflection.cache
5783 def _load_enums(self, connection, schema=None, **kw):
5784 if not self.supports_native_enum:
5785 return []
5787 result = connection.execute(self._enum_query(schema))
5789 enums = []
5790 for name, visible, schema, labels in result:
5791 enums.append(
5792 {
5793 "name": name,
5794 "schema": schema,
5795 "visible": visible,
5796 "labels": [] if labels is None else labels,
5797 }
5798 )
5799 return enums
5801 @lru_cache()
5802 def _domain_query(self, schema):
5803 con_sq = (
5804 select(
5805 pg_catalog.pg_constraint.c.contypid,
5806 sql.func.array_agg(
5807 pg_catalog.pg_get_constraintdef(
5808 pg_catalog.pg_constraint.c.oid, True
5809 )
5810 ).label("condefs"),
5811 sql.func.array_agg(
5812 # NOTE: cast since some postgresql derivatives may
5813 # not support array_agg on the name type
5814 pg_catalog.pg_constraint.c.conname.cast(TEXT)
5815 ).label("connames"),
5816 )
5817 # The domain this constraint is on; zero if not a domain constraint
5818 .where(pg_catalog.pg_constraint.c.contypid != 0)
5819 .group_by(pg_catalog.pg_constraint.c.contypid)
5820 .subquery("domain_constraints")
5821 )
5823 collation_namespace = pg_catalog.pg_namespace.alias(
5824 "collation_namespace"
5825 )
5827 query = (
5828 select(
5829 pg_catalog.pg_type.c.typname.label("name"),
5830 pg_catalog.format_type(
5831 pg_catalog.pg_type.c.typbasetype,
5832 pg_catalog.pg_type.c.typtypmod,
5833 ).label("attype"),
5834 (~pg_catalog.pg_type.c.typnotnull).label("nullable"),
5835 pg_catalog.pg_type.c.typdefault.label("default"),
5836 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid).label(
5837 "visible"
5838 ),
5839 pg_catalog.pg_namespace.c.nspname.label("schema"),
5840 con_sq.c.condefs,
5841 con_sq.c.connames,
5842 pg_catalog.pg_collation.c.collname,
5843 sql.case(
5844 (
5845 pg_catalog.pg_collation.c.oid.is_(None),
5846 sql.null(),
5847 ),
5848 (
5849 pg_catalog.pg_collation_is_visible(
5850 pg_catalog.pg_collation.c.oid
5851 ),
5852 sql.null(),
5853 ),
5854 else_=collation_namespace.c.nspname,
5855 ).label("collation_schema"),
5856 )
5857 .join(
5858 pg_catalog.pg_namespace,
5859 pg_catalog.pg_namespace.c.oid
5860 == pg_catalog.pg_type.c.typnamespace,
5861 )
5862 .outerjoin(
5863 pg_catalog.pg_collation,
5864 pg_catalog.pg_type.c.typcollation
5865 == pg_catalog.pg_collation.c.oid,
5866 )
5867 .outerjoin(
5868 collation_namespace,
5869 collation_namespace.c.oid
5870 == pg_catalog.pg_collation.c.collnamespace,
5871 )
5872 .outerjoin(
5873 con_sq,
5874 pg_catalog.pg_type.c.oid == con_sq.c.contypid,
5875 )
5876 .where(pg_catalog.pg_type.c.typtype == "d")
5877 .order_by(
5878 pg_catalog.pg_namespace.c.nspname, pg_catalog.pg_type.c.typname
5879 )
5880 )
5881 return self._pg_type_filter_schema(query, schema)
5883 @reflection.cache
5884 def _load_domains(self, connection, schema=None, **kw):
5885 result = connection.execute(self._domain_query(schema))
5887 domains: List[ReflectedDomain] = []
5888 for domain in result.mappings():
5889 # strip (30) from character varying(30)
5890 attype = re.search(r"([^\(]+)", domain["attype"]).group(1)
5891 constraints: List[ReflectedDomainConstraint] = []
5892 if domain["connames"]:
5893 # When a domain has multiple CHECK constraints, they will
5894 # be tested in alphabetical order by name.
5895 sorted_constraints = sorted(
5896 zip(domain["connames"], domain["condefs"]),
5897 key=lambda t: t[0],
5898 )
5899 for name, def_ in sorted_constraints:
5900 # constraint is in the form "CHECK (expression)"
5901 # or "NOT NULL". Ignore the "NOT NULL" and
5902 # remove "CHECK (" and the tailing ")".
5903 if def_.casefold().startswith("check"):
5904 check = def_[7:-1]
5905 constraints.append({"name": name, "check": check})
5906 domain_rec: ReflectedDomain = {
5907 "name": domain["name"],
5908 "schema": domain["schema"],
5909 "visible": domain["visible"],
5910 "type": attype,
5911 "nullable": domain["nullable"],
5912 "default": domain["default"],
5913 "constraints": constraints,
5914 "collation": domain["collname"],
5915 "collation_schema": domain["collation_schema"],
5916 }
5917 domains.append(domain_rec)
5919 return domains
5921 @util.memoized_property
5922 def _pg_am_query(self):
5923 return sql.select(pg_catalog.pg_am.c.oid, pg_catalog.pg_am.c.amname)
5925 @reflection.cache
5926 def _load_pg_am_dict(self, connection, **kw) -> dict[int, str]:
5927 rows = connection.execute(self._pg_am_query)
5928 return dict(rows.all())
5930 def _load_pg_am_btree_oid(self, connection):
5931 # this oid is assumed to be stable
5932 if self._pg_am_btree_oid == -1:
5933 self._pg_am_btree_oid = connection.scalar(
5934 self._pg_am_query.where(pg_catalog.pg_am.c.amname == "btree")
5935 )
5936 return self._pg_am_btree_oid
5938 @util.memoized_property
5939 def _pg_opclass_notdefault_query(self):
5940 return sql.select(
5941 pg_catalog.pg_opclass.c.oid, pg_catalog.pg_opclass.c.opcname
5942 ).where(~pg_catalog.pg_opclass.c.opcdefault)
5944 @reflection.cache
5945 def _load_pg_opclass_notdefault_dict(
5946 self, connection, **kw
5947 ) -> dict[int, str]:
5948 rows = connection.execute(self._pg_opclass_notdefault_query)
5949 return dict(rows.all())
5951 def _set_backslash_escapes(self, connection):
5952 # this method is provided as an override hook for descendant
5953 # dialects (e.g. Redshift), so removing it may break them
5954 std_string = connection.exec_driver_sql(
5955 "show standard_conforming_strings"
5956 ).scalar()
5957 self._backslash_escapes = std_string == "off"
5960class _NamedTypeLoader:
5961 """Helper class used for deferred loading of named types (enums, domains)
5962 only when needed.
5963 """
5965 def __init__(
5966 self, dialect: PGDialect, connection, kw: Dict[str, Any]
5967 ) -> None:
5968 self.dialect = dialect
5969 self.connection = connection
5970 self.kw = kw
5972 @util.memoized_property
5973 def enums(self) -> Dict[Tuple[str] | Tuple[str, str], ReflectedEnum]:
5974 # dictionary with (name, ) if default search path or (schema, name)
5975 # as keys
5976 enums = dict(
5977 (
5978 ((rec["name"],), rec)
5979 if rec["visible"]
5980 else ((rec["schema"], rec["name"]), rec)
5981 )
5982 for rec in self.dialect._load_enums(
5983 self.connection,
5984 schema="*",
5985 info_cache=self.kw.get("info_cache"),
5986 )
5987 )
5988 return enums
5990 @util.memoized_property
5991 def domains(self) -> Dict[Tuple[str] | Tuple[str, str], ReflectedDomain]:
5992 # dictionary with (name, ) if default search path or (schema, name)
5993 # as keys
5994 domains = {
5995 ((d["schema"], d["name"]) if not d["visible"] else (d["name"],)): d
5996 for d in self.dialect._load_domains(
5997 self.connection,
5998 schema="*",
5999 info_cache=self.kw.get("info_cache"),
6000 )
6001 }
6002 return domains