from datetime import datetime, timedelta from typing import List, Tuple import sqlite3 AUTO_CHECKOUT_HOURS = 6 class CheckinDB: def __init__(self, db_name: str) -> None: self.con = sqlite3.connect(db_name) self.con.row_factory = sqlite3.Row self.try_create_tables() def try_create_tables(self) -> bool: try: with self.con: self.con.execute( "CREATE TABLE checkin (ID INTEGER PRIMARY KEY AUTOINCREMENT, NAME VARCHAR UNIQUE, IS_PRESENT BOOL, UPDATED INT)" ) print("Creating new table") except sqlite3.OperationalError: print("Table already exists in database, nothing to create.") return True def update(self, name: str, is_present: bool) -> bool: present_int = 1 if is_present else 0 with self.con: self.con.execute( f"INSERT INTO checkin (NAME, IS_PRESENT, UPDATED) VALUES ('{name}', '{present_int}', strftime('%s','now')) ON CONFLICT(name) DO UPDATE SET is_present=excluded.is_present" ) return True return False def get_full_entries(self) -> List[Tuple]: with self.con: return [ (x["name"], x["is_present"], x["updated"]) for x in self.con.execute( "SELECT name, is_present, updated FROM checkin ORDER BY name" ) ] def get_checkin_names(self) -> List[str]: with self.con: return [ x["name"] for x in self.con.execute( "SELECT name FROM checkin WHERE is_present ORDER BY name" ) ] def get_checkin_count(self) -> int: with self.con: return self.con.execute( "SELECT COUNT(name) FROM checkin WHERE is_present" ).fetchone()[0] def clear_rows(self) -> None: with self.con: self.con.execute("DELETE FROM checkin") def get_updated_for_name(self, name: str) -> datetime: with self.con: return datetime.fromtimestamp( float( self.con.execute( f"SELECT updated FROM checkin WHERE name = '{name}'" ).fetchone()[0] ) ) def auto_checkout(self) -> None: filter_time = ( datetime.now() - timedelta(hours=AUTO_CHECKOUT_HOURS) ).timestamp() with self.con: self.con.execute( f"UPDATE checkin SET IS_PRESENT = 0 WHERE updated <= {filter_time}" )