notifications-api/app/dao/templates_dao.py

import uuid

from sqlalchemy import asc, desc, select

from app import db
from app.dao.dao_utils import VersionOptions, autocommit, version_class
from app.models import Template, TemplateHistory, TemplateRedacted
from app.utils import utc_now


@autocommit
@version_class(VersionOptions(Template, history_class=TemplateHistory))
def dao_create_template(template):
    template.id = (
        uuid.uuid4()
    )  # must be set now so version history model can use same id
    template.archived = False

    redacted_dict = {
        "template": template,
        "redact_personalisation": False,
    }
    if template.created_by:
        redacted_dict.update({"updated_by": template.created_by})
    else:
        redacted_dict.update({"updated_by_id": template.created_by_id})

    template.template_redacted = TemplateRedacted(**redacted_dict)

    db.session.add(template)


@autocommit
@version_class(VersionOptions(Template, history_class=TemplateHistory))
def dao_update_template(template):
    db.session.add(template)


@autocommit
def dao_redact_template(template, user_id):
    template.template_redacted.redact_personalisation = True
    template.template_redacted.updated_at = utc_now()
    template.template_redacted.updated_by_id = user_id
    db.session.add(template.template_redacted)


def dao_get_template_by_id_and_service_id(template_id, service_id, version=None):
    if version is not None:
        stmt = select(TemplateHistory).where(
            TemplateHistory.id == template_id,
            TemplateHistory.hidden == False,  # noqa
            TemplateHistory.service_id == service_id,
            TemplateHistory.version == version,
        )
        return db.session.execute(stmt).scalars().one()
    stmt = select(Template).where(
        Template.id == template_id,
        Template.hidden == False,  # noqa
        Template.service_id == service_id,
    )
    return db.session.execute(stmt).scalars().one()


def dao_get_template_by_id(template_id, version=None):
    if version is not None:
        stmt = select(TemplateHistory).where(
            TemplateHistory.id == template_id, TemplateHistory.version == version
        )
        return db.session.execute(stmt).scalars().one()
    stmt = select(Template).where(Template.id == template_id)
    return db.session.execute(stmt).scalars().one()


def dao_get_all_templates_for_service(service_id, template_type=None):
    if template_type is not None:
        stmt = (
            select(Template)
            .where(
                Template.service_id == service_id,
                Template.template_type == template_type,
                Template.hidden == False,  # noqa
                Template.archived == False,  # noqa
            )
            .order_by(
                asc(Template.name),
                asc(Template.template_type),
            )
        )
        return db.session.execute(stmt).scalars().all()
    stmt = (
        select(Template)
        .where(
            Template.service_id == service_id,
            Template.hidden == False,  # noqa
            Template.archived == False,  # noqa
        )
        .order_by(
            asc(Template.name),
            asc(Template.template_type),
        )
    )
    return db.session.execute(stmt).scalars().all()


def dao_get_template_versions(service_id, template_id):
    stmt = (
        select(TemplateHistory)
        .where(
            TemplateHistory.service_id == service_id,
            TemplateHistory.id == template_id,
            TemplateHistory.hidden == False,  # noqa
        )
        .order_by(desc(TemplateHistory.version))
    )
    return db.session.execute(stmt).scalars().all()