How to create a Union table inside the SQL database using SQLAlchemy?

Question

I have the following structure :

import sqlalchemy as sa
import sqlalchemy.orm as orm

Base = orm.declarative_base()

class Equity(Base) :
    __tablename__ = 'equities'
        
    id = sa.Column(sa.String, primary_key=True)
    name = sa.Column(sa.String, nullable=False)
    currency = sa.Column(sa.String, nullable=False)
    country = sa.Column(sa.String, nullable=False)
    sector = sa.Column(sa.String, nullable=True)
        
    def __repr__(self) :
        return f"Equity('Ticker: {self.id}', 'Name: {self.name}')"

class Bond(Base) :
    __tablename__ = 'bonds'
        
    id = sa.Column(sa.String, primary_key=True)
    name = sa.Column(sa.String, nullable=False)
    country = sa.Column(sa.String, nullable=False)
    currency = sa.Column(sa.String, nullable=False)
    sector = sa.Column(sa.String, nullable=False)
            
    def __repr__(self) :
        return f"Bond('Ticker: {self.id}', 'Name: {self.name}')"

And I want to create a new permanent table inside the database that is the UNION of those two tables (using the columns ID and CURRENCY)

I know I can create that outside using this :

results = session.query(Equity.id, Equity.currency).union(session.query(Bond.id, Bond.currency))

But I want (if possible) to have a relationship table inside my Database that automatically updates when I change anything on either on the EQUITIES or BONDS table. Something like this :

class NewUnionTable(Base) :
    <relationship><union>

Can someone help me create this, please? Appreciate very much

Sounds like you just need a VIEW which union's the two source tables? Then, anything that changes in either table is presented automatically by the view... — MatBailie, Jan 23 '23 at 18:42
It might be that. I am still new to SQLAlchemy. I saw an extensive way to create a VIEW using SQLAlchemy here https://stackoverflow.com/questions/9766940/how-to-create-an-sql-view-with-sqlalchemy but if you have any simpler way, I'd appreciate it — Gcedism, Jan 23 '23 at 22:53
There are many answers there, including references to the docs. If you don't find any of them useful, you're probably out of luck. — MatBailie, Jan 23 '23 at 22:57

score 1 · Answer 1 · answered Jan 24 '23 at 19:44

From the Views recipe and your classes (simplified), here is a functioning version of the view of the union of your sets of columns.

Read through the recipe for more details, but simply it's creating the necessary compiler extensions and using them when appropriate via an event.

from sqlalchemy import Column, String, create_engine, event, inspect, select, table, union
from sqlalchemy.ext import compiler
from sqlalchemy.orm import Session, declarative_base
from sqlalchemy.schema import DDLElement


# boilerplate from https://github.com/sqlalchemy/sqlalchemy/wiki/Views
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 _create_view(element, compiler, **kw):
    return "CREATE VIEW %s AS %s" % (
        element.name,
        compiler.sql_compiler.process(element.selectable, literal_binds=True),
    )


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


def view_exists(ddl, target, connection, **kw):
    return ddl.name in inspect(connection).get_view_names()


def view_doesnt_exist(ddl, target, connection, **kw):
    return not view_exists(ddl, target, connection, **kw)


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

    t._columns._populate_separate_keys(
        col._make_proxy(t) for col in selectable.selected_columns
    )

    event.listen(
        metadata,
        "after_create",
        CreateView(name, selectable).execute_if(callable_=view_doesnt_exist),
    )
    event.listen(
        metadata,
        "before_drop",
        DropView(name).execute_if(callable_=view_exists),
    )
    return t


# demo
Base = declarative_base()


class Equity(Base):
    __tablename__ = "equities"

    id = Column(String, primary_key=True)
    currency = Column(String, nullable=False)


class Bond(Base):
    __tablename__ = "bonds"

    id = Column(String, primary_key=True)
    currency = Column(String, nullable=False)


common_view = view(
    "bonds_equities_union_view",
    Base.metadata,
    union(
        select(Equity.id.label("id"), Equity.currency.label("currency")),
        select(Bond.id.label("id"), Bond.currency.label("currency")),
    ),
)


engine = create_engine("sqlite://", echo=True, future=True)

Base.metadata.create_all(engine)

with Session(engine) as session:
    session.add_all(
        [
            Equity(id="AAA", currency="EUR"),
            Equity(id="AAB", currency="USD"),
            Bond(id="AAA", currency="EUR"),
            Equity(id="EEF", currency="GBP"),
        ]
    )
    session.commit()

with Session(engine) as session:
    results = session.execute(select(common_view)).all()

print(results)  # [('AAA', 'EUR'), ('AAB', 'USD'), ('EEF', 'GBP')]

NB. this is a union, so it's only distinct values. Notice four instances inserted but only three in the view. This deduplication (sort + filter) is slow on large datasets, if you do not care use a union_all which allows duplicates.

How to create a Union table inside the SQL database using SQLAlchemy?

1 Answers1