sqlite 如何使用SQLAlchemy在SQL数据库中创建一个Union表？

我有以下结构：

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}')"

并且我想在数据库中创建一个新的永久表，它是这两个表的UNION（使用列ID和CURRENCY）
我知道我可以用这个在外面创造出这样的效果：

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

但我希望（如果可能的话）在数据库中有一个关系表，当我在EQUITIES或BONDS表中更改任何内容时，它会自动更新。

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

有人能帮我做这个吗？非常感谢

从Views recipe和您的类（简化）中，下面是列集合的并集视图的功能版本。
阅读recipe以了解更多细节，但它只是创建必要的编译器扩展，并在适当的时候通过事件使用它们。

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.**这是一个union，所以它只有不同的值。请注意，插入了四个示例，但视图中只有三个。这种重复数据删除（排序+筛选）在大型数据集上速度很慢，如果您不介意，请使用允许重复的union_all。

sqlite 如何使用SQLAlchemy在SQL数据库中创建一个Union表？

1条答案

相关问题

热门标签

最新问答