grocery-manager/app/txn_view.py

#
# Copyright (c) Daniel Sheffield 2021 - 2022
#
# All rights reserved
#
# THIS SOFTWARE IS PROVIDED AS IS WITHOUT WARRANTY
from psycopg.sql import (
    Identifier,
    SQL,
    Literal,
    Placeholder,
    Composed,
)
from collections import (
    OrderedDict,
)

def get_table_statement(alias):
    tables = {
        'product': 'products',
        'category': 'categories',
        'group': 'groups',
        'unit': 'units',
        'store': 'stores',
    }
    return SQL("SELECT {column} AS {alias} FROM {table}").format(
        column=Identifier(tables[alias], 'name'),
        table=Identifier(tables[alias]),
        alias=Identifier(alias),
    )

def get_transactions_statement():
    return SQL("SELECT * FROM transaction_view")

select = OrderedDict([
    ('id', Identifier('transactions','id')),
    ('ts', Identifier('transactions', 'ts')),
    ('store', Identifier('stores', 'name')),
    ('code', Identifier('stores', 'code')),
    ('description', Identifier('transactions','description')),
    ('volume', Identifier('quantity')),
    ('unit', Identifier('units', 'name')),
    ('price', Identifier('price')),
    ('$/unit', SQL("""TRUNC(price/quantity,4)""")),
    ('total', SQL("""sum(transactions.price) OVER (PARTITION BY transactions.ts::date ORDER BY transactions.id ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)""")),
    ('group', Identifier('groups','name')),
    ('category', Identifier('categories','name')),
    ('product', Identifier('products','name')),
    ('organic', Identifier('organic')),
])

def get_where(date, store, full_name=False, exact_time=False):
    where = [ ]
    if store is not None:
        where.append(SQL(' ').join([
            Identifier('stores', 'name' if full_name else 'code'),
            SQL('='),
            Literal(store)
        ]))
    where.append(
        SQL("{ts} at time zone 'utc' BETWEEN {date}::date AND {date}::date + {interval}::interval").format(
            ts=Identifier('ts'),
            date=Literal(str(date)),
            interval=Literal('23 hours 59 minutes 59 seconds'),
        ) if not exact_time else SQL("{ts} at time zone 'utc' = {date}").format(
            ts=Identifier('ts'),
            date=Literal(str(date)),
        )
    )
    return SQL('').join([
        SQL("WHERE"
            "\n      "),
        SQL("\n  AND ").join(where),
    ])

def get_session_transactions_statement(date, store, full_name=False, exact_time=False):
    statement = SQL('\n').join([
        SQL('').join([
            SQL("SELECT"
                "\n  "),
            SQL(','
                "\n  ").join([
                SQL(' ').join([v, SQL('AS'), Identifier(f'{k}')]) for k, v in select.items()
            ])
        ]),
        SQL('').join([
            SQL("FROM"
                "\n       "),
            SQL("\n  JOIN ").join([
                SQL("transactions"),
                SQL("{table} ON {table}.{key} = {index}").format(
                    table=Identifier('units'),
                    key=Identifier('id'),
                    index=Identifier('unit_id'),
                ),
                SQL("{table} ON {table}.{key} = {index}").format(
                    table=Identifier('stores'),
                    key=Identifier('id'),
                    index=Identifier('store_id')
                ),
                SQL("{table} ON {table}.{key} = {index}").format(
                    table=Identifier('products'),
                    key=Identifier('id'),
                    index=Identifier('product_id')
                ),
                SQL("{table} ON {table}.{key} = {index}").format(
                    table=Identifier('categories'),
                    key=Identifier('id'),
                    index=Identifier('category_id')
                ),
                SQL("{table} ON {table}.{key} = {index}").format(
                    table=Identifier('groups'),
                    key=Identifier('id'),
                    index=Identifier('group_id')
                ),
            ]),
        ]),
        get_where(date if exact_time else date.replace(
            hour=0, minute=0, second=0, microsecond=0
        ), store, full_name=full_name, exact_time=exact_time),
        SQL('ORDER BY {_id} DESC').format(_id=Identifier('transactions','id')),
    ])
    return statement