Source code for wutta_continuum.util

# -*- coding: utf-8; -*-
################################################################################
#
#  Wutta-Continuum -- SQLAlchemy Versioning for Wutta Framework
#  Copyright © 2024-2025 Lance Edgar
#
#  This file is part of Wutta Framework.
#
#  Wutta Framework is free software: you can redistribute it and/or modify it
#  under the terms of the GNU General Public License as published by the Free
#  Software Foundation, either version 3 of the License, or (at your option) any
#  later version.
#
#  Wutta Framework is distributed in the hope that it will be useful, but
#  WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
#  FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
#  more details.
#
#  You should have received a copy of the GNU General Public License along with
#  Wutta Framework.  If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
"""
SQLAlchemy-Continuum utilities
"""

import sqlalchemy as sa
from sqlalchemy import orm
import sqlalchemy_continuum as continuum


OPERATION_TYPES = {
    continuum.Operation.INSERT: "INSERT",
    continuum.Operation.UPDATE: "UPDATE",
    continuum.Operation.DELETE: "DELETE",
}




def render_operation_type(operation_type):
    """
    Render a SQLAlchemy-Continuum ``operation_type`` from a version
    record, for display to user.

    :param operation_type: Value of same name from a version record.
       Must be one of:

       * :attr:`sqlalchemy_continuum:sqlalchemy_continuum.operation.Operation.INSERT`
       * :attr:`sqlalchemy_continuum:sqlalchemy_continuum.operation.Operation.UPDATE`
       * :attr:`sqlalchemy_continuum:sqlalchemy_continuum.operation.Operation.DELETE`

    :returns: Display name for the operation type, as string.
    """
    return OPERATION_TYPES[operation_type]





def model_transaction_query(instance, session=None, model_class=None, joins=None):
    """
    Make a query capable of finding all SQLAlchemy-Continuum
    ``transaction`` records associated with the given model instance.

    :param instance: Instance of a versioned :term:`data model`.

    :param session: Optional :term:`db session` to use for the query.
       If not specified, will be obtained from the ``instance``.

    :param model_class: Optional :term:`data model` class to query.
       If not specified, will be obtained from the ``instance``.

    :param joins: Optional sequence of "join info tuples" - see
       further explanation below.

    :returns: SQLAlchemy query object.  Note that it will *not* have an
       ``ORDER BY`` clause yet.

    The default logic looks for any "version" records for the given
    instance, and returns the associated transactions.  But sometimes
    you need to look for more than one type of version record:

    If e.g. a core table provides common fields but a custom table
    adds extension fields for a record, you will want to find
    transactions involving *either* version table when showing a
    record's total history.

    This is accomplished here via the ``joins`` param.  If specified,
    each item in the ``joins`` sequence must be a 3-tuple::

       (related_class, related_attr, instance_attr)

    The meaning of those is as follows; assuming ``User`` is the main
    instance model class:

    * ``related_class`` - model class which is "related" to the
      instance model class, e.g. ``UserExtension``
    * ``related_attr`` - attribute name on the related class which
      serves as foreign key to the instance, e.g. ``"user_uuid"``
    * ``instance_attr`` - attribute name on the main instance class
      which serves as primary key, e.g. ``"uuid"``
    """
    if not session:
        session = orm.object_session(instance)
    if not model_class:
        model_class = type(instance)

    txncls = continuum.transaction_class(model_class)
    vercls = continuum.version_class(model_class)

    # basic query is for the *transaction* table
    query = session.query(txncls)

    # we'll do inner *or* outer join on main version table below
    join_args = (
        vercls,
        sa.and_(
            vercls.uuid == instance.uuid,
            vercls.transaction_id == txncls.id,
        ),
    )

    if joins:

        # we must *outer* join on main version table, since we will
        # also be joining on other version tables
        query = query.outerjoin(*join_args)

        # we'll collect "filter conditions" for use below...
        conditions = [vercls.uuid != None]  # pylint: disable=singleton-comparison

        # add join/filter for each requested by caller
        for child_class, foreign_attr, primary_attr in joins:
            child_vercls = continuum.version_class(child_class)
            foreign_attr = getattr(child_vercls, foreign_attr)
            query = query.outerjoin(
                child_vercls,
                sa.and_(
                    child_vercls.transaction_id == txncls.id,
                    foreign_attr == getattr(instance, primary_attr),
                ),
            )

            # and add the filter condition for use below...
            conditions.append(
                foreign_attr != None  # pylint: disable=singleton-comparison
            )

        # at this point we have *outer* joined on *all* version tables
        # involved, but that means basically "all transactions" will
        # match!  so we add explicit filter to make sure at least one
        # of them is related for a transaction to match
        query = query.filter(sa.or_(*conditions))

    else:

        # no joins were specified, so we can just do *inner* join on
        # the main version table and call it good
        query = query.join(*join_args)

    return query