daily-counter/app/queries/crud.py

import logging
import streamlit as st
from sqlalchemy.sql import text
from queries.connection import connection
from enums import CounterType

logger = logging.getLogger(__name__)

def create_counter(title:str, counter_type:CounterType, counter_color) -> None:
    user_id = int(st.session_state.user_id)
    logger.info("Adding counter %s for user %d", counter_type, user_id)
    with connection().session as session:
        try:
            query = text('INSERT INTO counters (user_id, name, type, color) VALUES (:user, :title, :type, :color)')
            session.execute(query, {
                'user': user_id,
                'title': title,
                'type': counter_type,
                'color': counter_color
            })
        except Exception as e:
            logger.error(e)
            session.rollback()

def get_counters():
    user_id = int(st.session_state.user_id)
    try:
        return connection().query("""
              SELECT id, name, type, color 
              FROM counters 
              WHERE user_id = :user
              """, params={'user': user_id })
    except Exception as e:
        logger.error(e)
        return st.dataframe()

def increment_counter(counter_id:int) -> None:
    user_id = int(st.session_state.user_id)
    logger.info("Incrementing counter %d for user %d", counter_id, user_id)
    with connection().session as session:
        try:
            query = text('INSERT INTO entries (counter_id, user_id) VALUES (:id, :user)')
            session.execute(query, {
                'id': counter_id,
                'user': user_id
            })
        except Exception as e:
            logger.error(e)
            session.rollback()

def remove_counter(counter_id:int) -> None:
    user_id = int(st.session_state.user_id)
    logger.info("Removing counter %d from user %d", counter_id, user_id)
    with connection().session as session:
        try:
            query = text('DELETE FROM counters WHERE id = :id AND user_id = :user')
            session.execute(query, {
                'id': counter_id,
                'user': user_id
            })
        except Exception as e:
            logger.error(e)
            session.rollback()

def get_counter(counter_id:int):
    user_id = int(st.session_state.user_id)
    try:
        counters = connection().query("""
                   SELECT * FROM counters 
                   WHERE id = :id AND user_id = :user
                   """, params={ 'id': counter_id, 'user': user_id}
                )
        if counters.empty:
            return None

        return counters.iloc[0]
    except Exception as e:
        logger.error(e)
        return None


def get_color_palettes():
    try:
        return connection().query('SELECT * FROM color_palettes''')
    except Exception as e:
        logger.error(e)
        return None


def get_color_palette() -> int:
    user_id = int(st.session_state.user_id)
    try:
        return int(connection().query('SELECT color_palette_id FROM users WHERE id = :id''', params={'id': user_id})['color_palette_id'][0])
    except Exception as e:
        logger.error(e)
        return None


def set_color_palette(palette_id:int):
    user_id = int(st.session_state.user_id)
    logger.info("Changing palette for user %d to %d", user_id, palette_id)
    with connection().session as session:
        try:
            query = text('UPDATE users SET color_palette_id = :palette WHERE id = :user')
            session.execute(query, {
                'palette': palette_id,
                'user': user_id
            })
        except Exception as e:
            logger.error(e)
            session.rollback()


def get_colors():
    user_id = int(st.session_state.user_id)
    try:
        return connection().query('''
              SELECT color1,color2,color3,color4,color5 
              FROM users u
              LEFT JOIN color_palettes p ON p.id = u.color_palette_id
              WHERE u.id = :id
        ''', params={'id': user_id})
    except Exception as e:
        logger.error(e)
        return None


def set_theme(theme:str):
    user_id = int(st.session_state.user_id)
    logger.info("Changing theme for user %d to %s", user_id, theme)
    with connection().session as session:
        try:
            query = text('UPDATE users SET theme = :theme WHERE id = :user')
            session.execute(query, {
                'theme': theme,
                'user': user_id
            })
        except Exception as e:
            logger.error(e)
            session.rollback()


def get_theme() -> str:
    user_id = int(st.session_state.user_id)
    try:
        return connection().query('SELECT theme FROM users u WHERE u.id = :id', params={'id': user_id})['theme'][0]
    except Exception as e:
        logger.error(e)
        return None