How to create an SQL View with SQLAlchemy?

Question

Is there a "Pythonic" way (I mean, no "pure SQL" query) to define an SQL view with SQLAlchemy?

Answer 1

Update: See also the SQLAlchemy usage recipe here

Creating a (read-only non-materialized) view is not supported out of the box as far as I know. But adding this functionality in SQLAlchemy 0.7 is straightforward (similar to the example I gave here). You just have to write a compiler extension CreateView. With this extension, you can then write (assuming that t is a table object with a column 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 

Here is a working example:

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 

If you want, you can also specialize for a dialect, e.g.

@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)          ) 

Answer 2

stephan's answer is a good one and covers most bases, but what left me unsatisfied was the lack of integration with the rest of SQLAlchemy (the ORM, automatic dropping etc.). After hours of experimenting and piecing together knowledge from all corners of the internet I came up with the following:

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' 

Note that I am utilizing the sqlalchemy_views package, just to simplify things.

Defining a view (e.g. globally like your Table models):

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] 

Mapping the views (enable ORM functionality):

Do when loading up your app, before any queries and after setting up the DB.

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

Creating the views:

Do when initializing the database, e.g. after a create_all() call.

from sqlalchemy import orm   for view in views:     db.engine.execute(CreateView(view)) 

How to query a view:

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

This would return exactly what you expect (a list of objects that each has a SomeModel object and a SampleView object).

Dropping a view:

SampleView.__view__.drop(db.engine) 

It will also automatically get dropped during a drop_all() call.

This is obviously a very hacky solution but in my eyes it is the best one and cleanest one out there at the moment. I have tested it these past few days and have not had any issues. I'm not sure how to add in relationships (ran into problems there) but it's not really necessary, as demonstrated above in the query.

If anyone has any input, finds any unexpected issues, or knows a better way to do things, please do leave a comment or let me know.

This was tested on SQLAlchemy 1.2.6 and Python 3.6.