drinks-manager/app/db_queries.py

#from datetime import datetime

from django.conf import settings
from django.db import connection
from django.utils.translation import gettext
from calendar import day_name


COMBINE_ALPHABET = "abcdefghijklmnopqrstuvwxyz"


def _db_select(sql_select:str):
    result = None
    with connection.cursor() as cursor:
        cursor.execute(sql_select)
        result = cursor.fetchall()
    return result

def _combine_results(results:list) -> dict:
    '''
    e.g.
    input: [
        [("x", 12), ("y", 13)],
        [("y", 10), ("z", 42)]
    ]
    output: {
        "x": {"a": 12},
        "y": {"a": 13, "b": 10},
        "z": {"b": 42}
    }
    '''
    result = {}
    for i, d in enumerate(results):
        a = COMBINE_ALPHABET[i]
        for r in d:
            r_0 = r[0]
            if r_0 not in result:
                result[r_0] = {}
            result[r_0][a] = r[1]
    return result


def select_history(user, language_code="en") -> list:
    # select order history and deposits
    user_id = user.pk
    result = _db_select(f"""
    select
        price_sum as "sum",
        concat(
            product_name,
            ' (',
            content_litres::real, -- converting to real removes trailing zeros
            'l) x ', amount
        ) as "text",
        datetime
    from app_order
    where user_id = {user_id}

    union

    select
        transaction_sum as "sum",
        '{gettext("Deposit")}' as "text",
        datetime
    from app_userdeposits_view
    where user_id = {user_id}

    union

    select
        transaction_sum as "sum",
        comment as "text",
        datetime
    from app_registertransaction
    where user_id = {user_id} and is_transfer = true

    order by datetime desc
    fetch first 30 rows only;
    """)
    result = [list(row) for row in result]
    return result

def select_orders_per_month(user) -> dict:
    # number of orders per month (last 12 months)
    result_user = _db_select(f"""
    select
        to_char(date_trunc('month', datetime), 'YYYY-MM') as "month",
        sum(amount) as "count"
    from app_order
        where user_id = {user.pk}
        and date_trunc('month', datetime) > date_trunc('month', now() - '12 months'::interval)
    group by "month"
    order by "month" desc;
    """)
    result_all = _db_select(f"""
    select
        to_char(date_trunc('month', datetime), 'YYYY-MM') as "month",
        sum(amount) as "count"
    from app_order
        where date_trunc('month', datetime) > date_trunc('month', now() - '12 months'::interval)
    group by "month"
    order by "month" desc;
    """)
    return _combine_results([result_all, result_user])

def select_orders_per_weekday(user) -> list:
    # number of orders per weekday (all time)
    result = _db_select(f"""
    with q_all as (
        select
            extract(isodow from datetime) as "d",
            sum(amount) as "c"
        from app_order
        group by d
    ), q_user as (
        select
            extract(isodow from datetime) as "d",
            sum(amount) as "c"
        from app_order
            where user_id = {user.pk}
        group by d
    )
    select q_all.d as "day", q_all.c, q_user.c from q_all full join q_user on q_all.d = q_user.d
    group by day, q_all.c, q_user.c
    order by day asc;
    """)
    for i in range(len(result)):
        day_, all_, user_ = result[i]
        result[i] = (day_name[int(day_)-1], all_, user_)
    return result

def select_orders_per_drink(user) -> dict:
    # number of orders per drink (all time)
    result_user = _db_select(f"""
    select
        d.product_name as "label",
        sum(o.amount) as "data"
    from app_drink d
        join app_order o on (d.id = o.drink_id)
    where o.user_id = {user.pk}
    group by d.product_name
    order by "data" desc;
    """)
    result_all = _db_select(f"""
    select
        d.product_name as "label",
        sum(o.amount) as "data"
    from app_drink d
        join app_order o on (d.id = o.drink_id)
    group by d.product_name
    order by "data" desc;
    """)
    return _combine_results([result_all, result_user])

def select_order_sum_per_user_all_users() -> list:
    # sum of all orders per user, for all users
    result = _db_select(f"""
    select
        app_user.username as user,
        sum(app_order.price_sum) as sum
    from app_user
        left outer join app_order on (app_user.id = app_order.user_id)
    group by app_user.id
    order by app_user asc;
    """)
    return result

def select_deposit_sum_per_user_all_users() -> list:
    # sum of all orders per user, for all users
    result = _db_select(f"""
    select
        app_user.username as user,
        sum(rt.transaction_sum) as sum
    from app_user
        left outer join app_registertransaction rt on (app_user.id = rt.user_id)
    where rt.is_user_deposit is true or rt.is_user_deposit is null
    group by app_user.id
    order by app_user asc;
    """)
    return result