maxenceg2m/ppit
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

black formatting

Browse Source
This commit is contained in:
Maxence G. de Montauzan
2018-07-21 15:41:48 +02:00
parent 5f9ea6f0b7
commit b18c4dcd18
8 changed files with 475 additions and 266 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
action.py
View File

@@ -9,106 +9,142 @@ from datetime import datetime
from enum import Enum
import sys
class TypeAction(Enum):
CREATE = 'created'
UPDATE = 'updated'
MOVE = 'moved'
DELETE = 'deleted'
CREATE = "created"
UPDATE = "updated"
MOVE = "moved"
DELETE = "deleted"
class Action:
def __init__(self, row):
if not row:
logging.debug('Action - empty row')
logging.debug("Action - empty row")
# TODO
# raise Exception('No action found - database need to be initialized.')
else:
logging.debug('Action - fill fields')
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(args, last_action, conn):
logging.info('>> handle action')
query = 'SELECT * FROM action;'
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')
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))
if taction == "init":
print("{} ({}): {}".format(formated_date, username, message))
else:
object_type = ''
id_object = ''
object_type = ""
id_object = ""
object_type = '' # TODO Enum ?
if (project_id):
object_type = 'project'
object_type = "" # TODO Enum ?
if project_id:
object_type = "project"
id_object = project_id
if (task_id):
object_type = 'task'
if task_id:
object_type = "task"
id_object = task_id
if (note_id):
object_type = 'note'
if note_id:
object_type = "note"
id_object = note_id
if taction == TypeAction.DELETE.value:
print('{} ({}): {} {} {}'.format(formated_date, username, taction, object_type, id_object))
print(
"{} ({}): {} {} {}".format(
formated_date, username, taction, object_type, id_object
)
)
else:
print('{} ({}): {} {} {}: {}'.format(formated_date, username, taction, object_type, id_object, message))
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):
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(),))
cursor.execute(
query,
(
project_id,
task_id,
note_id,
getpass.getuser(),
action_type.value,
message,
datetime.now(),
),
)
logging.debug('created action')
logging.debug("created action")
def set_active(cursor, project_id = None, task_id = None, note_id = None):
query = 'UPDATE current SET {} = ? WHERE id = 0'
def set_active(cursor, project_id=None, task_id=None, note_id=None):
query = "UPDATE current SET {} = ? WHERE id = 0"
if project_id:
query = query.format('project_id')
query = query.format("project_id")
obj_id = project_id
elif task_id:
query = query.format('task_id')
query = query.format("task_id")
obj_id = task_id
elif note_id:
query = query.format('note_id')
query = query.format("note_id")
obj_id = note_id
cursor.execute(query, (obj_id,))
def read_current(conn):
logging.info('> last_action')
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;'
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):
def _get_border_action(conn, last=False):
# Used for info
logging.info('> get_border_action')
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 '')
query = "SELECT username, created_at FROM action ORDER BY id {} LIMIT 1;"
query = query.format("DESC" if last else "")
logging.debug(query)
@@ -116,7 +152,9 @@ def _get_border_action(conn, last = False):
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')
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)
@@ -128,6 +166,7 @@ def get_first_action(conn):
"""
return _get_border_action(conn)
def get_last_action(conn):
"""
Get last action entry and return a tuple with username and readable date

17
args.py
View File

@@ -1,6 +1,7 @@
import sys
import logging
def get_string_arg(args, required):
"""
Take an array of element, take first element and return it stripped.
@@ -12,24 +13,26 @@ def get_string_arg(args, required):
:return: The arg stripped
"""
if not args:
print(f'Missing {required}')
print(f"Missing {required}")
sys.exit(1)
logging.debug(args)
return args[0].strip()
# TODO Move this function
def arg_string(arg, required = ''):
def arg_string(arg, required=""):
# TODO To meld with get_string_args
if required and arg.startswith('-'):
print('required ', required)
sys.exit('1')
if required and arg.startswith("-"):
print("required ", required)
sys.exit("1")
else:
return arg.strip()
def arg_number(arg, required = ''):
def arg_number(arg, required=""):
if not arg.isdigit():
print('Invalid value') # TODO
print("Invalid value") # TODO
sys.exit(1)
return int(arg)

99
note.py
View File

@@ -14,29 +14,33 @@ import project
from args import get_string_arg
from args import arg_number
class Note:
def __init__(self):
self.message = None
def handle_note(args, last_action, conn):
logging.info('> handle note')
logging.debug('args: ' + str(args))
logging.info("> handle note")
logging.debug("args: " + str(args))
if not args:
list_note(last_action.project_id, last_action.task_id, last_action.note_id, conn)
list_note(
last_action.project_id, last_action.task_id, last_action.note_id, conn
)
sys.exit(0)
if args[0] == '-c':
if args[0] == "-c":
if len(args) == 1:
print('missing note message')
print("missing note message")
sys.exit(1)
note = Note()
note.message = get_string_arg(args[1:], 'note message')
note.message = get_string_arg(args[1:], "note message")
create_note(note, last_action.project_id, last_action.task_id, conn)
elif args[0] == '-e':
elif args[0] == "-e":
if len(args) == 1:
print('nothing to update')
print("nothing to update")
note = Note()
args_i = 1
@@ -46,23 +50,24 @@ def handle_note(args, last_action, conn):
args_i += 1
if not note_id:
print('No active note to update')
print("No active note to update")
sys.exit(1)
note.message = get_string_arg(args[args_i:], 'note message')
note.message = get_string_arg(args[args_i:], "note message")
edit_note(note, note_id, last_action, conn)
elif args[0] == '-d':
elif args[0] == "-d":
# FIXME Duplicate code
if len(args) > 1:
delete_note(arg_number(args[1]), conn)
else:
delete_note(last_action.task_id, conn)
else:
print(f'Invalid note option: {args[0]}')
print(f"Invalid note option: {args[0]}")
def create_note(note, active_project_id, active_task_id, conn):
# TODO Don't create note if no project (nothing selected?) ==> foreign key is not a solution: there is no project ID :)
logging.info('>> Create note')
logging.info(">> Create note")
query = """
INSERT INTO note (project_id, task_id, username, message)
@@ -70,36 +75,57 @@ def create_note(note, active_project_id, active_task_id, conn):
"""
cursor = conn.cursor()
cursor.execute(query, (active_project_id, active_task_id, getpass.getuser(), note.message))
cursor.execute(
query, (active_project_id, active_task_id, getpass.getuser(), note.message)
)
note_id = cursor.lastrowid
action_message = '{} (task: {})'.format(note.message, active_task_id)
record_action(cursor, TypeAction.CREATE, action_message, active_project_id, active_task_id, note_id)
action_message = "{} (task: {})".format(note.message, active_task_id)
record_action(
cursor,
TypeAction.CREATE,
action_message,
active_project_id,
active_task_id,
note_id,
)
logging.debug(action_message)
print('created note {}: {} (task {})'.format(note_id, note.message, active_task_id))
print("created note {}: {} (task {})".format(note_id, note.message, active_task_id))
def edit_note(note, note_id, last_action, conn):
logging.info('>> Edit note')
logging.info(">> Edit note")
query = 'UPDATE note SET message = ? WHERE id = ?'
query = "UPDATE note SET message = ? WHERE id = ?"
cursor = conn.cursor()
cursor.execute(query, (note.message, note_id,))
cursor.execute(query, (note.message, note_id))
if cursor.rowcount != 1:
logging.error('UPDATE FAILED')
print('could not find note {}'.format(note_id))
logging.error("UPDATE FAILED")
print("could not find note {}".format(note_id))
sys.exit(1)
record_action(cursor, TypeAction.UPDATE, note.message, last_action.project_id,
last_action.task_id, last_action.note_id)
record_action(
cursor,
TypeAction.UPDATE,
note.message,
last_action.project_id,
last_action.task_id,
last_action.note_id,
)
print(
"updated note {}: (message: {}, task {})".format(
note_id, note.message, last_action.task_id
)
)
print('updated note {}: (message: {}, task {})'.format(note_id, note.message, last_action.task_id))
def delete_note(note_id, conn):
logging.info('>> Remove note')
logging.info(">> Remove note")
query = """
DELETE FROM note
@@ -109,20 +135,27 @@ def delete_note(note_id, conn):
cursor = conn.cursor()
cursor.execute(query, (note_id,))
if cursor.rowcount != 1:
logging.error('DELETE FAILED')
print('could not find note {}'.format(note_id))
logging.error("DELETE FAILED")
print("could not find note {}".format(note_id))
sys.exit(1)
record_action(cursor, TypeAction.DELETE, '', note_id=note_id)
record_action(cursor, TypeAction.DELETE, "", note_id=note_id)
print("deleted note {}: {}".format(note_id, "note_name"))
print('deleted note {}: {}'.format(note_id, 'note_name'))
def list_note(active_project_id, active_task_id, active_note_id, conn):
logging.info('>> No arguments')
query = "SELECT id, username, message FROM note WHERE project_id = ? AND task_id = ?;" #TODO Date & time
logging.info(">> No arguments")
query = (
"SELECT id, username, message FROM note WHERE project_id = ? AND task_id = ?;"
) # TODO Date & time
cursor = conn.cursor()
cursor.execute(query, (active_project_id, active_task_id))
for row in cursor.fetchall():
note_id, username, message = row
print('{:1} {:2d}: ({:8}) {}'.format('*' if active_note_id == note_id else '', note_id, username, message))
print(
"{:1} {:2d}: ({:8}) {}".format(
"*" if active_note_id == note_id else "", note_id, username, message
)
)

77
phelp.py
View File

@@ -1,20 +1,21 @@
import sys
from utils import strstr
def handle_help(args, *_):
if not args:
help_help()
sys.exit(0)
cmd_handler = [
('init', help_init),
('project', help_project),
('task', help_task),
('note', help_note),
('log', help_action),
('info', help_info),
('help', help_help),
('version', help_version)
("init", help_init),
("project", help_project),
("task", help_task),
("note", help_note),
("log", help_action),
("info", help_info),
("help", help_help),
("version", help_version),
]
candidate = -1
@@ -23,17 +24,23 @@ def handle_help(args, *_):
if candidate < 0:
candidate = i
else:
print('Ambiguous command ({})'.format(args[0]))
print("Ambiguous command ({})".format(args[0]))
sys.exit(1)
if candidate < 0:
print("Invalid command ({}), run '{} help' for help".format(args[0], os.path.basename(__file__)))
print(
"Invalid command ({}), run '{} help' for help".format(
args[0], os.path.basename(__file__)
)
)
sys.exit(1)
cmd_handler[candidate][1]()
def help_project():
print('''Pits project is basic entity used to group related tasks together. A project has name and status.
print(
"""Pits project is basic entity used to group related tasks together. A project has name and status.
Creating a project:
@@ -79,10 +86,13 @@ Examples:
$ pits p
* 1: (username) [current] Task and Bugs (0 tasks)
''')
"""
)
def help_task():
print('''In Pits a task belongs to a project and projects can have many tasks. Task attributes include name, status,
print(
"""In Pits a task belongs to a project and projects can have many tasks. Task attributes include name, status,
priority, date, and time. All attributes except the name are optional.
Creating a task:
@@ -142,10 +152,13 @@ Examples:
$ pits t
1: (username) [done] [normal] Oct 10, 2010 4:30 Hack this (0 notes)
''')
"""
)
def help_note():
print('''Pits notes are attached to a task. The only attribute is the note's message body.
print(
"""Pits notes are attached to a task. The only attribute is the note's message body.
Creating a note:
@@ -162,22 +175,31 @@ Deleting a note:
Listing notes:
$ pits note
''')
"""
)
def help_action():
print('''Show summary information about your Pits database. This command is as simple as:
print(
"""Show summary information about your Pits database. This command is as simple as:
pits log
''')
"""
)
def help_info():
print('''Show summary information about your Pits database. This command is as simple as:
print(
"""Show summary information about your Pits database. This command is as simple as:
pits info
''')
"""
)
def help_help():
print('''Pits is highly experimental software. Use it at your own risk. See LICENSE file for details.
print(
"""Pits is highly experimental software. Use it at your own risk. See LICENSE file for details.
usage: pits command [args]
The commands are:
@@ -192,13 +214,17 @@ The commands are:
All commands might be shortened as long as they remain unambiguous. See 'pits help <command>' for more
information on a specific command.
''')
"""
)
def help_version():
print('todo')
print("todo")
def help_init():
print('''Create empty Pits database or reinitialize an existing one. Default file name for the Pits database
print(
"""Create empty Pits database or reinitialize an existing one. Default file name for the Pits database
is ~/.pits.db -- you can override the default by setting PITFILE environment variable.
$ pits init [-f]
@@ -210,4 +236,5 @@ Example:
$ pits init
/home/user/.pits.db already exists, do you want to override it [y/N]: y
Created database /home/user/.pit
''')
"""
)

93
pits.py
View File

@@ -17,22 +17,29 @@ import utils
# logging.basicConfig(level=logging.ERROR, format='%(levelname)7s :: %(message)s')
# logging.basicConfig(level=logging.DEBUG, format='%(asctime)s :: %(levelname)s :: %(message)s')
logging.basicConfig(level=logging.ERROR, format='%(asctime)s :: {%(filename)10s:%(lineno)3d} :: %(funcName)s :: %(levelname)s :: %(message)s')
logging.basicConfig(
level=logging.ERROR,
format="%(asctime)s :: {%(filename)10s:%(lineno)3d} :: %(funcName)s :: %(levelname)s :: %(message)s",
)
DB_FILENAME = 'pits.db'
SCHEMA_FILENAME = 'pits_schema.sql'
DB_FILENAME = "pits.db"
SCHEMA_FILENAME = "pits_schema.sql"
PIT_VERSION = '0.0.1'
PIT_VERSION = "0.0.1"
SCHEMA_VERSION = 1
def pits_init():
db_path = utils.get_file_path(DB_FILENAME)
logging.debug('Search database in {}'.format(db_path))
logging.debug("Search database in {}".format(db_path))
if os.path.exists(db_path):
valid = {"yes": True, "y": True, "ye": True}
print('{} already exists, do you want to override it? [y/N]:'.format(db_path), end='')
print(
"{} already exists, do you want to override it? [y/N]:".format(db_path),
end="",
)
choice = input().lower()
if choice in valid:
os.remove(db_path)
@@ -40,58 +47,63 @@ def pits_init():
sys.exit(0)
with sqlite3.connect(db_path) as conn:
logging.info('Creating schema')
logging.info("Creating schema")
schema_path = utils.get_file_path(SCHEMA_FILENAME)
logging.debug('Schema file path: {}'.format(schema_path))
logging.debug("Schema file path: {}".format(schema_path))
with open(schema_path, 'rt') as f:
with open(schema_path, "rt") as f:
schema = f.read()
conn.executescript(schema)
action_query = '''
action_query = """
INSERT INTO action (username, action, message)
VALUES (?, ?, ?);
'''
action_msg = 'Initialized pits database - DB Schema version: {}'.format(SCHEMA_VERSION)
conn.execute(action_query, (getpass.getuser(), 'init', action_msg,))
"""
action_msg = "Initialized pits database - DB Schema version: {}".format(
SCHEMA_VERSION
)
conn.execute(action_query, (getpass.getuser(), "init", action_msg))
print('Created database ' + db_path)
print("Created database " + db_path)
# logging.info('Inserting initial data')
# conn.executescript("""
# insert into project values(1, 'me', 'test', 'in progress', 'now', 'now')""")
def pits_info(_1, _2, conn):
print('Pit version: {}'.format(PIT_VERSION))
print('Pit file name: {}'.format(utils.get_file_path(DB_FILENAME)))
print('Created by: {} on {}'.format(*action.get_first_action(conn)))
print('Last updated by: {} on {}'.format(*action.get_last_action(conn)))
print('Schema version: {}'.format(SCHEMA_VERSION))
print('Projects: {}'.format(utils.count_object(conn, 'project')))
print('Tasks: {}'.format(utils.count_object(conn, 'task')))
print('Notes: {}'.format(utils.count_object(conn, 'note')))
print('Log entries: {}'.format(utils.count_object(conn, 'action')))
print("Pit version: {}".format(PIT_VERSION))
print("Pit file name: {}".format(utils.get_file_path(DB_FILENAME)))
print("Created by: {} on {}".format(*action.get_first_action(conn)))
print("Last updated by: {} on {}".format(*action.get_last_action(conn)))
print("Schema version: {}".format(SCHEMA_VERSION))
print("Projects: {}".format(utils.count_object(conn, "project")))
print("Tasks: {}".format(utils.count_object(conn, "task")))
print("Notes: {}".format(utils.count_object(conn, "note")))
print("Log entries: {}".format(utils.count_object(conn, "action")))
def pits_version(*_):
print('Pits version 0.1.0')
print("Pits version 0.1.0")
def main():
cmd_handler = [
('project', project.handle_project),
('task', task.handle_task),
('note', note.handle_note),
('log', action.handle_action),
('info', pits_info),
('help', phelp.handle_help),
('version', pits_version),
('init', pits_init)
("project", project.handle_project),
("task", task.handle_task),
("note", note.handle_note),
("log", action.handle_action),
("info", pits_info),
("help", phelp.handle_help),
("version", pits_version),
("init", pits_init),
]
candidate = -1
argv = sys.argv
if len(argv) == 1:
argv.append('help')
argv.append("help")
argv.append(None)
for i in range(len(cmd_handler)):
@@ -99,20 +111,24 @@ def main():
if candidate < 0:
candidate = i
else:
print('Ambiguous command ({})'.format(argv[1]))
print("Ambiguous command ({})".format(argv[1]))
sys.exit(1)
if candidate < 0:
print("Invalid command ({}), run '{} help' for help".format(argv[1], utils.get_pits_path()))
print(
"Invalid command ({}), run '{} help' for help".format(
argv[1], utils.get_pits_path()
)
)
sys.exit(1)
if candidate == (len(cmd_handler) -1):
if candidate == (len(cmd_handler) - 1):
pits_init()
sys.exit(0)
conn = utils.create_connection(DB_FILENAME)
last_action = action.read_current(conn)
logging.debug('Last action: {}'.format(last_action))
logging.debug("Last action: {}".format(last_action))
logging.debug(argv)
logging.debug(argv[2:])
cmd_handler[candidate][1](argv[2:], last_action, conn)
@@ -120,5 +136,6 @@ def main():
conn.commit()
conn.close()
if __name__ == '__main__':
if __name__ == "__main__":
main()

125
project.py
View File

@@ -16,17 +16,19 @@ from args import arg_number
from utils import get_pits_path
class Project:
def __init__(self, status = None):
def __init__(self, status=None):
self.name = None
self.status = status
def __str__(self):
return str(self.__dict__)
def handle_project(args, last_action, conn):
logging.info('> handle project')
logging.debug('args: ' + str(args))
logging.info("> handle project")
logging.debug("args: " + str(args))
if not args:
list_project(last_action.project_id, conn)
@@ -34,18 +36,18 @@ def handle_project(args, last_action, conn):
if args[0].isdigit():
view_project_set_active(int(args[0]), last_action, conn)
elif args[0] == '-c':
elif args[0] == "-c":
if len(args) == 1:
print('missing project name')
print("missing project name")
sys.exit(1)
project = Project('active')
project.name = get_string_arg(args[1:], 'project name')
project = Project("active")
project.name = get_string_arg(args[1:], "project name")
parse_project_args(project, args[2:])
create_project(project, conn)
elif args[0] == '-e':
elif args[0] == "-e":
if len(args) == 1:
print('nothing to update')
print("nothing to update")
sys.exit(1)
project = Project()
@@ -56,18 +58,19 @@ def handle_project(args, last_action, conn):
else:
project_id = last_action.project_id
parse_project_args(project, args[1:])
logging.debug('Project: ({}) {}'.format(project_id, project))
logging.debug("Project: ({}) {}".format(project_id, project))
edit_project(project, project_id, conn)
# TODO To set as active project, us a -a option?
elif args[0] == '-d':
elif args[0] == "-d":
if len(args) > 1:
delete_project(arg_number(args[1]), conn)
else:
delete_project(last_action.project_id, conn)
else:
print(f'Invalid project option: {args[0]}')
print(f"Invalid project option: {args[0]}")
def parse_project_args(project, args):
if not args:
@@ -77,20 +80,21 @@ def parse_project_args(project, args):
while i < len(args):
logging.debug(args[i])
if args[i] == '-s':
if args[i] == "-s":
i += 1
project.status = get_string_arg(args[i:i+1], 'project status')
elif args[i] == '-n':
project.status = get_string_arg(args[i : i + 1], "project status")
elif args[i] == "-n":
i += 1
project.name = get_string_arg(args[i:i+1], 'project name')
project.name = get_string_arg(args[i : i + 1], "project name")
else:
print(f'Invalid project option: {args[i]}')
print(f"Invalid project option: {args[i]}")
sys.exit(1)
i += 1
def create_project(project, conn):
# TODO Project is same name is forbidden
logging.info('>> Create project')
logging.info(">> Create project")
query = """
INSERT INTO project (username, name, status, created_at)
@@ -98,44 +102,56 @@ def create_project(project, conn):
"""
cursor = conn.cursor()
cursor.execute(query, (getpass.getuser(), project.name, project.status, datetime.datetime.now(),))
cursor.execute(
query,
(getpass.getuser(), project.name, project.status, datetime.datetime.now()),
)
project_id = cursor.lastrowid
action_message = '{} (status: {})'.format(project.name, project.status)
action_message = "{} (status: {})".format(project.name, project.status)
record_action(cursor, TypeAction.CREATE, action_message, project_id=project_id)
logging.debug(action_message)
print('created project {}: {} (status: {})'.format(project_id, project.name, project.status))
print(
"created project {}: {} (status: {})".format(
project_id, project.name, project.status
)
)
def edit_project(project, project_id, conn):
logging.info('>> Edit project')
logging.info(">> Edit project")
update_args = [item for item in project.__dict__.items() if item[1] is not None]
if not update_args:
print("nothing to update")
print("Tips: if you want to active a project, just do '{} project <project_id>'"
.format(get_pits_path()))
print(
"Tips: if you want to active a project, just do '{} project <project_id>'".format(
get_pits_path()
)
)
sys.exit(1)
logging.debug('Project update args: {}'.format(update_args))
logging.debug("Project update args: {}".format(update_args))
query = 'UPDATE project SET {} WHERE id = ?'
query = query.format(', '.join("%s = '%s'" % (k, v) for k, v in update_args))
logging.debug('update project query: ' + query)
query = "UPDATE project SET {} WHERE id = ?"
query = query.format(", ".join("%s = '%s'" % (k, v) for k, v in update_args))
logging.debug("update project query: " + query)
cursor = conn.cursor()
logging.debug('Do a project update')
logging.debug("Do a project update")
cursor.execute(query, (project_id,))
log_args = ', '.join("%s: %s" % (k, v) for k, v in update_args)
print('updated project {}: ({})'.format(project_id, log_args))
log_args = ", ".join("%s: %s" % (k, v) for k, v in update_args)
print("updated project {}: ({})".format(project_id, log_args))
# TODO Print project name ?
record_action(cursor, TypeAction.UPDATE, '({})'.format(log_args), project_id)
record_action(cursor, TypeAction.UPDATE, "({})".format(log_args), project_id)
def delete_project(project_id, conn):
logging.info('>> Remove project')
logging.info(">> Remove project")
query = """
DELETE FROM project
@@ -145,13 +161,14 @@ def delete_project(project_id, conn):
cursor = conn.cursor()
cursor.execute(query, (project_id,))
if cursor.rowcount != 1:
logging.error('DELETE FAILED')
print('could not find project {}'.format(project_id))
logging.error("DELETE FAILED")
print("could not find project {}".format(project_id))
print('deleted project {}: {}'.format(project_id, 'project_name'))
record_action(cursor, TypeAction.DELETE, '', project_id)
print("deleted project {}: {}".format(project_id, "project_name"))
record_action(cursor, TypeAction.DELETE, "", project_id)
# TODO Cascade deleting
def list_project(active_project_id, conn):
query = """
SELECT id, username, name, status,
@@ -162,11 +179,20 @@ def list_project(active_project_id, conn):
cursor = conn.cursor()
cursor.execute(query)
for row in cursor.fetchall():
logging.debug('Project row: {}'.format(row))
logging.debug("Project row: {}".format(row))
project_id, username, name, status, nb_task = row
# TODO Formatting track: https://stackoverflow.com/questions/9989334/create-nice-column-output-in-python
print('{:1} {:2d}: ({:8}) | {} | {} ({} tasks)'.format('*' if active_project_id == project_id else '',
project_id, username, status, name, nb_task))
print(
"{:1} {:2d}: ({:8}) | {} | {} ({} tasks)".format(
"*" if active_project_id == project_id else "",
project_id,
username,
status,
name,
nb_task,
)
)
def view_project_set_active(project_id, last_action, conn):
# FIXME duplicate with list_project
@@ -180,10 +206,10 @@ def view_project_set_active(project_id, last_action, conn):
cursor.execute(query, (project_id,))
row = cursor.fetchone()
if not row:
print('Could not find project {}'.format(project_id))
print("Could not find project {}".format(project_id))
sys.exit(1)
print('* {:d}: ({}) {} (status: {}, {} tasks)'.format(*row))
print("* {:d}: ({}) {} (status: {}, {} tasks)".format(*row))
# FIXME duplicate with list_task
query = """
@@ -193,11 +219,20 @@ def view_project_set_active(project_id, last_action, conn):
"""
cursor.execute(query, (project_id,))
for row in cursor.fetchall():
logging.debug('Task row: {}'.format(row))
logging.debug("Task row: {}".format(row))
task_id, username, name, status, priority, date, time, nb_note = row
message = date + time + name
print(' {} {:d}: ({}) [{}] [{}] {} ({} tasks)'.format('*' if last_action.task_id == row[0] else ' ',
task_id, username, status, priority, message, nb_note))
print(
" {} {:d}: ({}) [{}] [{}] {} ({} tasks)".format(
"*" if last_action.task_id == row[0] else " ",
task_id,
username,
status,
priority,
message,
nb_note,
)
)
set_active(cursor, project_id=project_id)

198
task.py
View File

@@ -15,17 +15,19 @@ from args import arg_number
from utils import get_pits_path
class Task:
def __init__(self, status = None, priority = None):
def __init__(self, status=None, priority=None):
self.name = None
self.status = status
self.priority = priority
self.date = None
self.time = None
def handle_task(args, last_action, conn):
logging.info('> handle task')
logging.debug('args: ' + str(args))
logging.info("> handle task")
logging.debug("args: " + str(args))
if not args:
list_task(last_action.project_id, last_action.task_id, conn)
@@ -33,18 +35,18 @@ def handle_task(args, last_action, conn):
if args[0].isdigit():
view_task_set_active(arg_number(args[0]), last_action, conn)
elif args[0] == '-c':
elif args[0] == "-c":
if len(args) == 1:
print('missing task name')
print("missing task name")
sys.exit(1)
task = Task('open', 'normal')
task.name = get_string_arg(args[1:], 'task name')
task = Task("open", "normal")
task.name = get_string_arg(args[1:], "task name")
parse_task_args(task, args[2:])
create_task(task, last_action.project_id, conn)
elif args[0] == '-e':
elif args[0] == "-e":
if len(args) == 1:
print('nothing to update')
print("nothing to update")
sys.exit(1)
task = Task()
@@ -53,23 +55,25 @@ def handle_task(args, last_action, conn):
task_id = int(args[1])
parse_task_args(task, args[2:])
else:
task_id = last_action.task_id # FIXME If no active tast - use incremental args_i
task_id = (
last_action.task_id
) # FIXME If no active tast - use incremental args_i
parse_task_args(task, args[1:])
logging.debug('Task: ({}) {}'.format(task_id, task))
logging.debug("Task: ({}) {}".format(task_id, task))
edit_task(task, task_id, last_action.project_id, conn)
# TODO To set as active project, us a -a option?
elif args[0] == '-d':
elif args[0] == "-d":
# FIXME Duplicate code
if len(args) > 1:
delete_task(arg_number(args[1]),last_action.project_id, conn)
delete_task(arg_number(args[1]), last_action.project_id, conn)
else:
delete_task(last_action.task_id,last_action.project_id, conn)
elif args[0] == '-m':
delete_task(last_action.task_id, last_action.project_id, conn)
elif args[0] == "-m":
# Try another logic
args_i = 1
if len(args) < 3:
print('missing project number')
print("missing project number")
sys.exit(1)
task_id = last_action.task_id
@@ -80,22 +84,23 @@ def handle_task(args, last_action, conn):
args_i += 1
if not task_id:
print('No active task to move.')
print("No active task to move.")
sys.exit(1)
# Project option - if not fail
if args[args_i] != '-p':
print('missing project number')
if args[args_i] != "-p":
print("missing project number")
sys.exit(1)
else:
args_i += 1
moving_task(task_id, last_action.project_id, arg_number(args[args_i]), conn)
else:
print(f'Invalid task option: {args[0]}')
print(f"Invalid task option: {args[0]}")
# TODO Cascade deleting
def parse_task_args(task, args):
if not args:
return
@@ -104,30 +109,30 @@ def parse_task_args(task, args):
while i < len(args):
logging.debug(args[i])
if args[i] == '-s':
if args[i] == "-s":
i += 1
task.status = get_string_arg(args[i:i+1], 'task status')
elif args[i] == '-n':
task.status = get_string_arg(args[i : i + 1], "task status")
elif args[i] == "-n":
i += 1
task.name = get_string_arg(args[i:i+1], 'task name')
elif args[i] == '-p':
task.name = get_string_arg(args[i : i + 1], "task name")
elif args[i] == "-p":
i += 1
task.priority = get_string_arg(args[i:i+1], 'task priority')
elif args[i] == '-d':
task.priority = get_string_arg(args[i : i + 1], "task priority")
elif args[i] == "-d":
i += 1
task.date = get_string_arg(args[i:i+1], 'task date')
elif args[i] == '-t':
task.date = get_string_arg(args[i : i + 1], "task date")
elif args[i] == "-t":
i += 1
task.time = get_string_arg(args[i:i+1], 'task time')
task.time = get_string_arg(args[i : i + 1], "task time")
else:
print(f'Invalid task option: {args[i]}')
print(f"Invalid task option: {args[i]}")
sys.exit(1)
i += 1
def create_task(task, project_id, conn):
# TODO Don't create task if no project (nothing selected?) ==> foreign key is not a solution: there is no project ID :)
logging.info('>> Create task')
logging.info(">> Create task")
query = """
INSERT INTO
@@ -136,72 +141,101 @@ def create_task(task, project_id, conn):
"""
# TODO Date & time
date = ''
time = ''
date = ""
time = ""
cursor = conn.cursor()
cursor.execute(query, (project_id, getpass.getuser(), task.name, task.status,
task.priority, date, time, datetime.datetime.now(),))
cursor.execute(
query,
(
project_id,
getpass.getuser(),
task.name,
task.status,
task.priority,
date,
time,
datetime.datetime.now(),
),
)
task_id = cursor.lastrowid
logging.debug('CREATE ACTION ' + str(project_id) + str(task_id) )
logging.debug("CREATE ACTION " + str(project_id) + str(task_id))
task_details = [item for item in task.__dict__.items() if item[1] is not None and item[0] is not 'name']
log_args = ', '.join("%s: %s" % (k, v) for k, v in task_details)
task_details = [
item
for item in task.__dict__.items()
if item[1] is not None and item[0] is not "name"
]
log_args = ", ".join("%s: %s" % (k, v) for k, v in task_details)
action_message = '{} ({}, project: {})'.format(task.name, log_args, project_id)
action_message = "{} ({}, project: {})".format(task.name, log_args, project_id)
record_action(cursor, TypeAction.CREATE, action_message, project_id, task_id)
print('created task {}: {}'.format(task_id, action_message))
print("created task {}: {}".format(task_id, action_message))
def edit_task(task, task_id, project_id, conn):
logging.info('>> Edit task')
logging.info(">> Edit task")
update_args = [item for item in task.__dict__.items() if item[1] is not None]
if not update_args:
print("nothing to update")
print("Tips: if you want to activate a task, just do '{} task <task_id>'"
.format(get_pits_path()))
print(
"Tips: if you want to activate a task, just do '{} task <task_id>'".format(
get_pits_path()
)
)
sys.exit(1)
logging.debug('Task update args: {}'.format(update_args))
logging.debug("Task update args: {}".format(update_args))
query = 'UPDATE task SET {} WHERE id = ?'
query = query.format(', '.join("%s = '%s'" % (k, v) for k, v in update_args))
logging.debug('update task query: ' + query)
query = "UPDATE task SET {} WHERE id = ?"
query = query.format(", ".join("%s = '%s'" % (k, v) for k, v in update_args))
logging.debug("update task query: " + query)
cursor = conn.cursor()
logging.debug('Do a task update')
logging.debug("Do a task update")
cursor.execute(query, (task_id,))
log_args = ', '.join("%s: %s" % (k, v) for k, v in update_args)
log_args = ", ".join("%s: %s" % (k, v) for k, v in update_args)
# TODO Print task name ?
print('updated task {}: ({})'.format(task_id, log_args))
print("updated task {}: ({})".format(task_id, log_args))
# TODO Remove project id ?
logging.debug('UPDATE TASK - ACTION MESSAGE : ' + log_args)
record_action(cursor, TypeAction.UPDATE, '({})'.format(log_args), project_id, task_id)
logging.debug("UPDATE TASK - ACTION MESSAGE : " + log_args)
record_action(
cursor, TypeAction.UPDATE, "({})".format(log_args), project_id, task_id
)
def moving_task(task_id, old_project_id, project_id, conn):
logging.info('>> Moving task')
logging.info(">> Moving task")
query = """
UPDATE task SET project_id = ? WHERE id = ?;
"""
cursor = conn.cursor()
cursor.execute(query, (project_id, task_id,))
cursor.execute(query, (project_id, task_id))
if cursor.rowcount != 1:
print('Problem occur when moving task...')
print("Problem occur when moving task...")
sys.exit(1)
print('moved task {}: from project {} to project {}'.format(task_id, old_project_id, project_id))
record_action(cursor, TypeAction.MOVE, '', project_id=project_id, task_id=task_id) # TODO Message
print(
"moved task {}: from project {} to project {}".format(
task_id, old_project_id, project_id
)
)
record_action(
cursor, TypeAction.MOVE, "", project_id=project_id, task_id=task_id
) # TODO Message
def delete_task(task_id, project_id, conn):
logging.info('>> Remove task')
logging.info(">> Remove task")
query = """
DELETE FROM task
@@ -211,29 +245,40 @@ def delete_task(task_id, project_id, conn):
cursor = conn.cursor()
cursor.execute(query, (task_id,))
if cursor.rowcount != 1:
logging.error('DELETE FAILED')
print('could not find task {}'.format(task_id))
logging.error("DELETE FAILED")
print("could not find task {}".format(task_id))
sys.exit(1)
print('deleted task {}: {}'.format(task_id, 'task_name'))
record_action(cursor, TypeAction.DELETE, '', project_id, task_id) # TODO Message
print("deleted task {}: {}".format(task_id, "task_name"))
record_action(cursor, TypeAction.DELETE, "", project_id, task_id) # TODO Message
# TODO Cascade deleting
def list_task(active_project_id, active_task_id, conn):
logging.info('>> No arguments')
logging.info(">> No arguments")
query = """
SELECT id, username, status, priority, name,
(SELECT count(*) FROM note WHERE note.task_id = task.id)
FROM task WHERE project_id = ?;
""" #TODO Date & time
""" # TODO Date & time
cursor = conn.cursor()
cursor.execute(query, (active_project_id,))
for row in cursor.fetchall():
task_id, username, status, priority, name, nb_note = row
print('{:1} {:2d}: ({:8}) | {} | {} | {} ({} notes)'.format('*' if active_task_id == task_id else '', task_id,
username, status, priority, name, nb_note))
print(
"{:1} {:2d}: ({:8}) | {} | {} | {} ({} notes)".format(
"*" if active_task_id == task_id else "",
task_id,
username,
status,
priority,
name,
nb_note,
)
)
def view_task_set_active(task_id, last_action, conn):
# FIXME duplicate with list_task / view_project_set_active
@@ -241,16 +286,16 @@ def view_task_set_active(task_id, last_action, conn):
SELECT id, username, status, priority, name,
(SELECT count(*) FROM note WHERE note.task_id = task.id)
FROM task WHERE id = ?;
""" #TODO Date & time
""" # TODO Date & time
cursor = conn.cursor()
cursor.execute(query, (task_id,))
row = cursor.fetchone()
if not row:
print('Could not find task {}'.format(task_id))
print("Could not find task {}".format(task_id))
sys.exit(1)
print('* {:d}: ({}) {} (status: {}, {} tasks)'.format(*row))
print("* {:d}: ({}) {} (status: {}, {} tasks)".format(*row))
# FIXME duplicate with list_note
query = """
@@ -258,12 +303,19 @@ def view_task_set_active(task_id, last_action, conn):
"""
cursor.execute(query, (task_id,))
for row in cursor.fetchall():
logging.debug('Note row: {}'.format(row))
logging.debug("Note row: {}".format(row))
note_id, username, message = row
print(' {} {:d}: ({}) {}'.format('*' if last_action.note_id == row[0] else ' ',
note_id, username, message))
print(
" {} {:d}: ({}) {}".format(
"*" if last_action.note_id == row[0] else " ",
note_id,
username,
message,
)
)
set_active(cursor, task_id=task_id)
# $ pit t 11
# * 11: (budd) a task (project: 1, status: in progress, priority: high, date: Apr 26, 2018, time: 5:00, 2 notes)
# 6: (budd) bonjour le monde

3
utils.py
View File

@@ -3,6 +3,7 @@ import sys
import sqlite3
import logging
def get_file_path(file_name):
return os.path.join(os.path.dirname(os.path.realpath(__file__)), file_name)
@@ -34,6 +35,7 @@ def strstr(haystack, needle):
return haystack[pos:]
def create_connection(db_filename):
db_path = get_file_path(db_filename)
logging.debug("Try to connect to {}".format(db_path))
@@ -45,6 +47,7 @@ def create_connection(db_filename):
return None
def count_object(conn, table_name):
query = "SELECT count(*) FROM " + table_name
cursor = conn.cursor()