import getpass import datetime import logging from datetime import datetime from enum import Enum class TypeAction(Enum): CREATE = "created" UPDATE = "updated" MOVE = "moved" DELETE = "deleted" class Action: def __init__(self, row): if not row: logging.debug("Action - empty row") # TODO # raise Exception('No action found - database need to be initialized.') else: logging.debug("Action - fill fields") self.project_id, self.task_id, self.note_id = row def __str__(self): return str(self.__dict__) def handle_action(_1, _2, conn): logging.info(">> handle action") query = "SELECT * FROM action;" cursor = conn.cursor() cursor.execute(query) for row in cursor.fetchall(): _, project_id, task_id, note_id, username, taction, message, created_at = row logging.debug(row) formated_date = datetime.strptime( created_at[:26], "%Y-%m-%d %H:%M:%S.%f" ).strftime("%b %d, %Y %H:%M") if taction == "init": print("{} ({}): {}".format(formated_date, username, message)) else: object_type = "" id_object = "" object_type = "" # TODO Enum ? if project_id: object_type = "project" id_object = project_id if task_id: object_type = "task" id_object = task_id if note_id: object_type = "note" id_object = note_id print( "{} ({}): {} {} {}: {}".format( formated_date, username, taction, object_type, id_object, message ) ) # print('Sep 02, 2017 02:33 (dakota ): updated task 1: Passage à Angular 4 (status: in progress)') # Sep 02, 2017 02:33 (dakota ): updated task 1: Passage à Angular 4 (status: in progress) # Jun 13, 2018 01:55 (edelweiss ): updated note 67: (message: Fix anonymize when get a replica with ID [task 30, status:in progress] # Jul 05, 2018 00:40 (budd): moved task 2: from project 1 to project 2 def record_action( cursor, action_type, message, project_id="", task_id=None, note_id=None ): query = """ INSERT INTO action (project_id, task_id, note_id, username, action, message, created_at) VALUES (?, ?, ?, ?, ?, ?, ?); """ cursor.execute( query, ( project_id, task_id, note_id, getpass.getuser(), action_type.value, message, datetime.now(), ), ) logging.debug("created action") def set_active(cursor, project_id=None, task_id=None, note_id=None): query = "UPDATE current SET {} = ?" if project_id: query = query.format("project_id") obj_id = project_id elif task_id: query = query.format("task_id") obj_id = task_id elif note_id: query = query.format("note_id") obj_id = note_id cursor.execute(query, (obj_id,)) def read_current(conn): logging.info("> last_action") # query = 'SELECT * FROM action WHERE id = (SELECT MAX(id) FROM action' # query = 'SELECT project_id, task_id, note_id FROM action' query = "SELECT project_id, task_id, note_id FROM current;" cursor = conn.cursor() cursor.execute(query) last_action = Action(cursor.fetchone()) return last_action def _get_border_action(conn, last=False): # Used for info logging.info("> get_border_action") # query = 'SELECT * FROM action WHERE id = (SELECT MAX(id) FROM action' # query = 'SELECT project_id, task_id, note_id FROM action' query = "SELECT username, created_at FROM action ORDER BY id {} LIMIT 1;" query = query.format("DESC" if last else "") logging.debug(query) cursor = conn.cursor() cursor.execute(query) data = cursor.fetchone() formated_date = datetime.strptime(data[1][:26], "%Y-%m-%d %H:%M:%S.%f").strftime( "%b %d, %Y at %H:%M" ) # TODO Extrat a format date method return (data[0], formated_date) def get_first_action(conn): """ Get first action entry and return a tuple with username and readable date """ return _get_border_action(conn) def get_last_action(conn): """ Get last action entry and return a tuple with username and readable date """ return _get_border_action(conn, True)