No olvides que en las ciencias un problema casi siempere puede tener más de una soluciones, pero compartiremos lo más óptimo y mejor.
Solución:
Actualizar: Consulte también la receta de uso de SQLAlchemy aquí
La creación de una vista (no materializada de solo lectura) no es compatible de forma inmediata, hasta donde yo sé. Pero agregar esta funcionalidad en SQLAlchemy 0.7 es sencillo (similar al ejemplo que di aquí). Solo tienes que escribir una extensión del compilador CreateView. Con esta extensión, puede escribir (asumiendo que t es un objeto de tabla con una columna 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
Aquí hay un ejemplo de trabajo:
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
Si lo desea, también puede especializarse en un dialecto, p. Ej.
@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)
)
La respuesta de Stephan es buena y cubre la mayoría de las bases, pero lo que me dejó insatisfecho fue la falta de integración con el resto de SQLAlchemy (el ORM, caída automática, etc.). Después de horas de experimentar y juntar conocimientos de todos los rincones de Internet, se me ocurrió lo siguiente:
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'
Tenga en cuenta que estoy utilizando el sqlalchemy_views paquete, solo para simplificar las cosas.
Definición de una vista (por ejemplo, globalmente como sus modelos de tabla):
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]
Mapeo de las vistas (habilite la funcionalidad ORM):
Hágalo al cargar su aplicación, antes de cualquier consulta y después de configurar la base de datos.
for view in views:
if not hasattr(view, '_sa_class_manager'):
orm.mapper(view, view.__view__)
Creando las vistas:
Hágalo al inicializar la base de datos, por ejemplo, después de una llamada a create_all ().
from sqlalchemy import orm
for view in views:
db.engine.execute(CreateView(view))
Cómo consultar una vista:
results = db.session.query(SomeModel, SampleView).join(
SampleView,
SomeModel.id == SampleView.some_model_id
).all()
Esto devolvería exactamente lo que espera (una lista de objetos que cada uno tiene un objeto SomeModel y un objeto SampleView).
Dejar caer una vista:
SampleView.__view__.drop(db.engine)
También se eliminará automáticamente durante una llamada drop_all ().
Obviamente, esta es una solución muy peligrosa, pero en mi opinión, es la mejor y la más limpia que existe en este momento. Lo he probado estos últimos días y no he tenido ningún problema. No estoy seguro de cómo agregar relaciones (tuve problemas allí) pero no es realmente necesario, como se demostró anteriormente en la consulta.
Si alguien tiene algún comentario, encuentra algún problema inesperado o conoce una mejor manera de hacer las cosas, por favor deje un comentario o hágamelo saber.
Esto se probó en SQLAlchemy 1.2.6 y Python 3.6.
En estos días hay un paquete PyPI para eso: SQLAlchemy Views.
Desde su página de 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
Sin embargo, pediste un sin consulta “SQL puro”, entonces probablemente quieras el definition anterior para crearse con el objeto de consulta SQLAlchemy.
Afortunadamente, el text() en el ejemplo anterior deja claro que el definition parámetro a CreateView es un objeto de consulta de este tipo. Entonces, algo como esto debería funcionar:
>>> 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)
... )
Aquí está la parte interesante:
>>> 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
Sección de Reseñas y Valoraciones
Nos encantaría que puedieras dar visibilidad a este ensayo si te valió la pena.