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_ss_cursors:
119Server Side Cursors
120-------------------
122Server-side cursor support is available for the psycopg2, asyncpg
123dialects and may also be available in others.
125Server side cursors are enabled on a per-statement basis by using the
126:paramref:`.Connection.execution_options.stream_results` connection execution
127option::
129 with engine.connect() as conn:
130 result = conn.execution_options(stream_results=True).execute(
131 text("select * from table")
132 )
134Note that some kinds of SQL statements may not be supported with
135server side cursors; generally, only SQL statements that return rows should be
136used with this option.
138.. deprecated:: 1.4 The dialect-level server_side_cursors flag is deprecated
139 and will be removed in a future release. Please use the
140 :paramref:`_engine.Connection.stream_results` execution option for
141 unbuffered cursor support.
143.. seealso::
145 :ref:`engine_stream_results`
147.. _postgresql_isolation_level:
149Transaction Isolation Level
150---------------------------
152Most SQLAlchemy dialects support setting of transaction isolation level
153using the :paramref:`_sa.create_engine.isolation_level` parameter
154at the :func:`_sa.create_engine` level, and at the :class:`_engine.Connection`
155level via the :paramref:`.Connection.execution_options.isolation_level`
156parameter.
158For PostgreSQL dialects, this feature works either by making use of the
159DBAPI-specific features, such as psycopg2's isolation level flags which will
160embed the isolation level setting inline with the ``"BEGIN"`` statement, or for
161DBAPIs with no direct support by emitting ``SET SESSION CHARACTERISTICS AS
162TRANSACTION ISOLATION LEVEL <level>`` ahead of the ``"BEGIN"`` statement
163emitted by the DBAPI. For the special AUTOCOMMIT isolation level,
164DBAPI-specific techniques are used which is typically an ``.autocommit``
165flag on the DBAPI connection object.
167To set isolation level using :func:`_sa.create_engine`::
169 engine = create_engine(
170 "postgresql+pg8000://scott:tiger@localhost/test",
171 isolation_level="REPEATABLE READ",
172 )
174To set using per-connection execution options::
176 with engine.connect() as conn:
177 conn = conn.execution_options(isolation_level="REPEATABLE READ")
178 with conn.begin():
179 ... # work with transaction
181There are also more options for isolation level configurations, such as
182"sub-engine" objects linked to a main :class:`_engine.Engine` which each apply
183different isolation level settings. See the discussion at
184:ref:`dbapi_autocommit` for background.
186Valid values for ``isolation_level`` on most PostgreSQL dialects include:
188* ``READ COMMITTED``
189* ``READ UNCOMMITTED``
190* ``REPEATABLE READ``
191* ``SERIALIZABLE``
192* ``AUTOCOMMIT``
194.. seealso::
196 :ref:`dbapi_autocommit`
198 :ref:`postgresql_readonly_deferrable`
200 :ref:`psycopg2_isolation_level`
202 :ref:`pg8000_isolation_level`
204.. _postgresql_readonly_deferrable:
206Setting READ ONLY / DEFERRABLE
207------------------------------
209Most PostgreSQL dialects support setting the "READ ONLY" and "DEFERRABLE"
210characteristics of the transaction, which is in addition to the isolation level
211setting. These two attributes can be established either in conjunction with or
212independently of the isolation level by passing the ``postgresql_readonly`` and
213``postgresql_deferrable`` flags with
214:meth:`_engine.Connection.execution_options`. The example below illustrates
215passing the ``"SERIALIZABLE"`` isolation level at the same time as setting
216"READ ONLY" and "DEFERRABLE"::
218 with engine.connect() as conn:
219 conn = conn.execution_options(
220 isolation_level="SERIALIZABLE",
221 postgresql_readonly=True,
222 postgresql_deferrable=True,
223 )
224 with conn.begin():
225 ... # work with transaction
227Note that some DBAPIs such as asyncpg only support "readonly" with
228SERIALIZABLE isolation.
230.. versionadded:: 1.4 added support for the ``postgresql_readonly``
231 and ``postgresql_deferrable`` execution options.
233.. _postgresql_reset_on_return:
235Temporary Table / Resource Reset for Connection Pooling
236-------------------------------------------------------
238The :class:`.QueuePool` connection pool implementation used
239by the SQLAlchemy :class:`.Engine` object includes
240:ref:`reset on return <pool_reset_on_return>` behavior that will invoke
241the DBAPI ``.rollback()`` method when connections are returned to the pool.
242While this rollback will clear out the immediate state used by the previous
243transaction, it does not cover a wider range of session-level state, including
244temporary tables as well as other server state such as prepared statement
245handles and statement caches. The PostgreSQL database includes a variety
246of commands which may be used to reset this state, including
247``DISCARD``, ``RESET``, ``DEALLOCATE``, and ``UNLISTEN``.
250To install
251one or more of these commands as the means of performing reset-on-return,
252the :meth:`.PoolEvents.reset` event hook may be used, as demonstrated
253in the example below. The implementation
254will end transactions in progress as well as discard temporary tables
255using the ``CLOSE``, ``RESET`` and ``DISCARD`` commands; see the PostgreSQL
256documentation for background on what each of these statements do.
258The :paramref:`_sa.create_engine.pool_reset_on_return` parameter
259is set to ``None`` so that the custom scheme can replace the default behavior
260completely. The custom hook implementation calls ``.rollback()`` in any case,
261as it's usually important that the DBAPI's own tracking of commit/rollback
262will remain consistent with the state of the transaction::
265 from sqlalchemy import create_engine
266 from sqlalchemy import event
268 postgresql_engine = create_engine(
269 "postgresql+psycopg2://scott:tiger@hostname/dbname",
270 # disable default reset-on-return scheme
271 pool_reset_on_return=None,
272 )
275 @event.listens_for(postgresql_engine, "reset")
276 def _reset_postgresql(dbapi_connection, connection_record, reset_state):
277 if not reset_state.terminate_only:
278 dbapi_connection.execute("CLOSE ALL")
279 dbapi_connection.execute("RESET ALL")
280 dbapi_connection.execute("DISCARD TEMP")
282 # so that the DBAPI itself knows that the connection has been
283 # reset
284 dbapi_connection.rollback()
286.. versionchanged:: 2.0.0b3 Added additional state arguments to
287 the :meth:`.PoolEvents.reset` event and additionally ensured the event
288 is invoked for all "reset" occurrences, so that it's appropriate
289 as a place for custom "reset" handlers. Previous schemes which
290 use the :meth:`.PoolEvents.checkin` handler remain usable as well.
292.. seealso::
294 :ref:`pool_reset_on_return` - in the :ref:`pooling_toplevel` documentation
296.. _postgresql_alternate_search_path:
298Setting Alternate Search Paths on Connect
299------------------------------------------
301The PostgreSQL ``search_path`` variable refers to the list of schema names
302that will be implicitly referenced when a particular table or other
303object is referenced in a SQL statement. As detailed in the next section
304:ref:`postgresql_schema_reflection`, SQLAlchemy is generally organized around
305the concept of keeping this variable at its default value of ``public``,
306however, in order to have it set to any arbitrary name or names when connections
307are used automatically, the "SET SESSION search_path" command may be invoked
308for all connections in a pool using the following event handler, as discussed
309at :ref:`schema_set_default_connections`::
311 from sqlalchemy import event
312 from sqlalchemy import create_engine
314 engine = create_engine("postgresql+psycopg2://scott:tiger@host/dbname")
317 @event.listens_for(engine, "connect", insert=True)
318 def set_search_path(dbapi_connection, connection_record):
319 existing_autocommit = dbapi_connection.autocommit
320 dbapi_connection.autocommit = True
321 cursor = dbapi_connection.cursor()
322 cursor.execute("SET SESSION search_path='%s'" % schema_name)
323 cursor.close()
324 dbapi_connection.autocommit = existing_autocommit
326The reason the recipe is complicated by use of the ``.autocommit`` DBAPI
327attribute is so that when the ``SET SESSION search_path`` directive is invoked,
328it is invoked outside of the scope of any transaction and therefore will not
329be reverted when the DBAPI connection has a rollback.
331.. seealso::
333 :ref:`schema_set_default_connections` - in the :ref:`metadata_toplevel` documentation
335.. _postgresql_schema_reflection:
337Remote-Schema Table Introspection and PostgreSQL search_path
338------------------------------------------------------------
340.. admonition:: Section Best Practices Summarized
342 keep the ``search_path`` variable set to its default of ``public``, without
343 any other schema names. Ensure the username used to connect **does not**
344 match remote schemas, or ensure the ``"$user"`` token is **removed** from
345 ``search_path``. For other schema names, name these explicitly
346 within :class:`_schema.Table` definitions. Alternatively, the
347 ``postgresql_ignore_search_path`` option will cause all reflected
348 :class:`_schema.Table` objects to have a :attr:`_schema.Table.schema`
349 attribute set up.
351The PostgreSQL dialect can reflect tables from any schema, as outlined in
352:ref:`metadata_reflection_schemas`.
354In all cases, the first thing SQLAlchemy does when reflecting tables is
355to **determine the default schema for the current database connection**.
356It does this using the PostgreSQL ``current_schema()``
357function, illustated below using a PostgreSQL client session (i.e. using
358the ``psql`` tool):
360.. sourcecode:: sql
362 test=> select current_schema();
363 current_schema
364 ----------------
365 public
366 (1 row)
368Above we see that on a plain install of PostgreSQL, the default schema name
369is the name ``public``.
371However, if your database username **matches the name of a schema**, PostgreSQL's
372default is to then **use that name as the default schema**. Below, we log in
373using the username ``scott``. When we create a schema named ``scott``, **it
374implicitly changes the default schema**:
376.. sourcecode:: sql
378 test=> select current_schema();
379 current_schema
380 ----------------
381 public
382 (1 row)
384 test=> create schema scott;
385 CREATE SCHEMA
386 test=> select current_schema();
387 current_schema
388 ----------------
389 scott
390 (1 row)
392The behavior of ``current_schema()`` is derived from the
393`PostgreSQL search path
394<https://www.postgresql.org/docs/current/static/ddl-schemas.html#DDL-SCHEMAS-PATH>`_
395variable ``search_path``, which in modern PostgreSQL versions defaults to this:
397.. sourcecode:: sql
399 test=> show search_path;
400 search_path
401 -----------------
402 "$user", public
403 (1 row)
405Where above, the ``"$user"`` variable will inject the current username as the
406default schema, if one exists. Otherwise, ``public`` is used.
408When a :class:`_schema.Table` object is reflected, if it is present in the
409schema indicated by the ``current_schema()`` function, **the schema name assigned
410to the ".schema" attribute of the Table is the Python "None" value**. Otherwise, the
411".schema" attribute will be assigned the string name of that schema.
413With regards to tables which these :class:`_schema.Table`
414objects refer to via foreign key constraint, a decision must be made as to how
415the ``.schema`` is represented in those remote tables, in the case where that
416remote schema name is also a member of the current ``search_path``.
418By default, the PostgreSQL dialect mimics the behavior encouraged by
419PostgreSQL's own ``pg_get_constraintdef()`` builtin procedure. This function
420returns a sample definition for a particular foreign key constraint,
421omitting the referenced schema name from that definition when the name is
422also in the PostgreSQL schema search path. The interaction below
423illustrates this behavior:
425.. sourcecode:: sql
427 test=> CREATE TABLE test_schema.referred(id INTEGER PRIMARY KEY);
428 CREATE TABLE
429 test=> CREATE TABLE referring(
430 test(> id INTEGER PRIMARY KEY,
431 test(> referred_id INTEGER REFERENCES test_schema.referred(id));
432 CREATE TABLE
433 test=> SET search_path TO public, test_schema;
434 test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
435 test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n
436 test-> ON n.oid = c.relnamespace
437 test-> JOIN pg_catalog.pg_constraint r ON c.oid = r.conrelid
438 test-> WHERE c.relname='referring' AND r.contype = 'f'
439 test-> ;
440 pg_get_constraintdef
441 ---------------------------------------------------
442 FOREIGN KEY (referred_id) REFERENCES referred(id)
443 (1 row)
445Above, we created a table ``referred`` as a member of the remote schema
446``test_schema``, however when we added ``test_schema`` to the
447PG ``search_path`` and then asked ``pg_get_constraintdef()`` for the
448``FOREIGN KEY`` syntax, ``test_schema`` was not included in the output of
449the function.
451On the other hand, if we set the search path back to the typical default
452of ``public``:
454.. sourcecode:: sql
456 test=> SET search_path TO public;
457 SET
459The same query against ``pg_get_constraintdef()`` now returns the fully
460schema-qualified name for us:
462.. sourcecode:: sql
464 test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
465 test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n
466 test-> ON n.oid = c.relnamespace
467 test-> JOIN pg_catalog.pg_constraint r ON c.oid = r.conrelid
468 test-> WHERE c.relname='referring' AND r.contype = 'f';
469 pg_get_constraintdef
470 ---------------------------------------------------------------
471 FOREIGN KEY (referred_id) REFERENCES test_schema.referred(id)
472 (1 row)
474SQLAlchemy will by default use the return value of ``pg_get_constraintdef()``
475in order to determine the remote schema name. That is, if our ``search_path``
476were set to include ``test_schema``, and we invoked a table
477reflection process as follows::
479 >>> from sqlalchemy import Table, MetaData, create_engine, text
480 >>> engine = create_engine("postgresql+psycopg2://scott:tiger@localhost/test")
481 >>> with engine.connect() as conn:
482 ... conn.execute(text("SET search_path TO test_schema, public"))
483 ... metadata_obj = MetaData()
484 ... referring = Table("referring", metadata_obj, autoload_with=conn)
485 <sqlalchemy.engine.result.CursorResult object at 0x101612ed0>
487The above process would deliver to the :attr:`_schema.MetaData.tables`
488collection
489``referred`` table named **without** the schema::
491 >>> metadata_obj.tables["referred"].schema is None
492 True
494To alter the behavior of reflection such that the referred schema is
495maintained regardless of the ``search_path`` setting, use the
496``postgresql_ignore_search_path`` option, which can be specified as a
497dialect-specific argument to both :class:`_schema.Table` as well as
498:meth:`_schema.MetaData.reflect`::
500 >>> with engine.connect() as conn:
501 ... conn.execute(text("SET search_path TO test_schema, public"))
502 ... metadata_obj = MetaData()
503 ... referring = Table(
504 ... "referring",
505 ... metadata_obj,
506 ... autoload_with=conn,
507 ... postgresql_ignore_search_path=True,
508 ... )
509 <sqlalchemy.engine.result.CursorResult object at 0x1016126d0>
511We will now have ``test_schema.referred`` stored as schema-qualified::
513 >>> metadata_obj.tables["test_schema.referred"].schema
514 'test_schema'
516.. sidebar:: Best Practices for PostgreSQL Schema reflection
518 The description of PostgreSQL schema reflection behavior is complex, and
519 is the product of many years of dealing with widely varied use cases and
520 user preferences. But in fact, there's no need to understand any of it if
521 you just stick to the simplest use pattern: leave the ``search_path`` set
522 to its default of ``public`` only, never refer to the name ``public`` as
523 an explicit schema name otherwise, and refer to all other schema names
524 explicitly when building up a :class:`_schema.Table` object. The options
525 described here are only for those users who can't, or prefer not to, stay
526 within these guidelines.
528.. seealso::
530 :ref:`reflection_schema_qualified_interaction` - discussion of the issue
531 from a backend-agnostic perspective
533 `The Schema Search Path
534 <https://www.postgresql.org/docs/current/static/ddl-schemas.html#DDL-SCHEMAS-PATH>`_
535 - on the PostgreSQL website.
537INSERT/UPDATE...RETURNING
538-------------------------
540The dialect supports PG 8.2's ``INSERT..RETURNING``, ``UPDATE..RETURNING`` and
541``DELETE..RETURNING`` syntaxes. ``INSERT..RETURNING`` is used by default
542for single-row INSERT statements in order to fetch newly generated
543primary key identifiers. To specify an explicit ``RETURNING`` clause,
544use the :meth:`._UpdateBase.returning` method on a per-statement basis::
546 # INSERT..RETURNING
547 result = (
548 table.insert().returning(table.c.col1, table.c.col2).values(name="foo")
549 )
550 print(result.fetchall())
552 # UPDATE..RETURNING
553 result = (
554 table.update()
555 .returning(table.c.col1, table.c.col2)
556 .where(table.c.name == "foo")
557 .values(name="bar")
558 )
559 print(result.fetchall())
561 # DELETE..RETURNING
562 result = (
563 table.delete()
564 .returning(table.c.col1, table.c.col2)
565 .where(table.c.name == "foo")
566 )
567 print(result.fetchall())
569.. _postgresql_insert_on_conflict:
571INSERT...ON CONFLICT (Upsert)
572------------------------------
574Starting with version 9.5, PostgreSQL allows "upserts" (update or insert) of
575rows into a table via the ``ON CONFLICT`` clause of the ``INSERT`` statement. A
576candidate row will only be inserted if that row does not violate any unique
577constraints. In the case of a unique constraint violation, a secondary action
578can occur which can be either "DO UPDATE", indicating that the data in the
579target row should be updated, or "DO NOTHING", which indicates to silently skip
580this row.
582Conflicts are determined using existing unique constraints and indexes. These
583constraints may be identified either using their name as stated in DDL,
584or they may be inferred by stating the columns and conditions that comprise
585the indexes.
587SQLAlchemy provides ``ON CONFLICT`` support via the PostgreSQL-specific
588:func:`_postgresql.insert()` function, which provides
589the generative methods :meth:`_postgresql.Insert.on_conflict_do_update`
590and :meth:`~.postgresql.Insert.on_conflict_do_nothing`:
592.. sourcecode:: pycon+sql
594 >>> from sqlalchemy.dialects.postgresql import insert
595 >>> insert_stmt = insert(my_table).values(
596 ... id="some_existing_id", data="inserted value"
597 ... )
598 >>> do_nothing_stmt = insert_stmt.on_conflict_do_nothing(index_elements=["id"])
599 >>> print(do_nothing_stmt)
600 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
601 ON CONFLICT (id) DO NOTHING
602 {stop}
604 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
605 ... constraint="pk_my_table", set_=dict(data="updated value")
606 ... )
607 >>> print(do_update_stmt)
608 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
609 ON CONFLICT ON CONSTRAINT pk_my_table DO UPDATE SET data = %(param_1)s
611.. seealso::
613 `INSERT .. ON CONFLICT
614 <https://www.postgresql.org/docs/current/static/sql-insert.html#SQL-ON-CONFLICT>`_
615 - in the PostgreSQL documentation.
617Specifying the Target
618^^^^^^^^^^^^^^^^^^^^^
620Both methods supply the "target" of the conflict using either the
621named constraint or by column inference:
623* The :paramref:`_postgresql.Insert.on_conflict_do_update.index_elements` argument
624 specifies a sequence containing string column names, :class:`_schema.Column`
625 objects, and/or SQL expression elements, which would identify a unique
626 index:
628 .. sourcecode:: pycon+sql
630 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
631 ... index_elements=["id"], set_=dict(data="updated value")
632 ... )
633 >>> print(do_update_stmt)
634 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
635 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
636 {stop}
638 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
639 ... index_elements=[my_table.c.id], set_=dict(data="updated value")
640 ... )
641 >>> print(do_update_stmt)
642 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
643 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
645* When using :paramref:`_postgresql.Insert.on_conflict_do_update.index_elements` to
646 infer an index, a partial index can be inferred by also specifying the
647 use the :paramref:`_postgresql.Insert.on_conflict_do_update.index_where` parameter:
649 .. sourcecode:: pycon+sql
651 >>> stmt = insert(my_table).values(user_email="a@b.com", data="inserted data")
652 >>> stmt = stmt.on_conflict_do_update(
653 ... index_elements=[my_table.c.user_email],
654 ... index_where=my_table.c.user_email.like("%@gmail.com"),
655 ... set_=dict(data=stmt.excluded.data),
656 ... )
657 >>> print(stmt)
658 {printsql}INSERT INTO my_table (data, user_email)
659 VALUES (%(data)s, %(user_email)s) ON CONFLICT (user_email)
660 WHERE user_email LIKE %(user_email_1)s DO UPDATE SET data = excluded.data
662* The :paramref:`_postgresql.Insert.on_conflict_do_update.constraint` argument is
663 used to specify an index directly rather than inferring it. This can be
664 the name of a UNIQUE constraint, a PRIMARY KEY constraint, or an INDEX:
666 .. sourcecode:: pycon+sql
668 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
669 ... constraint="my_table_idx_1", set_=dict(data="updated value")
670 ... )
671 >>> print(do_update_stmt)
672 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
673 ON CONFLICT ON CONSTRAINT my_table_idx_1 DO UPDATE SET data = %(param_1)s
674 {stop}
676 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
677 ... constraint="my_table_pk", set_=dict(data="updated value")
678 ... )
679 >>> print(do_update_stmt)
680 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
681 ON CONFLICT ON CONSTRAINT my_table_pk DO UPDATE SET data = %(param_1)s
682 {stop}
684* The :paramref:`_postgresql.Insert.on_conflict_do_update.constraint` argument may
685 also refer to a SQLAlchemy construct representing a constraint,
686 e.g. :class:`.UniqueConstraint`, :class:`.PrimaryKeyConstraint`,
687 :class:`.Index`, or :class:`.ExcludeConstraint`. In this use,
688 if the constraint has a name, it is used directly. Otherwise, if the
689 constraint is unnamed, then inference will be used, where the expressions
690 and optional WHERE clause of the constraint will be spelled out in the
691 construct. This use is especially convenient
692 to refer to the named or unnamed primary key of a :class:`_schema.Table`
693 using the
694 :attr:`_schema.Table.primary_key` attribute:
696 .. sourcecode:: pycon+sql
698 >>> do_update_stmt = insert_stmt.on_conflict_do_update(
699 ... constraint=my_table.primary_key, set_=dict(data="updated value")
700 ... )
701 >>> print(do_update_stmt)
702 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
703 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s
705The SET Clause
706^^^^^^^^^^^^^^^
708``ON CONFLICT...DO UPDATE`` is used to perform an update of the already
709existing row, using any combination of new values as well as values
710from the proposed insertion. These values are specified using the
711:paramref:`_postgresql.Insert.on_conflict_do_update.set_` parameter. This
712parameter accepts a dictionary which consists of direct values
713for UPDATE:
715.. sourcecode:: pycon+sql
717 >>> stmt = insert(my_table).values(id="some_id", data="inserted value")
718 >>> do_update_stmt = stmt.on_conflict_do_update(
719 ... index_elements=["id"], 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 (id) DO UPDATE SET data = %(param_1)s
725.. warning::
727 The :meth:`_expression.Insert.on_conflict_do_update`
728 method does **not** take into
729 account Python-side default UPDATE values or generation functions, e.g.
730 those specified using :paramref:`_schema.Column.onupdate`.
731 These values will not be exercised for an ON CONFLICT style of UPDATE,
732 unless they are manually specified in the
733 :paramref:`_postgresql.Insert.on_conflict_do_update.set_` dictionary.
735Updating using the Excluded INSERT Values
736^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
738In order to refer to the proposed insertion row, the special alias
739:attr:`~.postgresql.Insert.excluded` is available as an attribute on
740the :class:`_postgresql.Insert` object; this object is a
741:class:`_expression.ColumnCollection`
742which alias contains all columns of the target
743table:
745.. sourcecode:: pycon+sql
747 >>> stmt = insert(my_table).values(
748 ... id="some_id", data="inserted value", author="jlh"
749 ... )
750 >>> do_update_stmt = stmt.on_conflict_do_update(
751 ... index_elements=["id"],
752 ... set_=dict(data="updated value", author=stmt.excluded.author),
753 ... )
754 >>> print(do_update_stmt)
755 {printsql}INSERT INTO my_table (id, data, author)
756 VALUES (%(id)s, %(data)s, %(author)s)
757 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author
759Additional WHERE Criteria
760^^^^^^^^^^^^^^^^^^^^^^^^^
762The :meth:`_expression.Insert.on_conflict_do_update` method also accepts
763a WHERE clause using the :paramref:`_postgresql.Insert.on_conflict_do_update.where`
764parameter, which will limit those rows which receive an UPDATE:
766.. sourcecode:: pycon+sql
768 >>> stmt = insert(my_table).values(
769 ... id="some_id", data="inserted value", author="jlh"
770 ... )
771 >>> on_update_stmt = stmt.on_conflict_do_update(
772 ... index_elements=["id"],
773 ... set_=dict(data="updated value", author=stmt.excluded.author),
774 ... where=(my_table.c.status == 2),
775 ... )
776 >>> print(on_update_stmt)
777 {printsql}INSERT INTO my_table (id, data, author)
778 VALUES (%(id)s, %(data)s, %(author)s)
779 ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author
780 WHERE my_table.status = %(status_1)s
782Skipping Rows with DO NOTHING
783^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
785``ON CONFLICT`` may be used to skip inserting a row entirely
786if any conflict with a unique or exclusion constraint occurs; below
787this is illustrated using the
788:meth:`~.postgresql.Insert.on_conflict_do_nothing` method:
790.. sourcecode:: pycon+sql
792 >>> stmt = insert(my_table).values(id="some_id", data="inserted value")
793 >>> stmt = stmt.on_conflict_do_nothing(index_elements=["id"])
794 >>> print(stmt)
795 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
796 ON CONFLICT (id) DO NOTHING
798If ``DO NOTHING`` is used without specifying any columns or constraint,
799it has the effect of skipping the INSERT for any unique or exclusion
800constraint violation which occurs:
802.. sourcecode:: pycon+sql
804 >>> stmt = insert(my_table).values(id="some_id", data="inserted value")
805 >>> stmt = stmt.on_conflict_do_nothing()
806 >>> print(stmt)
807 {printsql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
808 ON CONFLICT DO NOTHING
810.. _postgresql_match:
812Full Text Search
813----------------
815PostgreSQL's full text search system is available through the use of the
816:data:`.func` namespace, combined with the use of custom operators
817via the :meth:`.Operators.bool_op` method. For simple cases with some
818degree of cross-backend compatibility, the :meth:`.Operators.match` operator
819may also be used.
821.. _postgresql_simple_match:
823Simple plain text matching with ``match()``
824^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
826The :meth:`.Operators.match` operator provides for cross-compatible simple
827text matching. For the PostgreSQL backend, it's hardcoded to generate
828an expression using the ``@@`` operator in conjunction with the
829``plainto_tsquery()`` PostgreSQL function.
831On the PostgreSQL dialect, an expression like the following::
833 select(sometable.c.text.match("search string"))
835would emit to the database:
837.. sourcecode:: sql
839 SELECT text @@ plainto_tsquery('search string') FROM table
841Above, passing a plain string to :meth:`.Operators.match` will automatically
842make use of ``plainto_tsquery()`` to specify the type of tsquery. This
843establishes basic database cross-compatibility for :meth:`.Operators.match`
844with other backends.
846.. versionchanged:: 2.0 The default tsquery generation function used by the
847 PostgreSQL dialect with :meth:`.Operators.match` is ``plainto_tsquery()``.
849 To render exactly what was rendered in 1.4, use the following form::
851 from sqlalchemy import func
853 select(sometable.c.text.bool_op("@@")(func.to_tsquery("search string")))
855 Which would emit:
857 .. sourcecode:: sql
859 SELECT text @@ to_tsquery('search string') FROM table
861Using PostgreSQL full text functions and operators directly
862^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
864Text search operations beyond the simple use of :meth:`.Operators.match`
865may make use of the :data:`.func` namespace to generate PostgreSQL full-text
866functions, in combination with :meth:`.Operators.bool_op` to generate
867any boolean operator.
869For example, the query::
871 select(func.to_tsquery("cat").bool_op("@>")(func.to_tsquery("cat & rat")))
873would generate:
875.. sourcecode:: sql
877 SELECT to_tsquery('cat') @> to_tsquery('cat & rat')
880The :class:`_postgresql.TSVECTOR` type can provide for explicit CAST::
882 from sqlalchemy.dialects.postgresql import TSVECTOR
883 from sqlalchemy import select, cast
885 select(cast("some text", TSVECTOR))
887produces a statement equivalent to:
889.. sourcecode:: sql
891 SELECT CAST('some text' AS TSVECTOR) AS anon_1
893The ``func`` namespace is augmented by the PostgreSQL dialect to set up
894correct argument and return types for most full text search functions.
895These functions are used automatically by the :attr:`_sql.func` namespace
896assuming the ``sqlalchemy.dialects.postgresql`` package has been imported,
897or :func:`_sa.create_engine` has been invoked using a ``postgresql``
898dialect. These functions are documented at:
900* :class:`_postgresql.to_tsvector`
901* :class:`_postgresql.to_tsquery`
902* :class:`_postgresql.plainto_tsquery`
903* :class:`_postgresql.phraseto_tsquery`
904* :class:`_postgresql.websearch_to_tsquery`
905* :class:`_postgresql.ts_headline`
907Specifying the "regconfig" with ``match()`` or custom operators
908^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
910PostgreSQL's ``plainto_tsquery()`` function accepts an optional
911"regconfig" argument that is used to instruct PostgreSQL to use a
912particular pre-computed GIN or GiST index in order to perform the search.
913When using :meth:`.Operators.match`, this additional parameter may be
914specified using the ``postgresql_regconfig`` parameter, such as::
916 select(mytable.c.id).where(
917 mytable.c.title.match("somestring", postgresql_regconfig="english")
918 )
920Which would emit:
922.. sourcecode:: sql
924 SELECT mytable.id FROM mytable
925 WHERE mytable.title @@ plainto_tsquery('english', 'somestring')
927When using other PostgreSQL search functions with :data:`.func`, the
928"regconfig" parameter may be passed directly as the initial argument::
930 select(mytable.c.id).where(
931 func.to_tsvector("english", mytable.c.title).bool_op("@@")(
932 func.to_tsquery("english", "somestring")
933 )
934 )
936produces a statement equivalent to:
938.. sourcecode:: sql
940 SELECT mytable.id FROM mytable
941 WHERE to_tsvector('english', mytable.title) @@
942 to_tsquery('english', 'somestring')
944It is recommended that you use the ``EXPLAIN ANALYZE...`` tool from
945PostgreSQL to ensure that you are generating queries with SQLAlchemy that
946take full advantage of any indexes you may have created for full text search.
948.. seealso::
950 `Full Text Search <https://www.postgresql.org/docs/current/textsearch-controls.html>`_ - in the PostgreSQL documentation
953FROM ONLY ...
954-------------
956The dialect supports PostgreSQL's ONLY keyword for targeting only a particular
957table in an inheritance hierarchy. This can be used to produce the
958``SELECT ... FROM ONLY``, ``UPDATE ONLY ...``, and ``DELETE FROM ONLY ...``
959syntaxes. It uses SQLAlchemy's hints mechanism::
961 # SELECT ... FROM ONLY ...
962 result = table.select().with_hint(table, "ONLY", "postgresql")
963 print(result.fetchall())
965 # UPDATE ONLY ...
966 table.update(values=dict(foo="bar")).with_hint(
967 "ONLY", dialect_name="postgresql"
968 )
970 # DELETE FROM ONLY ...
971 table.delete().with_hint("ONLY", dialect_name="postgresql")
973.. _postgresql_indexes:
975PostgreSQL-Specific Index Options
976---------------------------------
978Several extensions to the :class:`.Index` construct are available, specific
979to the PostgreSQL dialect.
981.. _postgresql_covering_indexes:
983Covering Indexes
984^^^^^^^^^^^^^^^^
986A covering index includes additional columns that are not part of the index key
987but are stored in the index, allowing PostgreSQL to satisfy queries using only
988the index without accessing the table (an "index-only scan"). This is
989indicated on the index using the ``INCLUDE`` clause. The
990``postgresql_include`` option for :class:`.Index` (as well as
991:class:`.UniqueConstraint`) renders ``INCLUDE(colname)`` for the given string
992names::
994 Index("my_index", table.c.x, postgresql_include=["y"])
996would render the index as ``CREATE INDEX my_index ON table (x) INCLUDE (y)``
998Note that this feature requires PostgreSQL 11 or later.
1000.. seealso::
1002 :ref:`postgresql_constraint_options_include` - the same feature implemented
1003 for :class:`.UniqueConstraint`
1005.. versionadded:: 1.4 - support for covering indexes with :class:`.Index`.
1006 support for :class:`.UniqueConstraint` was in 2.0.41
1008.. _postgresql_partial_indexes:
1010Partial Indexes
1011^^^^^^^^^^^^^^^
1013Partial indexes add criterion to the index definition so that the index is
1014applied to a subset of rows. These can be specified on :class:`.Index`
1015using the ``postgresql_where`` keyword argument::
1017 Index("my_index", my_table.c.id, postgresql_where=my_table.c.value > 10)
1019.. _postgresql_operator_classes:
1021Operator Classes
1022^^^^^^^^^^^^^^^^
1024PostgreSQL allows the specification of an *operator class* for each column of
1025an index (see
1026https://www.postgresql.org/docs/current/interactive/indexes-opclass.html).
1027The :class:`.Index` construct allows these to be specified via the
1028``postgresql_ops`` keyword argument::
1030 Index(
1031 "my_index",
1032 my_table.c.id,
1033 my_table.c.data,
1034 postgresql_ops={"data": "text_pattern_ops", "id": "int4_ops"},
1035 )
1037Note that the keys in the ``postgresql_ops`` dictionaries are the
1038"key" name of the :class:`_schema.Column`, i.e. the name used to access it from
1039the ``.c`` collection of :class:`_schema.Table`, which can be configured to be
1040different than the actual name of the column as expressed in the database.
1042If ``postgresql_ops`` is to be used against a complex SQL expression such
1043as a function call, then to apply to the column it must be given a label
1044that is identified in the dictionary by name, e.g.::
1046 Index(
1047 "my_index",
1048 my_table.c.id,
1049 func.lower(my_table.c.data).label("data_lower"),
1050 postgresql_ops={"data_lower": "text_pattern_ops", "id": "int4_ops"},
1051 )
1053Operator classes are also supported by the
1054:class:`_postgresql.ExcludeConstraint` construct using the
1055:paramref:`_postgresql.ExcludeConstraint.ops` parameter. See that parameter for
1056details.
1058.. versionadded:: 1.3.21 added support for operator classes with
1059 :class:`_postgresql.ExcludeConstraint`.
1062Index Types
1063^^^^^^^^^^^
1065PostgreSQL provides several index types: B-Tree, Hash, GiST, and GIN, as well
1066as the ability for users to create their own (see
1067https://www.postgresql.org/docs/current/static/indexes-types.html). These can be
1068specified on :class:`.Index` using the ``postgresql_using`` keyword argument::
1070 Index("my_index", my_table.c.data, postgresql_using="gin")
1072The value passed to the keyword argument will be simply passed through to the
1073underlying CREATE INDEX command, so it *must* be a valid index type for your
1074version of PostgreSQL.
1076.. _postgresql_index_storage:
1078Index Storage Parameters
1079^^^^^^^^^^^^^^^^^^^^^^^^
1081PostgreSQL allows storage parameters to be set on indexes. The storage
1082parameters available depend on the index method used by the index. Storage
1083parameters can be specified on :class:`.Index` using the ``postgresql_with``
1084keyword argument::
1086 Index("my_index", my_table.c.data, postgresql_with={"fillfactor": 50})
1088PostgreSQL allows to define the tablespace in which to create the index.
1089The tablespace can be specified on :class:`.Index` using the
1090``postgresql_tablespace`` keyword argument::
1092 Index("my_index", my_table.c.data, postgresql_tablespace="my_tablespace")
1094Note that the same option is available on :class:`_schema.Table` as well.
1096.. _postgresql_index_concurrently:
1098Indexes with CONCURRENTLY
1099^^^^^^^^^^^^^^^^^^^^^^^^^
1101The PostgreSQL index option CONCURRENTLY is supported by passing the
1102flag ``postgresql_concurrently`` to the :class:`.Index` construct::
1104 tbl = Table("testtbl", m, Column("data", Integer))
1106 idx1 = Index("test_idx1", tbl.c.data, postgresql_concurrently=True)
1108The above index construct will render DDL for CREATE INDEX, assuming
1109PostgreSQL 8.2 or higher is detected or for a connection-less dialect, as:
1111.. sourcecode:: sql
1113 CREATE INDEX CONCURRENTLY test_idx1 ON testtbl (data)
1115For DROP INDEX, assuming PostgreSQL 9.2 or higher is detected or for
1116a connection-less dialect, it will emit:
1118.. sourcecode:: sql
1120 DROP INDEX CONCURRENTLY test_idx1
1122When using CONCURRENTLY, the PostgreSQL database requires that the statement
1123be invoked outside of a transaction block. The Python DBAPI enforces that
1124even for a single statement, a transaction is present, so to use this
1125construct, the DBAPI's "autocommit" mode must be used::
1127 metadata = MetaData()
1128 table = Table("foo", metadata, Column("id", String))
1129 index = Index("foo_idx", table.c.id, postgresql_concurrently=True)
1131 with engine.connect() as conn:
1132 with conn.execution_options(isolation_level="AUTOCOMMIT"):
1133 table.create(conn)
1135.. seealso::
1137 :ref:`postgresql_isolation_level`
1139.. _postgresql_index_reflection:
1141PostgreSQL Index Reflection
1142---------------------------
1144The PostgreSQL database creates a UNIQUE INDEX implicitly whenever the
1145UNIQUE CONSTRAINT construct is used. When inspecting a table using
1146:class:`_reflection.Inspector`, the :meth:`_reflection.Inspector.get_indexes`
1147and the :meth:`_reflection.Inspector.get_unique_constraints`
1148will report on these
1149two constructs distinctly; in the case of the index, the key
1150``duplicates_constraint`` will be present in the index entry if it is
1151detected as mirroring a constraint. When performing reflection using
1152``Table(..., autoload_with=engine)``, the UNIQUE INDEX is **not** returned
1153in :attr:`_schema.Table.indexes` when it is detected as mirroring a
1154:class:`.UniqueConstraint` in the :attr:`_schema.Table.constraints` collection
1155.
1157Special Reflection Options
1158--------------------------
1160The :class:`_reflection.Inspector`
1161used for the PostgreSQL backend is an instance
1162of :class:`.PGInspector`, which offers additional methods::
1164 from sqlalchemy import create_engine, inspect
1166 engine = create_engine("postgresql+psycopg2://localhost/test")
1167 insp = inspect(engine) # will be a PGInspector
1169 print(insp.get_enums())
1171.. autoclass:: PGInspector
1172 :members:
1174.. _postgresql_table_options:
1176PostgreSQL Table Options
1177------------------------
1179Several options for CREATE TABLE are supported directly by the PostgreSQL
1180dialect in conjunction with the :class:`_schema.Table` construct, listed in
1181the following sections.
1183.. seealso::
1185 `PostgreSQL CREATE TABLE options
1186 <https://www.postgresql.org/docs/current/static/sql-createtable.html>`_ -
1187 in the PostgreSQL documentation.
1189``INHERITS``
1190^^^^^^^^^^^^
1192Specifies one or more parent tables from which this table inherits columns and
1193constraints, enabling table inheritance hierarchies in PostgreSQL.
1195::
1197 Table("some_table", metadata, ..., postgresql_inherits="some_supertable")
1199 Table("some_table", metadata, ..., postgresql_inherits=("t1", "t2", ...))
1201For schema-qualified parent table names, use :class:`.quoted_name` with
1202``quote=False`` to prevent the dotted name from being quoted as a single
1203identifier::
1205 from sqlalchemy.sql import quoted_name
1207 Table(
1208 "some_table",
1209 metadata,
1210 ...,
1211 postgresql_inherits=quoted_name(
1212 "my_schema.some_supertable", quote=False
1213 ),
1214 )
1216SQLAlchemy does not automatically copy the columns from the inherited tables
1217mentioned in the ``postgresql_inherits`` argument into the new
1218:class:`_schema.Table`. To populate the new table columns reflection may be
1219used, or a function similar to the following one::
1221 def get_parent_columns(tbl: sa.Table) -> list[sa.Column]:
1222 return [
1223 sa.Column(
1224 c.name,
1225 c.type,
1226 key=c.key,
1227 nullable=c.nullable,
1228 # Set system=true to omit from the CREATE TABLE statement
1229 system=True,
1230 )
1231 for c in tbl.columns
1232 ]
1235 documents = Table(
1236 "some_table",
1237 metadata,
1238 *get_parent_columns(some_supertable),
1239 # ... # add other columns here normally if needed
1240 postgresql_inherits="some_supertable",
1241 )
1243``ON COMMIT``
1244^^^^^^^^^^^^^
1246Controls the behavior of temporary tables at transaction commit, with options
1247to preserve rows, delete rows, or drop the table.
1249::
1251 Table("some_table", metadata, ..., postgresql_on_commit="PRESERVE ROWS")
1253``PARTITION BY``
1254^^^^^^^^^^^^^^^^
1256Declares the table as a partitioned table using the specified partitioning
1257strategy (RANGE, LIST, or HASH) on the given column(s).
1259::
1261 Table(
1262 "some_table",
1263 metadata,
1264 ...,
1265 postgresql_partition_by="LIST (part_column)",
1266 )
1268``TABLESPACE``
1269^^^^^^^^^^^^^^
1271Specifies the tablespace where the table will be stored, allowing control over
1272the physical location of table data on disk.
1274::
1276 Table("some_table", metadata, ..., postgresql_tablespace="some_tablespace")
1278The above option is also available on the :class:`.Index` construct.
1280``USING``
1281^^^^^^^^^
1283Specifies the table access method to use for storing table data, such as
1284``heap`` (the default) or other custom access methods.
1286::
1288 Table("some_table", metadata, ..., postgresql_using="heap")
1290.. versionadded:: 2.0.26
1292``WITH OIDS``
1293^^^^^^^^^^^^^
1295Enables the legacy OID (object identifier) system column for the table, which
1296assigns a unique identifier to each row.
1298::
1300 Table("some_table", metadata, ..., postgresql_with_oids=True)
1302``WITHOUT OIDS``
1303^^^^^^^^^^^^^^^^
1305Explicitly disables the OID system column for the table (the default behavior
1306in modern PostgreSQL versions).
1308::
1310 Table("some_table", metadata, ..., postgresql_with_oids=False)
1312.. _postgresql_constraint_options:
1314PostgreSQL Constraint Options
1315-----------------------------
1317The following sections indicate options which are supported by the PostgreSQL
1318dialect in conjunction with selected constraint constructs.
1321``NOT VALID``
1322^^^^^^^^^^^^^
1324Allows a constraint to be added without validating existing rows, improving
1325performance when adding constraints to large tables. This option applies
1326towards CHECK and FOREIGN KEY constraints when the constraint is being added
1327to an existing table via ALTER TABLE, and has the effect that existing rows
1328are not scanned during the ALTER operation against the constraint being added.
1330When using a SQL migration tool such as `Alembic <https://alembic.sqlalchemy.org>`_
1331that renders ALTER TABLE constructs, the ``postgresql_not_valid`` argument
1332may be specified as an additional keyword argument within the operation
1333that creates the constraint, as in the following Alembic example::
1335 def update():
1336 op.create_foreign_key(
1337 "fk_user_address",
1338 "address",
1339 "user",
1340 ["user_id"],
1341 ["id"],
1342 postgresql_not_valid=True,
1343 )
1345The keyword is ultimately accepted directly by the
1346:class:`_schema.CheckConstraint`, :class:`_schema.ForeignKeyConstraint`
1347and :class:`_schema.ForeignKey` constructs; when using a tool like
1348Alembic, dialect-specific keyword arguments are passed through to
1349these constructs from the migration operation directives::
1351 CheckConstraint("some_field IS NOT NULL", postgresql_not_valid=True)
1353 ForeignKeyConstraint(
1354 ["some_id"], ["some_table.some_id"], postgresql_not_valid=True
1355 )
1357.. versionadded:: 1.4.32
1359.. seealso::
1361 `PostgreSQL ALTER TABLE options
1362 <https://www.postgresql.org/docs/current/static/sql-altertable.html>`_ -
1363 in the PostgreSQL documentation.
1365.. _postgresql_constraint_options_include:
1367``INCLUDE``
1368^^^^^^^^^^^
1370This keyword is applicable to both a ``UNIQUE`` constraint as well as an
1371``INDEX``. The ``postgresql_include`` option available for
1372:class:`.UniqueConstraint` as well as :class:`.Index` creates a covering index
1373by including additional columns in the underlying index without making them
1374part of the key constraint. This option adds one or more columns as a "payload"
1375to the index created automatically by PostgreSQL for the constraint. For
1376example, the following table definition::
1378 Table(
1379 "mytable",
1380 metadata,
1381 Column("id", Integer, nullable=False),
1382 Column("value", Integer, nullable=False),
1383 UniqueConstraint("id", postgresql_include=["value"]),
1384 )
1386would produce the DDL statement
1388.. sourcecode:: sql
1390 CREATE TABLE mytable (
1391 id INTEGER NOT NULL,
1392 value INTEGER NOT NULL,
1393 UNIQUE (id) INCLUDE (value)
1394 )
1396Note that this feature requires PostgreSQL 11 or later.
1398.. versionadded:: 2.0.41 - added support for ``postgresql_include`` to
1399 :class:`.UniqueConstraint`, to complement the existing feature in
1400 :class:`.Index`.
1402.. seealso::
1404 :ref:`postgresql_covering_indexes` - background on ``postgresql_include``
1405 for the :class:`.Index` construct.
1408Column list with foreign key ``ON DELETE SET`` actions
1409^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
1411Allows selective column updates when a foreign key action is triggered, limiting
1412which columns are set to NULL or DEFAULT upon deletion of a referenced row.
1413This applies to :class:`.ForeignKey` and :class:`.ForeignKeyConstraint`, the
1414:paramref:`.ForeignKey.ondelete` parameter will accept on the PostgreSQL
1415backend only a string list of column names inside parenthesis, following the
1416``SET NULL`` or ``SET DEFAULT`` phrases, which will limit the set of columns
1417that are subject to the action::
1419 fktable = Table(
1420 "fktable",
1421 metadata,
1422 Column("tid", Integer),
1423 Column("id", Integer),
1424 Column("fk_id_del_set_null", Integer),
1425 ForeignKeyConstraint(
1426 columns=["tid", "fk_id_del_set_null"],
1427 refcolumns=[pktable.c.tid, pktable.c.id],
1428 ondelete="SET NULL (fk_id_del_set_null)",
1429 ),
1430 )
1432.. versionadded:: 2.0.40
1435``NULLS NOT DISTINCT``
1436^^^^^^^^^^^^^^^^^^^^^^
1438By default, two ``null`` values are not considered equal for unique constraints
1439and indexes. Therefore, seemingly duplicate rows may be stored if one of the
1440values in the constraint is ``null``. This default behavior is implementation
1441defined, so other SQL dialects may behave differently than PostgreSQL.
1443The ``NULLS NOT DISTINCT`` clause can be used to change this behavior, treating
1444null values as equal and preventing unintended duplicate rows. The opposite
1445``NULLS DISTINCT`` clause can also be used to make PostgreSQL's default behavior
1446explict.
1448The ``postgresql_nulls_not_distinct`` parameter can be set to ``True`` to
1449add the ``NULLS NOT DISTINCT`` clause, or ``False`` to add ``NULLS DISTINCT``.
1450Not setting it, or passing ``None``, will not add a clause and keep the default
1451behavior.
1453This feature requires PostgreSQL 15 or later.
1455.. versionadded:: 2.0.16
1458.. _postgresql_table_valued_overview:
1460Table values, Table and Column valued functions, Row and Tuple objects
1461-----------------------------------------------------------------------
1463PostgreSQL makes great use of modern SQL forms such as table-valued functions,
1464tables and rows as values. These constructs are commonly used as part
1465of PostgreSQL's support for complex datatypes such as JSON, ARRAY, and other
1466datatypes. SQLAlchemy's SQL expression language has native support for
1467most table-valued and row-valued forms.
1469.. _postgresql_table_valued:
1471Table-Valued Functions
1472^^^^^^^^^^^^^^^^^^^^^^^
1474Many PostgreSQL built-in functions are intended to be used in the FROM clause
1475of a SELECT statement, and are capable of returning table rows or sets of table
1476rows. A large portion of PostgreSQL's JSON functions for example such as
1477``json_array_elements()``, ``json_object_keys()``, ``json_each_text()``,
1478``json_each()``, ``json_to_record()``, ``json_populate_recordset()`` use such
1479forms. These classes of SQL function calling forms in SQLAlchemy are available
1480using the :meth:`_functions.FunctionElement.table_valued` method in conjunction
1481with :class:`_functions.Function` objects generated from the :data:`_sql.func`
1482namespace.
1484Examples from PostgreSQL's reference documentation follow below:
1486* ``json_each()``:
1488 .. sourcecode:: pycon+sql
1490 >>> from sqlalchemy import select, func
1491 >>> stmt = select(
1492 ... func.json_each('{"a":"foo", "b":"bar"}').table_valued("key", "value")
1493 ... )
1494 >>> print(stmt)
1495 {printsql}SELECT anon_1.key, anon_1.value
1496 FROM json_each(:json_each_1) AS anon_1
1498* ``json_populate_record()``:
1500 .. sourcecode:: pycon+sql
1502 >>> from sqlalchemy import select, func, literal_column
1503 >>> stmt = select(
1504 ... func.json_populate_record(
1505 ... literal_column("null::myrowtype"), '{"a":1,"b":2}'
1506 ... ).table_valued("a", "b", name="x")
1507 ... )
1508 >>> print(stmt)
1509 {printsql}SELECT x.a, x.b
1510 FROM json_populate_record(null::myrowtype, :json_populate_record_1) AS x
1512* ``json_to_record()`` - this form uses a PostgreSQL specific form of derived
1513 columns in the alias, where we may make use of :func:`_sql.column` elements with
1514 types to produce them. The :meth:`_functions.FunctionElement.table_valued`
1515 method produces a :class:`_sql.TableValuedAlias` construct, and the method
1516 :meth:`_sql.TableValuedAlias.render_derived` method sets up the derived
1517 columns specification:
1519 .. sourcecode:: pycon+sql
1521 >>> from sqlalchemy import select, func, column, Integer, Text
1522 >>> stmt = select(
1523 ... func.json_to_record('{"a":1,"b":[1,2,3],"c":"bar"}')
1524 ... .table_valued(
1525 ... column("a", Integer),
1526 ... column("b", Text),
1527 ... column("d", Text),
1528 ... )
1529 ... .render_derived(name="x", with_types=True)
1530 ... )
1531 >>> print(stmt)
1532 {printsql}SELECT x.a, x.b, x.d
1533 FROM json_to_record(:json_to_record_1) AS x(a INTEGER, b TEXT, d TEXT)
1535* ``WITH ORDINALITY`` - part of the SQL standard, ``WITH ORDINALITY`` adds an
1536 ordinal counter to the output of a function and is accepted by a limited set
1537 of PostgreSQL functions including ``unnest()`` and ``generate_series()``. The
1538 :meth:`_functions.FunctionElement.table_valued` method accepts a keyword
1539 parameter ``with_ordinality`` for this purpose, which accepts the string name
1540 that will be applied to the "ordinality" column:
1542 .. sourcecode:: pycon+sql
1544 >>> from sqlalchemy import select, func
1545 >>> stmt = select(
1546 ... func.generate_series(4, 1, -1)
1547 ... .table_valued("value", with_ordinality="ordinality")
1548 ... .render_derived()
1549 ... )
1550 >>> print(stmt)
1551 {printsql}SELECT anon_1.value, anon_1.ordinality
1552 FROM generate_series(:generate_series_1, :generate_series_2, :generate_series_3)
1553 WITH ORDINALITY AS anon_1(value, ordinality)
1555.. versionadded:: 1.4.0b2
1557.. seealso::
1559 :ref:`tutorial_functions_table_valued` - in the :ref:`unified_tutorial`
1561.. _postgresql_column_valued:
1563Column Valued Functions
1564^^^^^^^^^^^^^^^^^^^^^^^
1566Similar to the table valued function, a column valued function is present
1567in the FROM clause, but delivers itself to the columns clause as a single
1568scalar value. PostgreSQL functions such as ``json_array_elements()``,
1569``unnest()`` and ``generate_series()`` may use this form. Column valued functions are available using the
1570:meth:`_functions.FunctionElement.column_valued` method of :class:`_functions.FunctionElement`:
1572* ``json_array_elements()``:
1574 .. sourcecode:: pycon+sql
1576 >>> from sqlalchemy import select, func
1577 >>> stmt = select(
1578 ... func.json_array_elements('["one", "two"]').column_valued("x")
1579 ... )
1580 >>> print(stmt)
1581 {printsql}SELECT x
1582 FROM json_array_elements(:json_array_elements_1) AS x
1584* ``unnest()`` - in order to generate a PostgreSQL ARRAY literal, the
1585 :func:`_postgresql.array` construct may be used:
1587 .. sourcecode:: pycon+sql
1589 >>> from sqlalchemy.dialects.postgresql import array
1590 >>> from sqlalchemy import select, func
1591 >>> stmt = select(func.unnest(array([1, 2])).column_valued())
1592 >>> print(stmt)
1593 {printsql}SELECT anon_1
1594 FROM unnest(ARRAY[%(param_1)s, %(param_2)s]) AS anon_1
1596 The function can of course be used against an existing table-bound column
1597 that's of type :class:`_types.ARRAY`:
1599 .. sourcecode:: pycon+sql
1601 >>> from sqlalchemy import table, column, ARRAY, Integer
1602 >>> from sqlalchemy import select, func
1603 >>> t = table("t", column("value", ARRAY(Integer)))
1604 >>> stmt = select(func.unnest(t.c.value).column_valued("unnested_value"))
1605 >>> print(stmt)
1606 {printsql}SELECT unnested_value
1607 FROM unnest(t.value) AS unnested_value
1609.. seealso::
1611 :ref:`tutorial_functions_column_valued` - in the :ref:`unified_tutorial`
1614Row Types
1615^^^^^^^^^
1617Built-in support for rendering a ``ROW`` may be approximated using
1618``func.ROW`` with the :attr:`_sa.func` namespace, or by using the
1619:func:`_sql.tuple_` construct:
1621.. sourcecode:: pycon+sql
1623 >>> from sqlalchemy import table, column, func, tuple_
1624 >>> t = table("t", column("id"), column("fk"))
1625 >>> stmt = (
1626 ... t.select()
1627 ... .where(tuple_(t.c.id, t.c.fk) > (1, 2))
1628 ... .where(func.ROW(t.c.id, t.c.fk) < func.ROW(3, 7))
1629 ... )
1630 >>> print(stmt)
1631 {printsql}SELECT t.id, t.fk
1632 FROM t
1633 WHERE (t.id, t.fk) > (:param_1, :param_2) AND ROW(t.id, t.fk) < ROW(:ROW_1, :ROW_2)
1635.. seealso::
1637 `PostgreSQL Row Constructors
1638 <https://www.postgresql.org/docs/current/sql-expressions.html#SQL-SYNTAX-ROW-CONSTRUCTORS>`_
1640 `PostgreSQL Row Constructor Comparison
1641 <https://www.postgresql.org/docs/current/functions-comparisons.html#ROW-WISE-COMPARISON>`_
1643Table Types passed to Functions
1644^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
1646PostgreSQL supports passing a table as an argument to a function, which is
1647known as a "record" type. SQLAlchemy :class:`_sql.FromClause` objects
1648such as :class:`_schema.Table` support this special form using the
1649:meth:`_sql.FromClause.table_valued` method, which is comparable to the
1650:meth:`_functions.FunctionElement.table_valued` method except that the collection
1651of columns is already established by that of the :class:`_sql.FromClause`
1652itself:
1654.. sourcecode:: pycon+sql
1656 >>> from sqlalchemy import table, column, func, select
1657 >>> a = table("a", column("id"), column("x"), column("y"))
1658 >>> stmt = select(func.row_to_json(a.table_valued()))
1659 >>> print(stmt)
1660 {printsql}SELECT row_to_json(a) AS row_to_json_1
1661 FROM a
1663.. versionadded:: 1.4.0b2
1667""" # noqa: E501
1669from __future__ import annotations
1671from collections import defaultdict
1672from functools import lru_cache
1673import re
1674from typing import Any
1675from typing import cast
1676from typing import Dict
1677from typing import List
1678from typing import Optional
1679from typing import Tuple
1680from typing import TYPE_CHECKING
1681from typing import Union
1683from . import arraylib as _array
1684from . import json as _json
1685from . import pg_catalog
1686from . import ranges as _ranges
1687from .ext import _regconfig_fn
1688from .ext import aggregate_order_by
1689from .hstore import HSTORE
1690from .named_types import CreateDomainType as CreateDomainType # noqa: F401
1691from .named_types import CreateEnumType as CreateEnumType # noqa: F401
1692from .named_types import DOMAIN as DOMAIN # noqa: F401
1693from .named_types import DropDomainType as DropDomainType # noqa: F401
1694from .named_types import DropEnumType as DropEnumType # noqa: F401
1695from .named_types import ENUM as ENUM # noqa: F401
1696from .named_types import NamedType as NamedType # noqa: F401
1697from .types import _DECIMAL_TYPES # noqa: F401
1698from .types import _FLOAT_TYPES # noqa: F401
1699from .types import _INT_TYPES # noqa: F401
1700from .types import BIT as BIT
1701from .types import BYTEA as BYTEA
1702from .types import CIDR as CIDR
1703from .types import CITEXT as CITEXT
1704from .types import INET as INET
1705from .types import INTERVAL as INTERVAL
1706from .types import MACADDR as MACADDR
1707from .types import MACADDR8 as MACADDR8
1708from .types import MONEY as MONEY
1709from .types import OID as OID
1710from .types import PGBit as PGBit # noqa: F401
1711from .types import PGCidr as PGCidr # noqa: F401
1712from .types import PGInet as PGInet # noqa: F401
1713from .types import PGInterval as PGInterval # noqa: F401
1714from .types import PGMacAddr as PGMacAddr # noqa: F401
1715from .types import PGMacAddr8 as PGMacAddr8 # noqa: F401
1716from .types import PGUuid as PGUuid
1717from .types import REGCLASS as REGCLASS
1718from .types import REGCONFIG as REGCONFIG # noqa: F401
1719from .types import TIME as TIME
1720from .types import TIMESTAMP as TIMESTAMP
1721from .types import TSVECTOR as TSVECTOR
1722from ... import exc
1723from ... import schema
1724from ... import select
1725from ... import sql
1726from ... import util
1727from ...engine import characteristics
1728from ...engine import default
1729from ...engine import interfaces
1730from ...engine import ObjectKind
1731from ...engine import ObjectScope
1732from ...engine import reflection
1733from ...engine import URL
1734from ...engine.reflection import ReflectionDefaults
1735from ...sql import bindparam
1736from ...sql import coercions
1737from ...sql import compiler
1738from ...sql import elements
1739from ...sql import expression
1740from ...sql import functions
1741from ...sql import roles
1742from ...sql import sqltypes
1743from ...sql import util as sql_util
1744from ...sql.compiler import InsertmanyvaluesSentinelOpts
1745from ...sql.visitors import InternalTraversal
1746from ...types import BIGINT
1747from ...types import BOOLEAN
1748from ...types import CHAR
1749from ...types import DATE
1750from ...types import DOUBLE_PRECISION
1751from ...types import FLOAT
1752from ...types import INTEGER
1753from ...types import NUMERIC
1754from ...types import REAL
1755from ...types import SMALLINT
1756from ...types import TEXT
1757from ...types import UUID as UUID
1758from ...types import VARCHAR
1759from ...util.typing import TypedDict
1761IDX_USING = re.compile(r"^(?:btree|hash|gist|gin|[\w_]+)$", re.I)
1763RESERVED_WORDS = {
1764 "all",
1765 "analyse",
1766 "analyze",
1767 "and",
1768 "any",
1769 "array",
1770 "as",
1771 "asc",
1772 "asymmetric",
1773 "both",
1774 "case",
1775 "cast",
1776 "check",
1777 "collate",
1778 "column",
1779 "constraint",
1780 "create",
1781 "current_catalog",
1782 "current_date",
1783 "current_role",
1784 "current_time",
1785 "current_timestamp",
1786 "current_user",
1787 "default",
1788 "deferrable",
1789 "desc",
1790 "distinct",
1791 "do",
1792 "else",
1793 "end",
1794 "except",
1795 "false",
1796 "fetch",
1797 "for",
1798 "foreign",
1799 "from",
1800 "grant",
1801 "group",
1802 "having",
1803 "in",
1804 "initially",
1805 "intersect",
1806 "into",
1807 "leading",
1808 "limit",
1809 "localtime",
1810 "localtimestamp",
1811 "new",
1812 "not",
1813 "null",
1814 "of",
1815 "off",
1816 "offset",
1817 "old",
1818 "on",
1819 "only",
1820 "or",
1821 "order",
1822 "placing",
1823 "primary",
1824 "references",
1825 "returning",
1826 "select",
1827 "session_user",
1828 "some",
1829 "symmetric",
1830 "table",
1831 "then",
1832 "to",
1833 "trailing",
1834 "true",
1835 "union",
1836 "unique",
1837 "user",
1838 "using",
1839 "variadic",
1840 "when",
1841 "where",
1842 "window",
1843 "with",
1844 "authorization",
1845 "between",
1846 "binary",
1847 "cross",
1848 "current_schema",
1849 "freeze",
1850 "full",
1851 "ilike",
1852 "inner",
1853 "is",
1854 "isnull",
1855 "join",
1856 "left",
1857 "like",
1858 "natural",
1859 "notnull",
1860 "outer",
1861 "over",
1862 "overlaps",
1863 "right",
1864 "similar",
1865 "verbose",
1866}
1869colspecs = {
1870 sqltypes.ARRAY: _array.ARRAY,
1871 sqltypes.Interval: INTERVAL,
1872 sqltypes.Enum: ENUM,
1873 sqltypes.JSON.JSONPathType: _json.JSONPATH,
1874 sqltypes.JSON: _json.JSON,
1875 sqltypes.Uuid: PGUuid,
1876}
1879ischema_names = {
1880 "_array": _array.ARRAY,
1881 "hstore": HSTORE,
1882 "json": _json.JSON,
1883 "jsonb": _json.JSONB,
1884 "int4range": _ranges.INT4RANGE,
1885 "int8range": _ranges.INT8RANGE,
1886 "numrange": _ranges.NUMRANGE,
1887 "daterange": _ranges.DATERANGE,
1888 "tsrange": _ranges.TSRANGE,
1889 "tstzrange": _ranges.TSTZRANGE,
1890 "int4multirange": _ranges.INT4MULTIRANGE,
1891 "int8multirange": _ranges.INT8MULTIRANGE,
1892 "nummultirange": _ranges.NUMMULTIRANGE,
1893 "datemultirange": _ranges.DATEMULTIRANGE,
1894 "tsmultirange": _ranges.TSMULTIRANGE,
1895 "tstzmultirange": _ranges.TSTZMULTIRANGE,
1896 "integer": INTEGER,
1897 "bigint": BIGINT,
1898 "smallint": SMALLINT,
1899 "character varying": VARCHAR,
1900 "character": CHAR,
1901 '"char"': sqltypes.String,
1902 "name": sqltypes.String,
1903 "text": TEXT,
1904 "numeric": NUMERIC,
1905 "float": FLOAT,
1906 "real": REAL,
1907 "inet": INET,
1908 "cidr": CIDR,
1909 "citext": CITEXT,
1910 "uuid": UUID,
1911 "bit": BIT,
1912 "bit varying": BIT,
1913 "macaddr": MACADDR,
1914 "macaddr8": MACADDR8,
1915 "money": MONEY,
1916 "oid": OID,
1917 "regclass": REGCLASS,
1918 "double precision": DOUBLE_PRECISION,
1919 "timestamp": TIMESTAMP,
1920 "timestamp with time zone": TIMESTAMP,
1921 "timestamp without time zone": TIMESTAMP,
1922 "time with time zone": TIME,
1923 "time without time zone": TIME,
1924 "date": DATE,
1925 "time": TIME,
1926 "bytea": BYTEA,
1927 "boolean": BOOLEAN,
1928 "interval": INTERVAL,
1929 "tsvector": TSVECTOR,
1930}
1933class PGCompiler(compiler.SQLCompiler):
1934 def visit_to_tsvector_func(self, element, **kw):
1935 return self._assert_pg_ts_ext(element, **kw)
1937 def visit_to_tsquery_func(self, element, **kw):
1938 return self._assert_pg_ts_ext(element, **kw)
1940 def visit_plainto_tsquery_func(self, element, **kw):
1941 return self._assert_pg_ts_ext(element, **kw)
1943 def visit_phraseto_tsquery_func(self, element, **kw):
1944 return self._assert_pg_ts_ext(element, **kw)
1946 def visit_websearch_to_tsquery_func(self, element, **kw):
1947 return self._assert_pg_ts_ext(element, **kw)
1949 def visit_ts_headline_func(self, element, **kw):
1950 return self._assert_pg_ts_ext(element, **kw)
1952 def _assert_pg_ts_ext(self, element, **kw):
1953 if not isinstance(element, _regconfig_fn):
1954 # other options here include trying to rewrite the function
1955 # with the correct types. however, that means we have to
1956 # "un-SQL-ize" the first argument, which can't work in a
1957 # generalized way. Also, parent compiler class has already added
1958 # the incorrect return type to the result map. So let's just
1959 # make sure the function we want is used up front.
1961 raise exc.CompileError(
1962 f'Can\'t compile "{element.name}()" full text search '
1963 f"function construct that does not originate from the "
1964 f'"sqlalchemy.dialects.postgresql" package. '
1965 f'Please ensure "import sqlalchemy.dialects.postgresql" is '
1966 f"called before constructing "
1967 f'"sqlalchemy.func.{element.name}()" to ensure registration '
1968 f"of the correct argument and return types."
1969 )
1971 return f"{element.name}{self.function_argspec(element, **kw)}"
1973 def render_bind_cast(self, type_, dbapi_type, sqltext):
1974 if dbapi_type._type_affinity is sqltypes.String and dbapi_type.length:
1975 # use VARCHAR with no length for VARCHAR cast.
1976 # see #9511
1977 dbapi_type = sqltypes.STRINGTYPE
1978 return f"""{sqltext}::{
1979 self.dialect.type_compiler_instance.process(
1980 dbapi_type, identifier_preparer=self.preparer
1981 )
1982 }"""
1984 def visit_array(self, element, **kw):
1985 if not element.clauses and not element.type.item_type._isnull:
1986 return "ARRAY[]::%s" % element.type.compile(self.dialect)
1987 return "ARRAY[%s]" % self.visit_clauselist(element, **kw)
1989 def visit_slice(self, element, **kw):
1990 return "%s:%s" % (
1991 self.process(element.start, **kw),
1992 self.process(element.stop, **kw),
1993 )
1995 def visit_bitwise_xor_op_binary(self, binary, operator, **kw):
1996 return self._generate_generic_binary(binary, " # ", **kw)
1998 def visit_json_getitem_op_binary(
1999 self, binary, operator, _cast_applied=False, **kw
2000 ):
2001 if (
2002 not _cast_applied
2003 and binary.type._type_affinity is not sqltypes.JSON
2004 ):
2005 kw["_cast_applied"] = True
2006 return self.process(sql.cast(binary, binary.type), **kw)
2008 kw["eager_grouping"] = True
2010 if (
2011 not _cast_applied
2012 and isinstance(binary.left.type, _json.JSONB)
2013 and self.dialect._supports_jsonb_subscripting
2014 ):
2015 left = binary.left
2016 if isinstance(left, (functions.FunctionElement, elements.Cast)):
2017 left = elements.Grouping(left)
2019 # for pg14+JSONB use subscript notation: col['key'] instead
2020 # of col -> 'key'
2021 return "%s[%s]" % (
2022 self.process(left, **kw),
2023 self.process(binary.right, **kw),
2024 )
2025 else:
2026 # Fall back to arrow notation for older versions or when cast
2027 # is applied
2028 return self._generate_generic_binary(
2029 binary, " -> " if not _cast_applied else " ->> ", **kw
2030 )
2032 def visit_json_path_getitem_op_binary(
2033 self, binary, operator, _cast_applied=False, **kw
2034 ):
2035 if (
2036 not _cast_applied
2037 and binary.type._type_affinity is not sqltypes.JSON
2038 ):
2039 kw["_cast_applied"] = True
2040 return self.process(sql.cast(binary, binary.type), **kw)
2042 kw["eager_grouping"] = True
2043 return self._generate_generic_binary(
2044 binary, " #> " if not _cast_applied else " #>> ", **kw
2045 )
2047 def visit_getitem_binary(self, binary, operator, **kw):
2048 return "%s[%s]" % (
2049 self.process(binary.left, **kw),
2050 self.process(binary.right, **kw),
2051 )
2053 def visit_aggregate_order_by(self, element, **kw):
2054 return "%s ORDER BY %s" % (
2055 self.process(element.target, **kw),
2056 self.process(element.order_by, **kw),
2057 )
2059 def visit_match_op_binary(self, binary, operator, **kw):
2060 if "postgresql_regconfig" in binary.modifiers:
2061 regconfig = self.render_literal_value(
2062 binary.modifiers["postgresql_regconfig"], sqltypes.STRINGTYPE
2063 )
2064 if regconfig:
2065 return "%s @@ plainto_tsquery(%s, %s)" % (
2066 self.process(binary.left, **kw),
2067 regconfig,
2068 self.process(binary.right, **kw),
2069 )
2070 return "%s @@ plainto_tsquery(%s)" % (
2071 self.process(binary.left, **kw),
2072 self.process(binary.right, **kw),
2073 )
2075 def visit_ilike_case_insensitive_operand(self, element, **kw):
2076 return element.element._compiler_dispatch(self, **kw)
2078 def visit_ilike_op_binary(self, binary, operator, **kw):
2079 escape = binary.modifiers.get("escape", None)
2081 return "%s ILIKE %s" % (
2082 self.process(binary.left, **kw),
2083 self.process(binary.right, **kw),
2084 ) + (
2085 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
2086 if escape is not None
2087 else ""
2088 )
2090 def visit_not_ilike_op_binary(self, binary, operator, **kw):
2091 escape = binary.modifiers.get("escape", None)
2092 return "%s NOT ILIKE %s" % (
2093 self.process(binary.left, **kw),
2094 self.process(binary.right, **kw),
2095 ) + (
2096 " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
2097 if escape is not None
2098 else ""
2099 )
2101 def _regexp_match(self, base_op, binary, operator, kw):
2102 flags = binary.modifiers["flags"]
2103 if flags is None:
2104 return self._generate_generic_binary(
2105 binary, " %s " % base_op, **kw
2106 )
2107 if flags == "i":
2108 return self._generate_generic_binary(
2109 binary, " %s* " % base_op, **kw
2110 )
2111 return "%s %s CONCAT('(?', %s, ')', %s)" % (
2112 self.process(binary.left, **kw),
2113 base_op,
2114 self.render_literal_value(flags, sqltypes.STRINGTYPE),
2115 self.process(binary.right, **kw),
2116 )
2118 def visit_regexp_match_op_binary(self, binary, operator, **kw):
2119 return self._regexp_match("~", binary, operator, kw)
2121 def visit_not_regexp_match_op_binary(self, binary, operator, **kw):
2122 return self._regexp_match("!~", binary, operator, kw)
2124 def visit_regexp_replace_op_binary(self, binary, operator, **kw):
2125 string = self.process(binary.left, **kw)
2126 pattern_replace = self.process(binary.right, **kw)
2127 flags = binary.modifiers["flags"]
2128 if flags is None:
2129 return "REGEXP_REPLACE(%s, %s)" % (
2130 string,
2131 pattern_replace,
2132 )
2133 else:
2134 return "REGEXP_REPLACE(%s, %s, %s)" % (
2135 string,
2136 pattern_replace,
2137 self.render_literal_value(flags, sqltypes.STRINGTYPE),
2138 )
2140 def visit_empty_set_expr(self, element_types, **kw):
2141 # cast the empty set to the type we are comparing against. if
2142 # we are comparing against the null type, pick an arbitrary
2143 # datatype for the empty set
2144 return "SELECT %s WHERE 1!=1" % (
2145 ", ".join(
2146 "CAST(NULL AS %s)"
2147 % self.dialect.type_compiler_instance.process(
2148 INTEGER() if type_._isnull else type_
2149 )
2150 for type_ in element_types or [INTEGER()]
2151 ),
2152 )
2154 def render_literal_value(self, value, type_):
2155 value = super().render_literal_value(value, type_)
2157 if self.dialect._backslash_escapes:
2158 value = value.replace("\\", "\\\\")
2159 return value
2161 def visit_aggregate_strings_func(self, fn, **kw):
2162 return "string_agg%s" % self.function_argspec(fn)
2164 def visit_sequence(self, seq, **kw):
2165 return "nextval(%s)" % self.preparer._format_sequence_string_literal(
2166 seq
2167 )
2169 def limit_clause(self, select, **kw):
2170 text = ""
2171 if select._limit_clause is not None:
2172 text += " \n LIMIT " + self.process(select._limit_clause, **kw)
2173 if select._offset_clause is not None:
2174 if select._limit_clause is None:
2175 text += "\n LIMIT ALL"
2176 text += " OFFSET " + self.process(select._offset_clause, **kw)
2177 return text
2179 def format_from_hint_text(self, sqltext, table, hint, iscrud):
2180 if hint.upper() != "ONLY":
2181 raise exc.CompileError("Unrecognized hint: %r" % hint)
2182 return "ONLY " + sqltext
2184 def get_select_precolumns(self, select, **kw):
2185 # Do not call super().get_select_precolumns because
2186 # it will warn/raise when distinct on is present
2187 if select._distinct or select._distinct_on:
2188 if select._distinct_on:
2189 return (
2190 "DISTINCT ON ("
2191 + ", ".join(
2192 [
2193 self.process(col, **kw)
2194 for col in select._distinct_on
2195 ]
2196 )
2197 + ") "
2198 )
2199 else:
2200 return "DISTINCT "
2201 else:
2202 return ""
2204 def for_update_clause(self, select, **kw):
2205 if select._for_update_arg.read:
2206 if select._for_update_arg.key_share:
2207 tmp = " FOR KEY SHARE"
2208 else:
2209 tmp = " FOR SHARE"
2210 elif select._for_update_arg.key_share:
2211 tmp = " FOR NO KEY UPDATE"
2212 else:
2213 tmp = " FOR UPDATE"
2215 if select._for_update_arg.of:
2216 tables = util.OrderedSet()
2217 for c in select._for_update_arg.of:
2218 tables.update(sql_util.surface_selectables_only(c))
2220 of_kw = dict(kw)
2221 of_kw.update(ashint=True, use_schema=False)
2222 tmp += " OF " + ", ".join(
2223 self.process(table, **of_kw) for table in tables
2224 )
2226 if select._for_update_arg.nowait:
2227 tmp += " NOWAIT"
2228 if select._for_update_arg.skip_locked:
2229 tmp += " SKIP LOCKED"
2231 return tmp
2233 def visit_substring_func(self, func, **kw):
2234 s = self.process(func.clauses.clauses[0], **kw)
2235 start = self.process(func.clauses.clauses[1], **kw)
2236 if len(func.clauses.clauses) > 2:
2237 length = self.process(func.clauses.clauses[2], **kw)
2238 return "SUBSTRING(%s FROM %s FOR %s)" % (s, start, length)
2239 else:
2240 return "SUBSTRING(%s FROM %s)" % (s, start)
2242 def _on_conflict_target(self, clause, **kw):
2243 if clause.constraint_target is not None:
2244 # target may be a name of an Index, UniqueConstraint or
2245 # ExcludeConstraint. While there is a separate
2246 # "max_identifier_length" for indexes, PostgreSQL uses the same
2247 # length for all objects so we can use
2248 # truncate_and_render_constraint_name
2249 target_text = (
2250 "ON CONSTRAINT %s"
2251 % self.preparer.truncate_and_render_constraint_name(
2252 clause.constraint_target
2253 )
2254 )
2255 elif clause.inferred_target_elements is not None:
2256 target_text = "(%s)" % ", ".join(
2257 (
2258 self.preparer.quote(c)
2259 if isinstance(c, str)
2260 else self.process(c, include_table=False, use_schema=False)
2261 )
2262 for c in clause.inferred_target_elements
2263 )
2264 if clause.inferred_target_whereclause is not None:
2265 whereclause_kw = dict(kw)
2266 whereclause_kw.update(include_table=False, use_schema=False)
2267 target_text += " WHERE %s" % self.process(
2268 clause.inferred_target_whereclause,
2269 **whereclause_kw,
2270 )
2271 else:
2272 target_text = ""
2274 return target_text
2276 def visit_on_conflict_do_nothing(self, on_conflict, **kw):
2277 target_text = self._on_conflict_target(on_conflict, **kw)
2279 if target_text:
2280 return "ON CONFLICT %s DO NOTHING" % target_text
2281 else:
2282 return "ON CONFLICT DO NOTHING"
2284 def visit_on_conflict_do_update(self, on_conflict, **kw):
2285 clause = on_conflict
2287 target_text = self._on_conflict_target(on_conflict, **kw)
2289 action_set_ops = []
2291 set_parameters = dict(clause.update_values_to_set)
2292 # create a list of column assignment clauses as tuples
2294 insert_statement = self.stack[-1]["selectable"]
2295 cols = insert_statement.table.c
2296 set_kw = dict(kw)
2297 set_kw.update(use_schema=False)
2298 for c in cols:
2299 col_key = c.key
2301 if col_key in set_parameters:
2302 value = set_parameters.pop(col_key)
2303 elif c in set_parameters:
2304 value = set_parameters.pop(c)
2305 else:
2306 continue
2308 # TODO: this coercion should be up front. we can't cache
2309 # SQL constructs with non-bound literals buried in them
2310 if coercions._is_literal(value):
2311 value = elements.BindParameter(None, value, type_=c.type)
2313 else:
2314 if (
2315 isinstance(value, elements.BindParameter)
2316 and value.type._isnull
2317 ):
2318 value = value._clone()
2319 value.type = c.type
2320 value_text = self.process(
2321 value.self_group(), is_upsert_set=True, **set_kw
2322 )
2324 key_text = self.preparer.quote(c.name)
2325 action_set_ops.append("%s = %s" % (key_text, value_text))
2327 # check for names that don't match columns
2328 if set_parameters:
2329 util.warn(
2330 "Additional column names not matching "
2331 "any column keys in table '%s': %s"
2332 % (
2333 self.current_executable.table.name,
2334 (", ".join("'%s'" % c for c in set_parameters)),
2335 )
2336 )
2337 for k, v in set_parameters.items():
2338 key_text = (
2339 self.preparer.quote(k)
2340 if isinstance(k, str)
2341 else self.process(k, use_schema=False)
2342 )
2343 value_text = self.process(
2344 coercions.expect(roles.ExpressionElementRole, v),
2345 is_upsert_set=True,
2346 **set_kw,
2347 )
2348 action_set_ops.append("%s = %s" % (key_text, value_text))
2350 action_text = ", ".join(action_set_ops)
2351 if clause.update_whereclause is not None:
2352 where_kw = dict(kw)
2353 where_kw.update(include_table=True, use_schema=False)
2354 action_text += " WHERE %s" % self.process(
2355 clause.update_whereclause, **where_kw
2356 )
2358 return "ON CONFLICT %s DO UPDATE SET %s" % (target_text, action_text)
2360 def update_from_clause(
2361 self, update_stmt, from_table, extra_froms, from_hints, **kw
2362 ):
2363 kw["asfrom"] = True
2364 return "FROM " + ", ".join(
2365 t._compiler_dispatch(self, fromhints=from_hints, **kw)
2366 for t in extra_froms
2367 )
2369 def delete_extra_from_clause(
2370 self, delete_stmt, from_table, extra_froms, from_hints, **kw
2371 ):
2372 """Render the DELETE .. USING clause specific to PostgreSQL."""
2373 kw["asfrom"] = True
2374 return "USING " + ", ".join(
2375 t._compiler_dispatch(self, fromhints=from_hints, **kw)
2376 for t in extra_froms
2377 )
2379 def fetch_clause(self, select, **kw):
2380 # pg requires parens for non literal clauses. It's also required for
2381 # bind parameters if a ::type casts is used by the driver (asyncpg),
2382 # so it's easiest to just always add it
2383 text = ""
2384 if select._offset_clause is not None:
2385 text += "\n OFFSET (%s) ROWS" % self.process(
2386 select._offset_clause, **kw
2387 )
2388 if select._fetch_clause is not None:
2389 text += "\n FETCH FIRST (%s)%s ROWS %s" % (
2390 self.process(select._fetch_clause, **kw),
2391 " PERCENT" if select._fetch_clause_options["percent"] else "",
2392 (
2393 "WITH TIES"
2394 if select._fetch_clause_options["with_ties"]
2395 else "ONLY"
2396 ),
2397 )
2398 return text
2401class PGDDLCompiler(compiler.DDLCompiler):
2402 def get_column_specification(self, column, **kwargs):
2403 colspec = self.preparer.format_column(column)
2404 impl_type = column.type.dialect_impl(self.dialect)
2405 if isinstance(impl_type, sqltypes.TypeDecorator):
2406 impl_type = impl_type.impl
2408 has_identity = (
2409 column.identity is not None
2410 and self.dialect.supports_identity_columns
2411 )
2413 if (
2414 column.primary_key
2415 and column is column.table._autoincrement_column
2416 and (
2417 self.dialect.supports_smallserial
2418 or not isinstance(impl_type, sqltypes.SmallInteger)
2419 )
2420 and not has_identity
2421 and (
2422 column.default is None
2423 or (
2424 isinstance(column.default, schema.Sequence)
2425 and column.default.optional
2426 )
2427 )
2428 ):
2429 if isinstance(impl_type, sqltypes.BigInteger):
2430 colspec += " BIGSERIAL"
2431 elif isinstance(impl_type, sqltypes.SmallInteger):
2432 colspec += " SMALLSERIAL"
2433 else:
2434 colspec += " SERIAL"
2435 else:
2436 colspec += " " + self.dialect.type_compiler_instance.process(
2437 column.type,
2438 type_expression=column,
2439 identifier_preparer=self.preparer,
2440 )
2441 default = self.get_column_default_string(column)
2442 if default is not None:
2443 colspec += " DEFAULT " + default
2445 if column.computed is not None:
2446 colspec += " " + self.process(column.computed)
2447 if has_identity:
2448 colspec += " " + self.process(column.identity)
2450 if not column.nullable and not has_identity:
2451 colspec += " NOT NULL"
2452 elif column.nullable and has_identity:
2453 colspec += " NULL"
2454 return colspec
2456 def _define_constraint_validity(self, constraint):
2457 not_valid = constraint.dialect_options["postgresql"]["not_valid"]
2458 return " NOT VALID" if not_valid else ""
2460 def _define_include(self, obj):
2461 includeclause = obj.dialect_options["postgresql"]["include"]
2462 if not includeclause:
2463 return ""
2464 inclusions = [
2465 obj.table.c[col] if isinstance(col, str) else col
2466 for col in includeclause
2467 ]
2468 return " INCLUDE (%s)" % ", ".join(
2469 [self.preparer.quote(c.name) for c in inclusions]
2470 )
2472 def visit_check_constraint(self, constraint, **kw):
2473 if constraint._type_bound:
2474 typ = list(constraint.columns)[0].type
2475 if (
2476 isinstance(typ, sqltypes.ARRAY)
2477 and isinstance(typ.item_type, sqltypes.Enum)
2478 and not typ.item_type.native_enum
2479 ):
2480 raise exc.CompileError(
2481 "PostgreSQL dialect cannot produce the CHECK constraint "
2482 "for ARRAY of non-native ENUM; please specify "
2483 "create_constraint=False on this Enum datatype."
2484 )
2486 text = super().visit_check_constraint(constraint)
2487 text += self._define_constraint_validity(constraint)
2488 return text
2490 def visit_foreign_key_constraint(self, constraint, **kw):
2491 text = super().visit_foreign_key_constraint(constraint)
2492 text += self._define_constraint_validity(constraint)
2493 return text
2495 def visit_primary_key_constraint(self, constraint, **kw):
2496 text = self.define_constraint_preamble(constraint, **kw)
2497 text += self.define_primary_key_body(constraint, **kw)
2498 text += self._define_include(constraint)
2499 text += self.define_constraint_deferrability(constraint)
2500 return text
2502 def visit_unique_constraint(self, constraint, **kw):
2503 if len(constraint) == 0:
2504 return ""
2505 text = self.define_constraint_preamble(constraint, **kw)
2506 text += self.define_unique_body(constraint, **kw)
2507 text += self._define_include(constraint)
2508 text += self.define_constraint_deferrability(constraint)
2509 return text
2511 @util.memoized_property
2512 def _fk_ondelete_pattern(self):
2513 return re.compile(
2514 r"^(?:RESTRICT|CASCADE|SET (?:NULL|DEFAULT)(?:\s*\(.+\))?"
2515 r"|NO ACTION)$",
2516 re.I,
2517 )
2519 def define_constraint_ondelete_cascade(self, constraint):
2520 return " ON DELETE %s" % self.preparer.validate_sql_phrase(
2521 constraint.ondelete, self._fk_ondelete_pattern
2522 )
2524 def visit_create_enum_type(self, create, **kw):
2525 type_ = create.element
2527 return "CREATE TYPE %s AS ENUM (%s)" % (
2528 self.preparer.format_type(type_),
2529 ", ".join(
2530 self.sql_compiler.process(sql.literal(e), literal_binds=True)
2531 for e in type_.enums
2532 ),
2533 )
2535 def visit_drop_enum_type(self, drop, **kw):
2536 type_ = drop.element
2538 return "DROP TYPE %s" % (self.preparer.format_type(type_))
2540 def visit_create_domain_type(self, create, **kw):
2541 domain: DOMAIN = create.element
2543 options = []
2544 if domain.collation is not None:
2545 options.append(f"COLLATE {self.preparer.quote(domain.collation)}")
2546 if domain.default is not None:
2547 default = self.render_default_string(domain.default)
2548 options.append(f"DEFAULT {default}")
2549 if domain.constraint_name is not None:
2550 name = self.preparer.truncate_and_render_constraint_name(
2551 domain.constraint_name
2552 )
2553 options.append(f"CONSTRAINT {name}")
2554 if domain.not_null:
2555 options.append("NOT NULL")
2556 if domain.check is not None:
2557 check = self.sql_compiler.process(
2558 domain.check, include_table=False, literal_binds=True
2559 )
2560 options.append(f"CHECK ({check})")
2562 return (
2563 f"CREATE DOMAIN {self.preparer.format_type(domain)} AS "
2564 f"{self.type_compiler.process(domain.data_type)} "
2565 f"{' '.join(options)}"
2566 )
2568 def visit_drop_domain_type(self, drop, **kw):
2569 domain = drop.element
2570 return f"DROP DOMAIN {self.preparer.format_type(domain)}"
2572 def visit_create_index(self, create, **kw):
2573 preparer = self.preparer
2574 index = create.element
2575 self._verify_index_table(index)
2576 text = "CREATE "
2577 if index.unique:
2578 text += "UNIQUE "
2580 text += "INDEX "
2582 if self.dialect._supports_create_index_concurrently:
2583 concurrently = index.dialect_options["postgresql"]["concurrently"]
2584 if concurrently:
2585 text += "CONCURRENTLY "
2587 if create.if_not_exists:
2588 text += "IF NOT EXISTS "
2590 text += "%s ON %s " % (
2591 self._prepared_index_name(index, include_schema=False),
2592 preparer.format_table(index.table),
2593 )
2595 using = index.dialect_options["postgresql"]["using"]
2596 if using:
2597 text += (
2598 "USING %s "
2599 % self.preparer.validate_sql_phrase(using, IDX_USING).lower()
2600 )
2602 ops = index.dialect_options["postgresql"]["ops"]
2603 text += "(%s)" % (
2604 ", ".join(
2605 [
2606 self.sql_compiler.process(
2607 (
2608 expr.self_group()
2609 if not isinstance(expr, expression.ColumnClause)
2610 else expr
2611 ),
2612 include_table=False,
2613 literal_binds=True,
2614 )
2615 + (
2616 (" " + ops[expr.key])
2617 if hasattr(expr, "key") and expr.key in ops
2618 else ""
2619 )
2620 for expr in index.expressions
2621 ]
2622 )
2623 )
2625 text += self._define_include(index)
2627 nulls_not_distinct = index.dialect_options["postgresql"][
2628 "nulls_not_distinct"
2629 ]
2630 if nulls_not_distinct is True:
2631 text += " NULLS NOT DISTINCT"
2632 elif nulls_not_distinct is False:
2633 text += " NULLS DISTINCT"
2635 withclause = index.dialect_options["postgresql"]["with"]
2636 if withclause:
2637 text += " WITH (%s)" % (
2638 ", ".join(
2639 [
2640 "%s = %s" % storage_parameter
2641 for storage_parameter in withclause.items()
2642 ]
2643 )
2644 )
2646 tablespace_name = index.dialect_options["postgresql"]["tablespace"]
2647 if tablespace_name:
2648 text += " TABLESPACE %s" % preparer.quote(tablespace_name)
2650 whereclause = index.dialect_options["postgresql"]["where"]
2651 if whereclause is not None:
2652 whereclause = coercions.expect(
2653 roles.DDLExpressionRole, whereclause
2654 )
2656 where_compiled = self.sql_compiler.process(
2657 whereclause, include_table=False, literal_binds=True
2658 )
2659 text += " WHERE " + where_compiled
2661 return text
2663 def define_unique_constraint_distinct(self, constraint, **kw):
2664 nulls_not_distinct = constraint.dialect_options["postgresql"][
2665 "nulls_not_distinct"
2666 ]
2667 if nulls_not_distinct is True:
2668 nulls_not_distinct_param = "NULLS NOT DISTINCT "
2669 elif nulls_not_distinct is False:
2670 nulls_not_distinct_param = "NULLS DISTINCT "
2671 else:
2672 nulls_not_distinct_param = ""
2673 return nulls_not_distinct_param
2675 def visit_drop_index(self, drop, **kw):
2676 index = drop.element
2678 text = "\nDROP INDEX "
2680 if self.dialect._supports_drop_index_concurrently:
2681 concurrently = index.dialect_options["postgresql"]["concurrently"]
2682 if concurrently:
2683 text += "CONCURRENTLY "
2685 if drop.if_exists:
2686 text += "IF EXISTS "
2688 text += self._prepared_index_name(index, include_schema=True)
2689 return text
2691 def visit_exclude_constraint(self, constraint, **kw):
2692 text = ""
2693 if constraint.name is not None:
2694 text += "CONSTRAINT %s " % self.preparer.format_constraint(
2695 constraint
2696 )
2697 elements = []
2698 kw["include_table"] = False
2699 kw["literal_binds"] = True
2700 for expr, name, op in constraint._render_exprs:
2701 exclude_element = self.sql_compiler.process(expr, **kw) + (
2702 (" " + constraint.ops[expr.key])
2703 if hasattr(expr, "key") and expr.key in constraint.ops
2704 else ""
2705 )
2707 elements.append("%s WITH %s" % (exclude_element, op))
2708 text += "EXCLUDE USING %s (%s)" % (
2709 self.preparer.validate_sql_phrase(
2710 constraint.using, IDX_USING
2711 ).lower(),
2712 ", ".join(elements),
2713 )
2714 if constraint.where is not None:
2715 text += " WHERE (%s)" % self.sql_compiler.process(
2716 constraint.where, literal_binds=True
2717 )
2718 text += self.define_constraint_deferrability(constraint)
2719 return text
2721 def post_create_table(self, table):
2722 table_opts = []
2723 pg_opts = table.dialect_options["postgresql"]
2725 inherits = pg_opts.get("inherits")
2726 if inherits is not None:
2727 if not isinstance(inherits, (list, tuple)):
2728 inherits = (inherits,)
2729 table_opts.append(
2730 "\n INHERITS ( "
2731 + ", ".join(self.preparer.quote(name) for name in inherits)
2732 + " )"
2733 )
2735 if pg_opts["partition_by"]:
2736 table_opts.append("\n PARTITION BY %s" % pg_opts["partition_by"])
2738 if pg_opts["using"]:
2739 table_opts.append("\n USING %s" % pg_opts["using"])
2741 if pg_opts["with_oids"] is True:
2742 table_opts.append("\n WITH OIDS")
2743 elif pg_opts["with_oids"] is False:
2744 table_opts.append("\n WITHOUT OIDS")
2746 if pg_opts["on_commit"]:
2747 on_commit_options = pg_opts["on_commit"].replace("_", " ").upper()
2748 table_opts.append("\n ON COMMIT %s" % on_commit_options)
2750 if pg_opts["tablespace"]:
2751 tablespace_name = pg_opts["tablespace"]
2752 table_opts.append(
2753 "\n TABLESPACE %s" % self.preparer.quote(tablespace_name)
2754 )
2756 return "".join(table_opts)
2758 def visit_computed_column(self, generated, **kw):
2759 if generated.persisted is False:
2760 raise exc.CompileError(
2761 "PostrgreSQL computed columns do not support 'virtual' "
2762 "persistence; set the 'persisted' flag to None or True for "
2763 "PostgreSQL support."
2764 )
2766 return "GENERATED ALWAYS AS (%s) STORED" % self.sql_compiler.process(
2767 generated.sqltext, include_table=False, literal_binds=True
2768 )
2770 def visit_create_sequence(self, create, **kw):
2771 prefix = None
2772 if create.element.data_type is not None:
2773 prefix = " AS %s" % self.type_compiler.process(
2774 create.element.data_type
2775 )
2777 return super().visit_create_sequence(create, prefix=prefix, **kw)
2779 def _can_comment_on_constraint(self, ddl_instance):
2780 constraint = ddl_instance.element
2781 if constraint.name is None:
2782 raise exc.CompileError(
2783 f"Can't emit COMMENT ON for constraint {constraint!r}: "
2784 "it has no name"
2785 )
2786 if constraint.table is None:
2787 raise exc.CompileError(
2788 f"Can't emit COMMENT ON for constraint {constraint!r}: "
2789 "it has no associated table"
2790 )
2792 def visit_set_constraint_comment(self, create, **kw):
2793 self._can_comment_on_constraint(create)
2794 return "COMMENT ON CONSTRAINT %s ON %s IS %s" % (
2795 self.preparer.format_constraint(create.element),
2796 self.preparer.format_table(create.element.table),
2797 self.sql_compiler.render_literal_value(
2798 create.element.comment, sqltypes.String()
2799 ),
2800 )
2802 def visit_drop_constraint_comment(self, drop, **kw):
2803 self._can_comment_on_constraint(drop)
2804 return "COMMENT ON CONSTRAINT %s ON %s IS NULL" % (
2805 self.preparer.format_constraint(drop.element),
2806 self.preparer.format_table(drop.element.table),
2807 )
2810class PGTypeCompiler(compiler.GenericTypeCompiler):
2811 def visit_TSVECTOR(self, type_, **kw):
2812 return "TSVECTOR"
2814 def visit_TSQUERY(self, type_, **kw):
2815 return "TSQUERY"
2817 def visit_INET(self, type_, **kw):
2818 return "INET"
2820 def visit_CIDR(self, type_, **kw):
2821 return "CIDR"
2823 def visit_CITEXT(self, type_, **kw):
2824 return "CITEXT"
2826 def visit_MACADDR(self, type_, **kw):
2827 return "MACADDR"
2829 def visit_MACADDR8(self, type_, **kw):
2830 return "MACADDR8"
2832 def visit_MONEY(self, type_, **kw):
2833 return "MONEY"
2835 def visit_OID(self, type_, **kw):
2836 return "OID"
2838 def visit_REGCONFIG(self, type_, **kw):
2839 return "REGCONFIG"
2841 def visit_REGCLASS(self, type_, **kw):
2842 return "REGCLASS"
2844 def visit_FLOAT(self, type_, **kw):
2845 if not type_.precision:
2846 return "FLOAT"
2847 else:
2848 return "FLOAT(%(precision)s)" % {"precision": type_.precision}
2850 def visit_double(self, type_, **kw):
2851 return self.visit_DOUBLE_PRECISION(type, **kw)
2853 def visit_BIGINT(self, type_, **kw):
2854 return "BIGINT"
2856 def visit_HSTORE(self, type_, **kw):
2857 return "HSTORE"
2859 def visit_JSON(self, type_, **kw):
2860 return "JSON"
2862 def visit_JSONB(self, type_, **kw):
2863 return "JSONB"
2865 def visit_INT4MULTIRANGE(self, type_, **kw):
2866 return "INT4MULTIRANGE"
2868 def visit_INT8MULTIRANGE(self, type_, **kw):
2869 return "INT8MULTIRANGE"
2871 def visit_NUMMULTIRANGE(self, type_, **kw):
2872 return "NUMMULTIRANGE"
2874 def visit_DATEMULTIRANGE(self, type_, **kw):
2875 return "DATEMULTIRANGE"
2877 def visit_TSMULTIRANGE(self, type_, **kw):
2878 return "TSMULTIRANGE"
2880 def visit_TSTZMULTIRANGE(self, type_, **kw):
2881 return "TSTZMULTIRANGE"
2883 def visit_INT4RANGE(self, type_, **kw):
2884 return "INT4RANGE"
2886 def visit_INT8RANGE(self, type_, **kw):
2887 return "INT8RANGE"
2889 def visit_NUMRANGE(self, type_, **kw):
2890 return "NUMRANGE"
2892 def visit_DATERANGE(self, type_, **kw):
2893 return "DATERANGE"
2895 def visit_TSRANGE(self, type_, **kw):
2896 return "TSRANGE"
2898 def visit_TSTZRANGE(self, type_, **kw):
2899 return "TSTZRANGE"
2901 def visit_json_int_index(self, type_, **kw):
2902 return "INT"
2904 def visit_json_str_index(self, type_, **kw):
2905 return "TEXT"
2907 def visit_datetime(self, type_, **kw):
2908 return self.visit_TIMESTAMP(type_, **kw)
2910 def visit_enum(self, type_, **kw):
2911 if not type_.native_enum or not self.dialect.supports_native_enum:
2912 return super().visit_enum(type_, **kw)
2913 else:
2914 return self.visit_ENUM(type_, **kw)
2916 def visit_ENUM(self, type_, identifier_preparer=None, **kw):
2917 if identifier_preparer is None:
2918 identifier_preparer = self.dialect.identifier_preparer
2919 return identifier_preparer.format_type(type_)
2921 def visit_DOMAIN(self, type_, identifier_preparer=None, **kw):
2922 if identifier_preparer is None:
2923 identifier_preparer = self.dialect.identifier_preparer
2924 return identifier_preparer.format_type(type_)
2926 def visit_TIMESTAMP(self, type_, **kw):
2927 return "TIMESTAMP%s %s" % (
2928 (
2929 "(%d)" % type_.precision
2930 if getattr(type_, "precision", None) is not None
2931 else ""
2932 ),
2933 (type_.timezone and "WITH" or "WITHOUT") + " TIME ZONE",
2934 )
2936 def visit_TIME(self, type_, **kw):
2937 return "TIME%s %s" % (
2938 (
2939 "(%d)" % type_.precision
2940 if getattr(type_, "precision", None) is not None
2941 else ""
2942 ),
2943 (type_.timezone and "WITH" or "WITHOUT") + " TIME ZONE",
2944 )
2946 def visit_INTERVAL(self, type_, **kw):
2947 text = "INTERVAL"
2948 if type_.fields is not None:
2949 text += " " + type_.fields
2950 if type_.precision is not None:
2951 text += " (%d)" % type_.precision
2952 return text
2954 def visit_BIT(self, type_, **kw):
2955 if type_.varying:
2956 compiled = "BIT VARYING"
2957 if type_.length is not None:
2958 compiled += "(%d)" % type_.length
2959 else:
2960 compiled = "BIT(%d)" % type_.length
2961 return compiled
2963 def visit_uuid(self, type_, **kw):
2964 if type_.native_uuid:
2965 return self.visit_UUID(type_, **kw)
2966 else:
2967 return super().visit_uuid(type_, **kw)
2969 def visit_UUID(self, type_, **kw):
2970 return "UUID"
2972 def visit_large_binary(self, type_, **kw):
2973 return self.visit_BYTEA(type_, **kw)
2975 def visit_BYTEA(self, type_, **kw):
2976 return "BYTEA"
2978 def visit_ARRAY(self, type_, **kw):
2979 inner = self.process(type_.item_type, **kw)
2980 return re.sub(
2981 r"((?: COLLATE.*)?)$",
2982 (
2983 r"%s\1"
2984 % (
2985 "[]"
2986 * (type_.dimensions if type_.dimensions is not None else 1)
2987 )
2988 ),
2989 inner,
2990 count=1,
2991 )
2993 def visit_json_path(self, type_, **kw):
2994 return self.visit_JSONPATH(type_, **kw)
2996 def visit_JSONPATH(self, type_, **kw):
2997 return "JSONPATH"
3000class _CompilerSequence:
3001 """Minimal stand-in for :class:`.Sequence`.
3003 Used for the implicit sequence behind a SERIAL column, where no
3004 :class:`.Sequence` object exists but a name still has to be rendered
3005 through :meth:`.IdentifierPreparer.format_sequence`.
3007 """
3009 __slots__ = ("name", "schema")
3011 # the schema handed to us has already been resolved against the
3012 # schema translate map, so don't let format_sequence() translate it
3013 # a second time
3014 _use_schema_map = False
3016 def __init__(self, name, schema=None):
3017 self.name = name
3018 self.schema = schema
3021class PGIdentifierPreparer(compiler.IdentifierPreparer):
3022 reserved_words = RESERVED_WORDS
3024 def _format_sequence_string_literal(self, sequence):
3025 """Render a sequence name as a SQL string literal.
3027 ``nextval()`` and friends take the sequence as a string rather than
3028 as an identifier, so the quoted identifier produced by
3029 :meth:`.format_sequence` has to be escaped a second time for the
3030 enclosing literal.
3032 """
3033 return "'%s'" % self.format_sequence(
3034 sequence, use_schema=True
3035 ).replace("'", "''")
3037 def _unquote_identifier(self, value):
3038 if value[0] == self.initial_quote:
3039 value = value[1:-1].replace(
3040 self.escape_to_quote, self.escape_quote
3041 )
3042 return value
3044 def format_type(self, type_, use_schema=True):
3045 if not type_.name:
3046 raise exc.CompileError(
3047 f"PostgreSQL {type_.__class__.__name__} type requires a name."
3048 )
3050 name = self.quote(type_.name)
3051 effective_schema = self.schema_for_object(type_)
3053 if (
3054 not self.omit_schema
3055 and use_schema
3056 and effective_schema is not None
3057 ):
3058 name = f"{self.quote_schema(effective_schema)}.{name}"
3059 return name
3062class ReflectedNamedType(TypedDict):
3063 """Represents a reflected named type."""
3065 name: str
3066 """Name of the type."""
3067 schema: str
3068 """The schema of the type."""
3069 visible: bool
3070 """Indicates if this type is in the current search path."""
3073class ReflectedDomainConstraint(TypedDict):
3074 """Represents a reflect check constraint of a domain."""
3076 name: str
3077 """Name of the constraint."""
3078 check: str
3079 """The check constraint text."""
3082class ReflectedDomain(ReflectedNamedType):
3083 """Represents a reflected enum."""
3085 type: str
3086 """The string name of the underlying data type of the domain."""
3087 nullable: bool
3088 """Indicates if the domain allows null or not."""
3089 default: Optional[str]
3090 """The string representation of the default value of this domain
3091 or ``None`` if none present.
3092 """
3093 constraints: List[ReflectedDomainConstraint]
3094 """The constraints defined in the domain, if any.
3095 The constraint are in order of evaluation by postgresql.
3096 """
3097 collation: Optional[str]
3098 """The collation for the domain."""
3101class ReflectedEnum(ReflectedNamedType):
3102 """Represents a reflected enum."""
3104 labels: List[str]
3105 """The labels that compose the enum."""
3108class PGInspector(reflection.Inspector):
3109 dialect: PGDialect
3111 def get_table_oid(
3112 self, table_name: str, schema: Optional[str] = None
3113 ) -> int:
3114 """Return the OID for the given table name.
3116 :param table_name: string name of the table. For special quoting,
3117 use :class:`.quoted_name`.
3119 :param schema: string schema name; if omitted, uses the default schema
3120 of the database connection. For special quoting,
3121 use :class:`.quoted_name`.
3123 """
3125 with self._operation_context() as conn:
3126 return self.dialect.get_table_oid(
3127 conn, table_name, schema, info_cache=self.info_cache
3128 )
3130 def get_domains(
3131 self, schema: Optional[str] = None
3132 ) -> List[ReflectedDomain]:
3133 """Return a list of DOMAIN objects.
3135 Each member is a dictionary containing these fields:
3137 * name - name of the domain
3138 * schema - the schema name for the domain.
3139 * visible - boolean, whether or not this domain is visible
3140 in the default search path.
3141 * type - the type defined by this domain.
3142 * nullable - Indicates if this domain can be ``NULL``.
3143 * default - The default value of the domain or ``None`` if the
3144 domain has no default.
3145 * constraints - A list of dict with the constraint defined by this
3146 domain. Each element contains two keys: ``name`` of the
3147 constraint and ``check`` with the constraint text.
3149 :param schema: schema name. If None, the default schema
3150 (typically 'public') is used. May also be set to ``'*'`` to
3151 indicate load domains for all schemas.
3153 .. versionadded:: 2.0
3155 """
3156 with self._operation_context() as conn:
3157 return self.dialect._load_domains(
3158 conn, schema, info_cache=self.info_cache
3159 )
3161 def get_enums(self, schema: Optional[str] = None) -> List[ReflectedEnum]:
3162 """Return a list of ENUM objects.
3164 Each member is a dictionary containing these fields:
3166 * name - name of the enum
3167 * schema - the schema name for the enum.
3168 * visible - boolean, whether or not this enum is visible
3169 in the default search path.
3170 * labels - a list of string labels that apply to the enum.
3172 :param schema: schema name. If None, the default schema
3173 (typically 'public') is used. May also be set to ``'*'`` to
3174 indicate load enums for all schemas.
3176 """
3177 with self._operation_context() as conn:
3178 return self.dialect._load_enums(
3179 conn, schema, info_cache=self.info_cache
3180 )
3182 def get_foreign_table_names(
3183 self, schema: Optional[str] = None
3184 ) -> List[str]:
3185 """Return a list of FOREIGN TABLE names.
3187 Behavior is similar to that of
3188 :meth:`_reflection.Inspector.get_table_names`,
3189 except that the list is limited to those tables that report a
3190 ``relkind`` value of ``f``.
3192 """
3193 with self._operation_context() as conn:
3194 return self.dialect._get_foreign_table_names(
3195 conn, schema, info_cache=self.info_cache
3196 )
3198 def has_type(
3199 self, type_name: str, schema: Optional[str] = None, **kw: Any
3200 ) -> bool:
3201 """Return if the database has the specified type in the provided
3202 schema.
3204 :param type_name: the type to check.
3205 :param schema: schema name. If None, the default schema
3206 (typically 'public') is used. May also be set to ``'*'`` to
3207 check in all schemas.
3209 .. versionadded:: 2.0
3211 """
3212 with self._operation_context() as conn:
3213 return self.dialect.has_type(
3214 conn, type_name, schema, info_cache=self.info_cache
3215 )
3218class PGExecutionContext(default.DefaultExecutionContext):
3219 def fire_sequence(self, seq, type_):
3220 return self._execute_scalar(
3221 (
3222 "select nextval(%s)"
3223 % self.identifier_preparer._format_sequence_string_literal(seq)
3224 ),
3225 type_,
3226 )
3228 def get_insert_default(self, column):
3229 if column.primary_key and column is column.table._autoincrement_column:
3230 if column.server_default and column.server_default.has_argument:
3231 # pre-execute passive defaults on primary key columns
3232 return self._execute_scalar(
3233 "select %s" % column.server_default.arg, column.type
3234 )
3236 elif column.default is None or (
3237 column.default.is_sequence and column.default.optional
3238 ):
3239 # execute the sequence associated with a SERIAL primary
3240 # key column. for non-primary-key SERIAL, the ID just
3241 # generates server side.
3243 try:
3244 seq_name = column._postgresql_seq_name
3245 except AttributeError:
3246 tab = column.table.name
3247 col = column.name
3248 tab = tab[0 : 29 + max(0, (29 - len(col)))]
3249 col = col[0 : 29 + max(0, (29 - len(tab)))]
3250 name = "%s_%s_seq" % (tab, col)
3251 column._postgresql_seq_name = seq_name = name
3253 if column.table is not None:
3254 effective_schema = self.connection.schema_for_object(
3255 column.table
3256 )
3257 else:
3258 effective_schema = None
3260 exc = (
3261 "select nextval(%s)"
3262 % self.identifier_preparer._format_sequence_string_literal(
3263 _CompilerSequence(seq_name, effective_schema)
3264 )
3265 )
3267 return self._execute_scalar(exc, column.type)
3269 return super().get_insert_default(column)
3272class PGReadOnlyConnectionCharacteristic(
3273 characteristics.ConnectionCharacteristic
3274):
3275 transactional = True
3277 def reset_characteristic(self, dialect, dbapi_conn):
3278 dialect.set_readonly(dbapi_conn, False)
3280 def set_characteristic(self, dialect, dbapi_conn, value):
3281 dialect.set_readonly(dbapi_conn, value)
3283 def get_characteristic(self, dialect, dbapi_conn):
3284 return dialect.get_readonly(dbapi_conn)
3287class PGDeferrableConnectionCharacteristic(
3288 characteristics.ConnectionCharacteristic
3289):
3290 transactional = True
3292 def reset_characteristic(self, dialect, dbapi_conn):
3293 dialect.set_deferrable(dbapi_conn, False)
3295 def set_characteristic(self, dialect, dbapi_conn, value):
3296 dialect.set_deferrable(dbapi_conn, value)
3298 def get_characteristic(self, dialect, dbapi_conn):
3299 return dialect.get_deferrable(dbapi_conn)
3302class PGDialect(default.DefaultDialect):
3303 name = "postgresql"
3304 supports_statement_cache = True
3305 supports_alter = True
3306 max_identifier_length = 63
3307 supports_sane_rowcount = True
3309 bind_typing = interfaces.BindTyping.RENDER_CASTS
3311 supports_native_enum = True
3312 supports_native_boolean = True
3313 supports_native_uuid = True
3314 supports_smallserial = True
3316 supports_sequences = True
3317 sequences_optional = True
3318 preexecute_autoincrement_sequences = True
3319 postfetch_lastrowid = False
3320 use_insertmanyvalues = True
3322 returns_native_bytes = True
3324 insertmanyvalues_implicit_sentinel = (
3325 InsertmanyvaluesSentinelOpts.ANY_AUTOINCREMENT
3326 | InsertmanyvaluesSentinelOpts.USE_INSERT_FROM_SELECT
3327 | InsertmanyvaluesSentinelOpts.RENDER_SELECT_COL_CASTS
3328 )
3330 supports_comments = True
3331 supports_constraint_comments = True
3332 supports_default_values = True
3334 supports_default_metavalue = True
3336 supports_empty_insert = False
3337 supports_multivalues_insert = True
3339 supports_identity_columns = True
3341 default_paramstyle = "pyformat"
3342 ischema_names = ischema_names
3343 colspecs = colspecs
3345 statement_compiler = PGCompiler
3346 ddl_compiler = PGDDLCompiler
3347 type_compiler_cls = PGTypeCompiler
3348 preparer = PGIdentifierPreparer
3349 execution_ctx_cls = PGExecutionContext
3350 inspector = PGInspector
3352 update_returning = True
3353 delete_returning = True
3354 insert_returning = True
3355 update_returning_multifrom = True
3356 delete_returning_multifrom = True
3358 connection_characteristics = (
3359 default.DefaultDialect.connection_characteristics
3360 )
3361 connection_characteristics = connection_characteristics.union(
3362 {
3363 "postgresql_readonly": PGReadOnlyConnectionCharacteristic(),
3364 "postgresql_deferrable": PGDeferrableConnectionCharacteristic(),
3365 }
3366 )
3368 construct_arguments = [
3369 (
3370 schema.Index,
3371 {
3372 "using": False,
3373 "include": None,
3374 "where": None,
3375 "ops": {},
3376 "concurrently": False,
3377 "with": {},
3378 "tablespace": None,
3379 "nulls_not_distinct": None,
3380 },
3381 ),
3382 (
3383 schema.Table,
3384 {
3385 "ignore_search_path": False,
3386 "tablespace": None,
3387 "partition_by": None,
3388 "with_oids": None,
3389 "on_commit": None,
3390 "inherits": None,
3391 "using": None,
3392 },
3393 ),
3394 (
3395 schema.CheckConstraint,
3396 {
3397 "not_valid": False,
3398 },
3399 ),
3400 (
3401 schema.ForeignKeyConstraint,
3402 {
3403 "not_valid": False,
3404 },
3405 ),
3406 (
3407 schema.PrimaryKeyConstraint,
3408 {"include": None},
3409 ),
3410 (
3411 schema.UniqueConstraint,
3412 {
3413 "include": None,
3414 "nulls_not_distinct": None,
3415 },
3416 ),
3417 ]
3419 reflection_options = ("postgresql_ignore_search_path",)
3421 _backslash_escapes = True
3422 _supports_create_index_concurrently = True
3423 _supports_drop_index_concurrently = True
3424 _supports_jsonb_subscripting = True
3426 def __init__(
3427 self,
3428 native_inet_types=None,
3429 json_serializer=None,
3430 json_deserializer=None,
3431 **kwargs,
3432 ):
3433 default.DefaultDialect.__init__(self, **kwargs)
3435 self._native_inet_types = native_inet_types
3436 self._json_deserializer = json_deserializer
3437 self._json_serializer = json_serializer
3439 def initialize(self, connection):
3440 super().initialize(connection)
3442 # https://www.postgresql.org/docs/9.3/static/release-9-2.html#AEN116689
3443 self.supports_smallserial = self.server_version_info >= (9, 2)
3445 self._set_backslash_escapes(connection)
3447 self._supports_drop_index_concurrently = self.server_version_info >= (
3448 9,
3449 2,
3450 )
3451 self.supports_identity_columns = self.server_version_info >= (10,)
3453 self._supports_jsonb_subscripting = self.server_version_info >= (14,)
3455 def get_isolation_level_values(self, dbapi_conn):
3456 # note the generic dialect doesn't have AUTOCOMMIT, however
3457 # all postgresql dialects should include AUTOCOMMIT.
3458 return (
3459 "SERIALIZABLE",
3460 "READ UNCOMMITTED",
3461 "READ COMMITTED",
3462 "REPEATABLE READ",
3463 )
3465 def set_isolation_level(self, dbapi_connection, level):
3466 cursor = dbapi_connection.cursor()
3467 cursor.execute(
3468 "SET SESSION CHARACTERISTICS AS TRANSACTION "
3469 f"ISOLATION LEVEL {level}"
3470 )
3471 cursor.execute("COMMIT")
3472 cursor.close()
3474 def get_isolation_level(self, dbapi_connection):
3475 cursor = dbapi_connection.cursor()
3476 cursor.execute("show transaction isolation level")
3477 val = cursor.fetchone()[0]
3478 cursor.close()
3479 return val.upper()
3481 def set_readonly(self, connection, value):
3482 raise NotImplementedError()
3484 def get_readonly(self, connection):
3485 raise NotImplementedError()
3487 def set_deferrable(self, connection, value):
3488 raise NotImplementedError()
3490 def get_deferrable(self, connection):
3491 raise NotImplementedError()
3493 def _split_multihost_from_url(self, url: URL) -> Union[
3494 Tuple[None, None],
3495 Tuple[Tuple[Optional[str], ...], Tuple[Optional[int], ...]],
3496 ]:
3497 hosts: Optional[Tuple[Optional[str], ...]] = None
3498 ports_str: Union[str, Tuple[Optional[str], ...], None] = None
3500 integrated_multihost = False
3502 if "host" in url.query:
3503 if isinstance(url.query["host"], (list, tuple)):
3504 integrated_multihost = True
3505 hosts, ports_str = zip(
3506 *[
3507 token.split(":") if ":" in token else (token, None)
3508 for token in url.query["host"]
3509 ]
3510 )
3512 elif isinstance(url.query["host"], str):
3513 hosts = tuple(url.query["host"].split(","))
3515 if (
3516 "port" not in url.query
3517 and len(hosts) == 1
3518 and ":" in hosts[0]
3519 ):
3520 # internet host is alphanumeric plus dots or hyphens.
3521 # this is essentially rfc1123, which refers to rfc952.
3522 # https://stackoverflow.com/questions/3523028/
3523 # valid-characters-of-a-hostname
3524 host_port_match = re.match(
3525 r"^([a-zA-Z0-9\-\.]*)(?:\:(\d*))?$", hosts[0]
3526 )
3527 if host_port_match:
3528 integrated_multihost = True
3529 h, p = host_port_match.group(1, 2)
3530 if TYPE_CHECKING:
3531 assert isinstance(h, str)
3532 assert isinstance(p, str)
3533 hosts = (h,)
3534 ports_str = cast(
3535 "Tuple[Optional[str], ...]", (p,) if p else (None,)
3536 )
3538 if "port" in url.query:
3539 if integrated_multihost:
3540 raise exc.ArgumentError(
3541 "Can't mix 'multihost' formats together; use "
3542 '"host=h1,h2,h3&port=p1,p2,p3" or '
3543 '"host=h1:p1&host=h2:p2&host=h3:p3" separately'
3544 )
3545 if isinstance(url.query["port"], (list, tuple)):
3546 ports_str = url.query["port"]
3547 elif isinstance(url.query["port"], str):
3548 ports_str = tuple(url.query["port"].split(","))
3550 ports: Optional[Tuple[Optional[int], ...]] = None
3552 if ports_str:
3553 try:
3554 ports = tuple(int(x) if x else None for x in ports_str)
3555 except ValueError:
3556 raise exc.ArgumentError(
3557 f"Received non-integer port arguments: {ports_str}"
3558 ) from None
3560 if ports and (
3561 (not hosts and len(ports) > 1)
3562 or (
3563 hosts
3564 and ports
3565 and len(hosts) != len(ports)
3566 and (len(hosts) > 1 or len(ports) > 1)
3567 )
3568 ):
3569 raise exc.ArgumentError("number of hosts and ports don't match")
3571 if hosts is not None:
3572 if ports is None:
3573 ports = tuple(None for _ in hosts)
3575 return hosts, ports # type: ignore
3577 def do_begin_twophase(self, connection, xid):
3578 self.do_begin(connection.connection)
3580 def do_prepare_twophase(self, connection, xid):
3581 connection.execute(
3582 sql.text("PREPARE TRANSACTION :xid").bindparams(
3583 sql.bindparam("xid", xid, literal_execute=True)
3584 )
3585 )
3587 def do_rollback_twophase(
3588 self, connection, xid, is_prepared=True, recover=False
3589 ):
3590 if is_prepared:
3591 if recover:
3592 # FIXME: ugly hack to get out of transaction
3593 # context when committing recoverable transactions
3594 # Must find out a way how to make the dbapi not
3595 # open a transaction.
3596 connection.exec_driver_sql("ROLLBACK")
3597 connection.execute(
3598 sql.text("ROLLBACK PREPARED :xid").bindparams(
3599 sql.bindparam("xid", xid, literal_execute=True)
3600 )
3601 )
3602 connection.exec_driver_sql("BEGIN")
3603 self.do_rollback(connection.connection)
3604 else:
3605 self.do_rollback(connection.connection)
3607 def do_commit_twophase(
3608 self, connection, xid, is_prepared=True, recover=False
3609 ):
3610 if is_prepared:
3611 if recover:
3612 connection.exec_driver_sql("ROLLBACK")
3613 connection.execute(
3614 sql.text("COMMIT PREPARED :xid").bindparams(
3615 sql.bindparam("xid", xid, literal_execute=True)
3616 )
3617 )
3618 connection.exec_driver_sql("BEGIN")
3619 self.do_rollback(connection.connection)
3620 else:
3621 self.do_commit(connection.connection)
3623 def do_recover_twophase(self, connection):
3624 return connection.scalars(
3625 sql.text("SELECT gid FROM pg_prepared_xacts")
3626 ).all()
3628 def _get_default_schema_name(self, connection):
3629 return connection.exec_driver_sql("select current_schema()").scalar()
3631 @reflection.cache
3632 def has_schema(self, connection, schema, **kw):
3633 query = select(pg_catalog.pg_namespace.c.nspname).where(
3634 pg_catalog.pg_namespace.c.nspname == schema
3635 )
3636 return bool(connection.scalar(query))
3638 def _pg_class_filter_scope_schema(
3639 self, query, schema, scope, pg_class_table=None
3640 ):
3641 if pg_class_table is None:
3642 pg_class_table = pg_catalog.pg_class
3643 query = query.join(
3644 pg_catalog.pg_namespace,
3645 pg_catalog.pg_namespace.c.oid == pg_class_table.c.relnamespace,
3646 )
3648 if scope is ObjectScope.DEFAULT:
3649 query = query.where(pg_class_table.c.relpersistence != "t")
3650 elif scope is ObjectScope.TEMPORARY:
3651 query = query.where(pg_class_table.c.relpersistence == "t")
3653 if schema is None:
3654 query = query.where(
3655 pg_catalog.pg_table_is_visible(pg_class_table.c.oid),
3656 # ignore pg_catalog schema
3657 pg_catalog.pg_namespace.c.nspname != "pg_catalog",
3658 )
3659 else:
3660 query = query.where(pg_catalog.pg_namespace.c.nspname == schema)
3661 return query
3663 def _pg_class_relkind_condition(self, relkinds, pg_class_table=None):
3664 if pg_class_table is None:
3665 pg_class_table = pg_catalog.pg_class
3666 # uses the any form instead of in otherwise postgresql complaings
3667 # that 'IN could not convert type character to "char"'
3668 return pg_class_table.c.relkind == sql.any_(_array.array(relkinds))
3670 @lru_cache()
3671 def _has_table_query(self, schema):
3672 query = select(pg_catalog.pg_class.c.relname).where(
3673 pg_catalog.pg_class.c.relname == bindparam("table_name"),
3674 self._pg_class_relkind_condition(
3675 pg_catalog.RELKINDS_ALL_TABLE_LIKE
3676 ),
3677 )
3678 return self._pg_class_filter_scope_schema(
3679 query, schema, scope=ObjectScope.ANY
3680 )
3682 @reflection.cache
3683 def has_table(self, connection, table_name, schema=None, **kw):
3684 self._ensure_has_table_connection(connection)
3685 query = self._has_table_query(schema)
3686 return bool(connection.scalar(query, {"table_name": table_name}))
3688 @reflection.cache
3689 def has_sequence(self, connection, sequence_name, schema=None, **kw):
3690 query = select(pg_catalog.pg_class.c.relname).where(
3691 pg_catalog.pg_class.c.relkind == "S",
3692 pg_catalog.pg_class.c.relname == sequence_name,
3693 )
3694 query = self._pg_class_filter_scope_schema(
3695 query, schema, scope=ObjectScope.ANY
3696 )
3697 return bool(connection.scalar(query))
3699 @reflection.cache
3700 def has_type(self, connection, type_name, schema=None, **kw):
3701 query = (
3702 select(pg_catalog.pg_type.c.typname)
3703 .join(
3704 pg_catalog.pg_namespace,
3705 pg_catalog.pg_namespace.c.oid
3706 == pg_catalog.pg_type.c.typnamespace,
3707 )
3708 .where(pg_catalog.pg_type.c.typname == type_name)
3709 )
3710 if schema is None:
3711 query = query.where(
3712 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid),
3713 # ignore pg_catalog schema
3714 pg_catalog.pg_namespace.c.nspname != "pg_catalog",
3715 )
3716 elif schema != "*":
3717 query = query.where(pg_catalog.pg_namespace.c.nspname == schema)
3719 return bool(connection.scalar(query))
3721 def _get_server_version_info(self, connection):
3722 v = connection.exec_driver_sql("select pg_catalog.version()").scalar()
3723 m = re.match(
3724 r".*(?:PostgreSQL|EnterpriseDB) "
3725 r"(\d+)\.?(\d+)?(?:\.(\d+))?(?:\.\d+)?(?:devel|beta)?",
3726 v,
3727 )
3728 if not m:
3729 raise AssertionError(
3730 "Could not determine version from string '%s'" % v
3731 )
3732 return tuple([int(x) for x in m.group(1, 2, 3) if x is not None])
3734 @reflection.cache
3735 def get_table_oid(self, connection, table_name, schema=None, **kw):
3736 """Fetch the oid for schema.table_name."""
3737 query = select(pg_catalog.pg_class.c.oid).where(
3738 pg_catalog.pg_class.c.relname == table_name,
3739 self._pg_class_relkind_condition(
3740 pg_catalog.RELKINDS_ALL_TABLE_LIKE
3741 ),
3742 )
3743 query = self._pg_class_filter_scope_schema(
3744 query, schema, scope=ObjectScope.ANY
3745 )
3746 table_oid = connection.scalar(query)
3747 if table_oid is None:
3748 raise exc.NoSuchTableError(
3749 f"{schema}.{table_name}" if schema else table_name
3750 )
3751 return table_oid
3753 @reflection.cache
3754 def get_schema_names(self, connection, **kw):
3755 query = (
3756 select(pg_catalog.pg_namespace.c.nspname)
3757 .where(pg_catalog.pg_namespace.c.nspname.not_like("pg_%"))
3758 .order_by(pg_catalog.pg_namespace.c.nspname)
3759 )
3760 return connection.scalars(query).all()
3762 def _get_relnames_for_relkinds(self, connection, schema, relkinds, scope):
3763 query = select(pg_catalog.pg_class.c.relname).where(
3764 self._pg_class_relkind_condition(relkinds)
3765 )
3766 query = self._pg_class_filter_scope_schema(query, schema, scope=scope)
3767 return connection.scalars(query).all()
3769 @reflection.cache
3770 def get_table_names(self, connection, schema=None, **kw):
3771 return self._get_relnames_for_relkinds(
3772 connection,
3773 schema,
3774 pg_catalog.RELKINDS_TABLE_NO_FOREIGN,
3775 scope=ObjectScope.DEFAULT,
3776 )
3778 @reflection.cache
3779 def get_temp_table_names(self, connection, **kw):
3780 return self._get_relnames_for_relkinds(
3781 connection,
3782 schema=None,
3783 relkinds=pg_catalog.RELKINDS_TABLE_NO_FOREIGN,
3784 scope=ObjectScope.TEMPORARY,
3785 )
3787 @reflection.cache
3788 def _get_foreign_table_names(self, connection, schema=None, **kw):
3789 return self._get_relnames_for_relkinds(
3790 connection, schema, relkinds=("f",), scope=ObjectScope.ANY
3791 )
3793 @reflection.cache
3794 def get_view_names(self, connection, schema=None, **kw):
3795 return self._get_relnames_for_relkinds(
3796 connection,
3797 schema,
3798 pg_catalog.RELKINDS_VIEW,
3799 scope=ObjectScope.DEFAULT,
3800 )
3802 @reflection.cache
3803 def get_materialized_view_names(self, connection, schema=None, **kw):
3804 return self._get_relnames_for_relkinds(
3805 connection,
3806 schema,
3807 pg_catalog.RELKINDS_MAT_VIEW,
3808 scope=ObjectScope.DEFAULT,
3809 )
3811 @reflection.cache
3812 def get_temp_view_names(self, connection, schema=None, **kw):
3813 return self._get_relnames_for_relkinds(
3814 connection,
3815 schema,
3816 # NOTE: do not include temp materialzied views (that do not
3817 # seem to be a thing at least up to version 14)
3818 pg_catalog.RELKINDS_VIEW,
3819 scope=ObjectScope.TEMPORARY,
3820 )
3822 @reflection.cache
3823 def get_sequence_names(self, connection, schema=None, **kw):
3824 return self._get_relnames_for_relkinds(
3825 connection, schema, relkinds=("S",), scope=ObjectScope.ANY
3826 )
3828 @reflection.cache
3829 def get_view_definition(self, connection, view_name, schema=None, **kw):
3830 query = (
3831 select(pg_catalog.pg_get_viewdef(pg_catalog.pg_class.c.oid))
3832 .select_from(pg_catalog.pg_class)
3833 .where(
3834 pg_catalog.pg_class.c.relname == view_name,
3835 self._pg_class_relkind_condition(
3836 pg_catalog.RELKINDS_VIEW + pg_catalog.RELKINDS_MAT_VIEW
3837 ),
3838 )
3839 )
3840 query = self._pg_class_filter_scope_schema(
3841 query, schema, scope=ObjectScope.ANY
3842 )
3843 res = connection.scalar(query)
3844 if res is None:
3845 raise exc.NoSuchTableError(
3846 f"{schema}.{view_name}" if schema else view_name
3847 )
3848 else:
3849 return res
3851 def _value_or_raise(self, data, table, schema):
3852 try:
3853 return dict(data)[(schema, table)]
3854 except KeyError:
3855 raise exc.NoSuchTableError(
3856 f"{schema}.{table}" if schema else table
3857 ) from None
3859 def _prepare_filter_names(self, filter_names):
3860 if filter_names:
3861 return True, {"filter_names": filter_names}
3862 else:
3863 return False, {}
3865 def _kind_to_relkinds(self, kind: ObjectKind) -> Tuple[str, ...]:
3866 if kind is ObjectKind.ANY:
3867 return pg_catalog.RELKINDS_ALL_TABLE_LIKE
3868 relkinds = ()
3869 if ObjectKind.TABLE in kind:
3870 relkinds += pg_catalog.RELKINDS_TABLE
3871 if ObjectKind.VIEW in kind:
3872 relkinds += pg_catalog.RELKINDS_VIEW
3873 if ObjectKind.MATERIALIZED_VIEW in kind:
3874 relkinds += pg_catalog.RELKINDS_MAT_VIEW
3875 return relkinds
3877 @reflection.cache
3878 def get_columns(self, connection, table_name, schema=None, **kw):
3879 data = self.get_multi_columns(
3880 connection,
3881 schema=schema,
3882 filter_names=[table_name],
3883 scope=ObjectScope.ANY,
3884 kind=ObjectKind.ANY,
3885 **kw,
3886 )
3887 return self._value_or_raise(data, table_name, schema)
3889 @lru_cache()
3890 def _columns_query(self, schema, has_filter_names, scope, kind):
3891 # NOTE: the query with the default and identity options scalar
3892 # subquery is faster than trying to use outer joins for them
3893 generated = (
3894 pg_catalog.pg_attribute.c.attgenerated.label("generated")
3895 if self.server_version_info >= (12,)
3896 else sql.null().label("generated")
3897 )
3898 if self.server_version_info >= (10,):
3899 # join lateral performs worse (~2x slower) than a scalar_subquery
3900 identity = (
3901 select(
3902 sql.func.json_build_object(
3903 "always",
3904 pg_catalog.pg_attribute.c.attidentity == "a",
3905 "start",
3906 pg_catalog.pg_sequence.c.seqstart,
3907 "increment",
3908 pg_catalog.pg_sequence.c.seqincrement,
3909 "minvalue",
3910 pg_catalog.pg_sequence.c.seqmin,
3911 "maxvalue",
3912 pg_catalog.pg_sequence.c.seqmax,
3913 "cache",
3914 pg_catalog.pg_sequence.c.seqcache,
3915 "cycle",
3916 pg_catalog.pg_sequence.c.seqcycle,
3917 type_=sqltypes.JSON(),
3918 )
3919 )
3920 .select_from(pg_catalog.pg_sequence)
3921 .where(
3922 # attidentity != '' is required or it will reflect also
3923 # serial columns as identity.
3924 pg_catalog.pg_attribute.c.attidentity != "",
3925 pg_catalog.pg_sequence.c.seqrelid
3926 == sql.cast(
3927 sql.cast(
3928 pg_catalog.pg_get_serial_sequence(
3929 sql.cast(
3930 sql.cast(
3931 pg_catalog.pg_attribute.c.attrelid,
3932 REGCLASS,
3933 ),
3934 TEXT,
3935 ),
3936 pg_catalog.pg_attribute.c.attname,
3937 ),
3938 REGCLASS,
3939 ),
3940 OID,
3941 ),
3942 )
3943 .correlate(pg_catalog.pg_attribute)
3944 .scalar_subquery()
3945 .label("identity_options")
3946 )
3947 else:
3948 identity = sql.null().label("identity_options")
3950 # join lateral performs the same as scalar_subquery here
3951 default = (
3952 select(
3953 pg_catalog.pg_get_expr(
3954 pg_catalog.pg_attrdef.c.adbin,
3955 pg_catalog.pg_attrdef.c.adrelid,
3956 )
3957 )
3958 .select_from(pg_catalog.pg_attrdef)
3959 .where(
3960 pg_catalog.pg_attrdef.c.adrelid
3961 == pg_catalog.pg_attribute.c.attrelid,
3962 pg_catalog.pg_attrdef.c.adnum
3963 == pg_catalog.pg_attribute.c.attnum,
3964 pg_catalog.pg_attribute.c.atthasdef,
3965 )
3966 .correlate(pg_catalog.pg_attribute)
3967 .scalar_subquery()
3968 .label("default")
3969 )
3971 # get the name of the collate when it's different from the default one
3972 collate = sql.case(
3973 (
3974 sql.and_(
3975 pg_catalog.pg_attribute.c.attcollation != 0,
3976 select(pg_catalog.pg_type.c.typcollation)
3977 .where(
3978 pg_catalog.pg_type.c.oid
3979 == pg_catalog.pg_attribute.c.atttypid,
3980 )
3981 .correlate(pg_catalog.pg_attribute)
3982 .scalar_subquery()
3983 != pg_catalog.pg_attribute.c.attcollation,
3984 ),
3985 select(pg_catalog.pg_collation.c.collname)
3986 .where(
3987 pg_catalog.pg_collation.c.oid
3988 == pg_catalog.pg_attribute.c.attcollation
3989 )
3990 .correlate(pg_catalog.pg_attribute)
3991 .scalar_subquery(),
3992 ),
3993 else_=sql.null(),
3994 ).label("collation")
3996 relkinds = self._kind_to_relkinds(kind)
3997 query = (
3998 select(
3999 pg_catalog.pg_attribute.c.attname.label("name"),
4000 pg_catalog.format_type(
4001 pg_catalog.pg_attribute.c.atttypid,
4002 pg_catalog.pg_attribute.c.atttypmod,
4003 ).label("format_type"),
4004 default,
4005 pg_catalog.pg_attribute.c.attnotnull.label("not_null"),
4006 pg_catalog.pg_class.c.relname.label("table_name"),
4007 pg_catalog.pg_description.c.description.label("comment"),
4008 generated,
4009 identity,
4010 collate,
4011 )
4012 .select_from(pg_catalog.pg_class)
4013 # NOTE: postgresql support table with no user column, meaning
4014 # there is no row with pg_attribute.attnum > 0. use a left outer
4015 # join to avoid filtering these tables.
4016 .outerjoin(
4017 pg_catalog.pg_attribute,
4018 sql.and_(
4019 pg_catalog.pg_class.c.oid
4020 == pg_catalog.pg_attribute.c.attrelid,
4021 pg_catalog.pg_attribute.c.attnum > 0,
4022 ~pg_catalog.pg_attribute.c.attisdropped,
4023 ),
4024 )
4025 .outerjoin(
4026 pg_catalog.pg_description,
4027 sql.and_(
4028 pg_catalog.pg_description.c.objoid
4029 == pg_catalog.pg_attribute.c.attrelid,
4030 pg_catalog.pg_description.c.objsubid
4031 == pg_catalog.pg_attribute.c.attnum,
4032 ),
4033 )
4034 .where(self._pg_class_relkind_condition(relkinds))
4035 .order_by(
4036 pg_catalog.pg_class.c.relname, pg_catalog.pg_attribute.c.attnum
4037 )
4038 )
4039 query = self._pg_class_filter_scope_schema(query, schema, scope=scope)
4040 if has_filter_names:
4041 query = query.where(
4042 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
4043 )
4044 return query
4046 def get_multi_columns(
4047 self, connection, schema, filter_names, scope, kind, **kw
4048 ):
4049 has_filter_names, params = self._prepare_filter_names(filter_names)
4050 query = self._columns_query(schema, has_filter_names, scope, kind)
4051 rows = connection.execute(query, params).mappings()
4053 # dictionary with (name, ) if default search path or (schema, name)
4054 # as keys
4055 domains = {
4056 ((d["schema"], d["name"]) if not d["visible"] else (d["name"],)): d
4057 for d in self._load_domains(
4058 connection, schema="*", info_cache=kw.get("info_cache")
4059 )
4060 }
4062 # dictionary with (name, ) if default search path or (schema, name)
4063 # as keys
4064 enums = dict(
4065 (
4066 ((rec["name"],), rec)
4067 if rec["visible"]
4068 else ((rec["schema"], rec["name"]), rec)
4069 )
4070 for rec in self._load_enums(
4071 connection, schema="*", info_cache=kw.get("info_cache")
4072 )
4073 )
4075 columns = self._get_columns_info(rows, domains, enums, schema)
4077 return columns.items()
4079 _format_type_args_pattern = re.compile(r"\((.*)\)")
4080 _format_type_args_delim = re.compile(r"\s*,\s*")
4081 _format_array_spec_pattern = re.compile(r"((?:\[\])*)$")
4083 def _reflect_type(
4084 self,
4085 format_type: Optional[str],
4086 domains: Dict[str, ReflectedDomain],
4087 enums: Dict[str, ReflectedEnum],
4088 type_description: str,
4089 collation: Optional[str],
4090 ) -> sqltypes.TypeEngine[Any]:
4091 """
4092 Attempts to reconstruct a column type defined in ischema_names based
4093 on the information available in the format_type.
4095 If the `format_type` cannot be associated with a known `ischema_names`,
4096 it is treated as a reference to a known PostgreSQL named `ENUM` or
4097 `DOMAIN` type.
4098 """
4099 type_description = type_description or "unknown type"
4100 if format_type is None:
4101 util.warn(
4102 "PostgreSQL format_type() returned NULL for %s"
4103 % type_description
4104 )
4105 return sqltypes.NULLTYPE
4107 attype_args_match = self._format_type_args_pattern.search(format_type)
4108 if attype_args_match and attype_args_match.group(1):
4109 attype_args = self._format_type_args_delim.split(
4110 attype_args_match.group(1)
4111 )
4112 else:
4113 attype_args = ()
4115 match_array_dim = self._format_array_spec_pattern.search(format_type)
4116 # Each "[]" in array specs corresponds to an array dimension
4117 array_dim = len(match_array_dim.group(1) or "") // 2
4119 # Remove all parameters and array specs from format_type to obtain an
4120 # ischema_name candidate
4121 attype = self._format_type_args_pattern.sub("", format_type)
4122 attype = self._format_array_spec_pattern.sub("", attype)
4124 schema_type = self.ischema_names.get(attype.lower(), None)
4125 args, kwargs = (), {}
4127 if attype == "numeric":
4128 if len(attype_args) == 2:
4129 precision, scale = map(int, attype_args)
4130 args = (precision, scale)
4132 elif attype == "double precision":
4133 args = (53,)
4135 elif attype == "integer":
4136 args = ()
4138 elif attype in ("timestamp with time zone", "time with time zone"):
4139 kwargs["timezone"] = True
4140 if len(attype_args) == 1:
4141 kwargs["precision"] = int(attype_args[0])
4143 elif attype in (
4144 "timestamp without time zone",
4145 "time without time zone",
4146 "time",
4147 ):
4148 kwargs["timezone"] = False
4149 if len(attype_args) == 1:
4150 kwargs["precision"] = int(attype_args[0])
4152 elif attype == "bit varying":
4153 kwargs["varying"] = True
4154 if len(attype_args) == 1:
4155 charlen = int(attype_args[0])
4156 args = (charlen,)
4158 # a domain or enum can start with interval, so be mindful of that.
4159 elif attype == "interval" or attype.startswith("interval "):
4160 schema_type = INTERVAL
4162 field_match = re.match(r"interval (.+)", attype)
4163 if field_match:
4164 kwargs["fields"] = field_match.group(1)
4166 if len(attype_args) == 1:
4167 kwargs["precision"] = int(attype_args[0])
4169 else:
4170 enum_or_domain_key = tuple(util.quoted_token_parser(attype))
4172 if enum_or_domain_key in enums:
4173 schema_type = ENUM
4174 enum = enums[enum_or_domain_key]
4176 kwargs["name"] = enum["name"]
4178 if not enum["visible"]:
4179 kwargs["schema"] = enum["schema"]
4180 args = tuple(enum["labels"])
4181 elif enum_or_domain_key in domains:
4182 schema_type = DOMAIN
4183 domain = domains[enum_or_domain_key]
4185 data_type = self._reflect_type(
4186 domain["type"],
4187 domains,
4188 enums,
4189 type_description="DOMAIN '%s'" % domain["name"],
4190 collation=domain["collation"],
4191 )
4192 args = (domain["name"], data_type)
4194 kwargs["collation"] = domain["collation"]
4195 kwargs["default"] = domain["default"]
4196 kwargs["not_null"] = not domain["nullable"]
4197 kwargs["create_type"] = False
4199 if domain["constraints"]:
4200 # We only support a single constraint
4201 check_constraint = domain["constraints"][0]
4203 kwargs["constraint_name"] = check_constraint["name"]
4204 kwargs["check"] = check_constraint["check"]
4206 if not domain["visible"]:
4207 kwargs["schema"] = domain["schema"]
4209 else:
4210 try:
4211 charlen = int(attype_args[0])
4212 args = (charlen, *attype_args[1:])
4213 except (ValueError, IndexError):
4214 args = attype_args
4216 if not schema_type:
4217 util.warn(
4218 "Did not recognize type '%s' of %s"
4219 % (attype, type_description)
4220 )
4221 return sqltypes.NULLTYPE
4223 if collation is not None:
4224 kwargs["collation"] = collation
4226 data_type = schema_type(*args, **kwargs)
4227 if array_dim >= 1:
4228 # postgres does not preserve dimensionality or size of array types.
4229 data_type = _array.ARRAY(data_type)
4231 return data_type
4233 def _get_columns_info(self, rows, domains, enums, schema):
4234 columns = defaultdict(list)
4235 for row_dict in rows:
4236 # ensure that each table has an entry, even if it has no columns
4237 if row_dict["name"] is None:
4238 columns[(schema, row_dict["table_name"])] = (
4239 ReflectionDefaults.columns()
4240 )
4241 continue
4242 table_cols = columns[(schema, row_dict["table_name"])]
4244 collation = row_dict["collation"]
4246 coltype = self._reflect_type(
4247 row_dict["format_type"],
4248 domains,
4249 enums,
4250 type_description="column '%s'" % row_dict["name"],
4251 collation=collation,
4252 )
4254 default = row_dict["default"]
4255 name = row_dict["name"]
4256 generated = row_dict["generated"]
4257 nullable = not row_dict["not_null"]
4259 if isinstance(coltype, DOMAIN):
4260 if not default:
4261 # domain can override the default value but
4262 # can't set it to None
4263 if coltype.default is not None:
4264 default = coltype.default
4266 nullable = nullable and not coltype.not_null
4268 identity = row_dict["identity_options"]
4270 # If a zero byte or blank string depending on driver (is also
4271 # absent for older PG versions), then not a generated column.
4272 # Otherwise, s = stored. (Other values might be added in the
4273 # future.)
4274 if generated not in (None, "", b"\x00"):
4275 computed = dict(
4276 sqltext=default, persisted=generated in ("s", b"s")
4277 )
4278 default = None
4279 else:
4280 computed = None
4282 # adjust the default value
4283 autoincrement = False
4284 if default is not None:
4285 match = re.search(r"""(nextval\(')([^']+)('.*$)""", default)
4286 if match is not None:
4287 if issubclass(coltype._type_affinity, sqltypes.Integer):
4288 autoincrement = True
4289 # the default is related to a Sequence
4290 if "." not in match.group(2) and schema is not None:
4291 # unconditionally quote the schema name. this could
4292 # later be enhanced to obey quoting rules /
4293 # "quote schema"
4294 default = (
4295 match.group(1)
4296 + ('"%s"' % schema)
4297 + "."
4298 + match.group(2)
4299 + match.group(3)
4300 )
4302 column_info = {
4303 "name": name,
4304 "type": coltype,
4305 "nullable": nullable,
4306 "default": default,
4307 "autoincrement": autoincrement or identity is not None,
4308 "comment": row_dict["comment"],
4309 }
4310 if computed is not None:
4311 column_info["computed"] = computed
4312 if identity is not None:
4313 column_info["identity"] = identity
4315 table_cols.append(column_info)
4317 return columns
4319 @lru_cache()
4320 def _table_oids_query(self, schema, has_filter_names, scope, kind):
4321 relkinds = self._kind_to_relkinds(kind)
4322 oid_q = select(
4323 pg_catalog.pg_class.c.oid, pg_catalog.pg_class.c.relname
4324 ).where(self._pg_class_relkind_condition(relkinds))
4325 oid_q = self._pg_class_filter_scope_schema(oid_q, schema, scope=scope)
4327 if has_filter_names:
4328 oid_q = oid_q.where(
4329 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
4330 )
4331 return oid_q
4333 @reflection.flexi_cache(
4334 ("schema", InternalTraversal.dp_string),
4335 ("filter_names", InternalTraversal.dp_string_list),
4336 ("kind", InternalTraversal.dp_plain_obj),
4337 ("scope", InternalTraversal.dp_plain_obj),
4338 )
4339 def _get_table_oids(
4340 self, connection, schema, filter_names, scope, kind, **kw
4341 ):
4342 has_filter_names, params = self._prepare_filter_names(filter_names)
4343 oid_q = self._table_oids_query(schema, has_filter_names, scope, kind)
4344 result = connection.execute(oid_q, params)
4345 return result.all()
4347 @util.memoized_property
4348 def _constraint_query(self):
4349 if self.server_version_info >= (11, 0):
4350 indnkeyatts = pg_catalog.pg_index.c.indnkeyatts
4351 else:
4352 indnkeyatts = pg_catalog.pg_index.c.indnatts.label("indnkeyatts")
4354 if self.server_version_info >= (15,):
4355 indnullsnotdistinct = pg_catalog.pg_index.c.indnullsnotdistinct
4356 else:
4357 indnullsnotdistinct = sql.false().label("indnullsnotdistinct")
4359 con_sq = (
4360 select(
4361 pg_catalog.pg_constraint.c.conrelid,
4362 pg_catalog.pg_constraint.c.conname,
4363 sql.func.unnest(pg_catalog.pg_index.c.indkey).label("attnum"),
4364 sql.func.generate_subscripts(
4365 pg_catalog.pg_index.c.indkey, 1
4366 ).label("ord"),
4367 indnkeyatts,
4368 indnullsnotdistinct,
4369 pg_catalog.pg_description.c.description,
4370 )
4371 .join(
4372 pg_catalog.pg_index,
4373 pg_catalog.pg_constraint.c.conindid
4374 == pg_catalog.pg_index.c.indexrelid,
4375 )
4376 .outerjoin(
4377 pg_catalog.pg_description,
4378 pg_catalog.pg_description.c.objoid
4379 == pg_catalog.pg_constraint.c.oid,
4380 )
4381 .where(
4382 pg_catalog.pg_constraint.c.contype == bindparam("contype"),
4383 pg_catalog.pg_constraint.c.conrelid.in_(bindparam("oids")),
4384 # NOTE: filtering also on pg_index.indrelid for oids does
4385 # not seem to have a performance effect, but it may be an
4386 # option if perf problems are reported
4387 )
4388 .subquery("con")
4389 )
4391 attr_sq = (
4392 select(
4393 con_sq.c.conrelid,
4394 con_sq.c.conname,
4395 con_sq.c.description,
4396 con_sq.c.ord,
4397 con_sq.c.indnkeyatts,
4398 con_sq.c.indnullsnotdistinct,
4399 pg_catalog.pg_attribute.c.attname,
4400 )
4401 .select_from(pg_catalog.pg_attribute)
4402 .join(
4403 con_sq,
4404 sql.and_(
4405 pg_catalog.pg_attribute.c.attnum == con_sq.c.attnum,
4406 pg_catalog.pg_attribute.c.attrelid == con_sq.c.conrelid,
4407 ),
4408 )
4409 .where(
4410 # NOTE: restate the condition here, since pg15 otherwise
4411 # seems to get confused on pscopg2 sometimes, doing
4412 # a sequential scan of pg_attribute.
4413 # The condition in the con_sq subquery is not actually needed
4414 # in pg15, but it may be needed in older versions. Keeping it
4415 # does not seems to have any impact in any case.
4416 con_sq.c.conrelid.in_(bindparam("oids"))
4417 )
4418 .subquery("attr")
4419 )
4421 return (
4422 select(
4423 attr_sq.c.conrelid,
4424 sql.func.array_agg(
4425 # NOTE: cast since some postgresql derivatives may
4426 # not support array_agg on the name type
4427 aggregate_order_by(
4428 attr_sq.c.attname.cast(TEXT), attr_sq.c.ord
4429 )
4430 ).label("cols"),
4431 attr_sq.c.conname,
4432 sql.func.min(attr_sq.c.description).label("description"),
4433 sql.func.min(attr_sq.c.indnkeyatts).label("indnkeyatts"),
4434 sql.func.bool_and(attr_sq.c.indnullsnotdistinct).label(
4435 "indnullsnotdistinct"
4436 ),
4437 )
4438 .group_by(attr_sq.c.conrelid, attr_sq.c.conname)
4439 .order_by(attr_sq.c.conrelid, attr_sq.c.conname)
4440 )
4442 def _reflect_constraint(
4443 self, connection, contype, schema, filter_names, scope, kind, **kw
4444 ):
4445 # used to reflect primary and unique constraint
4446 table_oids = self._get_table_oids(
4447 connection, schema, filter_names, scope, kind, **kw
4448 )
4449 batches = list(table_oids)
4450 is_unique = contype == "u"
4452 while batches:
4453 batch = batches[0:3000]
4454 batches[0:3000] = []
4456 result = connection.execute(
4457 self._constraint_query,
4458 {"oids": [r[0] for r in batch], "contype": contype},
4459 ).mappings()
4461 result_by_oid = defaultdict(list)
4462 for row_dict in result:
4463 result_by_oid[row_dict["conrelid"]].append(row_dict)
4465 for oid, tablename in batch:
4466 for_oid = result_by_oid.get(oid, ())
4467 if for_oid:
4468 for row in for_oid:
4469 # See note in get_multi_indexes
4470 all_cols = row["cols"]
4471 indnkeyatts = row["indnkeyatts"]
4472 if len(all_cols) > indnkeyatts:
4473 inc_cols = all_cols[indnkeyatts:]
4474 cst_cols = all_cols[:indnkeyatts]
4475 else:
4476 inc_cols = []
4477 cst_cols = all_cols
4479 opts = {}
4480 if self.server_version_info >= (11,):
4481 opts["postgresql_include"] = inc_cols
4482 if is_unique:
4483 opts["postgresql_nulls_not_distinct"] = row[
4484 "indnullsnotdistinct"
4485 ]
4486 yield (
4487 tablename,
4488 cst_cols,
4489 row["conname"],
4490 row["description"],
4491 opts,
4492 )
4493 else:
4494 yield tablename, None, None, None, None
4496 @reflection.cache
4497 def get_pk_constraint(self, connection, table_name, schema=None, **kw):
4498 data = self.get_multi_pk_constraint(
4499 connection,
4500 schema=schema,
4501 filter_names=[table_name],
4502 scope=ObjectScope.ANY,
4503 kind=ObjectKind.ANY,
4504 **kw,
4505 )
4506 return self._value_or_raise(data, table_name, schema)
4508 def get_multi_pk_constraint(
4509 self, connection, schema, filter_names, scope, kind, **kw
4510 ):
4511 result = self._reflect_constraint(
4512 connection, "p", schema, filter_names, scope, kind, **kw
4513 )
4515 # only a single pk can be present for each table. Return an entry
4516 # even if a table has no primary key
4517 default = ReflectionDefaults.pk_constraint
4519 def pk_constraint(pk_name, cols, comment, opts):
4520 info = {
4521 "constrained_columns": cols,
4522 "name": pk_name,
4523 "comment": comment,
4524 }
4525 if opts:
4526 info["dialect_options"] = opts
4527 return info
4529 return (
4530 (
4531 (schema, table_name),
4532 (
4533 pk_constraint(pk_name, cols, comment, opts)
4534 if pk_name is not None
4535 else default()
4536 ),
4537 )
4538 for table_name, cols, pk_name, comment, opts in result
4539 )
4541 @reflection.cache
4542 def get_foreign_keys(
4543 self,
4544 connection,
4545 table_name,
4546 schema=None,
4547 postgresql_ignore_search_path=False,
4548 **kw,
4549 ):
4550 data = self.get_multi_foreign_keys(
4551 connection,
4552 schema=schema,
4553 filter_names=[table_name],
4554 postgresql_ignore_search_path=postgresql_ignore_search_path,
4555 scope=ObjectScope.ANY,
4556 kind=ObjectKind.ANY,
4557 **kw,
4558 )
4559 return self._value_or_raise(data, table_name, schema)
4561 @lru_cache()
4562 def _foreing_key_query(self, schema, has_filter_names, scope, kind):
4563 pg_class_ref = pg_catalog.pg_class.alias("cls_ref")
4564 pg_namespace_ref = pg_catalog.pg_namespace.alias("nsp_ref")
4565 relkinds = self._kind_to_relkinds(kind)
4566 query = (
4567 select(
4568 pg_catalog.pg_class.c.relname,
4569 pg_catalog.pg_constraint.c.conname,
4570 # NOTE: avoid calling pg_get_constraintdef when not needed
4571 # to speed up the query
4572 sql.case(
4573 (
4574 pg_catalog.pg_constraint.c.oid.is_not(None),
4575 pg_catalog.pg_get_constraintdef(
4576 pg_catalog.pg_constraint.c.oid, True
4577 ),
4578 ),
4579 else_=None,
4580 ),
4581 pg_namespace_ref.c.nspname,
4582 pg_catalog.pg_description.c.description,
4583 )
4584 .select_from(pg_catalog.pg_class)
4585 .outerjoin(
4586 pg_catalog.pg_constraint,
4587 sql.and_(
4588 pg_catalog.pg_class.c.oid
4589 == pg_catalog.pg_constraint.c.conrelid,
4590 pg_catalog.pg_constraint.c.contype == "f",
4591 ),
4592 )
4593 .outerjoin(
4594 pg_class_ref,
4595 pg_class_ref.c.oid == pg_catalog.pg_constraint.c.confrelid,
4596 )
4597 .outerjoin(
4598 pg_namespace_ref,
4599 pg_class_ref.c.relnamespace == pg_namespace_ref.c.oid,
4600 )
4601 .outerjoin(
4602 pg_catalog.pg_description,
4603 pg_catalog.pg_description.c.objoid
4604 == pg_catalog.pg_constraint.c.oid,
4605 )
4606 .order_by(
4607 pg_catalog.pg_class.c.relname,
4608 pg_catalog.pg_constraint.c.conname,
4609 )
4610 .where(self._pg_class_relkind_condition(relkinds))
4611 )
4612 query = self._pg_class_filter_scope_schema(query, schema, scope)
4613 if has_filter_names:
4614 query = query.where(
4615 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
4616 )
4617 return query
4619 @util.memoized_property
4620 def _fk_regex_pattern(self):
4621 # optionally quoted token
4622 qtoken = r'(?:"(?:[^"]|"")+"|[\w]+?)'
4624 # https://www.postgresql.org/docs/current/static/sql-createtable.html
4625 return re.compile(
4626 r"FOREIGN KEY \((.*?)\) "
4627 rf"REFERENCES (?:({qtoken})\.)?({qtoken})\(((?:{qtoken}(?: *, *)?)+)\)" # noqa: E501
4628 r"[\s]?(MATCH (FULL|PARTIAL|SIMPLE)+)?"
4629 r"[\s]?(?:ON (UPDATE|DELETE) "
4630 r"(CASCADE|RESTRICT|NO ACTION|"
4631 r"SET (?:NULL|DEFAULT)(?:\s\(.+\))?)+)?"
4632 r"[\s]?(?:ON (UPDATE|DELETE) "
4633 r"(CASCADE|RESTRICT|NO ACTION|"
4634 r"SET (?:NULL|DEFAULT)(?:\s\(.+\))?)+)?"
4635 r"[\s]?(DEFERRABLE|NOT DEFERRABLE)?"
4636 r"[\s]?(INITIALLY (DEFERRED|IMMEDIATE)+)?"
4637 )
4639 def _parse_fk(self, condef):
4640 FK_REGEX = self._fk_regex_pattern
4641 m = re.search(FK_REGEX, condef).groups()
4643 (
4644 constrained_columns,
4645 referred_schema,
4646 referred_table,
4647 referred_columns,
4648 _,
4649 match,
4650 upddelkey1,
4651 upddelval1,
4652 upddelkey2,
4653 upddelval2,
4654 deferrable,
4655 _,
4656 initially,
4657 ) = m
4659 onupdate = (
4660 upddelval1
4661 if upddelkey1 == "UPDATE"
4662 else upddelval2 if upddelkey2 == "UPDATE" else None
4663 )
4664 ondelete = (
4665 upddelval1
4666 if upddelkey1 == "DELETE"
4667 else upddelval2 if upddelkey2 == "DELETE" else None
4668 )
4670 return (
4671 constrained_columns,
4672 referred_schema,
4673 referred_table,
4674 referred_columns,
4675 match,
4676 onupdate,
4677 ondelete,
4678 deferrable,
4679 initially,
4680 )
4682 def get_multi_foreign_keys(
4683 self,
4684 connection,
4685 schema,
4686 filter_names,
4687 scope,
4688 kind,
4689 postgresql_ignore_search_path=False,
4690 **kw,
4691 ):
4692 preparer = self.identifier_preparer
4694 has_filter_names, params = self._prepare_filter_names(filter_names)
4695 query = self._foreing_key_query(schema, has_filter_names, scope, kind)
4696 result = connection.execute(query, params)
4698 fkeys = defaultdict(list)
4699 default = ReflectionDefaults.foreign_keys
4700 for table_name, conname, condef, conschema, comment in result:
4701 # ensure that each table has an entry, even if it has
4702 # no foreign keys
4703 if conname is None:
4704 fkeys[(schema, table_name)] = default()
4705 continue
4706 table_fks = fkeys[(schema, table_name)]
4708 (
4709 constrained_columns,
4710 referred_schema,
4711 referred_table,
4712 referred_columns,
4713 match,
4714 onupdate,
4715 ondelete,
4716 deferrable,
4717 initially,
4718 ) = self._parse_fk(condef)
4720 if deferrable is not None:
4721 deferrable = True if deferrable == "DEFERRABLE" else False
4722 constrained_columns = [
4723 preparer._unquote_identifier(x)
4724 for x in re.split(r"\s*,\s*", constrained_columns)
4725 ]
4727 if postgresql_ignore_search_path:
4728 # when ignoring search path, we use the actual schema
4729 # provided it isn't the "default" schema
4730 if conschema != self.default_schema_name:
4731 referred_schema = conschema
4732 else:
4733 referred_schema = schema
4734 elif referred_schema:
4735 # referred_schema is the schema that we regexp'ed from
4736 # pg_get_constraintdef(). If the schema is in the search
4737 # path, pg_get_constraintdef() will give us None.
4738 referred_schema = preparer._unquote_identifier(referred_schema)
4739 elif schema is not None and schema == conschema:
4740 # If the actual schema matches the schema of the table
4741 # we're reflecting, then we will use that.
4742 referred_schema = schema
4744 referred_table = preparer._unquote_identifier(referred_table)
4745 referred_columns = [
4746 preparer._unquote_identifier(x)
4747 for x in re.split(r"\s*,\s", referred_columns)
4748 ]
4749 options = {
4750 k: v
4751 for k, v in [
4752 ("onupdate", onupdate),
4753 ("ondelete", ondelete),
4754 ("initially", initially),
4755 ("deferrable", deferrable),
4756 ("match", match),
4757 ]
4758 if v is not None and v != "NO ACTION"
4759 }
4760 fkey_d = {
4761 "name": conname,
4762 "constrained_columns": constrained_columns,
4763 "referred_schema": referred_schema,
4764 "referred_table": referred_table,
4765 "referred_columns": referred_columns,
4766 "options": options,
4767 "comment": comment,
4768 }
4769 table_fks.append(fkey_d)
4770 return fkeys.items()
4772 @reflection.cache
4773 def get_indexes(self, connection, table_name, schema=None, **kw):
4774 data = self.get_multi_indexes(
4775 connection,
4776 schema=schema,
4777 filter_names=[table_name],
4778 scope=ObjectScope.ANY,
4779 kind=ObjectKind.ANY,
4780 **kw,
4781 )
4782 return self._value_or_raise(data, table_name, schema)
4784 @util.memoized_property
4785 def _index_query(self):
4786 # NOTE: pg_index is used as from two times to improve performance,
4787 # since extraing all the index information from `idx_sq` to avoid
4788 # the second pg_index use leads to a worse performing query in
4789 # particular when querying for a single table (as of pg 17)
4790 # NOTE: repeating oids clause improve query performance
4792 # subquery to get the columns
4793 idx_sq = (
4794 select(
4795 pg_catalog.pg_index.c.indexrelid,
4796 pg_catalog.pg_index.c.indrelid,
4797 sql.func.unnest(pg_catalog.pg_index.c.indkey).label("attnum"),
4798 sql.func.unnest(pg_catalog.pg_index.c.indclass).label(
4799 "att_opclass"
4800 ),
4801 sql.func.generate_subscripts(
4802 pg_catalog.pg_index.c.indkey, 1
4803 ).label("ord"),
4804 )
4805 .where(
4806 ~pg_catalog.pg_index.c.indisprimary,
4807 pg_catalog.pg_index.c.indrelid.in_(bindparam("oids")),
4808 )
4809 .subquery("idx")
4810 )
4812 attr_sq = (
4813 select(
4814 idx_sq.c.indexrelid,
4815 idx_sq.c.indrelid,
4816 idx_sq.c.ord,
4817 # NOTE: always using pg_get_indexdef is too slow so just
4818 # invoke when the element is an expression
4819 sql.case(
4820 (
4821 idx_sq.c.attnum == 0,
4822 pg_catalog.pg_get_indexdef(
4823 idx_sq.c.indexrelid, idx_sq.c.ord + 1, True
4824 ),
4825 ),
4826 # NOTE: need to cast this since attname is of type "name"
4827 # that's limited to 63 bytes, while pg_get_indexdef
4828 # returns "text" so its output may get cut
4829 else_=pg_catalog.pg_attribute.c.attname.cast(TEXT),
4830 ).label("element"),
4831 (idx_sq.c.attnum == 0).label("is_expr"),
4832 pg_catalog.pg_opclass.c.opcname,
4833 pg_catalog.pg_opclass.c.opcdefault,
4834 )
4835 .select_from(idx_sq)
4836 .outerjoin(
4837 # do not remove rows where idx_sq.c.attnum is 0
4838 pg_catalog.pg_attribute,
4839 sql.and_(
4840 pg_catalog.pg_attribute.c.attnum == idx_sq.c.attnum,
4841 pg_catalog.pg_attribute.c.attrelid == idx_sq.c.indrelid,
4842 ),
4843 )
4844 .outerjoin(
4845 pg_catalog.pg_opclass,
4846 pg_catalog.pg_opclass.c.oid == idx_sq.c.att_opclass,
4847 )
4848 .where(idx_sq.c.indrelid.in_(bindparam("oids")))
4849 .subquery("idx_attr")
4850 )
4852 cols_sq = (
4853 select(
4854 attr_sq.c.indexrelid,
4855 sql.func.min(attr_sq.c.indrelid),
4856 sql.func.array_agg(
4857 aggregate_order_by(attr_sq.c.element, attr_sq.c.ord)
4858 ).label("elements"),
4859 sql.func.array_agg(
4860 aggregate_order_by(attr_sq.c.is_expr, attr_sq.c.ord)
4861 ).label("elements_is_expr"),
4862 sql.func.array_agg(
4863 aggregate_order_by(attr_sq.c.opcname, attr_sq.c.ord)
4864 ).label("elements_opclass"),
4865 sql.func.array_agg(
4866 aggregate_order_by(attr_sq.c.opcdefault, attr_sq.c.ord)
4867 ).label("elements_opdefault"),
4868 )
4869 .group_by(attr_sq.c.indexrelid)
4870 .subquery("idx_cols")
4871 )
4873 if self.server_version_info >= (11, 0):
4874 indnkeyatts = pg_catalog.pg_index.c.indnkeyatts
4875 else:
4876 indnkeyatts = pg_catalog.pg_index.c.indnatts.label("indnkeyatts")
4878 if self.server_version_info >= (15,):
4879 nulls_not_distinct = pg_catalog.pg_index.c.indnullsnotdistinct
4880 else:
4881 nulls_not_distinct = sql.false().label("indnullsnotdistinct")
4883 return (
4884 select(
4885 pg_catalog.pg_index.c.indrelid,
4886 pg_catalog.pg_class.c.relname,
4887 pg_catalog.pg_index.c.indisunique,
4888 pg_catalog.pg_constraint.c.conrelid.is_not(None).label(
4889 "has_constraint"
4890 ),
4891 pg_catalog.pg_index.c.indoption,
4892 pg_catalog.pg_class.c.reloptions,
4893 pg_catalog.pg_am.c.amname,
4894 # NOTE: pg_get_expr is very fast so this case has almost no
4895 # performance impact
4896 sql.case(
4897 (
4898 pg_catalog.pg_index.c.indpred.is_not(None),
4899 pg_catalog.pg_get_expr(
4900 pg_catalog.pg_index.c.indpred,
4901 pg_catalog.pg_index.c.indrelid,
4902 ),
4903 ),
4904 else_=None,
4905 ).label("filter_definition"),
4906 indnkeyatts,
4907 nulls_not_distinct,
4908 cols_sq.c.elements,
4909 cols_sq.c.elements_is_expr,
4910 cols_sq.c.elements_opclass,
4911 cols_sq.c.elements_opdefault,
4912 )
4913 .select_from(pg_catalog.pg_index)
4914 .where(
4915 pg_catalog.pg_index.c.indrelid.in_(bindparam("oids")),
4916 ~pg_catalog.pg_index.c.indisprimary,
4917 )
4918 .join(
4919 pg_catalog.pg_class,
4920 pg_catalog.pg_index.c.indexrelid == pg_catalog.pg_class.c.oid,
4921 )
4922 .join(
4923 pg_catalog.pg_am,
4924 pg_catalog.pg_class.c.relam == pg_catalog.pg_am.c.oid,
4925 )
4926 .outerjoin(
4927 cols_sq,
4928 pg_catalog.pg_index.c.indexrelid == cols_sq.c.indexrelid,
4929 )
4930 .outerjoin(
4931 pg_catalog.pg_constraint,
4932 sql.and_(
4933 pg_catalog.pg_index.c.indrelid
4934 == pg_catalog.pg_constraint.c.conrelid,
4935 pg_catalog.pg_index.c.indexrelid
4936 == pg_catalog.pg_constraint.c.conindid,
4937 pg_catalog.pg_constraint.c.contype
4938 == sql.any_(_array.array(("p", "u", "x"))),
4939 ),
4940 )
4941 .order_by(
4942 pg_catalog.pg_index.c.indrelid, pg_catalog.pg_class.c.relname
4943 )
4944 )
4946 def get_multi_indexes(
4947 self, connection, schema, filter_names, scope, kind, **kw
4948 ):
4949 table_oids = self._get_table_oids(
4950 connection, schema, filter_names, scope, kind, **kw
4951 )
4953 indexes = defaultdict(list)
4954 default = ReflectionDefaults.indexes
4956 batches = list(table_oids)
4958 while batches:
4959 batch = batches[0:3000]
4960 batches[0:3000] = []
4962 result = connection.execute(
4963 self._index_query, {"oids": [r[0] for r in batch]}
4964 ).mappings()
4966 result_by_oid = defaultdict(list)
4967 for row_dict in result:
4968 result_by_oid[row_dict["indrelid"]].append(row_dict)
4970 for oid, table_name in batch:
4971 if oid not in result_by_oid:
4972 # ensure that each table has an entry, even if reflection
4973 # is skipped because not supported
4974 indexes[(schema, table_name)] = default()
4975 continue
4977 for row in result_by_oid[oid]:
4978 index_name = row["relname"]
4980 table_indexes = indexes[(schema, table_name)]
4982 all_elements = row["elements"]
4983 all_elements_is_expr = row["elements_is_expr"]
4984 all_elements_opclass = row["elements_opclass"]
4985 all_elements_opdefault = row["elements_opdefault"]
4986 indnkeyatts = row["indnkeyatts"]
4987 # "The number of key columns in the index, not counting any
4988 # included columns, which are merely stored and do not
4989 # participate in the index semantics"
4990 if len(all_elements) > indnkeyatts:
4991 # this is a "covering index" which has INCLUDE columns
4992 # as well as regular index columns
4993 inc_cols = all_elements[indnkeyatts:]
4994 idx_elements = all_elements[:indnkeyatts]
4995 idx_elements_is_expr = all_elements_is_expr[
4996 :indnkeyatts
4997 ]
4998 # postgresql does not support expression on included
4999 # columns as of v14: "ERROR: expressions are not
5000 # supported in included columns".
5001 assert all(
5002 not is_expr
5003 for is_expr in all_elements_is_expr[indnkeyatts:]
5004 )
5005 idx_elements_opclass = all_elements_opclass[
5006 :indnkeyatts
5007 ]
5008 idx_elements_opdefault = all_elements_opdefault[
5009 :indnkeyatts
5010 ]
5011 else:
5012 idx_elements = all_elements
5013 idx_elements_is_expr = all_elements_is_expr
5014 inc_cols = []
5015 idx_elements_opclass = all_elements_opclass
5016 idx_elements_opdefault = all_elements_opdefault
5018 index = {"name": index_name, "unique": row["indisunique"]}
5019 if any(idx_elements_is_expr):
5020 index["column_names"] = [
5021 None if is_expr else expr
5022 for expr, is_expr in zip(
5023 idx_elements, idx_elements_is_expr
5024 )
5025 ]
5026 index["expressions"] = idx_elements
5027 else:
5028 index["column_names"] = idx_elements
5030 dialect_options = {}
5032 if not all(idx_elements_opdefault):
5033 dialect_options["postgresql_ops"] = {
5034 name: opclass
5035 for name, opclass, is_default in zip(
5036 idx_elements,
5037 idx_elements_opclass,
5038 idx_elements_opdefault,
5039 )
5040 if not is_default
5041 }
5043 sorting = {}
5044 for col_index, col_flags in enumerate(row["indoption"]):
5045 col_sorting = ()
5046 # try to set flags only if they differ from PG
5047 # defaults...
5048 if col_flags & 0x01:
5049 col_sorting += ("desc",)
5050 if not (col_flags & 0x02):
5051 col_sorting += ("nulls_last",)
5052 else:
5053 if col_flags & 0x02:
5054 col_sorting += ("nulls_first",)
5055 if col_sorting:
5056 sorting[idx_elements[col_index]] = col_sorting
5057 if sorting:
5058 index["column_sorting"] = sorting
5059 if row["has_constraint"]:
5060 index["duplicates_constraint"] = index_name
5062 if row["reloptions"]:
5063 dialect_options["postgresql_with"] = dict(
5064 [
5065 option.split("=", 1)
5066 for option in row["reloptions"]
5067 ]
5068 )
5069 # it *might* be nice to include that this is 'btree' in the
5070 # reflection info. But we don't want an Index object
5071 # to have a ``postgresql_using`` in it that is just the
5072 # default, so for the moment leaving this out.
5073 amname = row["amname"]
5074 if amname != "btree":
5075 dialect_options["postgresql_using"] = row["amname"]
5076 if row["filter_definition"]:
5077 dialect_options["postgresql_where"] = row[
5078 "filter_definition"
5079 ]
5080 if self.server_version_info >= (11,):
5081 # NOTE: this is legacy, this is part of
5082 # dialect_options now as of #7382
5083 index["include_columns"] = inc_cols
5084 dialect_options["postgresql_include"] = inc_cols
5085 if row["indnullsnotdistinct"]:
5086 # the default is False, so ignore it.
5087 dialect_options["postgresql_nulls_not_distinct"] = row[
5088 "indnullsnotdistinct"
5089 ]
5091 if dialect_options:
5092 index["dialect_options"] = dialect_options
5094 table_indexes.append(index)
5095 return indexes.items()
5097 @reflection.cache
5098 def get_unique_constraints(
5099 self, connection, table_name, schema=None, **kw
5100 ):
5101 data = self.get_multi_unique_constraints(
5102 connection,
5103 schema=schema,
5104 filter_names=[table_name],
5105 scope=ObjectScope.ANY,
5106 kind=ObjectKind.ANY,
5107 **kw,
5108 )
5109 return self._value_or_raise(data, table_name, schema)
5111 def get_multi_unique_constraints(
5112 self,
5113 connection,
5114 schema,
5115 filter_names,
5116 scope,
5117 kind,
5118 **kw,
5119 ):
5120 result = self._reflect_constraint(
5121 connection, "u", schema, filter_names, scope, kind, **kw
5122 )
5124 # each table can have multiple unique constraints
5125 uniques = defaultdict(list)
5126 default = ReflectionDefaults.unique_constraints
5127 for table_name, cols, con_name, comment, options in result:
5128 # ensure a list is created for each table. leave it empty if
5129 # the table has no unique constraint
5130 if con_name is None:
5131 uniques[(schema, table_name)] = default()
5132 continue
5134 uc_dict = {
5135 "column_names": cols,
5136 "name": con_name,
5137 "comment": comment,
5138 }
5139 if options:
5140 uc_dict["dialect_options"] = options
5142 uniques[(schema, table_name)].append(uc_dict)
5143 return uniques.items()
5145 @reflection.cache
5146 def get_table_comment(self, connection, table_name, schema=None, **kw):
5147 data = self.get_multi_table_comment(
5148 connection,
5149 schema,
5150 [table_name],
5151 scope=ObjectScope.ANY,
5152 kind=ObjectKind.ANY,
5153 **kw,
5154 )
5155 return self._value_or_raise(data, table_name, schema)
5157 @lru_cache()
5158 def _comment_query(self, schema, has_filter_names, scope, kind):
5159 relkinds = self._kind_to_relkinds(kind)
5160 query = (
5161 select(
5162 pg_catalog.pg_class.c.relname,
5163 pg_catalog.pg_description.c.description,
5164 )
5165 .select_from(pg_catalog.pg_class)
5166 .outerjoin(
5167 pg_catalog.pg_description,
5168 sql.and_(
5169 pg_catalog.pg_class.c.oid
5170 == pg_catalog.pg_description.c.objoid,
5171 pg_catalog.pg_description.c.objsubid == 0,
5172 pg_catalog.pg_description.c.classoid
5173 == sql.func.cast("pg_catalog.pg_class", REGCLASS),
5174 ),
5175 )
5176 .where(self._pg_class_relkind_condition(relkinds))
5177 )
5178 query = self._pg_class_filter_scope_schema(query, schema, scope)
5179 if has_filter_names:
5180 query = query.where(
5181 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
5182 )
5183 return query
5185 def get_multi_table_comment(
5186 self, connection, schema, filter_names, scope, kind, **kw
5187 ):
5188 has_filter_names, params = self._prepare_filter_names(filter_names)
5189 query = self._comment_query(schema, has_filter_names, scope, kind)
5190 result = connection.execute(query, params)
5192 default = ReflectionDefaults.table_comment
5193 return (
5194 (
5195 (schema, table),
5196 {"text": comment} if comment is not None else default(),
5197 )
5198 for table, comment in result
5199 )
5201 @reflection.cache
5202 def get_check_constraints(self, connection, table_name, schema=None, **kw):
5203 data = self.get_multi_check_constraints(
5204 connection,
5205 schema,
5206 [table_name],
5207 scope=ObjectScope.ANY,
5208 kind=ObjectKind.ANY,
5209 **kw,
5210 )
5211 return self._value_or_raise(data, table_name, schema)
5213 @lru_cache()
5214 def _check_constraint_query(self, schema, has_filter_names, scope, kind):
5215 relkinds = self._kind_to_relkinds(kind)
5216 query = (
5217 select(
5218 pg_catalog.pg_class.c.relname,
5219 pg_catalog.pg_constraint.c.conname,
5220 # NOTE: avoid calling pg_get_constraintdef when not needed
5221 # to speed up the query
5222 sql.case(
5223 (
5224 pg_catalog.pg_constraint.c.oid.is_not(None),
5225 pg_catalog.pg_get_constraintdef(
5226 pg_catalog.pg_constraint.c.oid, True
5227 ),
5228 ),
5229 else_=None,
5230 ),
5231 pg_catalog.pg_description.c.description,
5232 )
5233 .select_from(pg_catalog.pg_class)
5234 .outerjoin(
5235 pg_catalog.pg_constraint,
5236 sql.and_(
5237 pg_catalog.pg_class.c.oid
5238 == pg_catalog.pg_constraint.c.conrelid,
5239 pg_catalog.pg_constraint.c.contype == "c",
5240 ),
5241 )
5242 .outerjoin(
5243 pg_catalog.pg_description,
5244 pg_catalog.pg_description.c.objoid
5245 == pg_catalog.pg_constraint.c.oid,
5246 )
5247 .order_by(
5248 pg_catalog.pg_class.c.relname,
5249 pg_catalog.pg_constraint.c.conname,
5250 )
5251 .where(self._pg_class_relkind_condition(relkinds))
5252 )
5253 query = self._pg_class_filter_scope_schema(query, schema, scope)
5254 if has_filter_names:
5255 query = query.where(
5256 pg_catalog.pg_class.c.relname.in_(bindparam("filter_names"))
5257 )
5258 return query
5260 def get_multi_check_constraints(
5261 self, connection, schema, filter_names, scope, kind, **kw
5262 ):
5263 has_filter_names, params = self._prepare_filter_names(filter_names)
5264 query = self._check_constraint_query(
5265 schema, has_filter_names, scope, kind
5266 )
5267 result = connection.execute(query, params)
5269 check_constraints = defaultdict(list)
5270 default = ReflectionDefaults.check_constraints
5271 for table_name, check_name, src, comment in result:
5272 # only two cases for check_name and src: both null or both defined
5273 if check_name is None and src is None:
5274 check_constraints[(schema, table_name)] = default()
5275 continue
5276 # samples:
5277 # "CHECK (((a > 1) AND (a < 5)))"
5278 # "CHECK (((a = 1) OR ((a > 2) AND (a < 5))))"
5279 # "CHECK (((a > 1) AND (a < 5))) NOT VALID"
5280 # "CHECK (some_boolean_function(a))"
5281 # "CHECK (((a\n < 1)\n OR\n (a\n >= 5))\n)"
5282 # "CHECK (a NOT NULL) NO INHERIT"
5283 # "CHECK (a NOT NULL) NO INHERIT NOT VALID"
5285 m = re.match(
5286 r"^CHECK *\((.+)\)( NO INHERIT)?( NOT VALID)?$",
5287 src,
5288 flags=re.DOTALL,
5289 )
5290 if not m:
5291 util.warn("Could not parse CHECK constraint text: %r" % src)
5292 sqltext = ""
5293 else:
5294 sqltext = util.strip_outer_parens(m.group(1))
5295 entry = {
5296 "name": check_name,
5297 "sqltext": sqltext,
5298 "comment": comment,
5299 }
5300 if m:
5301 do = {}
5302 if " NOT VALID" in m.groups():
5303 do["not_valid"] = True
5304 if " NO INHERIT" in m.groups():
5305 do["no_inherit"] = True
5306 if do:
5307 entry["dialect_options"] = do
5309 check_constraints[(schema, table_name)].append(entry)
5310 return check_constraints.items()
5312 def _pg_type_filter_schema(self, query, schema):
5313 if schema is None:
5314 query = query.where(
5315 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid),
5316 # ignore pg_catalog schema
5317 pg_catalog.pg_namespace.c.nspname != "pg_catalog",
5318 )
5319 elif schema != "*":
5320 query = query.where(pg_catalog.pg_namespace.c.nspname == schema)
5321 return query
5323 @lru_cache()
5324 def _enum_query(self, schema):
5325 lbl_agg_sq = (
5326 select(
5327 pg_catalog.pg_enum.c.enumtypid,
5328 sql.func.array_agg(
5329 aggregate_order_by(
5330 # NOTE: cast since some postgresql derivatives may
5331 # not support array_agg on the name type
5332 pg_catalog.pg_enum.c.enumlabel.cast(TEXT),
5333 pg_catalog.pg_enum.c.enumsortorder,
5334 )
5335 ).label("labels"),
5336 )
5337 .group_by(pg_catalog.pg_enum.c.enumtypid)
5338 .subquery("lbl_agg")
5339 )
5341 query = (
5342 select(
5343 pg_catalog.pg_type.c.typname.label("name"),
5344 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid).label(
5345 "visible"
5346 ),
5347 pg_catalog.pg_namespace.c.nspname.label("schema"),
5348 lbl_agg_sq.c.labels.label("labels"),
5349 )
5350 .join(
5351 pg_catalog.pg_namespace,
5352 pg_catalog.pg_namespace.c.oid
5353 == pg_catalog.pg_type.c.typnamespace,
5354 )
5355 .outerjoin(
5356 lbl_agg_sq, pg_catalog.pg_type.c.oid == lbl_agg_sq.c.enumtypid
5357 )
5358 .where(pg_catalog.pg_type.c.typtype == "e")
5359 .order_by(
5360 pg_catalog.pg_namespace.c.nspname, pg_catalog.pg_type.c.typname
5361 )
5362 )
5364 return self._pg_type_filter_schema(query, schema)
5366 @reflection.cache
5367 def _load_enums(self, connection, schema=None, **kw):
5368 if not self.supports_native_enum:
5369 return []
5371 result = connection.execute(self._enum_query(schema))
5373 enums = []
5374 for name, visible, schema, labels in result:
5375 enums.append(
5376 {
5377 "name": name,
5378 "schema": schema,
5379 "visible": visible,
5380 "labels": [] if labels is None else labels,
5381 }
5382 )
5383 return enums
5385 @lru_cache()
5386 def _domain_query(self, schema):
5387 con_sq = (
5388 select(
5389 pg_catalog.pg_constraint.c.contypid,
5390 sql.func.array_agg(
5391 pg_catalog.pg_get_constraintdef(
5392 pg_catalog.pg_constraint.c.oid, True
5393 )
5394 ).label("condefs"),
5395 sql.func.array_agg(
5396 # NOTE: cast since some postgresql derivatives may
5397 # not support array_agg on the name type
5398 pg_catalog.pg_constraint.c.conname.cast(TEXT)
5399 ).label("connames"),
5400 )
5401 # The domain this constraint is on; zero if not a domain constraint
5402 .where(pg_catalog.pg_constraint.c.contypid != 0)
5403 .group_by(pg_catalog.pg_constraint.c.contypid)
5404 .subquery("domain_constraints")
5405 )
5407 query = (
5408 select(
5409 pg_catalog.pg_type.c.typname.label("name"),
5410 pg_catalog.format_type(
5411 pg_catalog.pg_type.c.typbasetype,
5412 pg_catalog.pg_type.c.typtypmod,
5413 ).label("attype"),
5414 (~pg_catalog.pg_type.c.typnotnull).label("nullable"),
5415 pg_catalog.pg_type.c.typdefault.label("default"),
5416 pg_catalog.pg_type_is_visible(pg_catalog.pg_type.c.oid).label(
5417 "visible"
5418 ),
5419 pg_catalog.pg_namespace.c.nspname.label("schema"),
5420 con_sq.c.condefs,
5421 con_sq.c.connames,
5422 pg_catalog.pg_collation.c.collname,
5423 )
5424 .join(
5425 pg_catalog.pg_namespace,
5426 pg_catalog.pg_namespace.c.oid
5427 == pg_catalog.pg_type.c.typnamespace,
5428 )
5429 .outerjoin(
5430 pg_catalog.pg_collation,
5431 pg_catalog.pg_type.c.typcollation
5432 == pg_catalog.pg_collation.c.oid,
5433 )
5434 .outerjoin(
5435 con_sq,
5436 pg_catalog.pg_type.c.oid == con_sq.c.contypid,
5437 )
5438 .where(pg_catalog.pg_type.c.typtype == "d")
5439 .order_by(
5440 pg_catalog.pg_namespace.c.nspname, pg_catalog.pg_type.c.typname
5441 )
5442 )
5443 return self._pg_type_filter_schema(query, schema)
5445 @reflection.cache
5446 def _load_domains(self, connection, schema=None, **kw):
5447 result = connection.execute(self._domain_query(schema))
5449 domains: List[ReflectedDomain] = []
5450 for domain in result.mappings():
5451 # strip (30) from character varying(30)
5452 attype = re.search(r"([^\(]+)", domain["attype"]).group(1)
5453 constraints: List[ReflectedDomainConstraint] = []
5454 if domain["connames"]:
5455 # When a domain has multiple CHECK constraints, they will
5456 # be tested in alphabetical order by name.
5457 sorted_constraints = sorted(
5458 zip(domain["connames"], domain["condefs"]),
5459 key=lambda t: t[0],
5460 )
5461 for name, def_ in sorted_constraints:
5462 # constraint is in the form "CHECK (expression)"
5463 # or "NOT NULL". Ignore the "NOT NULL" and
5464 # remove "CHECK (" and the tailing ")".
5465 if def_.casefold().startswith("check"):
5466 check = def_[7:-1]
5467 constraints.append({"name": name, "check": check})
5468 domain_rec: ReflectedDomain = {
5469 "name": domain["name"],
5470 "schema": domain["schema"],
5471 "visible": domain["visible"],
5472 "type": attype,
5473 "nullable": domain["nullable"],
5474 "default": domain["default"],
5475 "constraints": constraints,
5476 "collation": domain["collname"],
5477 }
5478 domains.append(domain_rec)
5480 return domains
5482 def _set_backslash_escapes(self, connection):
5483 # this method is provided as an override hook for descendant
5484 # dialects (e.g. Redshift), so removing it may break them
5485 std_string = connection.exec_driver_sql(
5486 "show standard_conforming_strings"
5487 ).scalar()
5488 self._backslash_escapes = std_string == "off"