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: logger.info("Adding counter %s", counter_type) with connection().session as session: try: query = text('INSERT INTO counters (name, type, color) VALUES (:title, :type, :color)') session.execute(query, {'title': title, 'type': counter_type, 'color': counter_color}) session.commit() except Exception as e: logger.error(e) session.rollback() def get_counters(): try: return connection().query('SELECT id, name, type, color FROM counters', ttl=0) except Exception as e: logger.error(e) return st.dataframe() def increment_counter(counter_id:int) -> None: logger.info("Incrementing counter %s", counter_id) with connection().session as session: try: query = text('INSERT INTO entries (counter_id) VALUES (:id)') session.execute(query, {'id': counter_id}) session.commit() except Exception as e: logger.error(e) session.rollback() def remove_counter(counter_id:int) -> None: logger.info("Removing counter %s", counter_id) with connection().session as session: try: query = text('DELETE FROM counters WHERE id = :id') session.execute(query, {'id': counter_id}) session.commit() except Exception as e: logger.error(e) session.rollback() def get_counter(counter_id:int): try: return connection().query('SELECT * FROM counters WHERE id = :id', params={'id': counter_id}, ttl=0).iloc[0] except Exception as e: logger.error(e) return None def get_colors(palette_id:int): try: return connection().query('''SELECT color1,color2,color3,color4,color5 FROM color_palettes WHERE id = :id''', params={'id': palette_id}) except Exception as e: logger.error(e) return None