sqlalchemy.schema DDLElement Example Code

# base.py
import functools

from sqlalchemy import exc
from sqlalchemy import Integer
from sqlalchemy import types as sqltypes
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.schema import Column
from sqlalchemy.schema import DDLElement
from sqlalchemy.sql.elements import quoted_name

from ..util import sqla_compat
from ..util.sqla_compat import _columns_for_constraint  # noqa
from ..util.sqla_compat import _find_columns  # noqa
from ..util.sqla_compat import _fk_spec  # noqa
from ..util.sqla_compat import _is_type_bound  # noqa
from ..util.sqla_compat import _table_for_constraint  # noqa


class AlterTable(DDLElement):


    def __init__(self, table_name, schema=None):
        self.table_name = table_name
        self.schema = schema


class RenameTable(AlterTable):
    def __init__(self, old_table_name, new_table_name, schema=None):
        super(RenameTable, self).__init__(old_table_name, schema=schema)
        self.new_table_name = new_table_name


class AlterColumn(AlterTable):
    def __init__(
        self,
        name,
        column_name,
        schema=None,
        existing_type=None,
        existing_nullable=None,
        existing_server_default=None,
        existing_comment=None,
    ):


## ... source file continues with no further DDLElement examples...

# ddl.py
import sqlalchemy as sa
from sqlalchemy.ext import compiler as sa_compiler
from sqlalchemy.schema import DDLElement

from .compat import string_types


def _check_if_key_exists(key):
    return isinstance(key, sa.Column) or key


def get_table_attributes(preparer,
                         diststyle=None,
                         distkey=None,
                         sortkey=None,
                         interleaved_sortkey=None):
    text = ""

    has_distkey = _check_if_key_exists(distkey)
    if diststyle:
        diststyle = diststyle.upper()
        if diststyle not in ('EVEN', 'KEY', 'ALL'):
            raise sa.exc.ArgumentError(
                u"diststyle {0} is invalid".format(diststyle)
            )
        if diststyle != 'KEY' and has_distkey:
            raise sa.exc.ArgumentError(


## ... source file abbreviated to get to DDLElement examples ...


    text = """\
        CREATE MATERIALIZED VIEW {name}
        {backup}
        {table_attributes}
        AS {selectable}\
    Drop the materialized view from the database.
    SEE:
    docs.aws.amazon.com/redshift/latest/dg/materialized-view-drop-sql-command

    This undoes the create command, as expected:

    >>> import sqlalchemy as sa
    >>> from sqlalchemy_redshift.dialect import DropMaterializedView
    >>> engine = sa.create_engine('redshift+psycopg2://example')
    >>> drop = DropMaterializedView(
    ...     'materialized_view_of_users',
    ...     if_exists=True
    ... )
    >>> print(drop.compile(engine))
    <BLANKLINE>
    DROP MATERIALIZED VIEW IF EXISTS materialized_view_of_users
    <BLANKLINE>
    <BLANKLINE>

    This can be included in any execute() statement.
        Build the DropMaterializedView DDLElement.

        Parameters
        ----------
        name: str
            name of the materialized view to drop
        if_exists: bool, optional
            if True, the IF EXISTS clause is added. This will make the query
            successful even if the view does not exist, i.e. it lets you drop
            a non-existant view. Defaults to False.
        cascade: bool, optional
            if True, the CASCADE clause is added. This will drop all
            views/objects in the DB that depend on this materialized view.
            Defaults to False.
    Formats and returns the drop statement for materialized views.

            distkey = distkey.name
        text += " DISTKEY ({0})".format(preparer.quote(distkey))

    has_sortkey = _check_if_key_exists(sortkey)
    has_interleaved = _check_if_key_exists(interleaved_sortkey)
    if has_sortkey and has_interleaved:
        raise sa.exc.ArgumentError(
            "Parameters sortkey and interleaved_sortkey are "
            "mutually exclusive; you may not specify both."
        )

    if has_sortkey or has_interleaved:
        keys = sortkey if has_sortkey else interleaved_sortkey
        if isinstance(keys, (string_types, sa.Column)):
            keys = [keys]
        keys = [key.name if isinstance(key, sa.Column) else key
                for key in keys]
        if has_interleaved:
            text += " INTERLEAVED"
        sortkey_string = ", ".join(preparer.quote(key)
                                   for key in keys)
        text += " SORTKEY ({0})".format(sortkey_string)
    return text


class CreateMaterializedView(DDLElement):
    def __init__(self, name, selectable, backup=True, diststyle=None,
                 distkey=None, sortkey=None, interleaved_sortkey=None):
        self.name = name
        self.selectable = selectable
        self.backup = backup
        self.diststyle = diststyle
        self.distkey = distkey
        self.sortkey = sortkey
        self.interleaved_sortkey = interleaved_sortkey


@sa_compiler.compiles(CreateMaterializedView)
def compile_create_materialized_view(element, compiler, **kw):



## ... source file continues with no further DDLElement examples...

# view.py
import sqlalchemy as sa
from sqlalchemy.ext import compiler
from sqlalchemy.schema import DDLElement, PrimaryKeyConstraint


class CreateView(DDLElement):
    def __init__(self, name, selectable, materialized=False):
        self.name = name
        self.selectable = selectable
        self.materialized = materialized


@compiler.compiles(CreateView)
def compile_create_materialized_view(element, compiler, **kw):
    return 'CREATE {}VIEW {} AS {}'.format(
        'MATERIALIZED ' if element.materialized else '',
        element.name,
        compiler.sql_compiler.process(element.selectable, literal_binds=True),
    )


class DropView(DDLElement):
    def __init__(self, name, materialized=False, cascade=True):
        self.name = name
        self.materialized = materialized
        self.cascade = cascade


@compiler.compiles(DropView)
def compile_drop_materialized_view(element, compiler, **kw):
    return 'DROP {}VIEW IF EXISTS {} {}'.format(
        'MATERIALIZED ' if element.materialized else '',
        element.name,
        'CASCADE' if element.cascade else ''
    )


def create_table_from_selectable(
    name,
    selectable,
    indexes=None,
    metadata=None,
    aliases=None
):
    if indexes is None:
        indexes = []


## ... source file continues with no further DDLElement examples...