grocery-manager/app/rest/query_to_xml.py

#
# Copyright (c) Daniel Sheffield 2023
# All rights reserved
#
# THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
from bottle import request, template, FormsDict
from pandas import DataFrame
from psycopg import Cursor
from psycopg.sql import SQL, Literal, Identifier
from typing import List

from ..data.filter import get_filter
from ..data.util import get_where_include_exclude
from ..data.QueryManager import get_data
from .form import get_form
from . import BOOLEAN, PARAMS


def get_product_rollup_statement(filters, orderby: List[str]) -> SQL:
    _map = { k: k[0] for k in ('product', 'category', 'group') }
    where = [ get_where_include_exclude(
        _map[k], "name", list(include), list(exclude)
    ) for k, (include, exclude) in filters.items() ]
    return SQL("""
SELECT
  count(DISTINCT p.id) AS "Products",
  count(DISTINCT c.id) AS "Categories",
  count(DISTINCT g.id) AS "Groups",
  p.name AS "product",
  c.name AS "category",
  g.name AS "group"
FROM products p
JOIN categories c ON p.category_id = c.id
JOIN groups g ON c.group_id = g.id
WHERE {where}
GROUP BY ROLLUP (g.name, c.name, p.name)
ORDER BY {orderby}
""").format(
        where=SQL("\nAND").join(where),
        orderby=SQL(", ").join(map(Identifier,orderby)),
    )


def get_inner_query(query: FormsDict, orderby: List[str]) -> SQL:
    filters = get_filter(query, allow=('group', 'category', 'product'))
    inner = get_product_rollup_statement(filters, orderby)
    return inner


def render_form(cur: Cursor, inner: str, query: FormsDict):
    _filter = get_filter(query, allow=PARAMS)
    data = DataFrame(get_data(cur, inner)).dropna()
    action = request.path.split('/')[-1]
    organic = BOOLEAN.get(query.organic, None)
    return get_form(action, 'post', _filter, organic, data)


def get_xml(cur: Cursor, sql: str):
    return cur.execute(SQL(
        "SELECT query_to_xml_and_xmlschema({q}, false, false, ''::text)"
    ).format(q=Literal(sql))).fetchone()[0]


def get_products(cur: Cursor, query: FormsDict):
    inner = get_inner_query(query, ['product', 'category', 'group'])
    form = render_form(cur, inner, query)
    sql = SQL("""
SELECT
    --"Transactions",
    COALESCE("product", "Products"||'') "Product",
    COALESCE("category", "Categories"||'') "Category",
    COALESCE("group", "Groups"||'') "Group"
FROM ({inner}) q
WHERE q.product IS NOT NULL OR q.group IS NULL
""").format(inner=inner).as_string(cur)
    xml = get_xml(cur, sql)
    return template("query-to-xml", title="Products", xml=xml, form=form)


def get_categories(cur: Cursor, query: FormsDict):
    inner = get_inner_query(query, ['category', 'group'])
    form = render_form(cur, inner, query)
    sql = SQL("""
SELECT
    "Products",
    COALESCE("category", "Categories"||'') "Category",
    COALESCE("group", "Groups"||'') "Group"
FROM ({inner}) q
WHERE q.product IS NULL AND (q.category IS NOT NULL OR q.group IS NULL)
""").format(inner=inner).as_string(cur)
    xml = get_xml(cur, sql)
    return template("query-to-xml", title="Categories", xml=xml, form=form)


def get_groups(cur: Cursor, query: FormsDict):
    inner = get_inner_query(query, ['group',])
    form = render_form(cur, inner, query)
    sql = SQL("""
SELECT
    "Products",
    "Categories",
    COALESCE("group", "Groups"||'') "Group"
FROM ({inner}) q
WHERE q.category IS NULL
""").format(inner=inner).as_string(cur)
    xml = get_xml(cur, sql)
    return template("query-to-xml", title="Groups", xml=xml, form=form)

def get_tags_statement(filters) -> SQL:
    _map = {
        k: k[0] for k in ('product', 'category', 'group')
    }
    _map.update({ 'tag': 'tg' })
    where = [ get_where_include_exclude(
        _map[k], "name", list(include), list(exclude)
    ) for k, (include, exclude) in filters.items() ]
    return SQL("""
SELECT * FROM (SELECT count(DISTINCT txn.id) AS "Uses", tg.name AS "Name"
FROM tags tg
JOIN tags_map tm ON tg.id = tm.tag_id
JOIN transactions txn ON txn.id = tm.transaction_id
WHERE {where}
GROUP BY tg.name
ORDER BY 1 DESC, 2) q
UNION ALL
SELECT count(DISTINCT txn.id) AS "Uses", count(DISTINCT tg.name)||'' AS "Name"
FROM tags tg
JOIN tags_map tm ON tg.id = tm.tag_id
JOIN transactions txn ON txn.id = tm.transaction_id
WHERE {where}
""").format(where=SQL("\nAND").join(where))

def get_inner_tags_query(query: FormsDict) -> SQL:
    filters = get_filter(query, allow=('tag',))
    inner = get_tags_statement(filters)
    return inner

def get_tags(cur: Cursor, query: FormsDict):
    inner = get_inner_tags_query(query)
    form = render_form(cur, inner, query)
    sql = inner.as_string(cur)
    xml = get_xml(cur, sql)
    return template("query-to-xml", title="Tags", xml=xml, form=form)