diff --git a/tlapbot/redeems.py b/tlapbot/redeems.py index 9e3a11d..a89d790 100644 --- a/tlapbot/redeems.py +++ b/tlapbot/redeems.py @@ -1,8 +1,10 @@ from flask import current_app -from sqlite3 import Error +from sqlite3 import Error, Connection +from typing import Tuple, Any from tlapbot.owncast_helpers import use_points +from tlapbot.types import Redeem, Redeems -def counter_exists(db, counter_name): +def counter_exists(db: Connection, counter_name: str) -> bool: try: cursor = db.execute( "SELECT count FROM counters WHERE name = ?", @@ -20,7 +22,7 @@ def counter_exists(db, counter_name): current_app.logger.error(f"For counter: {counter_name}") -def add_to_counter(db, counter_name): +def add_to_counter(db: Connection, counter_name: str) -> bool: if counter_exists(db, counter_name): try: db.execute( @@ -34,8 +36,8 @@ def add_to_counter(db, counter_name): current_app.logger.error(f"To counter: {counter_name}") return False - -def add_to_redeem_queue(db, user_id, redeem_name, note=None): +# TODO: test if the new default works +def add_to_redeem_queue(db: Connection, user_id: str, redeem_name: str, note: str="") -> bool: try: db.execute( "INSERT INTO redeem_queue(redeem, redeemer_id, note) VALUES(?, ?, ?)", @@ -49,7 +51,7 @@ def add_to_redeem_queue(db, user_id, redeem_name, note=None): return False -def add_to_milestone(db, user_id, redeem_name, points_donated): +def add_to_milestone(db: Connection, user_id: str, redeem_name: str, points_donated: int) -> bool: try: cursor = db.execute( "SELECT progress, goal FROM milestones WHERE name = ?", @@ -77,8 +79,8 @@ def add_to_milestone(db, user_id, redeem_name, points_donated): current_app.logger.error(f"Error occurred updating milestone: {e.args[0]}") return False - -def milestone_complete(db, redeem_name): +# TODO: milestone is complete when progress equals goal? +def milestone_complete(db: Connection, redeem_name: str) -> bool: try: cursor = db.execute( "SELECT complete FROM milestones WHERE name = ?", @@ -95,7 +97,7 @@ def milestone_complete(db, redeem_name): current_app.logger.error(f"Error occurred checking if milestone is complete: {e.args[0]}") -def check_apply_milestone_completion(db, redeem_name): +def check_apply_milestone_completion(db: Connection, redeem_name: str) -> bool: try: cursor = db.execute( "SELECT progress, goal FROM milestones WHERE name = ?", @@ -121,7 +123,7 @@ def check_apply_milestone_completion(db, redeem_name): return False -def all_milestones(db): +def all_milestones(db: Connection) -> list[Tuple[str, int, int]]: try: cursor = db.execute( """SELECT name, progress, goal FROM milestones""" @@ -131,26 +133,7 @@ def all_milestones(db): current_app.logger.error(f"Error occurred selecting all milestones: {e.args[0]}") -def all_counters(db): - try: - cursor = db.execute( - """SELECT counters.name, counters.count FROM counters""" - ) - return cursor.fetchall() - except Error as e: - current_app.logger.error(f"Error occurred selecting all counters: {e.args[0]}") - - -def all_active_counters(db): - counters = all_counters(db) - all_active_counters = [] - for name, count in counters: - if is_redeem_active(name): - all_active_counters.append((name, count)) - return all_active_counters - - -def all_active_milestones(db): +def all_active_milestones(db: Connection) -> list[Tuple[str, int, int]]: milestones = all_milestones(db) all_active_milestones = [] for name, progress, goal in milestones: @@ -159,7 +142,26 @@ def all_active_milestones(db): return all_active_milestones -def all_active_redeems(): +def all_counters(db: Connection) -> list[Tuple[str, int]]: + try: + cursor = db.execute( + """SELECT name, count FROM counters""" + ) + return cursor.fetchall() + except Error as e: + current_app.logger.error(f"Error occurred selecting all counters: {e.args[0]}") + + +def all_active_counters(db: Connection) -> list[Tuple[str, int]]: + counters = all_counters(db) + all_active_counters = [] + for name, count in counters: + if is_redeem_active(name): + all_active_counters.append((name, count)) + return all_active_counters + + +def all_active_redeems() -> dict[str, dict[str, Any]]: redeems = current_app.config['REDEEMS'] all_active_redeems = {} for redeem_name, redeem_dict in redeems.items(): @@ -173,7 +175,7 @@ def all_active_redeems(): return all_active_redeems -def pretty_redeem_queue(db): +def pretty_redeem_queue(db: Connection) -> list[Tuple[str, str, str, str]]: try: cursor = db.execute( """SELECT redeem_queue.created, redeem_queue.redeem, redeem_queue.note, points.name @@ -186,17 +188,17 @@ def pretty_redeem_queue(db): current_app.logger.error(f"Error occurred selecting pretty redeem queue: {e.args[0]}") -def whole_redeem_queue(db): +def whole_redeem_queue(db: Connection) -> list[Any]: try: cursor = db.execute( - "SELECT * from redeem_queue" + "SELECT * from redeem_queue" #TODO: specify columns to fetch ) return cursor.fetchall() except Error as e: current_app.logger.error(f"Error occurred selecting redeem queue: {e.args[0]}") -def is_redeem_active(redeem_name): +def is_redeem_active(redeem_name: str) -> bool | None: """Checks if redeem is active. Pulls the redeem by name from config.""" active_categories = current_app.config['ACTIVE_CATEGORIES'] redeem_dict = current_app.config['REDEEMS'].get(redeem_name, None) @@ -212,9 +214,9 @@ def is_redeem_active(redeem_name): -def is_redeem_from_config_active(redeem, active_categories): +def is_redeem_from_config_active(redeem: Redeem, active_categories: list[str]) -> bool: """Checks if redeem is active. `redeem` is a whole key:value pair from redeems config.""" - if "category" in redeem[1] and redeem[1]["category"]: + if isinstance(redeem[1].get("category"), list): for category in redeem[1]["category"]: if category in active_categories: return True @@ -222,6 +224,6 @@ def is_redeem_from_config_active(redeem, active_categories): return True -def remove_inactive_redeems(redeems, active_categories): +def remove_inactive_redeems(redeems: Redeems, active_categories: list[str]) -> Redeems]: return dict(filter(lambda redeem: is_redeem_from_config_active(redeem, active_categories), redeems.items())) \ No newline at end of file diff --git a/tlapbot/types.py b/tlapbot/types.py new file mode 100644 index 0000000..f79db4b --- /dev/null +++ b/tlapbot/types.py @@ -0,0 +1,4 @@ +from typing import Tuple, Any, TypeAlias + +Redeem: TypeAlias = Tuple[str, dict[str, Any]] +Redeems: TypeAlias = Tuple[str, Redeem] \ No newline at end of file