144 lines
5.2 KiB
Python
144 lines
5.2 KiB
Python
import logging
|
|
import sqlite3 as sq
|
|
|
|
from classes import Activity, Tag
|
|
from enums import ActivityType
|
|
|
|
|
|
class Database:
|
|
def __init__(self, db_name: str):
|
|
try:
|
|
self.con = sq.connect(db_name)
|
|
self.con.cursor().executescript('''
|
|
CREATE TABLE IF NOT EXISTS "activities" (
|
|
"id" VARCHAR NOT NULL UNIQUE,
|
|
"description" VARCHAR NOT NULL,
|
|
"time_spent" NUMERIC(3, 3) NOT NULL,
|
|
"date_start" DATETIME NOT NULL,
|
|
"date_end" DATETIME,
|
|
"is_tracked" BOOLEAN NOT NULL DEFAULT false,
|
|
"vetro_project_id" VARCHAR NOT NULL,
|
|
"tag_id" VARCHAR NOT NULL,
|
|
"project_id" VARCHAR NOT NULL,
|
|
PRIMARY KEY("id"),
|
|
FOREIGN KEY ("tag_id") REFERENCES "tags"("id")
|
|
FOREIGN KEY ("vetro_project_id") REFERENCES "vetro_projects"("id")
|
|
ON UPDATE NO ACTION ON DELETE NO ACTION
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS "tags" (
|
|
"id" VARCHAR NOT NULL UNIQUE,
|
|
"title" VARCHAR NOT NULL,
|
|
PRIMARY KEY("id")
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS "vetro_projects" (
|
|
"id" VARCHAR NOT NULL UNIQUE,
|
|
"title" VARCHAR NOT NULL,
|
|
PRIMARY KEY("id")
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS "activity_types" (
|
|
"id" INTEGER NOT NULL UNIQUE,
|
|
"title" VARCHAR NOT NULL,
|
|
PRIMARY KEY("id")
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS "time_entry" (
|
|
"id" INTEGER PRIMARY KEY,
|
|
"date" DATETIME NOT NULL,
|
|
"description" VARCHAR NOT NULL,
|
|
"activity_id" VARCHAR NOT NULL,
|
|
"time_spent" NUMERIC(3, 3) NOT NULL,
|
|
"task_id" INTEGER NOT NULL,
|
|
"activity_type_id" INTEGER NOT NULL,
|
|
FOREIGN KEY ("activity_id") REFERENCES "activities"("id")
|
|
ON UPDATE NO ACTION ON DELETE NO ACTION,
|
|
FOREIGN KEY ("task_id") REFERENCES "activities"("task_id")
|
|
ON UPDATE NO ACTION ON DELETE NO ACTION,
|
|
FOREIGN KEY ("activity_type_id") REFERENCES "activity_types"("id")
|
|
ON UPDATE NO ACTION ON DELETE NO ACTION
|
|
);
|
|
''')
|
|
except Exception as e:
|
|
raise(e)
|
|
|
|
def insert_activities(self, activities: list[Activity]):
|
|
cur = self.con.cursor()
|
|
for act in activities:
|
|
try:
|
|
cur.execute(
|
|
"INSERT INTO activities VALUES(?,?,?,?,?,?,?,?,?)",
|
|
(
|
|
act.id,
|
|
act.description,
|
|
act.time_spent,
|
|
act.date_start,
|
|
act.date_end,
|
|
act.is_tracked,
|
|
act.vetro_project_id,
|
|
act.tag_id,
|
|
act.project_id,
|
|
)
|
|
)
|
|
self.con.commit()
|
|
except Exception as e:
|
|
logging.error(e)
|
|
|
|
# Доделать
|
|
def insert_time_entries(self, activities: list[Activity]):
|
|
cur = self.con.cursor()
|
|
for act in activities:
|
|
try:
|
|
cur.execute(
|
|
"""INSERT INTO time_entry
|
|
(date, description, activity_id, time_spent, task_id, activity_type_id)
|
|
VALUES(?,?,?,?,?,?)""",
|
|
(
|
|
act.date_start.date(),
|
|
act.description,
|
|
act.id,
|
|
act.time_spent,
|
|
act.task_id,
|
|
ActivityType.DEVELOPMENT
|
|
)
|
|
)
|
|
self.con.commit()
|
|
except Exception as e:
|
|
logging.error(e)
|
|
|
|
|
|
def insert_tags(self, tags: list[Tag]):
|
|
cur = self.con.cursor()
|
|
for tag in tags:
|
|
try:
|
|
cur.execute(
|
|
"INSERT INTO tags VALUES(?, ?)",
|
|
(
|
|
tag.id,
|
|
tag.title
|
|
)
|
|
)
|
|
self.con.commit()
|
|
except Exception as e:
|
|
logging.error(e)
|
|
|
|
|
|
def insert_vetro_projects(self, vetro_projects: dict):
|
|
cur = self.con.cursor()
|
|
for k, v in vetro_projects.items():
|
|
try:
|
|
cur.execute("INSERT INTO vetro_projects VALUES(?, ?)", (k, v))
|
|
self.con.commit()
|
|
except Exception as e:
|
|
logging.error(e)
|
|
|
|
def insert_activity_types(self, activity_types: dict):
|
|
cur = self.con.cursor()
|
|
for k,v in activity_types.items():
|
|
try:
|
|
cur.execute("INSERT INTO activity_types VALUES(?, ?)", (k, v))
|
|
self.con.commit()
|
|
except Exception as e:
|
|
logging.error(e)
|