如何使用 SQLAlchemy 创建 SQL 视图?

Posted

技术标签:

【中文标题】如何使用 SQLAlchemy 创建 SQL 视图?【英文标题】:How to create an SQL View with SQLAlchemy? 【发布时间】:2012-04-03 17:56:42 【问题描述】:

是否有“Pythonic”方式(我的意思是,没有“纯 SQL”查询)用 SQLAlchemy 定义 SQL 视图?

【问题讨论】:

【参考方案1】:

更新:另请参阅 SQLAlchemy 使用配方here

据我所知,开箱即用不支持创建(只读非物化)视图。但是在 SQLAlchemy 0.7 中添加这个功能很简单(类似于我给here 的示例)。你只需要写一个compiler extensionCreateView。有了这个扩展,你就可以写(假设t是一个表对象,列id

createview = CreateView('viewname', t.select().where(t.c.id>5))
engine.execute(createview)

v = Table('viewname', metadata, autoload=True)
for r in engine.execute(v.select()):
    print r

这是一个工作示例:

from sqlalchemy import Table
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import Executable, ClauseElement

class CreateView(Executable, ClauseElement):
    def __init__(self, name, select):
        self.name = name
        self.select = select

@compiles(CreateView)
def visit_create_view(element, compiler, **kw):
    return "CREATE VIEW %s AS %s" % (
         element.name,
         compiler.process(element.select, literal_binds=True)
         )

# test data
from sqlalchemy import MetaData, Column, Integer
from sqlalchemy.engine import create_engine
engine = create_engine('sqlite://')
metadata = MetaData(engine)
t = Table('t',
          metadata,
          Column('id', Integer, primary_key=True),
          Column('number', Integer))
t.create()
engine.execute(t.insert().values(id=1, number=3))
engine.execute(t.insert().values(id=9, number=-3))

# create view
createview = CreateView('viewname', t.select().where(t.c.id>5))
engine.execute(createview)

# reflect view and print result
v = Table('viewname', metadata, autoload=True)
for r in engine.execute(v.select()):
    print r

如果你愿意,你也可以专攻一种方言,例如

@compiles(CreateView, 'sqlite')
def visit_create_view(element, compiler, **kw):
    return "CREATE VIEW IF NOT EXISTS %s AS %s" % (
         element.name,
         compiler.process(element.select, literal_binds=True)
         )

【讨论】:

我可以将 map v 与 orm.mapper 一起使用吗?像v = Table('viewname', metadata, autoload=True) class ViewName(object): def __init__(self, name): self.name = name mapper(ViewName, v) 以上可以吗?因为我会在会话中使用 View。 @SyedHabibM:是的,这是可能的。但是,您必须手动设置主键,例如orm.mapper(ViewName, v, primary_key=pk, properties=prop),其中pkprop 分别是您的主键(或键)和属性。见docs.sqlalchemy.org/en/latest/orm/…。 @SyedHabibM:当您通过覆盖列规范并指定 PK 来使用自动加载表时,您也可以执行 stephan 提到的操作:v = Table('viewname', metadata, Column('my_id_column', Integer, primary_key=True), autoload=True) @SyedHabibMI 现在确实用一个工作示例回答了您各自的问题***.com/q/20518521/92092。我也会在那里添加 van 的评论。【参考方案2】:

stephan 的回答很好,涵盖了大多数基础,但让我不满意的是缺乏与 SQLAlchemy 的其余部分(ORM、自动删除等)的集成。经过数小时的试验和拼凑来自互联网各个角落的知识,我得出了以下结论:

import sqlalchemy_views
from sqlalchemy import Table
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.ddl import DropTable


class View(Table):
    is_view = True


class CreateView(sqlalchemy_views.CreateView):
    def __init__(self, view):
        super().__init__(view.__view__, view.__definition__)


@compiles(DropTable, "postgresql")
def _compile_drop_table(element, compiler, **kwargs):
    if hasattr(element.element, 'is_view') and element.element.is_view:
        return compiler.visit_drop_view(element)

    # cascade seems necessary in case SQLA tries to drop 
    # the table a view depends on, before dropping the view
    return compiler.visit_drop_table(element) + ' CASCADE'

请注意,我正在使用 sqlalchemy_views 包,只是为了简化操作。

定义一个视图(例如,像您的表模型一样全局):

from sqlalchemy import MetaData, text, Text, Column


class SampleView:
    __view__ = View(
        'sample_view', MetaData(),
        Column('bar', Text, primary_key=True),
    )

    __definition__ = text('''select 'foo' as bar''')

# keeping track of your defined views makes things easier
views = [SampleView]

映射视图(启用 ORM 功能):

在加载您的应用程序、任何查询之前和设置数据库之后执行。

for view in views:
    if not hasattr(view, '_sa_class_manager'):
        orm.mapper(view, view.__view__)

创建视图:

在初始化数据库时执行,例如在 create_all() 调用之后。

from sqlalchemy import orm


for view in views:
    db.engine.execute(CreateView(view))

如何查询视图:

results = db.session.query(SomeModel, SampleView).join(
    SampleView,
    SomeModel.id == SampleView.some_model_id
).all()

这将完全返回您所期望的(对象列表,每个对象都有一个 SomeModel 对象和一个 SampleView 对象)。

删除视图:

SampleView.__view__.drop(db.engine)

它也会在 drop_all() 调用期间自动被丢弃。

这显然是一个非常老套的解决方案,但在我看来,它是目前最好、最干净的解决方案。这几天我已经测试过了,没有任何问题。我不确定如何添加关系(在那里遇到问题),但这并不是真正必要的,如上面的查询所示。

如果有人有任何意见、发现任何意外问题或知道更好的处理方式,请发表评论或告诉我。

这是在 SQLAlchemy 1.2.6 和 Python 3.6 上测试的。

【讨论】:

疯狂的时机,我自己一直在处理这个问题。对于 py 2.7 和 SQLa 1.1.2(不要问...),唯一需要的更改是 super(CreateView, self).__init__class SampleView(object) @Steven Dickinson 是的,听起来不错!是的,我认为这是一项非常常见的任务,这就是为什么我对它的文档如此糟糕/过时/浅薄感到惊讶的原因。但是,嘿,我想一次一步。 对于那些希望以声明方式执行此操作的人,我将视图定义在与表不同的元数据实例的单独文件中:Base = declarative_base(metadata=db.MetaData()) class ViewSample(Base): __tablename__ = 'view_sample' 我仍然包含 __definition__ 属性并调用 CreateView 来创建它正如原帖中所建议的那样。最后,我不得不修改 drop 装饰方法:if element.element.name.startswith('view_'): return compiler.visit_drop_view(element),因为我找不到将属性添加到嵌入表的方法。【参考方案3】:

现在有一个 PyPI 包:SQLAlchemy Views。

来自它的 PyPI 页面:

>>> from sqlalchemy import Table, MetaData
>>> from sqlalchemy.sql import text
>>> from sqlalchemy_views import CreateView, DropView

>>> view = Table('my_view', metadata)
>>> definition = text("SELECT * FROM my_table")

>>> create_view = CreateView(view, definition, or_replace=True)
>>> print(str(create_view.compile()).strip())
CREATE OR REPLACE VIEW my_view AS SELECT * FROM my_table

但是,您要求没有“纯 SQL”查询,因此您可能希望使用 SQLAlchemy 查询对象创建上面的 definition

幸运的是,上面示例中的text() 清楚地表明CreateViewdefinition 参数就是这样一个查询对象。所以这样的事情应该可以工作:

>>> from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
>>> from sqlalchemy.sql import select
>>> from sqlalchemy_views import CreateView, DropView

>>> metadata = MetaData()

>>> users = Table('users', metadata,
...     Column('id', Integer, primary_key=True),
...     Column('name', String),
...     Column('fullname', String),
... )

>>> addresses = Table('addresses', metadata,
...   Column('id', Integer, primary_key=True),
...   Column('user_id', None, ForeignKey('users.id')),
...   Column('email_address', String, nullable=False)
...  )

这是有趣的一点:

>>> view = Table('my_view', metadata)
>>> definition = select([users, addresses]).where(
...     users.c.id == addresses.c.user_id
... )
>>> create_view = CreateView(view, definition, or_replace=True)
>>> print(str(create_view.compile()).strip())
CREATE OR REPLACE VIEW my_view AS SELECT users.id, users.name,
users.fullname, addresses.id, addresses.user_id, addresses.email_address 
FROM users, addresses 
WHERE users.id = addresses.user_id

【讨论】:

【参考方案4】:

0.33.6 中的 SQLAlchemy-utils just added this functionality(在 pypi 中可用)。它具有视图、物化视图,并与 ORM 集成。它还没有记录,但我成功地使用了视图 + ORM。

您可以use their test as an example 使用 ORM 来获取常规视图和物化视图。

要创建视图,安装包后,请使用上面测试中的以下代码作为视图的基础:

class ArticleView(Base):
    __table__ = create_view(
        name='article_view',
        selectable=sa.select(
            [
                Article.id,
                Article.name,
                User.id.label('author_id'),
                User.name.label('author_name')
            ],
            from_obj=(
                Article.__table__
                    .join(User, Article.author_id == User.id)
            )
        ),
        metadata=Base.metadata
    )

其中Basedeclarative_basesaSQLAlchemy 包,create_view 是来自sqlalchemy_utils.view 的函数。

【讨论】:

你找到了和alembic一起使用的方法吗? @JorgeLeitao 我想知道同样的事情。打开了一张票 here 并包含了我目前正在尝试的一种解决方法。希望作者提供更好的解决方案,他通常很有帮助。【参考方案5】:

大致基于https://github.com/sqlalchemy/sqlalchemy/wiki/Views

仅包含sqlalchemy 的完整可执行示例,希望您不要为了让它运行而花费数小时。

import sqlalchemy as sa
import sqlalchemy.schema
import sqlalchemy.ext.compiler


engine = sa.create_engine('postgresql://localhost/postgres')
meta = sa.MetaData()

Session = sa.orm.sessionmaker(bind=engine)
session = Session()


class Drop(sa.schema.DDLElement):
    def __init__(self, name, schema):
        self.name   = name
        self.schema = schema


class Create(sa.schema.DDLElement):
    def __init__(self, name, select, schema='public'):
        self.name   = name
        self.schema = schema
        self.select = select

        sa.event.listen(meta, 'after_create', self)
        sa.event.listen(meta, 'before_drop', Drop(name, schema))


@sa.ext.compiler.compiles(Create)
def createGen(element, compiler, **kwargs):
    return 'CREATE OR REPLACE VIEW schema."name" AS select'.format(
        name   = element.name,
        schema = element.schema,
        select = compiler.sql_compiler.process(
            element.select,
            literal_binds = True
        ),
    )


@sa.ext.compiler.compiles(Drop)
def dropGen(element, compiler, **kw):
    return 'DROP VIEW schema."name"'.format(
        name   = element.name,
        schema = element.schema,
    )


if __name__ == '__main__':
    view = Create(
        name   = 'myview',
        select = sa.select(sa.literal_column('1 AS col'))
    )
    meta.create_all(bind=engine, checkfirst=True)
    print(session.execute('SELECT * FROM myview').all())
    session.close()

【讨论】:

【参考方案6】:

我找不到简短而方便的答案。

我不需要 View 的额外功能(如果有的话),所以我只是将视图视为普通表,就像其他表定义一样。

所以基本上我有 a.py 定义所有表和视图,sql 相关的东西,和 main.py 我从 a.py 导入这些类并使用它们。

这是我在 a.py 中添加的内容:

class A_View_From_Your_DataBase(Base):
    __tablename__ = 'View_Name'
    keyword = Column(String(100), nullable=False, primary_key=True)

值得注意的是,即使视图中没有主键,您也需要添加 primary_key 属性。

【讨论】:

这个解决方案不起作用,它说sqlalchemy.exc.NoSuchTableError【参考方案7】:

没有纯 SQL 的 SQL 视图? 您可以创建一个类或函数来实现定义的视图。

function get_view(con):
  return Table.query.filter(Table.name==con.name).first()

【讨论】:

抱歉,我不是这么问的。我的英语不是很完美,如果你误解了我很抱歉:)

以上是关于如何使用 SQLAlchemy 创建 SQL 视图?的主要内容,如果未能解决你的问题,请参考以下文章

Flask SQL性能测试:使用flask-sqlalchemy检测慢SQL语句

Flask-Sqlalchemy 使用 postgres 创建视图

使用 SQLAlchemy 刷新物化视图

无法使用 sqlalchemy 建立连接 SQL Server

如果数据库已经存在,你还需要创建 SQLAlchemy 模型吗?

使用 Python 的 Sqlalchemy 和 Singlestore sql 创建类似于现有的表