API¶

class sqlalchemy_declarative_extensions.schema.Schema¶

Represents a schema.

classmethod coerce_from_unknown(unknown)¶

Parameters:: unknown (Self | str)
Return type:: Self

name: str¶

to_sql_create()¶

Return type:: sqlalchemy.sql.base.Executable | str

to_sql_drop()¶

Return type:: sqlalchemy.sql.base.Executable | str

use_role: sqlalchemy_declarative_extensions.role.Role | str | None = None¶

class sqlalchemy_declarative_extensions.schema.Schemas¶

A collection of schemas and the settings for diff/collection.

Parameters:

schemas – The list of grants
ignore_unspecified – Optionally ignore detected grants which do not match the set of defined grants.

Examples

No schemas

>>> schemas = Schemas()

Some options set

>>> schemas = Schemas(ignore_unspecified=True)

With some actual schemas

>>> from sqlalchemy_declarative_extensions import Schema, Schemas
>>> schema = Schemas().are("foo", Schema("bar"), ...)

are(*schemas)¶

Declare the set of schemas which should exist.

Parameters:: schemas (Schema | str)

classmethod coerce_from_unknown(unknown)¶

Parameters:: unknown (None | Iterable[Schema | str] | Schemas)
Return type:: Schemas | None

classmethod extract(metadata)¶

Parameters:: metadata (sqlalchemy.MetaData | list[sqlalchemy.MetaData | None] | None)
Return type:: Self | None

ignore_unspecified: bool = False¶

schemas: Sequence[Schema] = ()¶

Decorate a class or declarative base model in order to register a View.

Given some object with the attributes: __tablename__, (optionally for schema) __table_args__, and __view__, registers a View object.

The __view__ attribute can be either a raw string query, a SQLAlchemy object capable of being compiled (namely text() or select()), or a no-argument function which returns either of the two.

This intentionally allows one to register a Model definition as a view, and have it register in the same way you might otherwise manually define it. This can be useful, to enable querying that view in native SQLAlchemy ORM-style, as though it were a table.

param base:: A declarative base object
param register_as_model:: Whether the view should be registered as a SQLAlchemy mapped object. Note this only works if the view defines mappable models columns (minimally a primary key), like a proper modeled table
param materialized:: Whether a view should be materialized or not. Accepts a bool for default options, a dialect-specific MaterializedOptions variant, or a dict describing that dialect-specific variant.

>>> try:
...     from sqlalchemy.orm import declarative_base
... except ImportError:
...     from sqlalchemy_declarative_extensions.sqlalchemy import declarative_base
>>> from sqlalchemy import Column, types
>>> from sqlalchemy_declarative_extensions import view
>>>
>>> Base = declarative_base()
>>>
>>> @view(Base)
... class Foo:
...     __tablename__ = "foo"
...     __view__ = "SELECT * from bar"
...
...     id = Column(types.Integer, primary_key=True)

class sqlalchemy_declarative_extensions.role.base.Roles¶

are(*roles)¶

Parameters:: roles (sqlalchemy_declarative_extensions.dialects.postgresql.Role | sqlalchemy_declarative_extensions.role.generic.Role | str)

classmethod coerce_from_unknown(unknown)¶

Parameters:: unknown (None | Iterable[sqlalchemy_declarative_extensions.dialects.postgresql.Role | sqlalchemy_declarative_extensions.role.generic.Role | str] | Roles)
Return type:: Roles | None

classmethod extract(metadata)¶

Parameters:: metadata (sqlalchemy.MetaData | list[sqlalchemy.MetaData | None] | None)
Return type:: Self | None

ignore_roles: list[str] = []¶

ignore_unspecified: bool = False¶

roles: Sequence[sqlalchemy_declarative_extensions.dialects.postgresql.Role | sqlalchemy_declarative_extensions.role.generic.Role] = []¶

class sqlalchemy_declarative_extensions.role.generic.Role¶

Represent a role.

Note a role can be defined as “external=True” so that it can be used in downstream DDL, without requiring that it be managed within the context of the library.

>>> base = Role("base", external=True)  # i.e. assumed to already exist, will not be created/updated
>>> foo = Role("base", in_roles=[base])

classmethod coerce_from_unknown(unknown)¶

Parameters:: unknown (str | Role)
Return type:: Role

external: bool = False¶

classmethod from_unknown_role(r)¶

Parameters:: r (Role)
Return type:: Role

property has_option¶

in_roles: list[Role | str] | None = None¶

property is_dynamic: bool¶

Return type:: bool

name: str¶

normalize()¶

property options¶

to_sql_create(raw=True)¶

Parameters:: raw (bool)
Return type:: list[str]

to_sql_drop(raw=True)¶

Parameters:: raw (bool)
Return type:: list[str]

abstract to_sql_update(to_role, raw=True)¶

Parameters:: raw (bool)
Return type:: list[str]

abstract to_sql_use(undo)¶

Parameters:: undo (bool)
Return type:: list[str]

use_role: Role | str | None = None¶

class sqlalchemy_declarative_extensions.grant.Grants¶

A collection of grants and the settings for diff/collection.

Parameters:

grants – The list of grants
ignore_unspecified – Defaults to False. When True, ignore detected grants which do not match the set of defined grants.
ignore_self_grants – Defaults to True. When True, ignores grants to the current user. It’s typical in migrations that the a single user performs migrations and will have implicitly granted grants on all objects. In this scenario, it can be tedious to define those permissions on every object, so they are ignored by default.
only_defined_roles – Defaults to True. When True, only applies to roles specified in the roles section.
default_grants_imply_grants – Defaults to True. When True, default grants also imply the set of expected actual grants. This allows one to specify only default grants, and per-object grants will be made to match the default set.

Examples

No grants

>>> grants = Grants()

Some options set

>>> grants = Grants(ignore_unspecified=True)

With some actual grants

>>> from sqlalchemy_declarative_extensions.dialects.postgresql import DefaultGrant
>>> grants = Grants().are(DefaultGrant(..., ...), ...)

are(*grants)¶

Parameters:: grants (G)

classmethod coerce_from_unknown(unknown)¶

Parameters:: unknown (None | Iterable[G] | Grants)
Return type:: Grants | None

default_grants_imply_grants: bool = True¶

classmethod extract(metadata)¶

Parameters:: metadata (sqlalchemy.MetaData | list[sqlalchemy.MetaData | None] | None)
Return type:: Self | None

grants: list[G] = []¶

ignore_self_grants: bool = True¶

ignore_unspecified: bool = False¶

only_defined_roles: bool = True¶

class sqlalchemy_declarative_extensions.function.Function¶

Describes a user defined function.

Many function attributes are not currently supported. Support is currently minimal due to being a means to an end for defining triggers.

definition: str¶

classmethod from_unknown_function(f)¶

Parameters:: f (Function)
Return type:: Self

language: str = 'sql'¶

name: str¶

abstract normalize()¶

Return type:: Self

parameters: Sequence[str] | None = None¶: List of parameter definitions as strings, e.g., [‘param1 int’, ‘param2 varchar’]. The comparison logic parses these strings to compare parameter names and types.

property qualified_name¶

returns: str = 'void'¶

schema: str | None = None¶

abstract to_sql_create()¶

Return type:: list[str]

to_sql_drop()¶

Return type:: list[str]

to_sql_update()¶

Return type:: list[str]

with_language(language)¶

Parameters:: language (str)

with_name(name)¶

Parameters:: name (str)

with_return_type(return_type)¶

Parameters:: return_type (str)

class sqlalchemy_declarative_extensions.function.Functions¶

The collection of functions and associated options comparisons.

Note: ignore option accepts a sequence of strings. Each string is individually: interpreted as a “glob”. This means a string like “foo.*” would ignore all views contained within the schema “foo”.

append(function)¶

Parameters:: function (Function)

are(*functions)¶

Parameters:: functions (Function)

classmethod coerce_from_unknown(unknown)¶

Parameters:: unknown (None | Iterable[Function] | Functions)
Return type:: Functions | None

classmethod extract(metadata)¶

Parameters:: metadata (sqlalchemy.MetaData | list[sqlalchemy.MetaData | None] | None)
Return type:: Self | None

functions: list[Function] = []¶

ignore: list[str] = []¶

ignore_unspecified: bool = False¶

include: list[str] | None = None¶

sqlalchemy_declarative_extensions.function.register_function(base_or_metadata, function)¶

This can be used instead of the static registration through Functions on a declarative base or MetaData, to imperitively register functions.

Parameters:

base_or_metadata (sqlalchemy_declarative_extensions.sqlalchemy.HasMetaData | sqlalchemy.MetaData)
function (Function)

class sqlalchemy_declarative_extensions.trigger.Trigger¶

Describes a generic trigger.

execute: str¶

property identity: Hashable¶

Return the dialect-specific identity used to compare triggers.

Return type:: Hashable

name: str¶

named(name)¶

Parameters:: name (str)

on: str¶

abstract to_sql_create()¶

to_sql_drop()¶

to_sql_update(connection=None)¶

Parameters:: connection (sqlalchemy.engine.Connection | None)

class sqlalchemy_declarative_extensions.trigger.Triggers¶

append(trigger)¶

Parameters:: trigger (Trigger)

are(*triggers)¶

Parameters:: triggers (Trigger)

classmethod coerce_from_unknown(unknown)¶

Parameters:: unknown (None | Iterable[Trigger] | Triggers)
Return type:: Triggers | None

classmethod extract(metadata)¶

Parameters:: metadata (sqlalchemy.MetaData | list[sqlalchemy.MetaData | None] | None)
Return type:: Self | None

ignore: list[str] = []¶

ignore_unspecified: bool = False¶

include: list[str] | None = None¶

triggers: list[Trigger] = []¶

sqlalchemy_declarative_extensions.trigger.register_trigger(base_or_metadata, trigger)¶

This can be used instead of the static registration through Triggers on a declarative base or MetaData, to imperitively register triggers.

Parameters:

base_or_metadata (sqlalchemy_declarative_extensions.sqlalchemy.HasMetaData | sqlalchemy.MetaData)
trigger (Trigger)

class sqlalchemy_declarative_extensions.row.Row(tablename, *, schema=None, **column_values)¶

Parameters:: schema (str | None)

column_values: dict[str, Any]¶

property qualified_name¶

qualify(schema)¶

Attach a schema to this row, if it doesn’t have one already.

Examples

>>> row = Row("foo")
>>> row.qualify("bar")
Row(schema='bar', tablename='foo', column_values={})

Parameters:: schema (str | None)
Return type:: Row

schema: str | None¶

tablename: str¶

class sqlalchemy_declarative_extensions.row.Rows¶

are(*rows)¶

Parameters:: rows (Row)

classmethod coerce_from_unknown(unknown)¶

Parameters:: unknown (None | Iterable[Row] | Rows)
Return type:: Rows | None

classmethod extract(metadata)¶

Parameters:: metadata (sqlalchemy.MetaData | list[sqlalchemy.MetaData | None] | None)
Return type:: tuple[Self, sqlalchemy.MetaData] | None

ignore_unspecified: bool = False¶

included_tables: list[str] = []¶

rows: list[Row] = []¶

class sqlalchemy_declarative_extensions.row.Table(name, **column_values)¶

Convenience class for producing multiple rows against the same table.

Examples

Rows might be created directly, like so:

>>> [
...     Row("users", id=1, name="John", active=True),
...     Row("users", id=2, name="Bob", active=True),
... ]
[Row(schema=None, tablename='users', column_values={'id': 1, 'name': 'John', 'active': True}), Row(schema=None, tablename='users', column_values={'id': 2, 'name': 'Bob', 'active': True})]

But use of Table can help elide repetition among those rows:

>>> users = Table("users", active=True)
>>> [
...     users.row(id=1, name="John"),
...     users.row(id=2, name="Bob"),
... ]
[Row(schema=None, tablename='users', column_values={'active': True, 'id': 1, 'name': 'John'}), Row(schema=None, tablename='users', column_values={'active': True, 'id': 2, 'name': 'Bob'})]

column_values: dict[str, Any]¶

name: str¶

row(**column_values)¶

Return type:: Row

Alembic¶

sqlalchemy_declarative_extensions.alembic.register_alembic_events(*, databases=True, schemas=True, views=True, roles=True, grants=True, functions=True, procedures=True, triggers=True, rows=True)¶

By default all object types are enabled, but each can be individually disabled.

Note this is the opposite of the defaults when registering against SQLAlchemy’s event system.

Parameters:

databases (bool)
schemas (bool)
views (bool)
roles (bool)
grants (bool)
functions (bool)
procedures (bool)
triggers (bool)
rows (bool)