Hello. I am currently working on an implementation of PostgreSQL materialized views in SQLAlchemy and Alembic. To give a basic explanation, materialized views are regular views but all the rows are computed statically, and you can refresh them with a REFRESH MATERIALIZED VIEW query. I've attached the code I have so far. It uses sqlalchemy.sql.table and column proxies to simulate a table, and also implements Alembic operations with autogenerate. I want to make MaterializedView into a proper class like Table. What would be the best way to go about this? What should I inherit from? Which methods should I override? Any input is appreciated.
-- SQLAlchemy - The Python SQL Toolkit and Object Relational Mapper http://www.sqlalchemy.org/ To post example code, please provide an MCVE: Minimal, Complete, and Verifiable Example. See http://stackoverflow.com/help/mcve for a full description. --- You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To unsubscribe from this group and stop receiving emails from it, send an email to sqlalchemy+unsubscr...@googlegroups.com. To view this discussion on the web visit https://groups.google.com/d/msgid/sqlalchemy/ee389a85-cba5-4773-9862-a412aa0e67ca%40googlegroups.com.
import importlib from typing import Union from sqlalchemy import MetaData from sqlalchemy.ext import compiler from sqlalchemy.schema import DDLElement from sqlalchemy.sql import Selectable, table from sqlalchemy.sql.compiler import DDLCompiler from alembic.operations import Operations, MigrateOperation from alembic.autogenerate import comparators, renderers from alembic.autogenerate.api import AutogenContext # http://www.jeffwidman.com/blog/847/using-sqlalchemy-to-create-and-manage-postgresql-materialized-views/ # Since that blogpost; # 1) focuses on Flask-SQLalchemy # 2) doesn't provide a way to migrate # I made this code use pure SQLAlchemy and also defined Alembic operations. __all__ = ( "CreateMaterializedView", "DropMaterializedView", "CreateMaterializedViewOp", "DropMaterializedViewOp", "MaterializedView", ) def _import_selectable(path: str) -> Selectable: """Imports a Selectable object from the given import path.""" module, obj = path.rsplit(".", 1) return getattr(importlib.import_module(module), obj) # SQLAlchemy custom DDL elements class CreateMaterializedView(DDLElement): """A DDL element which creates a materialized view on PostgreSQL.""" def __init__(self, name: str, selectable: Selectable): self.name = name self.selectable = selectable @compiler.compiles(CreateMaterializedView, "postgresql") def compile_create_materialized_view( element: CreateMaterializedView, compiler: DDLCompiler, **kw ): return "CREATE MATERIALIZED VIEW {:s} AS {:s}".format( element.name, compiler.sql_compiler.process(element.selectable, literal_binds=True), ) class DropMaterializedView(DDLElement): """A DDL element which drops a materialized view on PostgreSQL.""" def __init__(self, name: str): self.name = name @compiler.compiles(DropMaterializedView, "postgresql") def compile_drop_materialized_view( element: DropMaterializedView, compiler: DDLCompiler, **kw ): return "DROP MATERIALIZED VIEW {:s}".format(element.name) def MaterializedView(name: str, metadata: MetaData, import_path: str, schema=None): """Creates a PostgreSQL MATERIALIZED VIEW and returns it. Arguments: name -- The name of the view. import_path -- You need to pass the absolute path of the query expression to this. The query expression must be top-level. This is due to how Alembic renders migrations (migrations are just text templates, so all semantic information is lost). """ t = table(name) t.schema = schema CreateMaterializedView(name, _import_selectable(import_path)).execute_at( "after-create", metadata ) DropMaterializedView(name).execute_at("before-drop", metadata) # Add metadata for alembic metadata.info.setdefault("materialized_views", {})[(name, schema)] = import_path return t # Alembic operations QueryOrPath = Union[Selectable, str] @Operations.register_operation("create_materialized_view") class CreateMaterializedViewOp(MigrateOperation): """Implements a PostgreSQL CREATE MATERIALIZED VIEW operation.""" def __init__(self, view_name: str, qry: QueryOrPath, schema=None): if isinstance(qry, str): self.qry_path = qry qry = _import_selectable(qry) else: self.qry_path = None self.view_name = view_name self.qry = qry self.schema = schema @classmethod def create_materialized_view( cls, operations: Operations, view_name: str, selectable: Selectable, **kw ): return operations.invoke(cls(view_name, selectable, **kw)) def reverse(self): return DropMaterializedViewOp( self.view_name, schema=self.schema, _orig_path=self.qry_path ) @Operations.implementation_for(CreateMaterializedViewOp) def create_materialized_view(ops: Operations, op: CreateMaterializedViewOp): if op.schema is not None: name = "{}.{}".format(op.schema, op.view_name) else: name = op.view_name ops.execute(CreateMaterializedView(name, op.qry)) @Operations.register_operation("drop_materialized_view") class DropMaterializedViewOp(MigrateOperation): """Implements a PostgreSQL DROP MATERIALIZED VIEW operation.""" def __init__(self, view_name: str, schema=None, _orig_path: str = None): self.view_name = view_name self.schema = schema self._orig_path = _orig_path @classmethod def drop_materialized_view(cls, operations: Operations, view_name: str, **kw): return operations.invoke(cls(view_name, **kw)) def reverse(self): if self._orig_path is None: raise ValueError("cannot autogenerate; path to query doesn't exist") return CreateMaterializedViewOp( self.view_name, self._orig_path, schema=self.schema ) @Operations.implementation_for(DropMaterializedViewOp) def drop_materialized_view(ops: Operations, op: DropMaterializedViewOp): if op.schema is not None: name = "{}.{}".format(op.schema, op.view_name) else: name = op.view_name ops.execute(DropMaterializedView(name)) # Comparator implementation @comparators.dispatch_for("schema") def compare_materialized_views(ctx: AutogenContext, upgrade_ops, schemas): all_conn_matviews = set() # Collect all materialized views from all the tables for sch in schemas: all_conn_matviews.update( [ (row[0], sch) for row in ctx.connection.execute( # This query fetches the names of all materialized views # in this schema. "SELECT relname FROM pg_class c join " "pg_namespace n on n.oid=c.relnamespace where " "relkind='m' and n.nspname=%(nspname)s", nspname=(ctx.dialect.default_schema_name if sch is None else sch), ) ] ) # Get the materialized views that are registered locally meta_matviews: dict = ctx.metadata.info.setdefault("materialized_views", {}) matviews_set = set(meta_matviews.keys()) # Create new materialized views for those in the metadata but not in the # connection for key in matviews_set.difference(all_conn_matviews): name, schema = key upgrade_ops.ops.append( CreateMaterializedViewOp(name, meta_matviews[key], schema=schema) ) # Drop materialized views for those in the connection but not in the metadata # anymore # TODO: Implement fetching of the original selectable, so that drops are reversible for key in all_conn_matviews.difference(matviews_set): name, schema = key upgrade_ops.ops.append(DropMaterializedViewOp(name, schema=schema)) # Renderer implementations @renderers.dispatch_for(CreateMaterializedViewOp) def render_create_materialized_view( ctx: AutogenContext, op: CreateMaterializedViewOp ) -> str: module, obj_path = op.qry_path.split(".", 1) return "op.create_materialized_view({}, __import__({}).{}, **{})".format( repr(op.view_name), repr(module), obj_path, {"schema": op.schema} ) @renderers.dispatch_for(DropMaterializedViewOp) def render_drop_materialized_view( ctx: AutogenContext, op: DropMaterializedViewOp ) -> str: return "op.drop_materialized_view({}, **{})".format( repr(op.view_name), {"schema": op.schema} )
