Source code for codegen_database.views.view

"""Generic view factories."""

from __future__ import annotations

from typing import TYPE_CHECKING

from sqlalchemy import Column, MetaData, Table
from sqlalchemy import types as sa_types
from sqlalchemy_declarative_extensions import View, register_view

from codegen_database.utils.query import compile_query

if TYPE_CHECKING:
    from sqlalchemy import Select
    from sqlalchemy.sql.type_api import TypeEngine


def _column_type(col: object) -> TypeEngine:
    """Extract the SA type from a selected column."""
    return getattr(col, "type", sa_types.NullType())


def _construct_view_proxy_table(
    name: str,
    schema: str,
    query: Select,
) -> Table:
    """Build a joinable ``Table`` proxy from a query.

    The table lives on a private ``MetaData`` so it does not
    interfere with Alembic autogeneration.
    """
    cols = [
        Column(c.key, _column_type(c))
        for c in query.selected_columns
        if c.key is not None
    ]
    return Table(name, MetaData(), *cols, schema=schema)




class CodegenDatabasePlainView:
    """Create a plain PostgreSQL view from a SQLAlchemy select.

    Prefer the declarative
    :class:`~codegen_database.declarative.CodegenDatabaseView`
    base class for new views; this class exists for cases where an
    imperative registration with a ``self.table`` handle is needed.

    After construction, ``self.table`` is a joinable
    SQLAlchemy ``Table`` whose columns mirror the query.

    Args:
        name: View name.
        schema: PostgreSQL schema for the view.
        metadata: SQLAlchemy ``MetaData`` to register on.
        query: A SQLAlchemy ``Select`` defining the view body.

    """

    def __init__(
        self,
        name: str,
        schema: str,
        metadata: MetaData,
        query: Select,
    ) -> None:
        """Create and register the view."""
        definition = compile_query(query)
        self.view = View(name, definition, schema=schema)
        register_view(metadata, self.view)
        self.name = name
        self.schema = schema
        self.metadata = metadata
        self.table = _construct_view_proxy_table(name, schema, query)





class CodegenDatabaseMaterializedView:
    """Create a materialized view with an auto-generated refresh.

    After construction, ``self.table`` is a joinable
    SQLAlchemy ``Table`` whose columns mirror the query.

    Args:
        name: View name.
        schema: PostgreSQL schema for the view.
        metadata: SQLAlchemy ``MetaData`` to register on.
        query: A SQLAlchemy ``Select`` defining the view body.

    """

    def __init__(
        self,
        name: str,
        schema: str,
        metadata: MetaData,
        query: Select,
    ) -> None:
        """Create and register the materialized view."""
        definition = compile_query(query)
        self.view = View(
            name,
            definition,
            schema=schema,
            materialized=True,
        )
        register_view(metadata, self.view)
        self.name = name
        self.schema = schema
        self.metadata = metadata
        self.table = _construct_view_proxy_table(name, schema, query)