import traceback from datetime import datetime from datetime import timedelta from geruecht.exceptions import DatabaseExecption class Base: def get_free_drink_list_config(self, id): try: cursor = self.db.connection.cursor() cursor.execute(f'select * from free_drink_list_config where id={id}') data = cursor.fetchone() if data['drink_id'] != None: data['drink'] = self.getDrinkPrice(data['drink_id']) data['free_drink_types'] = self.get_free_drink_list_types_for_drink(data['id']) return data except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def get_free_drink_list_configs(self): try: cursor = self.db.connection.cursor() cursor.execute("select * from free_drink_list_config") retVal = cursor.fetchall() for data in retVal: if data['drink_id'] != None: data['drink'] = self.getDrinkPrice(data['drink_id']) data['free_drink_types'] = self.get_free_drink_list_types_for_drink(data['id']) return retVal except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def set_free_drink_list_config(self, free_drink_list_config): try: cursor = self.db.connection.cursor() cursor.execute(f'insert into free_drink_list_config (drink_id, label, price) values ({free_drink_list_config["drink"]["id"]}, "{free_drink_list_config["label"]}", {free_drink_list_config["price"]})') self.db.connection.commit() cursor.execute(f'select id from free_drink_list_config where drink_id={free_drink_list_config["drink"]["id"]} and label="{free_drink_list_config["label"]}" and price={free_drink_list_config["price"]}') data = cursor.fetchone() for free_drink_type in free_drink_list_config["free_drink_types"]: cursor.execute( f'insert into free_drink_list_type_config (free_drink_list_config_id, free_drink_list_type_id) values ({data["id"]},{free_drink_type["id"]})') self.db.connection.commit() return self.get_free_drink_list_configs() except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def update_free_drink_list_config(self, free_drink_list_config): try: cursor = self.db.connection.cursor() cursor.execute(f'update free_drink_list_config set drink_id={free_drink_list_config["drink"]["id"]}, label="{free_drink_list_config["label"]}", price={free_drink_list_config["price"]} where id={free_drink_list_config["id"]}') cursor.execute(f'delete from free_drink_list_type_config where free_drink_list_config_id={free_drink_list_config["id"]}') for free_drink_type in free_drink_list_config["free_drink_types"]: cursor.execute(f'insert into free_drink_list_type_config (free_drink_list_config_id, free_drink_list_type_id) values ({free_drink_list_config["id"]},{free_drink_type["id"]})') self.db.connection.commit() return self.get_free_drink_list_configs() except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def delete_free_drink_list_config(self, free_drink_list_config): try: cursor = self.db.connection.cursor() cursor.execute(f'delete from free_drink_list_type_config where free_drink_list_config_id={free_drink_list_config["id"]}') cursor.execute(f'delete from free_drink_list_history where free_drink_config_id={free_drink_list_config["id"]}') cursor.execute(f'delete from free_drink_list_config where id={free_drink_list_config["id"]}') self.db.connection.commit() return self.get_free_drink_list_configs() except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def get_free_drink_list_types(self): try: cursor = self.db.connection.cursor() cursor.execute('select * from free_drink_list_type') return cursor.fetchall() except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def get_free_drink_list_types_for_drink(self, id): try: cursor = self.db.connection.cursor() cursor.execute(f'select a.* from free_drink_list_type a, free_drink_list_type_config b where free_drink_list_config_id={id} and b.free_drink_list_type_id=a.id') return cursor.fetchall() except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def get_free_drink_list_type(self, name): try: cursor = self.db.connection.cursor() if type(name) == str: sql = f'select * from free_drink_list_type where name={name}' elif type(name) == int: sql = f'select * from free_drink_list_type where id={name}' else: raise DatabaseExecption("name as no type int or str. name={}, type={}".format(name, type(name))) cursor.execute(sql) return cursor.fetchone() except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def set_free_drink_list_history(self, user, free_drink_list_config): try: cursor = self.db.connection.cursor() if 'free_drink_list_reason_id' in free_drink_list_config and 'description' in free_drink_list_config: sql = f'insert into free_drink_list_history (timestamp, free_drink_config_id, user_id, free_drink_type_id, free_drink_list_reason_id, description) values ("{datetime.now()}", {free_drink_list_config["id"]}, {user.id}, {free_drink_list_config["free_drink_type_id"]}, {free_drink_list_config["free_drink_list_reason_id"]}, "{free_drink_list_config["description"]}")' else: sql = f'insert into free_drink_list_history (timestamp, free_drink_config_id, user_id, free_drink_type_id) values ("{datetime.now()}", {free_drink_list_config["id"]}, {user.id}, {free_drink_list_config["free_drink_type_id"]})' cursor.execute(sql) self.db.connection.commit() return self.get_free_drink_list_history_by_user(user) except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def get_free_drink_list_history_by_user(self, user): try: cursor = self.db.connection.cursor() now = datetime.now() worker = self.getWorker(user, now) if worker: timestamp = worker["startdatetime"] else: timestamp = datetime.now() - timedelta(minutes=30) cursor.execute(f'select * from free_drink_list_history where timestamp>="{timestamp}" and (user_id={user.id} or free_drink_type_id=3)') retVal = cursor.fetchall() for data in retVal: data['timestamp'] = {'year': data['timestamp'].year, 'month': data['timestamp'].month, 'day': data['timestamp'].day, 'hour': data['timestamp'].hour, 'minute': data['timestamp'].minute, 'second': data['timestamp'].second} data['free_drink_config'] = self.get_free_drink_list_config(data['free_drink_config_id']) data['free_drink_type'] = self.get_free_drink_list_type(data['free_drink_type_id']) data['free_drink_list_reason'] = self.get_free_drink_list_reason(data['free_drink_list_reason_id']) if data['free_drink_list_reason_id'] else None return retVal except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def update_free_drink_list_history(self, free_drink_list_history): try: cursor = self.db.connection.cursor() cursor.execute(f'update free_drink_list_history set canceled={free_drink_list_history["canceled"]} where id={free_drink_list_history["id"]}') self.db.connection.commit() return True except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def delete_free_drink_list_history(self, free_drink_list_history): try: cursor = self.db.connection.cursor() cursor.execute(f'delete from free_drink_list_history where id={free_drink_list_history["id"]}') self.db.connection.commit() return True except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def get_free_drink_list_reason(self, id): try: cursor = self.db.connection.cursor() cursor.execute(f'select * from free_drink_list_reason where id={id}') return cursor.fetchone() except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err)) def get_free_drink_list_reasons(self): try: cursor = self.db.connection.cursor() cursor.execute(f'select * from free_drink_list_reason') return cursor.fetchall() except Exception as err: traceback.print_exc() self.db.connection.rollback() raise DatabaseExecption("Something went wrong with Database: {}".format(err))