simple_view_sample.py

# -*- coding:utf-8 -*-

import sqlalchemy as sa
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.schema import DDLElement
from sqlalchemy.sql import table
from sqlalchemy.ext import compiler

# see:https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/Views

Session = scoped_session(sessionmaker())
engine = sa.create_engine('sqlite://', echo=True)
Session.configure(bind=engine)

Base = declarative_base(bind=engine)


class X(Base):
    __tablename__ = "X"
    query = Session.query_property()

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String(255), default="", nullable=False)


class Y(Base):
    __tablename__ = "Y"
    query = Session.query_property()

    id = sa.Column(sa.Integer, primary_key=True)
    name = sa.Column(sa.String(255), default="", nullable=False)


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


class DropView(DDLElement):
    def __init__(self, name):
        self.name = name


@compiler.compiles(CreateView)
def compile(element, compiler, **kw):
    return "CREATE VIEW %s AS %s" % (element.name, compiler.sql_compiler.process(element.selectable))


@compiler.compiles(DropView)
def compile_(element, compiler, **kw):
    return "DROP VIEW %s" % (element.name)


def view(name, metadata, selectable):
    t = table(name)

    for c in selectable.c:
        c._make_proxy(t)

    CreateView(name, selectable).execute_at('after-create', metadata)
    DropView(name).execute_at('before-drop', metadata)
    return t


class XorY(Base):
    lhs = sa.select([X.name, X.id.label("x_id"), sa.literal_column("-1").label("y_id")]).select_from(X.__table__)
    rhs = sa.select([Y.name, sa.literal_column("-1").label("x_id"), Y.id.label("y_id")]).select_from(Y.__table__)
    __table__ = view("XorY", Base.metadata, lhs.union_all(rhs))

    query = Session.query_property()

Base.metadata.create_all()

Session.add(X(name="foo0"))
Session.add(X(name="foo1"))
Session.add(X(name="foo2"))

Session.add(Y(name="boo0"))
Session.add(Y(name="boo1"))
Session.add(Y(name="boo2"))

Session.commit()

for xory in XorY.query.all():
    print(xory, xory.name)