sqlalchemy.databases.postgresql

Sequences/SERIAL/IDENTITY

PostgreSQL supports sequences, and SQLAlchemy uses these as the default means of creating new primary key values for integer-based primary key columns. When creating tables, SQLAlchemy will issue the SERIAL datatype for integer-based primary key columns, which generates a sequence and server side default corresponding to the column.

To specify a specific named sequence to be used for primary key generation, use the ~sqlalchemy.schema.Sequence construct:

Table('sometable', metadata,
        Column('id', Integer, Sequence('some_id_seq'), primary_key=True)
    )

When SQLAlchemy issues a single INSERT statement, to fulfill the contract of having the "last insert identifier" available, a RETURNING clause is added to the INSERT statement which specifies the primary key columns should be returned after the statement completes. The RETURNING functionality only takes place if PostgreSQL 8.2 or later is in use. As a fallback approach, the sequence, whether specified explicitly or implicitly via SERIAL, is executed independently beforehand, the returned value to be used in the subsequent insert. Note that when an ~sqlalchemy.sql.expression.insert() construct is executed using "executemany" semantics, the "last inserted identifier" functionality does not apply; no RETURNING clause is emitted nor is the sequence pre-executed in this case.

To force the usage of RETURNING by default off, specify the flag implicit_returning=False to _sa.create_engine.

from sqlalchemy import Table, Column, MetaData, Integer, Computed

metadata = MetaData()

data = Table(
    "data",
    metadata,
    Column(
        'id', Integer, Identity(start=42, cycle=True), primary_key=True
    ),
    Column('data', String)
)

CREATE TABLE data (
    id INTEGER GENERATED BY DEFAULT AS IDENTITY (START WITH 42 CYCLE),
    data VARCHAR,
    PRIMARY KEY (id)
)

from sqlalchemy.schema import CreateColumn
from sqlalchemy.ext.compiler import compiles


@compiles(CreateColumn, 'postgresql')
def use_identity(element, compiler, **kw):
    text = compiler.visit_create_column(element, **kw)
    text = text.replace(
        "SERIAL", "INT GENERATED BY DEFAULT AS IDENTITY"
     )
    return text

t = Table(
    't', m,
    Column('id', Integer, primary_key=True),
    Column('data', String)
)

CREATE TABLE t (
    id INT GENERATED BY DEFAULT AS IDENTITY,
    data VARCHAR,
    PRIMARY KEY (id)
)

Server Side Cursors

Server-side cursor support is available for the psycopg2, asyncpg dialects and may also be available in others.

Server side cursors are enabled on a per-statement basis by using the :paramref:`.Connection.execution_options.stream_results` connection execution option:

with engine.connect() as conn:
    result = conn.execution_options(stream_results=True).execute(text("select * from table"))

Note that some kinds of SQL statements may not be supported with server side cursors; generally, only SQL statements that return rows should be used with this option.

Transaction Isolation Level

Most SQLAlchemy dialects support setting of transaction isolation level using the :paramref:`_sa.create_engine.execution_options` parameter at the _sa.create_engine level, and at the _engine.Connection level via the :paramref:`.Connection.execution_options.isolation_level` parameter.

For PostgreSQL dialects, this feature works either by making use of the DBAPI-specific features, such as psycopg2's isolation level flags which will embed the isolation level setting inline with the "BEGIN" statement, or for DBAPIs with no direct support by emitting SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL <level> ahead of the "BEGIN" statement emitted by the DBAPI. For the special AUTOCOMMIT isolation level, DBAPI-specific techniques are used which is typically an .autocommit flag on the DBAPI connection object.

To set isolation level using _sa.create_engine:

engine = create_engine(
    "postgresql+pg8000://scott:tiger@localhost/test",
    execution_options={
        "isolation_level": "REPEATABLE READ"
    }
)

To set using per-connection execution options:

with engine.connect() as conn:
    conn = conn.execution_options(
        isolation_level="REPEATABLE READ"
    )
    with conn.begin():
        # ... work with transaction

Valid values for isolation_level on most PostgreSQL dialects include:

• READ COMMITTED
• READ UNCOMMITTED
• REPEATABLE READ
• SERIALIZABLE
• AUTOCOMMIT

Setting READ ONLY / DEFERRABLE

Most PostgreSQL dialects support setting the "READ ONLY" and "DEFERRABLE" characteristics of the transaction, which is in addition to the isolation level setting. These two attributes can be established either in conjunction with or independently of the isolation level by passing the postgresql_readonly and postgresql_deferrable flags with _engine.Connection.execution_options. The example below illustrates passing the "SERIALIZABLE" isolation level at the same time as setting "READ ONLY" and "DEFERRABLE":

with engine.connect() as conn:
    conn = conn.execution_options(
        isolation_level="SERIALIZABLE",
        postgresql_readonly=True,
        postgresql_deferrable=True
    )
    with conn.begin():
        #  ... work with transaction

Note that some DBAPIs such as asyncpg only support "readonly" with SERIALIZABLE isolation.

Setting Alternate Search Paths on Connect

The PostgreSQL search_path variable refers to the list of schema names that will be implicitly referred towards when a particular table or other object is referenced in a SQL statement. As detailed in the next section :ref:`postgresql_schema_reflection`, SQLAlchemy is generally organized around the concept of keeping this variable at its default value of public, however, in order to have it set to any arbitrary name or names when connections are used automatically, the "SET SESSION search_path" command may be invoked for all connections in a pool using the following event handler, as discussed at :ref:`schema_set_default_connections`:

from sqlalchemy import event
from sqlalchemy import create_engine

engine = create_engine("postgresql+psycopg2://scott:tiger@host/dbname")

@event.listens_for(engine, "connect", insert=True)
def set_search_path(dbapi_connection, connection_record):
    existing_autocommit = dbapi_connection.autocommit
    dbapi_connection.autocommit = True
    cursor = dbapi_connection.cursor()
    cursor.execute("SET SESSION search_path='%s'" % schema_name)
    cursor.close()
    dbapi_connection.autocommit = existing_autocommit

The reason the recipe is complicated by use of the .autocommit DBAPI attribute is so that when the SET SESSION search_path directive is invoked, it is invoked outside of the scope of any transaction and therefore will not be reverted when the DBAPI connection has a rollback.

Remote-Schema Table Introspection and PostgreSQL search_path

The PostgreSQL dialect can reflect tables from any schema, as outlined in :ref:`schema_table_reflection`.

With regards to tables which these _schema.Table objects refer to via foreign key constraint, a decision must be made as to how the .schema is represented in those remote tables, in the case where that remote schema name is also a member of the current PostgreSQL search path.

By default, the PostgreSQL dialect mimics the behavior encouraged by PostgreSQL's own pg_get_constraintdef() builtin procedure. This function returns a sample definition for a particular foreign key constraint, omitting the referenced schema name from that definition when the name is also in the PostgreSQL schema search path. The interaction below illustrates this behavior:

test=> CREATE TABLE test_schema.referred(id INTEGER PRIMARY KEY);
CREATE TABLE
test=> CREATE TABLE referring(
test(>         id INTEGER PRIMARY KEY,
test(>         referred_id INTEGER REFERENCES test_schema.referred(id));
CREATE TABLE
test=> SET search_path TO public, test_schema;
test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n
test-> ON n.oid = c.relnamespace
test-> JOIN pg_catalog.pg_constraint r  ON c.oid = r.conrelid
test-> WHERE c.relname='referring' AND r.contype = 'f'
test-> ;
               pg_get_constraintdef
---------------------------------------------------
 FOREIGN KEY (referred_id) REFERENCES referred(id)
(1 row)

Above, we created a table referred as a member of the remote schema test_schema, however when we added test_schema to the PG search_path and then asked pg_get_constraintdef() for the FOREIGN KEY syntax, test_schema was not included in the output of the function.

On the other hand, if we set the search path back to the typical default of public:

test=> SET search_path TO public;
SET

The same query against pg_get_constraintdef() now returns the fully schema-qualified name for us:

test=> SELECT pg_catalog.pg_get_constraintdef(r.oid, true) FROM
test-> pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n
test-> ON n.oid = c.relnamespace
test-> JOIN pg_catalog.pg_constraint r  ON c.oid = r.conrelid
test-> WHERE c.relname='referring' AND r.contype = 'f';
                     pg_get_constraintdef
---------------------------------------------------------------
 FOREIGN KEY (referred_id) REFERENCES test_schema.referred(id)
(1 row)

SQLAlchemy will by default use the return value of pg_get_constraintdef() in order to determine the remote schema name. That is, if our search_path were set to include test_schema, and we invoked a table reflection process as follows:

>>> from sqlalchemy import Table, MetaData, create_engine, text
>>> engine = create_engine("postgresql://scott:tiger@localhost/test")
>>> with engine.connect() as conn:
...     conn.execute(text("SET search_path TO test_schema, public"))
...     metadata_obj = MetaData()
...     referring = Table('referring', metadata_obj,
...                       autoload_with=conn)
...
<sqlalchemy.engine.result.CursorResult object at 0x101612ed0>

The above process would deliver to the _schema.MetaData.tables collection referred table named without the schema:

>>> metadata_obj.tables['referred'].schema is None
True

To alter the behavior of reflection such that the referred schema is maintained regardless of the search_path setting, use the postgresql_ignore_search_path option, which can be specified as a dialect-specific argument to both _schema.Table as well as _schema.MetaData.reflect:

>>> with engine.connect() as conn:
...     conn.execute(text("SET search_path TO test_schema, public"))
...     metadata_obj = MetaData()
...     referring = Table('referring', metadata_obj,
...                       autoload_with=conn,
...                       postgresql_ignore_search_path=True)
...
<sqlalchemy.engine.result.CursorResult object at 0x1016126d0>

We will now have test_schema.referred stored as schema-qualified:

>>> metadata_obj.tables['test_schema.referred'].schema
'test_schema'

Note that in all cases, the "default" schema is always reflected as None. The "default" schema on PostgreSQL is that which is returned by the PostgreSQL current_schema() function. On a typical PostgreSQL installation, this is the name public. So a table that refers to another which is in the public (i.e. default) schema will always have the .schema attribute set to None.

INSERT/UPDATE...RETURNING

The dialect supports PG 8.2's INSERT..RETURNING, UPDATE..RETURNING and DELETE..RETURNING syntaxes. INSERT..RETURNING is used by default for single-row INSERT statements in order to fetch newly generated primary key identifiers. To specify an explicit RETURNING clause, use the ._UpdateBase.returning method on a per-statement basis:

# INSERT..RETURNING
result = table.insert().returning(table.c.col1, table.c.col2).\
    values(name='foo')
print(result.fetchall())

# UPDATE..RETURNING
result = table.update().returning(table.c.col1, table.c.col2).\
    where(table.c.name=='foo').values(name='bar')
print(result.fetchall())

# DELETE..RETURNING
result = table.delete().returning(table.c.col1, table.c.col2).\
    where(table.c.name=='foo')
print(result.fetchall())

INSERT...ON CONFLICT (Upsert)

Starting with version 9.5, PostgreSQL allows "upserts" (update or insert) of rows into a table via the ON CONFLICT clause of the INSERT statement. A candidate row will only be inserted if that row does not violate any unique constraints. In the case of a unique constraint violation, a secondary action can occur which can be either "DO UPDATE", indicating that the data in the target row should be updated, or "DO NOTHING", which indicates to silently skip this row.

Conflicts are determined using existing unique constraints and indexes. These constraints may be identified either using their name as stated in DDL, or they may be inferred by stating the columns and conditions that comprise the indexes.

SQLAlchemy provides ON CONFLICT support via the PostgreSQL-specific _postgresql.insert() function, which provides the generative methods _postgresql.Insert.on_conflict_do_update and ~.postgresql.Insert.on_conflict_do_nothing:

>>> from sqlalchemy.dialects.postgresql import insert
>>> insert_stmt = insert(my_table).values(
...     id='some_existing_id',
...     data='inserted value')
>>> do_nothing_stmt = insert_stmt.on_conflict_do_nothing(
...     index_elements=['id']
... )
>>> print(do_nothing_stmt)
{opensql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
ON CONFLICT (id) DO NOTHING
{stop}

>>> do_update_stmt = insert_stmt.on_conflict_do_update(
...     constraint='pk_my_table',
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
{opensql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
ON CONFLICT ON CONSTRAINT pk_my_table DO UPDATE SET data = %(param_1)s

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value')
... )
>>> print(do_update_stmt)
{opensql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
ON CONFLICT (id) DO UPDATE SET data = %(param_1)s

>>> stmt = insert(my_table).values(
...     id='some_id',
...     data='inserted value',
...     author='jlh'
... )
>>> do_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value', author=stmt.excluded.author)
... )
>>> print(do_update_stmt)
{opensql}INSERT INTO my_table (id, data, author)
VALUES (%(id)s, %(data)s, %(author)s)
ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author

>>> stmt = insert(my_table).values(
...     id='some_id',
...     data='inserted value',
...     author='jlh'
... )
>>> on_update_stmt = stmt.on_conflict_do_update(
...     index_elements=['id'],
...     set_=dict(data='updated value', author=stmt.excluded.author),
...     where=(my_table.c.status == 2)
... )
>>> print(on_update_stmt)
{opensql}INSERT INTO my_table (id, data, author)
VALUES (%(id)s, %(data)s, %(author)s)
ON CONFLICT (id) DO UPDATE SET data = %(param_1)s, author = excluded.author
WHERE my_table.status = %(status_1)s

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> stmt = stmt.on_conflict_do_nothing(index_elements=['id'])
>>> print(stmt)
{opensql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
ON CONFLICT (id) DO NOTHING

>>> stmt = insert(my_table).values(id='some_id', data='inserted value')
>>> stmt = stmt.on_conflict_do_nothing()
>>> print(stmt)
{opensql}INSERT INTO my_table (id, data) VALUES (%(id)s, %(data)s)
ON CONFLICT DO NOTHING

Full Text Search

SQLAlchemy makes available the PostgreSQL @@ operator via the _expression.ColumnElement.match method on any textual column expression.

On the PostgreSQL dialect, an expression like the following:

select(sometable.c.text.match("search string"))

will emit to the database:

SELECT text @@ to_tsquery('search string') FROM table

Various other PostgreSQL text search functions such as to_tsquery(), to_tsvector(), and plainto_tsquery() are available by explicitly using the standard SQLAlchemy .func construct.

For example:

select(func.to_tsvector('fat cats ate rats').match('cat & rat'))

Emits the equivalent of:

SELECT to_tsvector('fat cats ate rats') @@ to_tsquery('cat & rat')

The _postgresql.TSVECTOR type can provide for explicit CAST:

from sqlalchemy.dialects.postgresql import TSVECTOR
from sqlalchemy import select, cast
select(cast("some text", TSVECTOR))

produces a statement equivalent to:

SELECT CAST('some text' AS TSVECTOR) AS anon_1

Full Text Searches in PostgreSQL are influenced by a combination of: the PostgreSQL setting of default_text_search_config, the regconfig used to build the GIN/GiST indexes, and the regconfig optionally passed in during a query.

When performing a Full Text Search against a column that has a GIN or GiST index that is already pre-computed (which is common on full text searches) one may need to explicitly pass in a particular PostgreSQL regconfig value to ensure the query-planner utilizes the index and does not re-compute the column on demand.

In order to provide for this explicit query planning, or to use different search strategies, the match method accepts a postgresql_regconfig keyword argument:

select(mytable.c.id).where(
    mytable.c.title.match('somestring', postgresql_regconfig='english')
)

Emits the equivalent of:

SELECT mytable.id FROM mytable
WHERE mytable.title @@ to_tsquery('english', 'somestring')

One can also specifically pass in a 'regconfig' value to the to_tsvector() command as the initial argument:

select(mytable.c.id).where(
        func.to_tsvector('english', mytable.c.title )\
        .match('somestring', postgresql_regconfig='english')
    )

produces a statement equivalent to:

SELECT mytable.id FROM mytable
WHERE to_tsvector('english', mytable.title) @@
    to_tsquery('english', 'somestring')

It is recommended that you use the EXPLAIN ANALYZE... tool from PostgreSQL to ensure that you are generating queries with SQLAlchemy that take full advantage of any indexes you may have created for full text search.

FROM ONLY ...

The dialect supports PostgreSQL's ONLY keyword for targeting only a particular table in an inheritance hierarchy. This can be used to produce the SELECT ... FROM ONLY, UPDATE ONLY ..., and DELETE FROM ONLY ... syntaxes. It uses SQLAlchemy's hints mechanism:

# SELECT ... FROM ONLY ...
result = table.select().with_hint(table, 'ONLY', 'postgresql')
print(result.fetchall())

# UPDATE ONLY ...
table.update(values=dict(foo='bar')).with_hint('ONLY',
                                               dialect_name='postgresql')

# DELETE FROM ONLY ...
table.delete().with_hint('ONLY', dialect_name='postgresql')

PostgreSQL-Specific Index Options

Several extensions to the .Index construct are available, specific to the PostgreSQL dialect.

Index("my_index", table.c.x, postgresql_include=['y'])

Index('my_index', my_table.c.id, postgresql_where=my_table.c.value > 10)

Index(
    'my_index', my_table.c.id, my_table.c.data,
    postgresql_ops={
        'data': 'text_pattern_ops',
        'id': 'int4_ops'
    })

Index(
    'my_index', my_table.c.id,
    func.lower(my_table.c.data).label('data_lower'),
    postgresql_ops={
        'data_lower': 'text_pattern_ops',
        'id': 'int4_ops'
    })

Index('my_index', my_table.c.data, postgresql_using='gin')

Index('my_index', my_table.c.data, postgresql_with={"fillfactor": 50})

Index('my_index', my_table.c.data, postgresql_tablespace='my_tablespace')

tbl = Table('testtbl', m, Column('data', Integer))

idx1 = Index('test_idx1', tbl.c.data, postgresql_concurrently=True)

CREATE INDEX CONCURRENTLY test_idx1 ON testtbl (data)

DROP INDEX CONCURRENTLY test_idx1

metadata = MetaData()
table = Table(
    "foo", metadata,
    Column("id", String))
index = Index(
    "foo_idx", table.c.id, postgresql_concurrently=True)

with engine.connect() as conn:
    with conn.execution_options(isolation_level='AUTOCOMMIT'):
        table.create(conn)

PostgreSQL Index Reflection

The PostgreSQL database creates a UNIQUE INDEX implicitly whenever the UNIQUE CONSTRAINT construct is used. When inspecting a table using _reflection.Inspector, the _reflection.Inspector.get_indexes and the _reflection.Inspector.get_unique_constraints will report on these two constructs distinctly; in the case of the index, the key duplicates_constraint will be present in the index entry if it is detected as mirroring a constraint. When performing reflection using Table(..., autoload_with=engine), the UNIQUE INDEX is not returned in _schema.Table.indexes when it is detected as mirroring a .UniqueConstraint in the _schema.Table.constraints collection .

Special Reflection Options

The _reflection.Inspector used for the PostgreSQL backend is an instance of .PGInspector, which offers additional methods:

from sqlalchemy import create_engine, inspect

engine = create_engine("postgresql+psycopg2://localhost/test")
insp = inspect(engine)  # will be a PGInspector

print(insp.get_enums())

PostgreSQL Table Options

Several options for CREATE TABLE are supported directly by the PostgreSQL dialect in conjunction with the _schema.Table construct:

• TABLESPACE:

  Table("some_table", metadata, ..., postgresql_tablespace='some_tablespace')
  

  The above option is also available on the .Index construct.

• ON COMMIT:

  Table("some_table", metadata, ..., postgresql_on_commit='PRESERVE ROWS')
  

• WITH OIDS:

  Table("some_table", metadata, ..., postgresql_with_oids=True)
  

• WITHOUT OIDS:

  Table("some_table", metadata, ..., postgresql_with_oids=False)
  

• INHERITS:

  Table("some_table", metadata, ..., postgresql_inherits="some_supertable")
  
  Table("some_table", metadata, ..., postgresql_inherits=("t1", "t2", ...))
  
  .. versionadded:: 1.0.0
  

• PARTITION BY:

  Table("some_table", metadata, ...,
        postgresql_partition_by='LIST (part_column)')
  
  .. versionadded:: 1.2.6
  

Table values, Table and Column valued functions, Row and Tuple objects

PostgreSQL makes great use of modern SQL forms such as table-valued functions, tables and rows as values. These constructs are commonly used as part of PostgreSQL's support for complex datatypes such as JSON, ARRAY, and other datatypes. SQLAlchemy's SQL expression language has native support for most table-valued and row-valued forms.

>>> from sqlalchemy import table, column, func, tuple_
>>> t = table("t", column("id"), column("fk"))
>>> stmt = t.select().where(
...     tuple_(t.c.id, t.c.fk) > (1,2)
... ).where(
...     func.ROW(t.c.id, t.c.fk) < func.ROW(3, 7)
... )
>>> print(stmt)
SELECT t.id, t.fk
FROM t
WHERE (t.id, t.fk) > (:param_1, :param_2) AND ROW(t.id, t.fk) < ROW(:ROW_1, :ROW_2)

>>> from sqlalchemy import table, column, func, select
>>> a = table( "a", column("id"), column("x"), column("y"))
>>> stmt = select(func.row_to_json(a.table_valued()))
>>> print(stmt)
SELECT row_to_json(a) AS row_to_json_1
FROM a

ARRAY Types

The PostgreSQL dialect supports arrays, both as multidimensional column types as well as array literals:

• _postgresql.ARRAY - ARRAY datatype
• _postgresql.array - array literal
• _postgresql.array_agg - ARRAY_AGG SQL function
• _postgresql.aggregate_order_by - helper for PG's ORDER BY aggregate function syntax.

JSON Types

The PostgreSQL dialect supports both JSON and JSONB datatypes, including psycopg2's native support and support for all of PostgreSQL's special operators:

• _postgresql.JSON
• _postgresql.JSONB

HSTORE Type

The PostgreSQL HSTORE type as well as hstore literals are supported:

• _postgresql.HSTORE - HSTORE datatype
• _postgresql.hstore - hstore literal

ENUM Types

PostgreSQL has an independently creatable TYPE structure which is used to implement an enumerated type. This approach introduces significant complexity on the SQLAlchemy side in terms of when this type should be CREATED and DROPPED. The type object is also an independently reflectable entity. The following sections should be consulted:

• _postgresql.ENUM - DDL and typing support for ENUM.
• .PGInspector.get_enums - retrieve a listing of current ENUM types
• .postgresql.ENUM.create , .postgresql.ENUM.drop - individual CREATE and DROP commands for ENUM.

from sqlalchemy import TypeDecorator
from sqlalchemy.dialects.postgresql import ARRAY

class ArrayOfEnum(TypeDecorator):
    impl = ARRAY

    def bind_expression(self, bindvalue):
        return sa.cast(bindvalue, self)

    def result_processor(self, dialect, coltype):
        super_rp = super(ArrayOfEnum, self).result_processor(
            dialect, coltype)

        def handle_raw_string(value):
            inner = re.match(r"^{(.*)}$", value).group(1)
            return inner.split(",") if inner else []

        def process(value):
            if value is None:
                return None
            return super_rp(handle_raw_string(value))
        return process

Table(
    'mydata', metadata,
    Column('id', Integer, primary_key=True),
    Column('data', ArrayOfEnum(ENUM('a', 'b, 'c', name='myenum')))

)

class CastingArray(ARRAY):
    def bind_expression(self, bindvalue):
        return sa.cast(bindvalue, self)

Table(
    'mydata', metadata,
    Column('id', Integer, primary_key=True),
    Column('data', CastingArray(JSONB))
)