API Reference¶

JSON Functions¶

json.agg¶

sqla_ext.func.json.agg(expression: Union[sqlalchemy.sql.elements.ColumnClause, sqlalchemy.sql.elements.TextClause]) → None[source]¶

JSON array aggregation

Dialects

mysql
postgresql
sqlite

Parameters

expression – A SQL expression, such as a ColumnElement expression or a TextClause with elements that are castable as JSON

Returns

FuntionElement

E.g.:

from sqlalchemy import select, Integer, text, table, column
from sqla_ext import func as func_ext

t = table('some_table', column('q', Integer))

query = select([
    func_ext.json.agg(t.q)
])

The above statement will produce SQL resembling:

SELECT
    jsonb_agg(some_table.q)
FROM
    some_table

json.build_object¶

sqla_ext.func.json.build_object(*clauses, **kwargs)[source]¶

JSON object creation

Dialects

mysql
postgresql
sqlite

Parameters

expression – A SQL expression, such as a ColumnElement expression or a TextClause with elements that are castable as JSON

Returns

FuntionElement

E.g.:

from sqlalchemy import select
from sqla_ext import func as func_ext

query = select([
    func_ext.json.build_object("first_key", "first_value", "second_key", 2)
])

The above statement will produce SQL resembling:

SELECT
    jsonb_build_object('first_key', 'first_value', 'second_key', 2)

json.to_array¶

sqla_ext.func.json.to_array(expression: Union[ColumnClause, TextClause], type_: TypeEngine) → None[source]¶

Cast a json array to a native array

Dialects

postgresql

Parameters

expression – A SQL expression, such as a ColumnElement expression or a TextClause that can be coerced into a JSONB array
type_ – A TypeEngine class or instance indicating the type to CAST elements of the native array as.

Returns

FuntionElement

E.g.:

from sqlalchemy import select, Integer, text
from sqla_ext import func as func_ext

query = select([
    func_ext.json.to_array(text("'[1,2,3,4]'::jsonb"), Integer)
])

The above statement will produce SQL resembling:

SELECT
    array_agg(CAST(anon_1 AS INTEGER)) AS array_agg_1
FROM
    jsonb_array_elements(CAST('[1,2,3,4]'::jsonb AS JSONB)) AS anon_1

Datetime Functions¶

datetime.utc_now¶

sqla_ext.func.datetime.utc_now() → None[source]¶

Current timestamp in UTC timezone

Dialects

mysql
postgresql
sqlite

Returns

FuntionElement

E.g.:

from sqlalchemy import select
from sqla_ext import func as func_ext

query = select([
    func_ext.datetime.utc_now()
])

The above statement will produce SQL resembling:

SELECT
    timezone('utc', current_timestamp)

Inspect¶

to_core_table¶

sqla_ext.inspect.to_core_table(entity: Union[sqlalchemy.sql.schema.Table, sqla_ext.protocols.ORMTableProtocol, sqlalchemy.orm.mapper.Mapper, Callable[], Union[sqla_ext.protocols.ORMTableProtocol, sqlalchemy.sql.schema.Table]]]) → sqlalchemy.sql.schema.Table[source]¶

Coerces multiple SQLA Table-like entities to a core sqlalcehmy.Table

Parameters: entity – A sqlalchemy Table, DeclarativeBase or Mapper
Returns: Table

to_schema_name¶

sqla_ext.inspect.to_schema_name(entity: Union[sqlalchemy.sql.schema.Table, sqla_ext.protocols.ORMTableProtocol, sqlalchemy.orm.mapper.Mapper, Callable[], Union[sqla_ext.protocols.ORMTableProtocol, sqlalchemy.sql.schema.Table]]]) → Optional[str][source]¶

Get the schema of a sqlalchemy table-like entity

Parameters: entity – A sqlalchemy Table, DeclarativeBase or Mapper
Returns: Optional[str]

to_table_name¶

sqla_ext.inspect.to_table_name(entity: Union[sqlalchemy.sql.schema.Table, sqla_ext.protocols.ORMTableProtocol, sqlalchemy.orm.mapper.Mapper, Callable[], Union[sqla_ext.protocols.ORMTableProtocol, sqlalchemy.sql.schema.Table]]]) → str[source]¶

Get the name of a sqlalchemy table-like entity

Parameters: entity – A sqlalchemy Table, DeclarativeBase or Mapper
Returns: str

Types¶

CITEXT¶

class sqla_ext.types.postgresql.CITEXT[source]¶

Case Insensitive Text Type

Dialects

postgresql

https://www.postgresql.org/docs/current/citext.html

E.g.:

from sqlalchemy import Column, Integer
from sqlalchemy.orm import declarative_base
from sqla_ext.types.postgresql import CITEXT


class User(Base):
    __tablename__ = 'user'
    id = sa.Column(Integer, primary_key=True)
    email = sa.Column(CITEXT())


user = User(id=1, email='John.Smith@example.com')
session.add(user)
session.commit()

# Note: query email is lowercase
user = (
    session.query(User)
    .filter(User.email == 'john.smith@example.com')
    .one()
)
assert user.id == 1