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 return re.compile(
2787 r"^(?:RESTRICT|CASCADE|SET (?:NULL|DEFAULT)(?:\s*\(.+\))?"
2788 r"|NO ACTION)$",
2789 re.I,
2790 )
2792 def define_constraint_ondelete_cascade(self, constraint):
2793 return " ON DELETE %s" % self.preparer.validate_sql_phrase(
2794 constraint.ondelete, self._fk_ondelete_pattern
2795 )
2797 def visit_create_enum_type(self, create, **kw):
2798 type_ = create.element
2800 return "CREATE TYPE %s AS ENUM (%s)" % (
2801 self.preparer.format_type(type_),
2802 ", ".join(
2803 self.sql_compiler.process(sql.literal(e), literal_binds=True)
2804 for e in type_.enums
2805 ),
2806 )
2808 def visit_drop_enum_type(self, drop, **kw):
2809 type_ = drop.element
2811 return "DROP TYPE %s" % (self.preparer.format_type(type_))
2813 def visit_create_domain_type(self, create, **kw):
2814 domain: DOMAIN = create.element
2816 options = []
2817 if domain.collation is not None:
2818 collation = self.preparer.format_collation(
2819 domain.collation, domain.collation_schema
2820 )
2821 options.append(f"COLLATE {collation}")
2822 if domain.default is not None:
2823 default = self.render_default_string(domain.default)
2824 options.append(f"DEFAULT {default}")
2825 if domain.constraint_name is not None:
2826 name = self.preparer.truncate_and_render_constraint_name(
2827 domain.constraint_name
2828 )
2829 options.append(f"CONSTRAINT {name}")
2830 if domain.not_null:
2831 options.append("NOT NULL")
2832 if domain.check is not None:
2833 check = self.sql_compiler.process(
2834 domain.check, include_table=False, literal_binds=True
2835 )
2836 options.append(f"CHECK ({check})")
2838 return (
2839 f"CREATE DOMAIN {self.preparer.format_type(domain)} AS "
2840 f"{self.type_compiler.process(domain.data_type)} "
2841 f"{' '.join(options)}"
2842 )
2844 def visit_drop_domain_type(self, drop, **kw):
2845 domain = drop.element
2846 return f"DROP DOMAIN {self.preparer.format_type(domain)}"
2848 def _prepare_withclause_opts(self, withclause):
2849 with_opts = []
2850 for param, value in withclause.items():
2851 if value is None:
2852 with_opts.append(param)
2853 elif isinstance(value, bool):
2854 with_opts.append(
2855 "%s = %s" % (param, "true" if value else "false")
2856 )
2857 else:
2858 with_opts.append("%s = %s" % (param, value))
2859 return ", ".join(with_opts)
2861 def create_table_select_suffixes(self, element, type_, **kw):
2862 if type_ == "view":
2863 withclause = element.dialect_options["postgresql"]["with"]
2864 if withclause:
2865 return "WITH (%s)" % self._prepare_withclause_opts(withclause)
2866 return ""
2868 def visit_create_index(self, create, **kw):
2869 preparer = self.preparer
2870 index = create.element
2871 self._verify_index_table(index)
2872 text = "CREATE "
2873 if index.unique:
2874 text += "UNIQUE "
2876 text += "INDEX "
2878 if self.dialect._supports_create_index_concurrently:
2879 concurrently = index.dialect_options["postgresql"]["concurrently"]
2880 if concurrently:
2881 text += "CONCURRENTLY "
2883 if create.if_not_exists:
2884 text += "IF NOT EXISTS "
2886 text += "%s ON %s " % (
2887 self._prepared_index_name(index, include_schema=False),
2888 preparer.format_table(index.table),
2889 )
2891 using = index.dialect_options["postgresql"]["using"]
2892 if using:
2893 text += (
2894 "USING %s "
2895 % self.preparer.validate_sql_phrase(using, IDX_USING).lower()
2896 )
2898 ops = index.dialect_options["postgresql"]["ops"]
2899 text += "(%s)" % (
2900 ", ".join(
2901 [
2902 self.sql_compiler.process(
2903 (
2904 expr.self_group()
2905 if not isinstance(expr, expression.ColumnClause)
2906 else expr
2907 ),
2908 include_table=False,
2909 literal_binds=True,
2910 )
2911 + (
2912 (" " + ops[expr.key])
2913 if hasattr(expr, "key") and expr.key in ops
2914 else ""
2915 )
2916 for expr in index.expressions
2917 ]
2918 )
2919 )
2921 text += self._define_include(index)
2923 nulls_not_distinct = index.dialect_options["postgresql"][
2924 "nulls_not_distinct"
2925 ]
2926 if nulls_not_distinct is True:
2927 text += " NULLS NOT DISTINCT"
2928 elif nulls_not_distinct is False:
2929 text += " NULLS DISTINCT"
2931 withclause = index.dialect_options["postgresql"]["with"]
2932 if withclause:
2933 text += " WITH (%s)" % self._prepare_withclause_opts(withclause)
2935 tablespace_name = index.dialect_options["postgresql"]["tablespace"]
2936 if tablespace_name:
2937 text += " TABLESPACE %s" % preparer.quote(tablespace_name)
2939 whereclause = index.dialect_options["postgresql"]["where"]
2940 if whereclause is not None:
2941 whereclause = coercions.expect(
2942 roles.DDLExpressionRole, whereclause
2943 )
2945 where_compiled = self.sql_compiler.process(
2946 whereclause, include_table=False, literal_binds=True
2947 )
2948 text += " WHERE " + where_compiled
2950 return text
2952 def define_unique_constraint_distinct(self, constraint, **kw):
2953 nulls_not_distinct = constraint.dialect_options["postgresql"][
2954 "nulls_not_distinct"
2955 ]
2956 if nulls_not_distinct is True:
2957 nulls_not_distinct_param = "NULLS NOT DISTINCT "
2958 elif nulls_not_distinct is False:
2959 nulls_not_distinct_param = "NULLS DISTINCT "
2960 else:
2961 nulls_not_distinct_param = ""
2962 return nulls_not_distinct_param
2964 def visit_drop_index(self, drop, **kw):
2965 index = drop.element
2967 text = "\nDROP INDEX "
2969 if self.dialect._supports_drop_index_concurrently:
2970 concurrently = index.dialect_options["postgresql"]["concurrently"]
2971 if concurrently:
2972 text += "CONCURRENTLY "
2974 if drop.if_exists:
2975 text += "IF EXISTS "
2977 text += self._prepared_index_name(index, include_schema=True)
2978 return text
2980 def visit_exclude_constraint(self, constraint, **kw):
2981 text = ""
2982 if constraint.name is not None:
2983 text += "CONSTRAINT %s " % self.preparer.format_constraint(
2984 constraint
2985 )
2986 elements = []
2987 kw["include_table"] = False
2988 kw["literal_binds"] = True
2989 for expr, name, op in constraint._render_exprs:
2990 exclude_element = self.sql_compiler.process(expr, **kw) + (
2991 (" " + constraint.ops[expr.key])
2992 if hasattr(expr, "key") and expr.key in constraint.ops
2993 else ""
2994 )
2996 elements.append("%s WITH %s" % (exclude_element, op))
2997 text += "EXCLUDE USING %s (%s)" % (
2998 self.preparer.validate_sql_phrase(
2999 constraint.using, IDX_USING
3000 ).lower(),
3001 ", ".join(elements),
3002 )
3003 if constraint.where is not None:
3004 text += " WHERE (%s)" % self.sql_compiler.process(
3005 constraint.where, literal_binds=True
3006 )
3007 text += self.define_constraint_deferrability(constraint)
3008 return text
3010 def post_create_table(self, table):
3011 table_opts = []
3012 pg_opts = table.dialect_options["postgresql"]
3014 inherits = pg_opts.get("inherits")
3015 if inherits is not None:
3016 if not isinstance(inherits, (list, tuple)):
3017 inherits = (inherits,)
3018 table_opts.append(
3019 "\n INHERITS ("
3020 + ", ".join(self.preparer.quote(name) for name in inherits)
3021 + ")"
3022 )
3024 if pg_opts["partition_by"]:
3025 table_opts.append("\n PARTITION BY %s" % pg_opts["partition_by"])
3027 if pg_opts["using"]:
3028 table_opts.append("\n USING %s" % pg_opts["using"])
3030 if pg_opts["with"]:
3031 table_opts.append(
3032 "\n WITH (%s)" % self._prepare_withclause_opts(pg_opts["with"])
3033 )
3035 if pg_opts["with_oids"] is True:
3036 table_opts.append("\n WITH OIDS")
3037 elif pg_opts["with_oids"] is False:
3038 table_opts.append("\n WITHOUT OIDS")
3040 if pg_opts["on_commit"]:
3041 on_commit_options = pg_opts["on_commit"].replace("_", " ").upper()
3042 table_opts.append("\n ON COMMIT %s" % on_commit_options)
3044 if pg_opts["tablespace"]:
3045 tablespace_name = pg_opts["tablespace"]
3046 table_opts.append(
3047 "\n TABLESPACE %s" % self.preparer.quote(tablespace_name)
3048 )
3050 return "".join(table_opts)
3052 def visit_computed_column(self, generated, **kw):
3053 if self.dialect.supports_virtual_generated_columns:
3054 return super().visit_computed_column(generated, **kw)
3055 if generated.persisted is False:
3056 raise exc.CompileError(
3057 "PostrgreSQL computed columns do not support 'virtual' "
3058 "persistence; set the 'persisted' flag to None or True for "
3059 "PostgreSQL support."
3060 )
3061 elif generated.persisted is None:
3062 util.warn(
3063 f"Computed column {generated.column} is being created as "
3064 "'STORED' since the current PostgreSQL version does not "
3065 "support VIRTUAL columns. On PostgreSQL 18+, when "
3066 "'persisted' is not "
3067 "specified, no keyword will be rendered and VIRTUAL will be "
3068 "used by default. Set 'persisted=True' to ensure STORED "
3069 "behavior across all PostgreSQL versions."
3070 )
3072 return "GENERATED ALWAYS AS (%s) STORED" % self.sql_compiler.process(
3073 generated.sqltext, include_table=False, literal_binds=True
3074 )
3076 def visit_create_sequence(self, create, **kw):
3077 prefix = None
3078 if create.element.data_type is not None:
3079 prefix = " AS %s" % self.type_compiler.process(
3080 create.element.data_type
3081 )
3083 return super().visit_create_sequence(create, prefix=prefix, **kw)
3085 def _can_comment_on_constraint(self, ddl_instance):
3086 constraint = ddl_instance.element
3087 if constraint.name is None:
3088 raise exc.CompileError(
3089 f"Can't emit COMMENT ON for constraint {constraint!r}: "
3090 "it has no name"
3091 )
3092 if constraint.table is None:
3093 raise exc.CompileError(
3094 f"Can't emit COMMENT ON for constraint {constraint!r}: "
3095 "it has no associated table"
3096 )
3098 def visit_set_constraint_comment(self, create, **kw):
3099 self._can_comment_on_constraint(create)
3100 return "COMMENT ON CONSTRAINT %s ON %s IS %s" % (
3101 self.preparer.format_constraint(create.element),
3102 self.preparer.format_table(create.element.table),
3103 self.sql_compiler.render_literal_value(
3104 create.element.comment, sqltypes.String()
3105 ),
3106 )
3108 def visit_drop_constraint_comment(self, drop, **kw):
3109 self._can_comment_on_constraint(drop)
3110 return "COMMENT ON CONSTRAINT %s ON %s IS NULL" % (
3111 self.preparer.format_constraint(drop.element),
3112 self.preparer.format_table(drop.element.table),
3113 )
3116class PGTypeCompiler(compiler.GenericTypeCompiler):
3117 def visit_TSVECTOR(self, type_, **kw):
3118 return "TSVECTOR"
3120 def visit_TSQUERY(self, type_, **kw):
3121 return "TSQUERY"
3123 def visit_INET(self, type_, **kw):
3124 return "INET"
3126 def visit_CIDR(self, type_, **kw):
3127 return "CIDR"
3129 def visit_CITEXT(self, type_, **kw):
3130 return "CITEXT"
3132 def visit_MACADDR(self, type_, **kw):
3133 return "MACADDR"
3135 def visit_MACADDR8(self, type_, **kw):
3136 return "MACADDR8"
3138 def visit_MONEY(self, type_, **kw):
3139 return "MONEY"
3141 def visit_OID(self, type_, **kw):
3142 return "OID"
3144 def visit_REGCONFIG(self, type_, **kw):
3145 return "REGCONFIG"
3147 def visit_REGCLASS(self, type_, **kw):
3148 return "REGCLASS"
3150 def visit_FLOAT(self, type_, **kw):
3151 if not type_.precision:
3152 return "FLOAT"
3153 else:
3154 return "FLOAT(%(precision)s)" % {"precision": type_.precision}
3156 def visit_double(self, type_, **kw):
3157 return self.visit_DOUBLE_PRECISION(type, **kw)
3159 def visit_BIGINT(self, type_, **kw):
3160 return "BIGINT"
3162 def visit_HSTORE(self, type_, **kw):
3163 return "HSTORE"
3165 def visit_JSON(self, type_, **kw):
3166 return "JSON"
3168 def visit_JSONB(self, type_, **kw):
3169 return "JSONB"
3171 def visit_INT4MULTIRANGE(self, type_, **kw):
3172 return "INT4MULTIRANGE"
3174 def visit_INT8MULTIRANGE(self, type_, **kw):
3175 return "INT8MULTIRANGE"
3177 def visit_NUMMULTIRANGE(self, type_, **kw):
3178 return "NUMMULTIRANGE"
3180 def visit_DATEMULTIRANGE(self, type_, **kw):
3181 return "DATEMULTIRANGE"
3183 def visit_TSMULTIRANGE(self, type_, **kw):
3184 return "TSMULTIRANGE"
3186 def visit_TSTZMULTIRANGE(self, type_, **kw):
3187 return "TSTZMULTIRANGE"
3189 def visit_INT4RANGE(self, type_, **kw):
3190 return "INT4RANGE"
3192 def visit_INT8RANGE(self, type_, **kw):
3193 return "INT8RANGE"
3195 def visit_NUMRANGE(self, type_, **kw):
3196 return "NUMRANGE"
3198 def visit_DATERANGE(self, type_, **kw):
3199 return "DATERANGE"
3201 def visit_TSRANGE(self, type_, **kw):
3202 return "TSRANGE"
3204 def visit_TSTZRANGE(self, type_, **kw):
3205 return "TSTZRANGE"
3207 def visit_json_int_index(self, type_, **kw):
3208 return "INT"
3210 def visit_json_str_index(self, type_, **kw):
3211 return "TEXT"
3213 def visit_datetime(self, type_, **kw):
3214 return self.visit_TIMESTAMP(type_, **kw)
3216 def visit_enum(self, type_, **kw):
3217 if not type_.native_enum or not self.dialect.supports_native_enum:
3218 return super().visit_enum(type_, **kw)
3219 else:
3220 return self.visit_ENUM(type_, **kw)
3222 def visit_ENUM(self, type_, identifier_preparer=None, **kw):
3223 if identifier_preparer is None:
3224 identifier_preparer = self.dialect.identifier_preparer
3225 return identifier_preparer.format_type(type_)
3227 def visit_DOMAIN(self, type_, identifier_preparer=None, **kw):
3228 if identifier_preparer is None:
3229 identifier_preparer = self.dialect.identifier_preparer
3230 return identifier_preparer.format_type(type_)
3232 def visit_TIMESTAMP(self, type_, **kw):
3233 return "TIMESTAMP%s %s" % (
3234 (
3235 "(%d)" % type_.precision
3236 if getattr(type_, "precision", None) is not None
3237 else ""
3238 ),
3239 (type_.timezone and "WITH" or "WITHOUT") + " TIME ZONE",
3240 )
3242 def visit_TIME(self, type_, **kw):
3243 return "TIME%s %s" % (
3244 (
3245 "(%d)" % type_.precision
3246 if getattr(type_, "precision", None) is not None
3247 else ""
3248 ),
3249 (type_.timezone and "WITH" or "WITHOUT") + " TIME ZONE",
3250 )
3252 def visit_INTERVAL(self, type_, **kw):
3253 text = "INTERVAL"
3254 if type_.fields is not None:
3255 text += " " + type_.fields
3256 if type_.precision is not None:
3257 text += " (%d)" % type_.precision
3258 return text
3260 def visit_BIT(self, type_, **kw):
3261 if type_.varying:
3262 compiled = "BIT VARYING"
3263 if type_.length is not None:
3264 compiled += "(%d)" % type_.length
3265 else:
3266 compiled = "BIT(%d)" % type_.length
3267 return compiled
3269 def visit_uuid(self, type_, **kw):
3270 if type_.native_uuid:
3271 return self.visit_UUID(type_, **kw)
3272 else:
3273 return super().visit_uuid(type_, **kw)
3275 def visit_UUID(self, type_, **kw):
3276 return "UUID"
3278 def visit_large_binary(self, type_, **kw):
3279 return self.visit_BYTEA(type_, **kw)
3281 def visit_BYTEA(self, type_, **kw):
3282 return "BYTEA"
3284 def visit_ARRAY(self, type_, **kw):
3285 inner = self.process(type_.item_type, **kw)
3286 return re.sub(
3287 r"((?: COLLATE.*)?)$",
3288 (
3289 r"%s\1"
3290 % (
3291 "[]"
3292 * (type_.dimensions if type_.dimensions is not None else 1)
3293 )
3294 ),
3295 inner,
3296 count=1,
3297 )
3299 def visit_json_path(self, type_, **kw):
3300 return self.visit_JSONPATH(type_, **kw)
3302 def visit_JSONPATH(self, type_, **kw):
3303 return "JSONPATH"
3306class _CompilerSequence:
3307 """Minimal stand-in for :class:`.Sequence`.
3309 Used for the implicit sequence behind a SERIAL column, where no
3310 :class:`.Sequence` object exists but a name still has to be rendered
3311 through :meth:`.IdentifierPreparer.format_sequence`.
3313 """
3315 __slots__ = ("name", "schema")
3317 # the schema handed to us has already been resolved against the
3318 # schema translate map, so don't let format_sequence() translate it
3319 # a second time
3320 _use_schema_map = False
3322 def __init__(self, name, schema=None):
3323 self.name = name
3324 self.schema = schema
3327class PGIdentifierPreparer(compiler.IdentifierPreparer):
3328 reserved_words = RESERVED_WORDS
3330 def _format_sequence_string_literal(self, sequence):
3331 """Render a sequence name as a SQL string literal.
3333 ``nextval()`` and friends take the sequence as a string rather than
3334 as an identifier, so the quoted identifier produced by
3335 :meth:`.format_sequence` has to be escaped a second time for the
3336 enclosing literal.
3338 """
3339 return "'%s'" % self.format_sequence(
3340 sequence, use_schema=True
3341 ).replace("'", "''")
3343 def _unquote_identifier(self, value):
3344 if value[0] == self.initial_quote:
3345 value = value[1:-1].replace(
3346 self.escape_to_quote, self.escape_quote
3347 )
3348 return value
3350 def format_type(self, type_, use_schema=True):
3351 if not type_.name:
3352 raise exc.CompileError(
3353 f"PostgreSQL {type_.__class__.__name__} type requires a name."
3354 )
3356 name = self.quote(type_.name)
3357 effective_schema = self.schema_for_object(type_)
3359 # a built-in type with the same name will obscure this type, so raise
3360 # for that case. this applies really to any visible type with the same
3361 # name in any other visible schema that would not be appropriate for
3362 # us to check against, so this is not a robust check, but
3363 # at least do something for an obvious built-in name conflict
3364 if (
3365 effective_schema is None
3366 and type_.name in self.dialect.ischema_names
3367 ):
3368 raise exc.CompileError(
3369 f"{type_!r} has name "
3370 f"'{type_.name}' that matches an existing type, and "
3371 "requires an explicit schema name in order to be rendered "
3372 "in DDL."
3373 )
3375 if (
3376 not self.omit_schema
3377 and use_schema
3378 and effective_schema is not None
3379 ):
3380 name = f"{self.quote_schema(effective_schema)}.{name}"
3381 return name
3384class ReflectedNamedType(TypedDict):
3385 """Represents a reflected named type."""
3387 name: str
3388 """Name of the type."""
3389 schema: str
3390 """The schema of the type."""
3391 visible: bool
3392 """Indicates if this type is in the current search path."""
3395class ReflectedDomainConstraint(TypedDict):
3396 """Represents a reflect check constraint of a domain."""
3398 name: str
3399 """Name of the constraint."""
3400 check: str
3401 """The check constraint text."""
3404class ReflectedDomain(ReflectedNamedType):
3405 """Represents a reflected enum."""
3407 type: str
3408 """The string name of the underlying data type of the domain."""
3409 nullable: bool
3410 """Indicates if the domain allows null or not."""
3411 default: Optional[str]
3412 """The string representation of the default value of this domain
3413 or ``None`` if none present.
3414 """
3415 constraints: List[ReflectedDomainConstraint]
3416 """The constraints defined in the domain, if any.
3417 The constraint are in order of evaluation by postgresql.
3418 """
3419 collation: Optional[str]
3420 """The collation for the domain."""
3421 collation_schema: Optional[str]
3422 """The schema of the collation for the domain, if the collation is not
3423 visible on the current search_path."""
3426class ReflectedEnum(ReflectedNamedType):
3427 """Represents a reflected enum."""
3429 labels: List[str]
3430 """The labels that compose the enum."""
3433class PGInspector(reflection.Inspector):
3434 dialect: PGDialect
3436 def get_table_oid(
3437 self, table_name: str, schema: Optional[str] = None
3438 ) -> int:
3439 """Return the OID for the given table name.
3441 :param table_name: string name of the table. For special quoting,
3442 use :class:`.quoted_name`.
3444 :param schema: string schema name; if omitted, uses the default schema
3445 of the database connection. For special quoting,
3446 use :class:`.quoted_name`.
3448 """
3450 with self._operation_context() as conn:
3451 return self.dialect.get_table_oid(
3452 conn, table_name, schema, info_cache=self.info_cache
3453 )
3455 def get_domains(
3456 self, schema: Optional[str] = None
3457 ) -> List[ReflectedDomain]:
3458 """Return a list of DOMAIN objects.
3460 Each member is a dictionary containing these fields:
3462 * name - name of the domain
3463 * schema - the schema name for the domain.
3464 * visible - boolean, whether or not this domain is visible
3465 in the default search path.
3466 * type - the type defined by this domain.
3467 * nullable - Indicates if this domain can be ``NULL``.
3468 * default - The default value of the domain or ``None`` if the
3469 domain has no default.
3470 * constraints - A list of dict with the constraint defined by this
3471 domain. Each element contains two keys: ``name`` of the
3472 constraint and ``check`` with the constraint text.
3474 :param schema: schema name. If None, the default schema
3475 (typically 'public') is used. May also be set to ``'*'`` to
3476 indicate load domains for all schemas.
3478 .. versionadded:: 2.0
3480 """
3481 with self._operation_context() as conn:
3482 return self.dialect._load_domains(
3483 conn, schema, info_cache=self.info_cache
3484 )
3486 def get_enums(self, schema: Optional[str] = None) -> List[ReflectedEnum]:
3487 """Return a list of ENUM objects.
3489 Each member is a dictionary containing these fields:
3491 * name - name of the enum
3492 * schema - the schema name for the enum.
3493 * visible - boolean, whether or not this enum is visible
3494 in the default search path.
3495 * labels - a list of string labels that apply to the enum.
3497 :param schema: schema name. If None, the default schema
3498 (typically 'public') is used. May also be set to ``'*'`` to
3499 indicate load enums for all schemas.
3501 """
3502 with self._operation_context() as conn:
3503 return self.dialect._load_enums(
3504 conn, schema, info_cache=self.info_cache
3505 )
3507 def get_foreign_table_names(
3508 self, schema: Optional[str] = None
3509 ) -> List[str]:
3510 """Return a list of FOREIGN TABLE names.
3512 Behavior is similar to that of
3513 :meth:`_reflection.Inspector.get_table_names`,
3514 except that the list is limited to those tables that report a
3515 ``relkind`` value of ``f``.
3517 """
3518 with self._operation_context() as conn:
3519 return self.dialect._get_foreign_table_names(
3520 conn, schema, info_cache=self.info_cache
3521 )
3523 def has_type(
3524 self, type_name: str, schema: Optional[str] = None, **kw: Any
3525 ) -> bool:
3526 """Return if the database has the specified type in the provided
3527 schema.
3529 :param type_name: the type to check.
3530 :param schema: schema name. If None, the default schema
3531 (typically 'public') is used. May also be set to ``'*'`` to
3532 check in all schemas.
3534 .. versionadded:: 2.0
3536 """
3537 with self._operation_context() as conn:
3538 return self.dialect.has_type(
3539 conn, type_name, schema, info_cache=self.info_cache
3540 )
3543class PGExecutionContext(default.DefaultExecutionContext):
3544 def fire_sequence(self, seq, type_):
3545 return self._execute_scalar(
3546 (
3547 "select nextval(%s)"
3548 % self.identifier_preparer._format_sequence_string_literal(seq)
3549 ),
3550 type_,
3551 )
3553 def get_insert_default(self, column):
3554 if column.primary_key and column is column.table._autoincrement_column:
3555 if column.server_default and column.server_default.has_argument:
3556 # pre-execute passive defaults on primary key columns
3557 return self._execute_scalar(
3558 "select %s" % column.server_default.arg, column.type
3559 )
3561 elif column.default is None or (
3562 column.default.is_sequence and column.default.optional
3563 ):
3564 # execute the sequence associated with a SERIAL primary
3565 # key column. for non-primary-key SERIAL, the ID just
3566 # generates server side.
3568 try:
3569 seq_name = column._postgresql_seq_name
3570 except AttributeError:
3571 tab = column.table.name
3572 col = column.name
3573 tab = tab[0 : 29 + max(0, (29 - len(col)))]
3574 col = col[0 : 29 + max(0, (29 - len(tab)))]
3575 name = "%s_%s_seq" % (tab, col)
3576 column._postgresql_seq_name = seq_name = name
3578 if column.table is not None:
3579 effective_schema = self.connection.schema_for_object(
3580 column.table
3581 )
3582 else:
3583 effective_schema = None
3585 exc = (
3586 "select nextval(%s)"
3587 % self.identifier_preparer._format_sequence_string_literal(
3588 _CompilerSequence(seq_name, effective_schema)
3589 )
3590 )
3592 return self._execute_scalar(exc, column.type)
3594 return super().get_insert_default(column)
3597class PGReadOnlyConnectionCharacteristic(
3598 characteristics.ConnectionCharacteristic
3599):
3600 transactional = True
3602 def reset_characteristic(self, dialect, dbapi_conn):
3603 dialect.set_readonly(dbapi_conn, False)
3605 def set_characteristic(self, dialect, dbapi_conn, value):
3606 dialect.set_readonly(dbapi_conn, value)
3608 def get_characteristic(self, dialect, dbapi_conn):
3609 return dialect.get_readonly(dbapi_conn)
3612class PGDeferrableConnectionCharacteristic(
3613 characteristics.ConnectionCharacteristic
3614):
3615 transactional = True
3617 def reset_characteristic(self, dialect, dbapi_conn):
3618 dialect.set_deferrable(dbapi_conn, False)
3620 def set_characteristic(self, dialect, dbapi_conn, value):
3621 dialect.set_deferrable(dbapi_conn, value)
3623 def get_characteristic(self, dialect, dbapi_conn):
3624 return dialect.get_deferrable(dbapi_conn)
3627class PGDialect(default._BackendsMultiReflection, default.DefaultDialect):
3628 name = "postgresql"
3629 supports_statement_cache = True
3630 supports_alter = True
3631 max_identifier_length = 63
3632 supports_sane_rowcount = True
3634 bind_typing = interfaces.BindTyping.RENDER_CASTS
3636 supports_native_enum = True
3637 supports_native_boolean = True
3638 supports_native_uuid = True
3639 supports_smallserial = True
3640 supports_virtual_generated_columns = True
3642 supports_sequences = True
3643 sequences_optional = True
3644 preexecute_autoincrement_sequences = True
3645 postfetch_lastrowid = False
3646 use_insertmanyvalues = True
3648 returns_native_bytes = True
3650 insertmanyvalues_implicit_sentinel = (
3651 InsertmanyvaluesSentinelOpts.ANY_AUTOINCREMENT
3652 | InsertmanyvaluesSentinelOpts.USE_INSERT_FROM_SELECT
3653 | InsertmanyvaluesSentinelOpts.RENDER_SELECT_COL_CASTS
3654 )
3656 supports_comments = True
3657 supports_constraint_comments = True
3658 supports_default_values = True
3660 supports_default_metavalue = True
3662 supports_empty_insert = False
3663 supports_multivalues_insert = True
3665 supports_identity_columns = True
3667 default_paramstyle = "pyformat"
3668 ischema_names = ischema_names
3669 colspecs = colspecs
3671 statement_compiler = PGCompiler
3672 ddl_compiler = PGDDLCompiler
3673 type_compiler_cls = PGTypeCompiler
3674 preparer = PGIdentifierPreparer
3675 execution_ctx_cls = PGExecutionContext
3676 inspector = PGInspector
3678 update_returning = True
3679 delete_returning = True
3680 insert_returning = True
3681 update_returning_multifrom = True
3682 delete_returning_multifrom = True
3684 connection_characteristics = (
3685 default.DefaultDialect.connection_characteristics
3686 )
3687 connection_characteristics = connection_characteristics.union(
3688 {
3689 "postgresql_readonly": PGReadOnlyConnectionCharacteristic(),
3690 "postgresql_deferrable": PGDeferrableConnectionCharacteristic(),
3691 }
3692 )
3694 construct_arguments = [
3695 (
3696 schema.Index,
3697 {
3698 "using": False,
3699 "include": None,
3700 "where": None,
3701 "ops": {},
3702 "concurrently": False,
3703 "with": {},
3704 "tablespace": None,
3705 "nulls_not_distinct": None,
3706 "invalid": DialectKWArgConst.REFLECTED_ONLY,
3707 },
3708 ),
3709 (
3710 schema.Table,
3711 {
3712 "ignore_search_path": False,
3713 "tablespace": None,
3714 "partition_by": None,
3715 "with_oids": None,
3716 "with": None,
3717 "on_commit": None,
3718 "inherits": None,
3719 "using": None,
3720 },
3721 ),
3722 (
3723 schema.CheckConstraint,
3724 {
3725 "not_valid": False,
3726 },
3727 ),
3728 (
3729 schema.ForeignKeyConstraint,
3730 {
3731 "not_valid": False,
3732 },
3733 ),
3734 (
3735 schema.PrimaryKeyConstraint,
3736 {"include": None},
3737 ),
3738 (
3739 schema.UniqueConstraint,
3740 {
3741 "include": None,
3742 "nulls_not_distinct": None,
3743 },
3744 ),
3745 (
3746 schema.CreateView,
3747 {
3748 "with": None,
3749 },
3750 ),
3751 ]
3753 reflection_options = ("postgresql_ignore_search_path",)
3755 _backslash_escapes = False
3756 _supports_create_index_concurrently = True
3757 _supports_drop_index_concurrently = True
3758 _supports_jsonb_subscripting = True
3759 _pg_am_btree_oid = -1
3761 def __init__(
3762 self,
3763 native_inet_types=None,
3764 json_serializer=None,
3765 json_deserializer=None,
3766 **kwargs,
3767 ):
3768 default.DefaultDialect.__init__(self, **kwargs)
3770 self._native_inet_types = native_inet_types
3771 self._json_deserializer = json_deserializer
3772 self._json_serializer = json_serializer
3774 def initialize(self, connection):
3775 super().initialize(connection)
3777 # https://www.postgresql.org/docs/9.3/static/release-9-2.html#AEN116689
3778 self.supports_smallserial = self.server_version_info >= (9, 2)
3780 self._set_backslash_escapes(connection)
3782 self._supports_drop_index_concurrently = self.server_version_info >= (
3783 9,
3784 2,
3785 )
3786 self.supports_identity_columns = self.server_version_info >= (10,)
3788 self._supports_jsonb_subscripting = self.server_version_info >= (14,)
3790 self.supports_virtual_generated_columns = self.server_version_info >= (
3791 18,
3792 )
3794 def get_isolation_level_values(self, dbapi_conn):
3795 # note the generic dialect doesn't have AUTOCOMMIT, however
3796 # all postgresql dialects should include AUTOCOMMIT.
3797 return (
3798 "SERIALIZABLE",
3799 "READ UNCOMMITTED",
3800 "READ COMMITTED",
3801 "REPEATABLE READ",
3802 )
3804 def set_isolation_level(self, dbapi_connection, level):
3805 cursor = dbapi_connection.cursor()
3806 cursor.execute(
3807 "SET SESSION CHARACTERISTICS AS TRANSACTION "
3808 f"ISOLATION LEVEL {level}"
3809 )
3810 cursor.execute("COMMIT")
3811 cursor.close()
3813 def get_isolation_level(self, dbapi_connection):
3814 cursor = dbapi_connection.cursor()
3815 cursor.execute("show transaction isolation level")
3816 val = cursor.fetchone()[0]
3817 cursor.close()
3818 return val.upper()
3820 def set_readonly(self, connection, value):
3821 raise NotImplementedError()
3823 def get_readonly(self, connection):
3824 raise NotImplementedError()
3826 def set_deferrable(self, connection, value):
3827 raise NotImplementedError()
3829 def get_deferrable(self, connection):
3830 raise NotImplementedError()
3832 def _split_multihost_from_url(self, url: URL) -> Union[
3833 Tuple[None, None],
3834 Tuple[Tuple[Optional[str], ...], Tuple[Optional[int], ...]],
3835 ]:
3836 hosts: Optional[Tuple[Optional[str], ...]] = None
3837 ports_str: Union[str, Tuple[Optional[str], ...], None] = None
3839 integrated_multihost = False
3841 if "host" in url.query:
3842 if isinstance(url.query["host"], (list, tuple)):
3843 integrated_multihost = True
3844 hosts, ports_str = zip(
3845 *[
3846 token.split(":") if ":" in token else (token, None)
3847 for token in url.query["host"]
3848 ]
3849 )
3851 elif isinstance(url.query["host"], str):
3852 hosts = tuple(url.query["host"].split(","))
3854 if (
3855 "port" not in url.query
3856 and len(hosts) == 1
3857 and ":" in hosts[0]
3858 ):
3859 # internet host is alphanumeric plus dots or hyphens.
3860 # this is essentially rfc1123, which refers to rfc952.
3861 # https://stackoverflow.com/questions/3523028/
3862 # valid-characters-of-a-hostname
3863 host_port_match = re.match(
3864 r"^([a-zA-Z0-9\-\.]*)(?:\:(\d*))?$", hosts[0]
3865 )
3866 if host_port_match:
3867 integrated_multihost = True
3868 h, p = host_port_match.group(1, 2)
3869 if TYPE_CHECKING:
3870 assert isinstance(h, str)
3871 assert isinstance(p, str)
3872 hosts = (h,)
3873 ports_str = cast(
3874 "Tuple[Optional[str], ...]", (p,) if p else (None,)
3875 )
3877 if "port" in url.query:
3878 if integrated_multihost:
3879 raise exc.ArgumentError(
3880 "Can't mix 'multihost' formats together; use "
3881 '"host=h1,h2,h3&port=p1,p2,p3" or '
3882 '"host=h1:p1&host=h2:p2&host=h3:p3" separately'
3883 )
3884 if isinstance(url.query["port"], (list, tuple)):
3885 ports_str = url.query["port"]
3886 elif isinstance(url.query["port"], str):
3887 ports_str = tuple(url.query["port"].split(","))
3889 ports: Optional[Tuple[Optional[int], ...]] = None
3891 if ports_str:
3892 try:
3893 ports = tuple(int(x) if x else None for x in ports_str)
3894 except ValueError:
3895 raise exc.ArgumentError(
3896 f"Received non-integer port arguments: {ports_str}"
3897 ) from None
3899 if ports and (
3900 (not hosts and len(ports) > 1)
3901 or (
3902 hosts
3903 and ports
3904 and len(hosts) != len(ports)
3905 and (len(hosts) > 1 or len(ports) > 1)
3906 )
3907 ):
3908 raise exc.ArgumentError("number of hosts and ports don't match")
3910 if hosts is not None:
3911 if ports is None:
3912 ports = tuple(None for _ in hosts)
3914 return hosts, ports # type: ignore
3916 def do_begin_twophase(self, connection, xid):
3917 self.do_begin(connection.connection)
3919 def do_prepare_twophase(self, connection, xid):
3920 connection.execute(
3921 sql.text("PREPARE TRANSACTION :xid").bindparams(
3922 sql.bindparam("xid", xid, literal_execute=True)
3923 )
3924 )
3926 def do_rollback_twophase(
3927 self, connection, xid, is_prepared=True, recover=False
3928 ):
3929 if is_prepared:
3930 if recover:
3931 # FIXME: ugly hack to get out of transaction
3932 # context when committing recoverable transactions
3933 # Must find out a way how to make the dbapi not
3934 # open a transaction.
3935 connection.exec_driver_sql("ROLLBACK")
3936 connection.execute(
3937 sql.text("ROLLBACK PREPARED :xid").bindparams(
3938 sql.bindparam("xid", xid, literal_execute=True)
3939 )
3940 )
3941 connection.exec_driver_sql("BEGIN")
3942 self.do_rollback(connection.connection)
3943 else:
3944 self.do_rollback(connection.connection)
3946 def do_commit_twophase(
3947 self, connection, xid, is_prepared=True, recover=False
3948 ):
3949 if is_prepared:
3950 if recover:
3951 connection.exec_driver_sql("ROLLBACK")
3952 connection.execute(
3953 sql.text("COMMIT PREPARED :xid").bindparams(
3954 sql.bindparam("xid", xid, literal_execute=True)
3955 )
3956 )
3957 connection.exec_driver_sql("BEGIN")
3958 self.do_rollback(connection.connection)
3959 else:
3960 self.do_commit(connection.connection)
3962 def do_recover_twophase(self, connection):
3963 return connection.scalars(
3964 sql.text("SELECT gid FROM pg_prepared_xacts")
3965 ).all()
3967 def _get_default_schema_name(self, connection):
3968 return connection.exec_driver_sql("select current_schema()").scalar()
3970 @reflection.cache
3971 def has_schema(self, connection, schema, **kw):
3972 query = select(pg_catalog.pg_namespace.c.nspname).where(
3973 pg_catalog.pg_namespace.c.nspname == schema
3974 )
3975 return bool(connection.scalar(query))
3977 def _pg_class_filter_scope_schema(
3978 self, query, schema, scope, pg_class_table=None
3979 ):
3980 if pg_class_table is None:
3981 pg_class_table = pg_catalog.pg_class
3982 query = query.join(
3983 pg_catalog.pg_namespace,
3984 pg_catalog.pg_namespace.c.oid == pg_class_table.c.relnamespace,
3985 )
3987 if scope is ObjectScope.DEFAULT:
3988 query = query.where(pg_class_table.c.relpersistence != "t")
3989 elif scope is ObjectScope.TEMPORARY:
3990 query = query.where(pg_class_table.c.relpersistence == "t")
3992 if schema is None:
3993 query = query.where(
3994 pg_catalog.pg_table_is_visible(pg_class_table.c.oid),
3995 # ignore pg_catalog schema
3996 pg_catalog.pg_namespace.c.nspname != "pg_catalog",
3997 )
3998 else:
3999 query = query.where(pg_catalog.pg_namespace.c.nspname == schema)
4000 return query
4002 def _pg_class_relkind_condition(self, relkinds, pg_class_table=None):
4003 if pg_class_table is None:
4004 pg_class_table = pg_catalog.pg_class
4005 # uses the any form instead of in otherwise postgresql complaings
4006 # that 'IN could not convert type character to "char"'
4007 return pg_class_table.c.relkind == sql.any_(_array.array(relkinds))
4009 @lru_cache()
4010 def _has_multi_table_query(self, schema):
4011 query = select(pg_catalog.pg_class.c.relname).where(
4012 pg_catalog.pg_class.c.relname.in_(
4013 bindparam("table_names", expanding=True)
4014 ),
4015 self._pg_class_relkind_condition(
4016 pg_catalog.RELKINDS_ALL_TABLE_LIKE
4017 ),
4018 )
4019 return self._pg_class_filter_scope_schema(
4020 query, schema, scope=ObjectScope.ANY
4021 )
4023 @reflection.cache
4024 def has_table(self, connection, table_name, schema=None, **kw):
4025 # NOTE: it's not worth calling into the multi table since the query
4026 # is compatible also with this single case.
4027 self._ensure_has_table_connection(connection)
4028 query = self._has_multi_table_query(schema)
4029 return bool(connection.scalar(query, {"table_names": [table_name]}))
4031 def has_multi_table(self, connection, table_names, schema=None, **kw):
4032 query = self._has_multi_table_query(schema)
4033 params = {"table_names": table_names}
4034 existing = set(connection.scalars(query, params).all())
4035 retval = {(schema, table): table in existing for table in table_names}
4036 return retval.items()
4038 @reflection.cache
4039 def has_sequence(self, connection, sequence_name, schema=None, **kw):
4040 query = select(pg_catalog.pg_class.c.relname).where(
4041 pg_catalog.pg_class.c.relkind == "S",
4042 pg_catalog.pg_class.c.relname == sequence_name,
4043 )
4044 query = self._pg_class_filter_scope_schema(
4045 query, schema, scope=ObjectScope.ANY
4046 )
4047 return bool(connection.scalar(query))
4049 @reflection.cache
4050 def has_type(self, connection, type_name, schema=None, **kw):
4051 query = (
4052 select(pg_catalog.pg_type.c.typname)
4053 .join(
4054 pg_catalog.pg_namespace,
4055 pg_catalog.pg_namespace.c.oid
4056 == pg_catalog.pg_type.c.typnamespace,
4057 )
4058 .where(pg_catalog.pg_type.c.typname == type_name)
4059 )
4060 if schema is None:
4061 query = query.where(
4062 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid),
4063 # ignore pg_catalog schema
4064 pg_catalog.pg_namespace.c.nspname != "pg_catalog",
4065 )
4066 elif schema != "*":
4067 query = query.where(pg_catalog.pg_namespace.c.nspname == schema)
4069 return bool(connection.scalar(query))
4071 def _get_server_version_info(self, connection):
4072 v = connection.exec_driver_sql("select pg_catalog.version()").scalar()
4073 m = re.match(
4074 r".*(?:PostgreSQL|EnterpriseDB) "
4075 r"(\d+)\.?(\d+)?(?:\.(\d+))?(?:\.\d+)?(?:devel|beta)?",
4076 v,
4077 )
4078 if not m:
4079 raise AssertionError(
4080 "Could not determine version from string '%s'" % v
4081 )
4082 return tuple([int(x) for x in m.group(1, 2, 3) if x is not None])
4084 @reflection.cache
4085 def get_table_oid(self, connection, table_name, schema=None, **kw):
4086 """Fetch the oid for schema.table_name."""
4087 query = select(pg_catalog.pg_class.c.oid).where(
4088 pg_catalog.pg_class.c.relname == table_name,
4089 self._pg_class_relkind_condition(
4090 pg_catalog.RELKINDS_ALL_TABLE_LIKE
4091 ),
4092 )
4093 query = self._pg_class_filter_scope_schema(
4094 query, schema, scope=ObjectScope.ANY
4095 )
4096 table_oid = connection.scalar(query)
4097 if table_oid is None:
4098 raise exc.NoSuchTableError(
4099 f"{schema}.{table_name}" if schema else table_name
4100 )
4101 return table_oid
4103 @reflection.cache
4104 def get_schema_names(self, connection, **kw):
4105 query = (
4106 select(pg_catalog.pg_namespace.c.nspname)
4107 .where(
4108 ~pg_catalog.pg_namespace.c.nspname.startswith(
4109 "pg_", autoescape=True
4110 )
4111 )
4112 .order_by(pg_catalog.pg_namespace.c.nspname)
4113 )
4114 return connection.scalars(query).all()
4116 def _get_relnames_for_relkinds(self, connection, schema, relkinds, scope):
4117 query = select(pg_catalog.pg_class.c.relname).where(
4118 self._pg_class_relkind_condition(relkinds)
4119 )
4120 query = self._pg_class_filter_scope_schema(query, schema, scope=scope)
4121 return connection.scalars(query).all()
4123 @reflection.cache
4124 def get_table_names(self, connection, schema=None, **kw):
4125 return self._get_relnames_for_relkinds(
4126 connection,
4127 schema,
4128 pg_catalog.RELKINDS_TABLE_NO_FOREIGN,
4129 scope=ObjectScope.DEFAULT,
4130 )
4132 @reflection.cache
4133 def get_temp_table_names(self, connection, **kw):
4134 return self._get_relnames_for_relkinds(
4135 connection,
4136 schema=None,
4137 relkinds=pg_catalog.RELKINDS_TABLE_NO_FOREIGN,
4138 scope=ObjectScope.TEMPORARY,
4139 )
4141 @reflection.cache
4142 def _get_foreign_table_names(self, connection, schema=None, **kw):
4143 return self._get_relnames_for_relkinds(
4144 connection, schema, relkinds=("f",), scope=ObjectScope.ANY
4145 )
4147 @reflection.cache
4148 def get_view_names(self, connection, schema=None, **kw):
4149 return self._get_relnames_for_relkinds(
4150 connection,
4151 schema,
4152 pg_catalog.RELKINDS_VIEW,
4153 scope=ObjectScope.DEFAULT,
4154 )
4156 @reflection.cache
4157 def get_materialized_view_names(self, connection, schema=None, **kw):
4158 return self._get_relnames_for_relkinds(
4159 connection,
4160 schema,
4161 pg_catalog.RELKINDS_MAT_VIEW,
4162 scope=ObjectScope.DEFAULT,
4163 )
4165 @reflection.cache
4166 def get_temp_view_names(self, connection, schema=None, **kw):
4167 return self._get_relnames_for_relkinds(
4168 connection,
4169 schema,
4170 # NOTE: do not include temp materialzied views (that do not
4171 # seem to be a thing at least up to version 14)
4172 pg_catalog.RELKINDS_VIEW,
4173 scope=ObjectScope.TEMPORARY,
4174 )
4176 @reflection.cache
4177 def get_sequence_names(self, connection, schema=None, **kw):
4178 return self._get_relnames_for_relkinds(
4179 connection, schema, relkinds=("S",), scope=ObjectScope.ANY
4180 )
4182 @reflection.cache
4183 def get_view_definition(self, connection, view_name, schema=None, **kw):
4184 query = (
4185 select(pg_catalog.pg_get_viewdef(pg_catalog.pg_class.c.oid))
4186 .select_from(pg_catalog.pg_class)
4187 .where(
4188 pg_catalog.pg_class.c.relname == view_name,
4189 self._pg_class_relkind_condition(
4190 pg_catalog.RELKINDS_VIEW + pg_catalog.RELKINDS_MAT_VIEW
4191 ),
4192 )
4193 )
4194 query = self._pg_class_filter_scope_schema(
4195 query, schema, scope=ObjectScope.ANY
4196 )
4197 res = connection.scalar(query)
4198 if res is None:
4199 raise exc.NoSuchTableError(
4200 f"{schema}.{view_name}" if schema else view_name
4201 )
4202 else:
4203 return res
4205 def _prepare_filter_names(self, filter_names):
4206 if filter_names:
4207 return True, {"filter_names": filter_names}
4208 else:
4209 return False, {}
4211 def _kind_to_relkinds(self, kind: ObjectKind) -> Tuple[str, ...]:
4212 if kind is ObjectKind.ANY:
4213 return pg_catalog.RELKINDS_ALL_TABLE_LIKE
4214 relkinds = ()
4215 if ObjectKind.TABLE in kind:
4216 relkinds += pg_catalog.RELKINDS_TABLE
4217 if ObjectKind.VIEW in kind:
4218 relkinds += pg_catalog.RELKINDS_VIEW
4219 if ObjectKind.MATERIALIZED_VIEW in kind:
4220 relkinds += pg_catalog.RELKINDS_MAT_VIEW
4221 return relkinds
4223 @lru_cache()
4224 def _table_options_query(self, schema, has_filter_names, scope, kind):
4225 inherits_sq = (
4226 select(
4227 pg_catalog.pg_inherits.c.inhrelid,
4228 sql.func.array_agg(
4229 aggregate_order_by(
4230 pg_catalog.pg_class.c.relname,
4231 pg_catalog.pg_inherits.c.inhseqno,
4232 )
4233 ).label("parent_table_names"),
4234 )
4235 .select_from(pg_catalog.pg_inherits)
4236 .join(
4237 pg_catalog.pg_class,
4238 pg_catalog.pg_inherits.c.inhparent
4239 == pg_catalog.pg_class.c.oid,
4240 )
4241 .group_by(pg_catalog.pg_inherits.c.inhrelid)
4242 .subquery("inherits")
4243 )
4245 if self.server_version_info < (12,):
4246 # this is not in the pg_catalog.pg_class since it was
4247 # removed in PostgreSQL version 12
4248 has_oids = sql.column("relhasoids", BOOLEAN)
4249 else:
4250 has_oids = sql.null().label("relhasoids")
4252 relkinds = self._kind_to_relkinds(kind)
4253 query = (
4254 select(
4255 pg_catalog.pg_class.c.oid,
4256 pg_catalog.pg_class.c.relname,
4257 pg_catalog.pg_class.c.reloptions,
4258 has_oids,
4259 sql.case(
4260 (
4261 sql.and_(
4262 pg_catalog.pg_am.c.amname.is_not(None),
4263 pg_catalog.pg_am.c.amname
4264 != sql.func.current_setting(
4265 "default_table_access_method"
4266 ),
4267 ),
4268 pg_catalog.pg_am.c.amname,
4269 ),
4270 else_=sql.null(),
4271 ).label("access_method_name"),
4272 pg_catalog.pg_tablespace.c.spcname.label("tablespace_name"),
4273 inherits_sq.c.parent_table_names,
4274 )
4275 .select_from(pg_catalog.pg_class)
4276 .outerjoin(
4277 # NOTE: on postgresql < 12, this could be avoided
4278 # since relam is always 0 so nothing is joined.
4279 pg_catalog.pg_am,
4280 pg_catalog.pg_class.c.relam == pg_catalog.pg_am.c.oid,
4281 )
4282 .outerjoin(
4283 inherits_sq,
4284 pg_catalog.pg_class.c.oid == inherits_sq.c.inhrelid,
4285 )
4286 .outerjoin(
4287 pg_catalog.pg_tablespace,
4288 pg_catalog.pg_tablespace.c.oid
4289 == pg_catalog.pg_class.c.reltablespace,
4290 )
4291 .where(self._pg_class_relkind_condition(relkinds))
4292 )
4293 query = self._pg_class_filter_scope_schema(query, schema, scope=scope)
4294 if has_filter_names:
4295 query = query.where(
4296 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
4297 )
4298 return query
4300 def get_multi_table_options(
4301 self, connection, schema, filter_names, scope, kind, **kw
4302 ):
4303 has_filter_names, params = self._prepare_filter_names(filter_names)
4304 query = self._table_options_query(
4305 schema, has_filter_names, scope, kind
4306 )
4307 rows = connection.execute(query, params).mappings()
4308 table_options = {}
4310 for row in rows:
4311 current: dict[str, Any] = {}
4312 if row["access_method_name"] is not None:
4313 current["postgresql_using"] = row["access_method_name"]
4315 if row["parent_table_names"]:
4316 current["postgresql_inherits"] = tuple(
4317 row["parent_table_names"]
4318 )
4320 if row["reloptions"]:
4321 current["postgresql_with"] = dict(
4322 option.split("=", 1) for option in row["reloptions"]
4323 )
4325 if row["relhasoids"]:
4326 current["postgresql_with_oids"] = True
4328 if row["tablespace_name"] is not None:
4329 current["postgresql_tablespace"] = row["tablespace_name"]
4331 table_options[(schema, row["relname"])] = current
4333 return table_options.items()
4335 @lru_cache()
4336 def _columns_query(self, schema, has_filter_names, scope, kind):
4337 # NOTE: the query with the default and identity options scalar
4338 # subquery is faster than trying to use outer joins for them
4339 generated = (
4340 pg_catalog.pg_attribute.c.attgenerated.label("generated")
4341 if self.server_version_info >= (12,)
4342 else sql.null().label("generated")
4343 )
4344 if self.server_version_info >= (10,):
4345 # join lateral performs worse (~2x slower) than a scalar_subquery
4346 # also the subquery can be run only if the column is an identity
4347 identity = sql.case(
4348 ( # attidentity != '' is required or it will reflect also
4349 # serial columns as identity.
4350 pg_catalog.pg_attribute.c.attidentity != "",
4351 select(
4352 sql.func.json_build_object(
4353 "always",
4354 pg_catalog.pg_attribute.c.attidentity == "a",
4355 "start",
4356 pg_catalog.pg_sequence.c.seqstart,
4357 "increment",
4358 pg_catalog.pg_sequence.c.seqincrement,
4359 "minvalue",
4360 pg_catalog.pg_sequence.c.seqmin,
4361 "maxvalue",
4362 pg_catalog.pg_sequence.c.seqmax,
4363 "cache",
4364 pg_catalog.pg_sequence.c.seqcache,
4365 "cycle",
4366 pg_catalog.pg_sequence.c.seqcycle,
4367 type_=sqltypes.JSON(),
4368 )
4369 )
4370 .select_from(pg_catalog.pg_sequence)
4371 .where(
4372 # not needed but pg seems to like it
4373 pg_catalog.pg_attribute.c.attidentity != "",
4374 pg_catalog.pg_sequence.c.seqrelid
4375 == sql.cast(
4376 sql.cast(
4377 pg_catalog.pg_get_serial_sequence(
4378 sql.cast(
4379 sql.cast(
4380 pg_catalog.pg_attribute.c.attrelid,
4381 REGCLASS,
4382 ),
4383 TEXT,
4384 ),
4385 pg_catalog.pg_attribute.c.attname,
4386 ),
4387 REGCLASS,
4388 ),
4389 OID,
4390 ),
4391 )
4392 .correlate(pg_catalog.pg_attribute)
4393 .scalar_subquery(),
4394 ),
4395 else_=sql.null(),
4396 ).label("identity_options")
4397 else:
4398 identity = sql.null().label("identity_options")
4400 # join lateral performs the same as scalar_subquery here, also
4401 # the subquery can be run only if the column has a default
4402 default = sql.case(
4403 (
4404 pg_catalog.pg_attribute.c.atthasdef,
4405 select(
4406 pg_catalog.pg_get_expr(
4407 pg_catalog.pg_attrdef.c.adbin,
4408 pg_catalog.pg_attrdef.c.adrelid,
4409 )
4410 )
4411 .select_from(pg_catalog.pg_attrdef)
4412 .where(
4413 # not needed but pg seems to like it
4414 pg_catalog.pg_attribute.c.atthasdef,
4415 pg_catalog.pg_attrdef.c.adrelid
4416 == pg_catalog.pg_attribute.c.attrelid,
4417 pg_catalog.pg_attrdef.c.adnum
4418 == pg_catalog.pg_attribute.c.attnum,
4419 )
4420 .correlate(pg_catalog.pg_attribute)
4421 .scalar_subquery(),
4422 ),
4423 else_=sql.null(),
4424 ).label("default")
4426 # get the name and schema of the collate when it's different from
4427 # the default one; the schema is only included when the collation
4428 # is not visible on the current search_path
4429 collate = sql.case(
4430 (
4431 sql.and_(
4432 pg_catalog.pg_attribute.c.attcollation != 0,
4433 select(pg_catalog.pg_type.c.typcollation)
4434 .where(
4435 pg_catalog.pg_type.c.oid
4436 == pg_catalog.pg_attribute.c.atttypid,
4437 )
4438 .correlate(pg_catalog.pg_attribute)
4439 .scalar_subquery()
4440 != pg_catalog.pg_attribute.c.attcollation,
4441 ),
4442 select(
4443 sql.func.json_build_object(
4444 "name",
4445 pg_catalog.pg_collation.c.collname,
4446 "schema",
4447 sql.case(
4448 (
4449 pg_catalog.pg_collation_is_visible(
4450 pg_catalog.pg_collation.c.oid
4451 ),
4452 sql.null(),
4453 ),
4454 else_=pg_catalog.pg_namespace.c.nspname,
4455 ),
4456 type_=sqltypes.JSON(),
4457 )
4458 )
4459 .select_from(pg_catalog.pg_collation)
4460 .join(
4461 pg_catalog.pg_namespace,
4462 pg_catalog.pg_namespace.c.oid
4463 == pg_catalog.pg_collation.c.collnamespace,
4464 )
4465 .where(
4466 pg_catalog.pg_collation.c.oid
4467 == pg_catalog.pg_attribute.c.attcollation
4468 )
4469 .correlate(pg_catalog.pg_attribute)
4470 .scalar_subquery(),
4471 ),
4472 else_=sql.null(),
4473 ).label("collation")
4475 relkinds = self._kind_to_relkinds(kind)
4476 query = (
4477 select(
4478 pg_catalog.pg_attribute.c.attname.label("name"),
4479 pg_catalog.format_type(
4480 pg_catalog.pg_attribute.c.atttypid,
4481 pg_catalog.pg_attribute.c.atttypmod,
4482 ).label("format_type"),
4483 default,
4484 pg_catalog.pg_attribute.c.attnotnull.label("not_null"),
4485 pg_catalog.pg_class.c.relname.label("table_name"),
4486 pg_catalog.pg_description.c.description.label("comment"),
4487 generated,
4488 identity,
4489 collate,
4490 )
4491 .select_from(pg_catalog.pg_class)
4492 # NOTE: postgresql support table with no user column, meaning
4493 # there is no row with pg_attribute.attnum > 0. use a left outer
4494 # join to avoid filtering these tables.
4495 .outerjoin(
4496 pg_catalog.pg_attribute,
4497 sql.and_(
4498 pg_catalog.pg_class.c.oid
4499 == pg_catalog.pg_attribute.c.attrelid,
4500 pg_catalog.pg_attribute.c.attnum > 0,
4501 ~pg_catalog.pg_attribute.c.attisdropped,
4502 ),
4503 )
4504 .outerjoin(
4505 pg_catalog.pg_description,
4506 sql.and_(
4507 pg_catalog.pg_description.c.objoid
4508 == pg_catalog.pg_attribute.c.attrelid,
4509 pg_catalog.pg_description.c.objsubid
4510 == pg_catalog.pg_attribute.c.attnum,
4511 ),
4512 )
4513 .where(self._pg_class_relkind_condition(relkinds))
4514 .order_by(
4515 pg_catalog.pg_class.c.relname, pg_catalog.pg_attribute.c.attnum
4516 )
4517 )
4518 query = self._pg_class_filter_scope_schema(query, schema, scope=scope)
4519 if has_filter_names:
4520 query = query.where(
4521 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
4522 )
4523 return query
4525 def get_multi_columns(
4526 self, connection, schema, filter_names, scope, kind, **kw
4527 ):
4528 has_filter_names, params = self._prepare_filter_names(filter_names)
4529 query = self._columns_query(schema, has_filter_names, scope, kind)
4530 rows = connection.execute(query, params).mappings()
4532 named_type_loader = _NamedTypeLoader(self, connection, kw)
4533 columns = self._get_columns_info(rows, named_type_loader, schema)
4535 return columns.items()
4537 _format_type_args_pattern = re.compile(r"\((.*)\)")
4538 _format_type_args_delim = re.compile(r"\s*,\s*")
4539 _format_array_spec_pattern = re.compile(r"((?:\[\])*)$")
4541 def _reflect_type(
4542 self,
4543 format_type: Optional[str],
4544 named_type_loader: _NamedTypeLoader,
4545 type_description: str,
4546 collation: Optional[str],
4547 collation_schema: Optional[str] = None,
4548 ) -> sqltypes.TypeEngine[Any]:
4549 """
4550 Attempts to reconstruct a column type defined in ischema_names based
4551 on the information available in the format_type.
4553 If the `format_type` cannot be associated with a known `ischema_names`,
4554 it is treated as a reference to a known PostgreSQL named `ENUM` or
4555 `DOMAIN` type.
4556 """
4557 type_description = type_description or "unknown type"
4558 if format_type is None:
4559 util.warn(
4560 "PostgreSQL format_type() returned NULL for %s"
4561 % type_description
4562 )
4563 return sqltypes.NULLTYPE
4565 attype_args_match = self._format_type_args_pattern.search(format_type)
4566 if attype_args_match and attype_args_match.group(1):
4567 attype_args = self._format_type_args_delim.split(
4568 attype_args_match.group(1)
4569 )
4570 else:
4571 attype_args = ()
4573 match_array_dim = self._format_array_spec_pattern.search(format_type)
4574 # Each "[]" in array specs corresponds to an array dimension
4575 array_dim = len(match_array_dim.group(1) or "") // 2
4577 # Remove all parameters and array specs from format_type to obtain an
4578 # ischema_name candidate
4579 attype = self._format_type_args_pattern.sub("", format_type)
4580 attype = self._format_array_spec_pattern.sub("", attype)
4582 schema_type = self.ischema_names.get(attype.lower(), None)
4583 args, kwargs = (), {}
4585 if attype == "numeric":
4586 if len(attype_args) == 2:
4587 precision, scale = map(int, attype_args)
4588 args = (precision, scale)
4590 elif attype == "double precision":
4591 args = (53,)
4593 elif attype == "integer":
4594 args = ()
4596 elif attype in ("timestamp with time zone", "time with time zone"):
4597 kwargs["timezone"] = True
4598 if len(attype_args) == 1:
4599 kwargs["precision"] = int(attype_args[0])
4601 elif attype in (
4602 "timestamp without time zone",
4603 "time without time zone",
4604 "time",
4605 ):
4606 kwargs["timezone"] = False
4607 if len(attype_args) == 1:
4608 kwargs["precision"] = int(attype_args[0])
4610 elif attype == "bit varying":
4611 kwargs["varying"] = True
4612 if len(attype_args) == 1:
4613 charlen = int(attype_args[0])
4614 args = (charlen,)
4616 # a domain or enum can start with interval, so be mindful of that.
4617 elif attype == "interval" or attype.startswith("interval "):
4618 schema_type = INTERVAL
4620 field_match = re.match(r"interval (.+)", attype)
4621 if field_match:
4622 kwargs["fields"] = field_match.group(1)
4624 if len(attype_args) == 1:
4625 kwargs["precision"] = int(attype_args[0])
4627 else:
4628 enum_or_domain_key = tuple(util.quoted_token_parser(attype))
4630 if (
4631 schema_type is None
4632 and enum_or_domain_key in named_type_loader.enums
4633 ):
4634 schema_type = ENUM
4635 enum = named_type_loader.enums[enum_or_domain_key]
4637 kwargs["name"] = enum["name"]
4639 if not enum["visible"]:
4640 kwargs["schema"] = enum["schema"]
4641 args = tuple(enum["labels"])
4642 elif (
4643 schema_type is None
4644 and enum_or_domain_key in named_type_loader.domains
4645 ):
4646 schema_type = DOMAIN
4647 domain = named_type_loader.domains[enum_or_domain_key]
4649 data_type = self._reflect_type(
4650 domain["type"],
4651 named_type_loader,
4652 type_description="DOMAIN '%s'" % domain["name"],
4653 collation=domain["collation"],
4654 collation_schema=domain["collation_schema"],
4655 )
4656 args = (domain["name"], data_type)
4658 kwargs["collation"] = domain["collation"]
4659 kwargs["collation_schema"] = domain["collation_schema"]
4660 kwargs["default"] = domain["default"]
4661 kwargs["not_null"] = not domain["nullable"]
4662 kwargs["create_type"] = False
4664 if domain["constraints"]:
4665 # We only support a single constraint
4666 check_constraint = domain["constraints"][0]
4668 kwargs["constraint_name"] = check_constraint["name"]
4669 kwargs["check"] = check_constraint["check"]
4671 if not domain["visible"]:
4672 kwargs["schema"] = domain["schema"]
4674 else:
4675 try:
4676 charlen = int(attype_args[0])
4677 args = (charlen, *attype_args[1:])
4678 except (ValueError, IndexError):
4679 args = attype_args
4681 if not schema_type:
4682 util.warn(
4683 "Did not recognize type '%s' of %s"
4684 % (attype, type_description)
4685 )
4686 return sqltypes.NULLTYPE
4688 if collation is not None:
4689 kwargs["collation"] = collation
4690 if collation_schema is not None:
4691 kwargs["collation_schema"] = collation_schema
4693 data_type = schema_type(*args, **kwargs)
4694 if array_dim >= 1:
4695 # postgres does not preserve dimensionality or size of array types.
4696 data_type = _array.ARRAY(data_type)
4698 return data_type
4700 def _get_columns_info(self, rows, named_type_loader, schema):
4701 columns = defaultdict(list)
4702 for row_dict in rows:
4703 # ensure that each table has an entry, even if it has no columns
4704 if row_dict["name"] is None:
4705 columns[(schema, row_dict["table_name"])] = (
4706 ReflectionDefaults.columns()
4707 )
4708 continue
4709 table_cols = columns[(schema, row_dict["table_name"])]
4711 collation_info = row_dict["collation"]
4712 if collation_info is not None:
4713 collation = collation_info["name"]
4714 collation_schema = collation_info["schema"]
4715 else:
4716 collation = collation_schema = None
4718 coltype = self._reflect_type(
4719 row_dict["format_type"],
4720 named_type_loader,
4721 type_description="column '%s'" % row_dict["name"],
4722 collation=collation,
4723 collation_schema=collation_schema,
4724 )
4726 default = row_dict["default"]
4727 name = row_dict["name"]
4728 generated = row_dict["generated"]
4729 nullable = not row_dict["not_null"]
4731 if isinstance(coltype, DOMAIN):
4732 if not default:
4733 # domain can override the default value but
4734 # can't set it to None
4735 if coltype.default is not None:
4736 default = coltype.default
4738 nullable = nullable and not coltype.not_null
4740 identity = row_dict["identity_options"]
4742 # If a zero byte or blank string depending on driver (is also
4743 # absent for older PG versions), then not a generated column.
4744 # Otherwise, s = stored, v = virtual.
4745 if generated not in (None, "", b"\x00"):
4746 computed = dict(
4747 sqltext=default, persisted=generated in ("s", b"s")
4748 )
4749 default = None
4750 else:
4751 computed = None
4753 # adjust the default value
4754 autoincrement = False
4755 if default is not None:
4756 match = re.search(r"""(nextval\(')([^']+)('.*$)""", default)
4757 if match is not None:
4758 if issubclass(coltype._type_affinity, sqltypes.Integer):
4759 autoincrement = True
4760 # the default is related to a Sequence
4761 if "." not in match.group(2) and schema is not None:
4762 # unconditionally quote the schema name. this could
4763 # later be enhanced to obey quoting rules /
4764 # "quote schema"
4765 default = (
4766 match.group(1)
4767 + ('"%s"' % schema)
4768 + "."
4769 + match.group(2)
4770 + match.group(3)
4771 )
4773 column_info = {
4774 "name": name,
4775 "type": coltype,
4776 "nullable": nullable,
4777 "default": default,
4778 "autoincrement": autoincrement or identity is not None,
4779 "comment": row_dict["comment"],
4780 }
4781 if computed is not None:
4782 column_info["computed"] = computed
4783 if identity is not None:
4784 column_info["identity"] = identity
4786 table_cols.append(column_info)
4788 return columns
4790 @lru_cache()
4791 def _table_oids_query(self, schema, has_filter_names, scope, kind):
4792 relkinds = self._kind_to_relkinds(kind)
4793 oid_q = select(
4794 pg_catalog.pg_class.c.oid, pg_catalog.pg_class.c.relname
4795 ).where(self._pg_class_relkind_condition(relkinds))
4796 oid_q = self._pg_class_filter_scope_schema(oid_q, schema, scope=scope)
4798 if has_filter_names:
4799 oid_q = oid_q.where(
4800 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
4801 )
4802 return oid_q
4804 @reflection.flexi_cache(
4805 ("schema", InternalTraversal.dp_string),
4806 ("filter_names", InternalTraversal.dp_string_list),
4807 ("kind", InternalTraversal.dp_plain_obj),
4808 ("scope", InternalTraversal.dp_plain_obj),
4809 )
4810 def _get_table_oids(
4811 self, connection, schema, filter_names, scope, kind, **kw
4812 ):
4813 has_filter_names, params = self._prepare_filter_names(filter_names)
4814 oid_q = self._table_oids_query(schema, has_filter_names, scope, kind)
4815 result = connection.execute(oid_q, params)
4816 return result.all()
4818 @util.memoized_property
4819 def _constraint_query(self):
4820 if self.server_version_info >= (11, 0):
4821 indnkeyatts = pg_catalog.pg_index.c.indnkeyatts
4822 else:
4823 indnkeyatts = pg_catalog.pg_index.c.indnatts.label("indnkeyatts")
4825 if self.server_version_info >= (15,):
4826 indnullsnotdistinct = pg_catalog.pg_index.c.indnullsnotdistinct
4827 else:
4828 indnullsnotdistinct = sql.false().label("indnullsnotdistinct")
4830 con_sq = (
4831 select(
4832 pg_catalog.pg_constraint.c.conrelid,
4833 pg_catalog.pg_constraint.c.conname,
4834 sql.func.unnest(pg_catalog.pg_index.c.indkey).label("attnum"),
4835 sql.func.generate_subscripts(
4836 pg_catalog.pg_index.c.indkey, 1
4837 ).label("ord"),
4838 indnkeyatts,
4839 indnullsnotdistinct,
4840 pg_catalog.pg_description.c.description,
4841 )
4842 .join(
4843 pg_catalog.pg_index,
4844 pg_catalog.pg_constraint.c.conindid
4845 == pg_catalog.pg_index.c.indexrelid,
4846 )
4847 .outerjoin(
4848 pg_catalog.pg_description,
4849 pg_catalog.pg_description.c.objoid
4850 == pg_catalog.pg_constraint.c.oid,
4851 )
4852 .where(
4853 pg_catalog.pg_constraint.c.contype == bindparam("contype"),
4854 pg_catalog.pg_constraint.c.conrelid.in_(bindparam("oids")),
4855 # NOTE: filtering also on pg_index.indrelid for oids does
4856 # not seem to have a performance effect, but it may be an
4857 # option if perf problems are reported
4858 )
4859 .subquery("con")
4860 )
4862 attr_sq = (
4863 select(
4864 con_sq.c.conrelid,
4865 con_sq.c.conname,
4866 con_sq.c.description,
4867 con_sq.c.ord,
4868 con_sq.c.indnkeyatts,
4869 con_sq.c.indnullsnotdistinct,
4870 pg_catalog.pg_attribute.c.attname,
4871 )
4872 .select_from(pg_catalog.pg_attribute)
4873 .join(
4874 con_sq,
4875 sql.and_(
4876 pg_catalog.pg_attribute.c.attnum == con_sq.c.attnum,
4877 pg_catalog.pg_attribute.c.attrelid == con_sq.c.conrelid,
4878 ),
4879 )
4880 .where(
4881 # NOTE: restate the condition here, since pg15 otherwise
4882 # seems to get confused on pscopg2 sometimes, doing
4883 # a sequential scan of pg_attribute.
4884 # The condition in the con_sq subquery is not actually needed
4885 # in pg15, but it may be needed in older versions. Keeping it
4886 # does not seems to have any impact in any case.
4887 con_sq.c.conrelid.in_(bindparam("oids"))
4888 )
4889 .subquery("attr")
4890 )
4892 return (
4893 select(
4894 attr_sq.c.conrelid,
4895 sql.func.array_agg(
4896 # NOTE: cast since some postgresql derivatives may
4897 # not support array_agg on the name type
4898 aggregate_order_by(
4899 attr_sq.c.attname.cast(TEXT), attr_sq.c.ord
4900 )
4901 ).label("cols"),
4902 attr_sq.c.conname,
4903 sql.func.min(attr_sq.c.description).label("description"),
4904 sql.func.min(attr_sq.c.indnkeyatts).label("indnkeyatts"),
4905 sql.func.bool_and(attr_sq.c.indnullsnotdistinct).label(
4906 "indnullsnotdistinct"
4907 ),
4908 )
4909 .group_by(attr_sq.c.conrelid, attr_sq.c.conname)
4910 .order_by(attr_sq.c.conrelid, attr_sq.c.conname)
4911 )
4913 def _reflect_constraint(
4914 self, connection, contype, schema, filter_names, scope, kind, **kw
4915 ):
4916 # used to reflect primary and unique constraint
4917 table_oids = self._get_table_oids(
4918 connection, schema, filter_names, scope, kind, **kw
4919 )
4920 batches = list(table_oids)
4921 is_unique = contype == "u"
4923 while batches:
4924 batch = batches[0:3000]
4925 batches[0:3000] = []
4927 result = connection.execute(
4928 self._constraint_query,
4929 {"oids": [r[0] for r in batch], "contype": contype},
4930 ).mappings()
4932 result_by_oid = defaultdict(list)
4933 for row_dict in result:
4934 result_by_oid[row_dict["conrelid"]].append(row_dict)
4936 for oid, tablename in batch:
4937 for_oid = result_by_oid.get(oid, ())
4938 if for_oid:
4939 for row in for_oid:
4940 # See note in get_multi_indexes
4941 all_cols = row["cols"]
4942 indnkeyatts = row["indnkeyatts"]
4943 if len(all_cols) > indnkeyatts:
4944 inc_cols = all_cols[indnkeyatts:]
4945 cst_cols = all_cols[:indnkeyatts]
4946 else:
4947 inc_cols = []
4948 cst_cols = all_cols
4950 opts = {}
4951 if self.server_version_info >= (11,):
4952 opts["postgresql_include"] = inc_cols
4953 if is_unique:
4954 opts["postgresql_nulls_not_distinct"] = row[
4955 "indnullsnotdistinct"
4956 ]
4957 yield (
4958 tablename,
4959 cst_cols,
4960 row["conname"],
4961 row["description"],
4962 opts,
4963 )
4964 else:
4965 yield tablename, None, None, None, None
4967 def get_multi_pk_constraint(
4968 self, connection, schema, filter_names, scope, kind, **kw
4969 ):
4970 result = self._reflect_constraint(
4971 connection, "p", schema, filter_names, scope, kind, **kw
4972 )
4974 # only a single pk can be present for each table. Return an entry
4975 # even if a table has no primary key
4976 default = ReflectionDefaults.pk_constraint
4978 def pk_constraint(pk_name, cols, comment, opts):
4979 info = {
4980 "constrained_columns": cols,
4981 "name": pk_name,
4982 "comment": comment,
4983 }
4984 if opts:
4985 info["dialect_options"] = opts
4986 return info
4988 return (
4989 (
4990 (schema, table_name),
4991 (
4992 pk_constraint(pk_name, cols, comment, opts)
4993 if pk_name is not None
4994 else default()
4995 ),
4996 )
4997 for table_name, cols, pk_name, comment, opts in result
4998 )
5000 @lru_cache()
5001 def _foreing_key_query(self, schema, has_filter_names, scope, kind):
5002 pg_class_ref = pg_catalog.pg_class.alias("cls_ref")
5003 pg_namespace_ref = pg_catalog.pg_namespace.alias("nsp_ref")
5004 relkinds = self._kind_to_relkinds(kind)
5005 query = (
5006 select(
5007 pg_catalog.pg_class.c.relname,
5008 pg_catalog.pg_constraint.c.conname,
5009 # NOTE: avoid calling pg_get_constraintdef when not needed
5010 # to speed up the query
5011 sql.case(
5012 (
5013 pg_catalog.pg_constraint.c.oid.is_not(None),
5014 pg_catalog.pg_get_constraintdef(
5015 pg_catalog.pg_constraint.c.oid, True
5016 ),
5017 ),
5018 else_=None,
5019 ),
5020 pg_namespace_ref.c.nspname,
5021 pg_catalog.pg_description.c.description,
5022 )
5023 .select_from(pg_catalog.pg_class)
5024 .outerjoin(
5025 pg_catalog.pg_constraint,
5026 sql.and_(
5027 pg_catalog.pg_class.c.oid
5028 == pg_catalog.pg_constraint.c.conrelid,
5029 pg_catalog.pg_constraint.c.contype == "f",
5030 ),
5031 )
5032 .outerjoin(
5033 pg_class_ref,
5034 pg_class_ref.c.oid == pg_catalog.pg_constraint.c.confrelid,
5035 )
5036 .outerjoin(
5037 pg_namespace_ref,
5038 pg_class_ref.c.relnamespace == pg_namespace_ref.c.oid,
5039 )
5040 .outerjoin(
5041 pg_catalog.pg_description,
5042 pg_catalog.pg_description.c.objoid
5043 == pg_catalog.pg_constraint.c.oid,
5044 )
5045 .order_by(
5046 pg_catalog.pg_class.c.relname,
5047 pg_catalog.pg_constraint.c.conname,
5048 )
5049 .where(self._pg_class_relkind_condition(relkinds))
5050 )
5051 query = self._pg_class_filter_scope_schema(query, schema, scope)
5052 if has_filter_names:
5053 query = query.where(
5054 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
5055 )
5056 return query
5058 @util.memoized_property
5059 def _fk_regex_pattern(self):
5060 # optionally quoted token
5061 qtoken = r'(?:"(?:[^"]|"")+"|[\w]+?)'
5063 # https://www.postgresql.org/docs/current/static/sql-createtable.html
5064 return re.compile(
5065 r"FOREIGN KEY \((.*?)\) "
5066 rf"REFERENCES (?:({qtoken})\.)?({qtoken})\(((?:{qtoken}(?: *, *)?)+)\)" # noqa: E501
5067 r"[\s]?(MATCH (FULL|PARTIAL|SIMPLE)+)?"
5068 r"[\s]?(?:ON (UPDATE|DELETE) "
5069 r"(CASCADE|RESTRICT|NO ACTION|"
5070 r"SET (?:NULL|DEFAULT)(?:\s\(.+\))?)+)?"
5071 r"[\s]?(?:ON (UPDATE|DELETE) "
5072 r"(CASCADE|RESTRICT|NO ACTION|"
5073 r"SET (?:NULL|DEFAULT)(?:\s\(.+\))?)+)?"
5074 r"[\s]?(DEFERRABLE|NOT DEFERRABLE)?"
5075 r"[\s]?(INITIALLY (DEFERRED|IMMEDIATE)+)?"
5076 )
5078 def _parse_fk(self, condef):
5079 FK_REGEX = self._fk_regex_pattern
5080 m = re.search(FK_REGEX, condef).groups()
5082 (
5083 constrained_columns,
5084 referred_schema,
5085 referred_table,
5086 referred_columns,
5087 _,
5088 match,
5089 upddelkey1,
5090 upddelval1,
5091 upddelkey2,
5092 upddelval2,
5093 deferrable,
5094 _,
5095 initially,
5096 ) = m
5098 onupdate = (
5099 upddelval1
5100 if upddelkey1 == "UPDATE"
5101 else upddelval2 if upddelkey2 == "UPDATE" else None
5102 )
5103 ondelete = (
5104 upddelval1
5105 if upddelkey1 == "DELETE"
5106 else upddelval2 if upddelkey2 == "DELETE" else None
5107 )
5109 return (
5110 constrained_columns,
5111 referred_schema,
5112 referred_table,
5113 referred_columns,
5114 match,
5115 onupdate,
5116 ondelete,
5117 deferrable,
5118 initially,
5119 )
5121 def get_multi_foreign_keys(
5122 self,
5123 connection,
5124 schema,
5125 filter_names,
5126 scope,
5127 kind,
5128 postgresql_ignore_search_path=False,
5129 **kw,
5130 ):
5131 preparer = self.identifier_preparer
5133 has_filter_names, params = self._prepare_filter_names(filter_names)
5134 query = self._foreing_key_query(schema, has_filter_names, scope, kind)
5135 result = connection.execute(query, params)
5137 fkeys = defaultdict(list)
5138 default = ReflectionDefaults.foreign_keys
5139 for table_name, conname, condef, conschema, comment in result:
5140 # ensure that each table has an entry, even if it has
5141 # no foreign keys
5142 if conname is None:
5143 fkeys[(schema, table_name)] = default()
5144 continue
5145 table_fks = fkeys[(schema, table_name)]
5147 (
5148 constrained_columns,
5149 referred_schema,
5150 referred_table,
5151 referred_columns,
5152 match,
5153 onupdate,
5154 ondelete,
5155 deferrable,
5156 initially,
5157 ) = self._parse_fk(condef)
5159 if deferrable is not None:
5160 deferrable = True if deferrable == "DEFERRABLE" else False
5161 constrained_columns = [
5162 preparer._unquote_identifier(x)
5163 for x in re.split(r"\s*,\s*", constrained_columns)
5164 ]
5166 if postgresql_ignore_search_path:
5167 # when ignoring search path, we use the actual schema
5168 # provided it isn't the "default" schema
5169 if conschema != self.default_schema_name:
5170 referred_schema = conschema
5171 else:
5172 referred_schema = schema
5173 elif referred_schema:
5174 # referred_schema is the schema that we regexp'ed from
5175 # pg_get_constraintdef(). If the schema is in the search
5176 # path, pg_get_constraintdef() will give us None.
5177 referred_schema = preparer._unquote_identifier(referred_schema)
5178 elif schema is not None and schema == conschema:
5179 # If the actual schema matches the schema of the table
5180 # we're reflecting, then we will use that.
5181 referred_schema = schema
5183 referred_table = preparer._unquote_identifier(referred_table)
5184 referred_columns = [
5185 preparer._unquote_identifier(x)
5186 for x in re.split(r"\s*,\s", referred_columns)
5187 ]
5188 options = {
5189 k: v
5190 for k, v in [
5191 ("onupdate", onupdate),
5192 ("ondelete", ondelete),
5193 ("initially", initially),
5194 ("deferrable", deferrable),
5195 ("match", match),
5196 ]
5197 if v is not None and v != "NO ACTION"
5198 }
5199 fkey_d = {
5200 "name": conname,
5201 "constrained_columns": constrained_columns,
5202 "referred_schema": referred_schema,
5203 "referred_table": referred_table,
5204 "referred_columns": referred_columns,
5205 "options": options,
5206 "comment": comment,
5207 }
5208 table_fks.append(fkey_d)
5209 return fkeys.items()
5211 @util.memoized_property
5212 def _index_query(self):
5213 # NOTE: pg_index is used as from two times to improve performance,
5214 # since extraing all the index information from `idx_sq` to avoid
5215 # the second pg_index use leads to a worse performing query in
5216 # particular when querying for a single table (as of pg 17)
5217 # NOTE: repeating oids clause improve query performance
5219 # subquery to get the columns
5220 idx_sq = (
5221 select(
5222 pg_catalog.pg_index.c.indexrelid,
5223 pg_catalog.pg_index.c.indrelid,
5224 sql.func.unnest(pg_catalog.pg_index.c.indkey).label("attnum"),
5225 sql.func.unnest(pg_catalog.pg_index.c.indclass).label(
5226 "att_opclass"
5227 ),
5228 sql.func.generate_subscripts(
5229 pg_catalog.pg_index.c.indkey, 1
5230 ).label("ord"),
5231 )
5232 .where(
5233 ~pg_catalog.pg_index.c.indisprimary,
5234 pg_catalog.pg_index.c.indrelid.in_(bindparam("oids")),
5235 )
5236 .subquery("idx")
5237 )
5239 attr_sq = (
5240 select(
5241 idx_sq.c.indexrelid,
5242 idx_sq.c.indrelid,
5243 idx_sq.c.ord,
5244 # NOTE: always using pg_get_indexdef is too slow so just
5245 # invoke when the element is an expression
5246 sql.case(
5247 (
5248 idx_sq.c.attnum == 0,
5249 pg_catalog.pg_get_indexdef(
5250 idx_sq.c.indexrelid, idx_sq.c.ord + 1, True
5251 ),
5252 ),
5253 # NOTE: need to cast this since attname is of type "name"
5254 # that's limited to 63 bytes, while pg_get_indexdef
5255 # returns "text" so its output may get cut
5256 else_=pg_catalog.pg_attribute.c.attname.cast(TEXT),
5257 ).label("element"),
5258 (idx_sq.c.attnum == 0).label("is_expr"),
5259 # since it's converted to array cast it to bigint (oid are
5260 # "unsigned four-byte integer") to make it easier for
5261 # dialects to interpret
5262 idx_sq.c.att_opclass.cast(BIGINT),
5263 )
5264 .select_from(idx_sq)
5265 .outerjoin(
5266 # do not remove rows where idx_sq.c.attnum is 0
5267 pg_catalog.pg_attribute,
5268 sql.and_(
5269 pg_catalog.pg_attribute.c.attnum == idx_sq.c.attnum,
5270 pg_catalog.pg_attribute.c.attrelid == idx_sq.c.indrelid,
5271 ),
5272 )
5273 .where(idx_sq.c.indrelid.in_(bindparam("oids")))
5274 .subquery("idx_attr")
5275 )
5277 cols_sq = (
5278 select(
5279 attr_sq.c.indexrelid,
5280 sql.func.min(attr_sq.c.indrelid),
5281 sql.func.array_agg(
5282 aggregate_order_by(attr_sq.c.element, attr_sq.c.ord)
5283 ).label("elements"),
5284 sql.func.array_agg(
5285 aggregate_order_by(attr_sq.c.is_expr, attr_sq.c.ord)
5286 ).label("elements_is_expr"),
5287 sql.func.array_agg(
5288 aggregate_order_by(attr_sq.c.att_opclass, attr_sq.c.ord)
5289 ).label("elements_opclass"),
5290 )
5291 .group_by(attr_sq.c.indexrelid)
5292 .subquery("idx_cols")
5293 )
5295 if self.server_version_info >= (11, 0):
5296 indnkeyatts = pg_catalog.pg_index.c.indnkeyatts
5297 else:
5298 indnkeyatts = pg_catalog.pg_index.c.indnatts.label("indnkeyatts")
5300 if self.server_version_info >= (15,):
5301 nulls_not_distinct = pg_catalog.pg_index.c.indnullsnotdistinct
5302 else:
5303 nulls_not_distinct = sql.false().label("indnullsnotdistinct")
5305 return (
5306 select(
5307 pg_catalog.pg_index.c.indrelid,
5308 pg_catalog.pg_class.c.relname,
5309 pg_catalog.pg_index.c.indisunique,
5310 pg_catalog.pg_index.c.indisvalid,
5311 pg_catalog.pg_constraint.c.conrelid.is_not(None).label(
5312 "has_constraint"
5313 ),
5314 pg_catalog.pg_index.c.indoption,
5315 pg_catalog.pg_class.c.reloptions,
5316 # will get the value using the pg_am cached dict
5317 pg_catalog.pg_class.c.relam,
5318 # NOTE: pg_get_expr is very fast so this case has almost no
5319 # performance impact
5320 sql.case(
5321 (
5322 pg_catalog.pg_index.c.indpred.is_not(None),
5323 pg_catalog.pg_get_expr(
5324 pg_catalog.pg_index.c.indpred,
5325 pg_catalog.pg_index.c.indrelid,
5326 ),
5327 ),
5328 else_=None,
5329 ).label("filter_definition"),
5330 indnkeyatts,
5331 nulls_not_distinct,
5332 cols_sq.c.elements,
5333 cols_sq.c.elements_is_expr,
5334 # will get the value using the pg_opclass cached dict
5335 cols_sq.c.elements_opclass,
5336 )
5337 .select_from(pg_catalog.pg_index)
5338 .where(
5339 pg_catalog.pg_index.c.indrelid.in_(bindparam("oids")),
5340 ~pg_catalog.pg_index.c.indisprimary,
5341 )
5342 .join(
5343 pg_catalog.pg_class,
5344 pg_catalog.pg_index.c.indexrelid == pg_catalog.pg_class.c.oid,
5345 )
5346 .outerjoin(
5347 cols_sq,
5348 pg_catalog.pg_index.c.indexrelid == cols_sq.c.indexrelid,
5349 )
5350 .outerjoin(
5351 pg_catalog.pg_constraint,
5352 sql.and_(
5353 pg_catalog.pg_index.c.indrelid
5354 == pg_catalog.pg_constraint.c.conrelid,
5355 pg_catalog.pg_index.c.indexrelid
5356 == pg_catalog.pg_constraint.c.conindid,
5357 pg_catalog.pg_constraint.c.contype
5358 == sql.any_(_array.array(("p", "u", "x"))),
5359 ),
5360 )
5361 .order_by(
5362 pg_catalog.pg_index.c.indrelid, pg_catalog.pg_class.c.relname
5363 )
5364 )
5366 def get_multi_indexes(
5367 self, connection, schema, filter_names, scope, kind, **kw
5368 ):
5369 table_oids = self._get_table_oids(
5370 connection, schema, filter_names, scope, kind, **kw
5371 )
5373 pg_am_btree_oid = self._load_pg_am_btree_oid(connection)
5374 # lazy load only if needed, the assumption is that most indexes
5375 # will use btree so it may not be needed at all
5376 pg_am_dict = None
5377 pg_opclass_dict = self._load_pg_opclass_notdefault_dict(
5378 connection, **kw
5379 )
5381 indexes = defaultdict(list)
5382 default = ReflectionDefaults.indexes
5384 batches = list(table_oids)
5386 while batches:
5387 batch = batches[0:3000]
5388 batches[0:3000] = []
5390 result = connection.execute(
5391 self._index_query, {"oids": [r[0] for r in batch]}
5392 ).mappings()
5394 result_by_oid = defaultdict(list)
5395 for row_dict in result:
5396 result_by_oid[row_dict["indrelid"]].append(row_dict)
5398 for oid, table_name in batch:
5399 if oid not in result_by_oid:
5400 # ensure that each table has an entry, even if reflection
5401 # is skipped because not supported
5402 indexes[(schema, table_name)] = default()
5403 continue
5405 for row in result_by_oid[oid]:
5406 index_name = row["relname"]
5408 table_indexes = indexes[(schema, table_name)]
5410 all_elements = row["elements"]
5411 all_elements_is_expr = row["elements_is_expr"]
5412 all_elements_opclass = row["elements_opclass"]
5413 indnkeyatts = row["indnkeyatts"]
5414 # "The number of key columns in the index, not counting any
5415 # included columns, which are merely stored and do not
5416 # participate in the index semantics"
5417 if len(all_elements) > indnkeyatts:
5418 # this is a "covering index" which has INCLUDE columns
5419 # as well as regular index columns
5420 inc_cols = all_elements[indnkeyatts:]
5421 idx_elements = all_elements[:indnkeyatts]
5422 idx_elements_is_expr = all_elements_is_expr[
5423 :indnkeyatts
5424 ]
5425 # postgresql does not support expression on included
5426 # columns as of v14: "ERROR: expressions are not
5427 # supported in included columns".
5428 assert all(
5429 not is_expr
5430 for is_expr in all_elements_is_expr[indnkeyatts:]
5431 )
5432 idx_elements_opclass = all_elements_opclass[
5433 :indnkeyatts
5434 ]
5435 else:
5436 idx_elements = all_elements
5437 idx_elements_is_expr = all_elements_is_expr
5438 inc_cols = []
5439 idx_elements_opclass = all_elements_opclass
5441 index = {"name": index_name, "unique": row["indisunique"]}
5442 if any(idx_elements_is_expr):
5443 index["column_names"] = [
5444 None if is_expr else expr
5445 for expr, is_expr in zip(
5446 idx_elements, idx_elements_is_expr
5447 )
5448 ]
5449 index["expressions"] = idx_elements
5450 else:
5451 index["column_names"] = idx_elements
5453 dialect_options = {}
5455 postgresql_ops = {}
5456 for name, opclass in zip(
5457 idx_elements, idx_elements_opclass
5458 ):
5459 # is not in the dict if the opclass is the default one
5460 opclass_name = pg_opclass_dict.get(opclass)
5461 if opclass_name is not None:
5462 postgresql_ops[name] = opclass_name
5464 if postgresql_ops:
5465 dialect_options["postgresql_ops"] = postgresql_ops
5467 sorting = {}
5468 for col_index, col_flags in enumerate(row["indoption"]):
5469 col_sorting = ()
5470 # try to set flags only if they differ from PG
5471 # defaults...
5472 if col_flags & 0x01:
5473 col_sorting += ("desc",)
5474 if not (col_flags & 0x02):
5475 col_sorting += ("nulls_last",)
5476 else:
5477 if col_flags & 0x02:
5478 col_sorting += ("nulls_first",)
5479 if col_sorting:
5480 sorting[idx_elements[col_index]] = col_sorting
5481 if sorting:
5482 index["column_sorting"] = sorting
5483 if row["has_constraint"]:
5484 index["duplicates_constraint"] = index_name
5486 if row["reloptions"]:
5487 dialect_options["postgresql_with"] = dict(
5488 [
5489 option.split("=", 1)
5490 for option in row["reloptions"]
5491 ]
5492 )
5493 # it *might* be nice to include that this is 'btree' in the
5494 # reflection info. But we don't want an Index object
5495 # to have a ``postgresql_using`` in it that is just the
5496 # default, so for the moment leaving this out.
5497 if row["relam"] != pg_am_btree_oid:
5498 if pg_am_dict is None:
5499 pg_am_dict = self._load_pg_am_dict(
5500 connection, **kw
5501 )
5502 dialect_options["postgresql_using"] = pg_am_dict[
5503 row["relam"]
5504 ]
5505 if row["filter_definition"]:
5506 dialect_options["postgresql_where"] = row[
5507 "filter_definition"
5508 ]
5509 if self.server_version_info >= (11,):
5510 dialect_options["postgresql_include"] = inc_cols
5511 if row["indnullsnotdistinct"]:
5512 # the default is False, so ignore it.
5513 dialect_options["postgresql_nulls_not_distinct"] = row[
5514 "indnullsnotdistinct"
5515 ]
5517 if not row["indisvalid"]:
5518 dialect_options["postgresql_invalid"] = True
5520 if dialect_options:
5521 index["dialect_options"] = dialect_options
5523 table_indexes.append(index)
5524 return indexes.items()
5526 def get_multi_unique_constraints(
5527 self,
5528 connection,
5529 schema,
5530 filter_names,
5531 scope,
5532 kind,
5533 **kw,
5534 ):
5535 result = self._reflect_constraint(
5536 connection, "u", schema, filter_names, scope, kind, **kw
5537 )
5539 # each table can have multiple unique constraints
5540 uniques = defaultdict(list)
5541 default = ReflectionDefaults.unique_constraints
5542 for table_name, cols, con_name, comment, options in result:
5543 # ensure a list is created for each table. leave it empty if
5544 # the table has no unique constraint
5545 if con_name is None:
5546 uniques[(schema, table_name)] = default()
5547 continue
5549 uc_dict = {
5550 "column_names": cols,
5551 "name": con_name,
5552 "comment": comment,
5553 }
5554 if options:
5555 uc_dict["dialect_options"] = options
5557 uniques[(schema, table_name)].append(uc_dict)
5558 return uniques.items()
5560 @lru_cache()
5561 def _comment_query(self, schema, has_filter_names, scope, kind):
5562 relkinds = self._kind_to_relkinds(kind)
5563 query = (
5564 select(
5565 pg_catalog.pg_class.c.relname,
5566 pg_catalog.pg_description.c.description,
5567 )
5568 .select_from(pg_catalog.pg_class)
5569 .outerjoin(
5570 pg_catalog.pg_description,
5571 sql.and_(
5572 pg_catalog.pg_class.c.oid
5573 == pg_catalog.pg_description.c.objoid,
5574 pg_catalog.pg_description.c.objsubid == 0,
5575 pg_catalog.pg_description.c.classoid
5576 == sql.func.cast("pg_catalog.pg_class", REGCLASS),
5577 ),
5578 )
5579 .where(self._pg_class_relkind_condition(relkinds))
5580 )
5581 query = self._pg_class_filter_scope_schema(query, schema, scope)
5582 if has_filter_names:
5583 query = query.where(
5584 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
5585 )
5586 return query
5588 def get_multi_table_comment(
5589 self, connection, schema, filter_names, scope, kind, **kw
5590 ):
5591 has_filter_names, params = self._prepare_filter_names(filter_names)
5592 query = self._comment_query(schema, has_filter_names, scope, kind)
5593 result = connection.execute(query, params)
5595 default = ReflectionDefaults.table_comment
5596 return (
5597 (
5598 (schema, table),
5599 {"text": comment} if comment is not None else default(),
5600 )
5601 for table, comment in result
5602 )
5604 @lru_cache()
5605 def _check_constraint_query(self, schema, has_filter_names, scope, kind):
5606 relkinds = self._kind_to_relkinds(kind)
5607 query = (
5608 select(
5609 pg_catalog.pg_class.c.relname,
5610 pg_catalog.pg_constraint.c.conname,
5611 # NOTE: avoid calling pg_get_constraintdef when not needed
5612 # to speed up the query
5613 sql.case(
5614 (
5615 pg_catalog.pg_constraint.c.oid.is_not(None),
5616 pg_catalog.pg_get_constraintdef(
5617 pg_catalog.pg_constraint.c.oid, True
5618 ),
5619 ),
5620 else_=None,
5621 ),
5622 pg_catalog.pg_description.c.description,
5623 )
5624 .select_from(pg_catalog.pg_class)
5625 .outerjoin(
5626 pg_catalog.pg_constraint,
5627 sql.and_(
5628 pg_catalog.pg_class.c.oid
5629 == pg_catalog.pg_constraint.c.conrelid,
5630 pg_catalog.pg_constraint.c.contype == "c",
5631 ),
5632 )
5633 .outerjoin(
5634 pg_catalog.pg_description,
5635 pg_catalog.pg_description.c.objoid
5636 == pg_catalog.pg_constraint.c.oid,
5637 )
5638 .order_by(
5639 pg_catalog.pg_class.c.relname,
5640 pg_catalog.pg_constraint.c.conname,
5641 )
5642 .where(self._pg_class_relkind_condition(relkinds))
5643 )
5644 query = self._pg_class_filter_scope_schema(query, schema, scope)
5645 if has_filter_names:
5646 query = query.where(
5647 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
5648 )
5649 return query
5651 def get_multi_check_constraints(
5652 self, connection, schema, filter_names, scope, kind, **kw
5653 ):
5654 has_filter_names, params = self._prepare_filter_names(filter_names)
5655 query = self._check_constraint_query(
5656 schema, has_filter_names, scope, kind
5657 )
5658 result = connection.execute(query, params)
5660 check_constraints = defaultdict(list)
5661 default = ReflectionDefaults.check_constraints
5662 for table_name, check_name, src, comment in result:
5663 # only two cases for check_name and src: both null or both defined
5664 if check_name is None and src is None:
5665 check_constraints[(schema, table_name)] = default()
5666 continue
5667 # samples:
5668 # "CHECK (((a > 1) AND (a < 5)))"
5669 # "CHECK (((a = 1) OR ((a > 2) AND (a < 5))))"
5670 # "CHECK (((a > 1) AND (a < 5))) NOT VALID"
5671 # "CHECK (some_boolean_function(a))"
5672 # "CHECK (((a\n < 1)\n OR\n (a\n >= 5))\n)"
5673 # "CHECK (a NOT NULL) NO INHERIT"
5674 # "CHECK (a NOT NULL) NO INHERIT NOT VALID"
5676 m = re.match(
5677 r"^CHECK *\((.+)\)( NO INHERIT)?( NOT VALID)?$",
5678 src,
5679 flags=re.DOTALL,
5680 )
5681 if not m:
5682 util.warn("Could not parse CHECK constraint text: %r" % src)
5683 sqltext = ""
5684 else:
5685 sqltext = util.strip_outer_parens(m.group(1))
5686 entry = {
5687 "name": check_name,
5688 "sqltext": sqltext,
5689 "comment": comment,
5690 }
5691 if m:
5692 do = {}
5693 if " NOT VALID" in m.groups():
5694 do["not_valid"] = True
5695 if " NO INHERIT" in m.groups():
5696 do["no_inherit"] = True
5697 if do:
5698 entry["dialect_options"] = do
5700 check_constraints[(schema, table_name)].append(entry)
5701 return check_constraints.items()
5703 def _pg_type_filter_schema(self, query, schema):
5704 if schema is None:
5705 query = query.where(
5706 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid),
5707 # ignore pg_catalog schema
5708 pg_catalog.pg_namespace.c.nspname != "pg_catalog",
5709 )
5710 elif schema != "*":
5711 query = query.where(pg_catalog.pg_namespace.c.nspname == schema)
5712 return query
5714 @lru_cache()
5715 def _enum_query(self, schema):
5716 lbl_agg_sq = (
5717 select(
5718 pg_catalog.pg_enum.c.enumtypid,
5719 sql.func.array_agg(
5720 aggregate_order_by(
5721 # NOTE: cast since some postgresql derivatives may
5722 # not support array_agg on the name type
5723 pg_catalog.pg_enum.c.enumlabel.cast(TEXT),
5724 pg_catalog.pg_enum.c.enumsortorder,
5725 )
5726 ).label("labels"),
5727 )
5728 .group_by(pg_catalog.pg_enum.c.enumtypid)
5729 .subquery("lbl_agg")
5730 )
5732 query = (
5733 select(
5734 pg_catalog.pg_type.c.typname.label("name"),
5735 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid).label(
5736 "visible"
5737 ),
5738 pg_catalog.pg_namespace.c.nspname.label("schema"),
5739 lbl_agg_sq.c.labels.label("labels"),
5740 )
5741 .join(
5742 pg_catalog.pg_namespace,
5743 pg_catalog.pg_namespace.c.oid
5744 == pg_catalog.pg_type.c.typnamespace,
5745 )
5746 .outerjoin(
5747 lbl_agg_sq, pg_catalog.pg_type.c.oid == lbl_agg_sq.c.enumtypid
5748 )
5749 .where(pg_catalog.pg_type.c.typtype == "e")
5750 .order_by(
5751 pg_catalog.pg_namespace.c.nspname, pg_catalog.pg_type.c.typname
5752 )
5753 )
5755 return self._pg_type_filter_schema(query, schema)
5757 @reflection.cache
5758 def _load_enums(self, connection, schema=None, **kw):
5759 if not self.supports_native_enum:
5760 return []
5762 result = connection.execute(self._enum_query(schema))
5764 enums = []
5765 for name, visible, schema, labels in result:
5766 enums.append(
5767 {
5768 "name": name,
5769 "schema": schema,
5770 "visible": visible,
5771 "labels": [] if labels is None else labels,
5772 }
5773 )
5774 return enums
5776 @lru_cache()
5777 def _domain_query(self, schema):
5778 con_sq = (
5779 select(
5780 pg_catalog.pg_constraint.c.contypid,
5781 sql.func.array_agg(
5782 pg_catalog.pg_get_constraintdef(
5783 pg_catalog.pg_constraint.c.oid, True
5784 )
5785 ).label("condefs"),
5786 sql.func.array_agg(
5787 # NOTE: cast since some postgresql derivatives may
5788 # not support array_agg on the name type
5789 pg_catalog.pg_constraint.c.conname.cast(TEXT)
5790 ).label("connames"),
5791 )
5792 # The domain this constraint is on; zero if not a domain constraint
5793 .where(pg_catalog.pg_constraint.c.contypid != 0)
5794 .group_by(pg_catalog.pg_constraint.c.contypid)
5795 .subquery("domain_constraints")
5796 )
5798 collation_namespace = pg_catalog.pg_namespace.alias(
5799 "collation_namespace"
5800 )
5802 query = (
5803 select(
5804 pg_catalog.pg_type.c.typname.label("name"),
5805 pg_catalog.format_type(
5806 pg_catalog.pg_type.c.typbasetype,
5807 pg_catalog.pg_type.c.typtypmod,
5808 ).label("attype"),
5809 (~pg_catalog.pg_type.c.typnotnull).label("nullable"),
5810 pg_catalog.pg_type.c.typdefault.label("default"),
5811 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid).label(
5812 "visible"
5813 ),
5814 pg_catalog.pg_namespace.c.nspname.label("schema"),
5815 con_sq.c.condefs,
5816 con_sq.c.connames,
5817 pg_catalog.pg_collation.c.collname,
5818 sql.case(
5819 (
5820 pg_catalog.pg_collation.c.oid.is_(None),
5821 sql.null(),
5822 ),
5823 (
5824 pg_catalog.pg_collation_is_visible(
5825 pg_catalog.pg_collation.c.oid
5826 ),
5827 sql.null(),
5828 ),
5829 else_=collation_namespace.c.nspname,
5830 ).label("collation_schema"),
5831 )
5832 .join(
5833 pg_catalog.pg_namespace,
5834 pg_catalog.pg_namespace.c.oid
5835 == pg_catalog.pg_type.c.typnamespace,
5836 )
5837 .outerjoin(
5838 pg_catalog.pg_collation,
5839 pg_catalog.pg_type.c.typcollation
5840 == pg_catalog.pg_collation.c.oid,
5841 )
5842 .outerjoin(
5843 collation_namespace,
5844 collation_namespace.c.oid
5845 == pg_catalog.pg_collation.c.collnamespace,
5846 )
5847 .outerjoin(
5848 con_sq,
5849 pg_catalog.pg_type.c.oid == con_sq.c.contypid,
5850 )
5851 .where(pg_catalog.pg_type.c.typtype == "d")
5852 .order_by(
5853 pg_catalog.pg_namespace.c.nspname, pg_catalog.pg_type.c.typname
5854 )
5855 )
5856 return self._pg_type_filter_schema(query, schema)
5858 @reflection.cache
5859 def _load_domains(self, connection, schema=None, **kw):
5860 result = connection.execute(self._domain_query(schema))
5862 domains: List[ReflectedDomain] = []
5863 for domain in result.mappings():
5864 # strip (30) from character varying(30)
5865 attype = re.search(r"([^\(]+)", domain["attype"]).group(1)
5866 constraints: List[ReflectedDomainConstraint] = []
5867 if domain["connames"]:
5868 # When a domain has multiple CHECK constraints, they will
5869 # be tested in alphabetical order by name.
5870 sorted_constraints = sorted(
5871 zip(domain["connames"], domain["condefs"]),
5872 key=lambda t: t[0],
5873 )
5874 for name, def_ in sorted_constraints:
5875 # constraint is in the form "CHECK (expression)"
5876 # or "NOT NULL". Ignore the "NOT NULL" and
5877 # remove "CHECK (" and the tailing ")".
5878 if def_.casefold().startswith("check"):
5879 check = def_[7:-1]
5880 constraints.append({"name": name, "check": check})
5881 domain_rec: ReflectedDomain = {
5882 "name": domain["name"],
5883 "schema": domain["schema"],
5884 "visible": domain["visible"],
5885 "type": attype,
5886 "nullable": domain["nullable"],
5887 "default": domain["default"],
5888 "constraints": constraints,
5889 "collation": domain["collname"],
5890 "collation_schema": domain["collation_schema"],
5891 }
5892 domains.append(domain_rec)
5894 return domains
5896 @util.memoized_property
5897 def _pg_am_query(self):
5898 return sql.select(pg_catalog.pg_am.c.oid, pg_catalog.pg_am.c.amname)
5900 @reflection.cache
5901 def _load_pg_am_dict(self, connection, **kw) -> dict[int, str]:
5902 rows = connection.execute(self._pg_am_query)
5903 return dict(rows.all())
5905 def _load_pg_am_btree_oid(self, connection):
5906 # this oid is assumed to be stable
5907 if self._pg_am_btree_oid == -1:
5908 self._pg_am_btree_oid = connection.scalar(
5909 self._pg_am_query.where(pg_catalog.pg_am.c.amname == "btree")
5910 )
5911 return self._pg_am_btree_oid
5913 @util.memoized_property
5914 def _pg_opclass_notdefault_query(self):
5915 return sql.select(
5916 pg_catalog.pg_opclass.c.oid, pg_catalog.pg_opclass.c.opcname
5917 ).where(~pg_catalog.pg_opclass.c.opcdefault)
5919 @reflection.cache
5920 def _load_pg_opclass_notdefault_dict(
5921 self, connection, **kw
5922 ) -> dict[int, str]:
5923 rows = connection.execute(self._pg_opclass_notdefault_query)
5924 return dict(rows.all())
5926 def _set_backslash_escapes(self, connection):
5927 # this method is provided as an override hook for descendant
5928 # dialects (e.g. Redshift), so removing it may break them
5929 std_string = connection.exec_driver_sql(
5930 "show standard_conforming_strings"
5931 ).scalar()
5932 self._backslash_escapes = std_string == "off"
5935class _NamedTypeLoader:
5936 """Helper class used for deferred loading of named types (enums, domains)
5937 only when needed.
5938 """
5940 def __init__(
5941 self, dialect: PGDialect, connection, kw: Dict[str, Any]
5942 ) -> None:
5943 self.dialect = dialect
5944 self.connection = connection
5945 self.kw = kw
5947 @util.memoized_property
5948 def enums(self) -> Dict[Tuple[str] | Tuple[str, str], ReflectedEnum]:
5949 # dictionary with (name, ) if default search path or (schema, name)
5950 # as keys
5951 enums = dict(
5952 (
5953 ((rec["name"],), rec)
5954 if rec["visible"]
5955 else ((rec["schema"], rec["name"]), rec)
5956 )
5957 for rec in self.dialect._load_enums(
5958 self.connection,
5959 schema="*",
5960 info_cache=self.kw.get("info_cache"),
5961 )
5962 )
5963 return enums
5965 @util.memoized_property
5966 def domains(self) -> Dict[Tuple[str] | Tuple[str, str], ReflectedDomain]:
5967 # dictionary with (name, ) if default search path or (schema, name)
5968 # as keys
5969 domains = {
5970 ((d["schema"], d["name"]) if not d["visible"] else (d["name"],)): d
5971 for d in self.dialect._load_domains(
5972 self.connection,
5973 schema="*",
5974 info_cache=self.kw.get("info_cache"),
5975 )
5976 }
5977 return domains