import os
from datetime import datetime as dt
-from sqlmodel import Field, Session, SQLModel, create_engine, select, text
+from sqlmodel import Field, Session, SQLModel, create_engine, select, text, Relationship
## Todo List
# - [ ] goal / vision / task
class Vision(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
- complete_date: dt = Field(default=None)
+ complete_date: dt | None = Field(default=None)
+ goals: list["Goal"] = Relationship(back_populates="vision")
def __str__(self):
return f'{self.name} ({self.id})'
class Goal(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
- complete_date: dt = Field(default=None)
+ complete_date: dt | None = Field(default=None)
vision_id: int | None = Field(default=None, foreign_key="vision.id")
- notes: str = Field(default=None)
+ notes: str | None = Field(default=None)
+ tasks: list["Task"] = Relationship(back_populates="goal")
+ vision: Vision = Relationship(back_populates="goals")
def __str__(self):
return f'{self.name} ({self.id})'
id: int | None = Field(default=None, primary_key=True)
name: str = Field(default=None, unique=True)
goal_id: int | None = Field(default=None, foreign_key="goal.id")
- complete_date: dt = Field(default=None)
+ complete_date: dt | None = Field(default=None)
+ goal: Goal = Relationship(back_populates="tasks")
+ def __str__(self):
+ return f'{self.name} ({self.id})'
+
+
+def pretty_print(goal, tasks):
+ results = []
+ results.append(f' ## {goal}')
+ for task in tasks:
+ if(task.complete_date):
+ results.append(f' - [x] {task}')
+ else:
+ results.append(f' - [ ] {task}')
+ else:
+ results.append('-- Nothing to list --')
+ return results
+
class Data():
def __init__(self, db_file = DB_FILE):
self.file = db_file
self.engine = self.get_engine()
+
def get_engine(self):
return create_engine(f"sqlite:///{self.file}")
def setup(self):
SQLModel.metadata.create_all(self.engine)
+ return 'Success'
- def print_all_visions(self):
- pass
-
- def print_all_goals(self):
- pass
+ def get_all_visions(self):
+ with Session(self.engine) as session:
+ cmd = select(Vision).where(Vision.complete_date == None)
+ return session.exec(cmd).all()
def print_ophan_tasks(self):
pass
+ def get_all_goals(self):
+ '''List all goals'''
+ with Session(self.engine) as session:
+ cmd = select(Goal).where(Goal.complete_date == None)
+ return session.exec(cmd).all()
+
+
+ def get_goal(self, id):
+ with Session(self.engine) as session:
+ cmd = select(Goal).where(Goal.id == id)
+ goal = session.exec(cmd).first()
+ return (goal, goal.tasks)
+
+ def get_vision(self, id):
+ with Session(self.engine) as session:
+ cmd = select(Vision).where(Vision.id == id)
+ vision = session.exec(cmd).first()
+ if vision:
+ return (vision, vision.goals)
+ else:
+ return ('Not found', [])
- def print_goal(self, goal, include_finished=False):
- '''Includes tasks'''
- pass
def complete_vision(self, id):
pass
def complete_goal(self, id):
- pass
+ with Session(self.engine) as session:
+ cmd = select(Goal).where(Goal.id == id)
+ goal = session.exec(cmd).first()
+ goal.complete_date = dt.today()
+ session.commit()
def complete_task(self, id):
- pass
+ with Session(self.engine) as session:
+ cmd = select(Task).where(Task.id == id)
+ task = session.exec(cmd).first()
+ task.complete_date = dt.today()
+ goal_id = task.goal_id
+ session.commit()
+ return goal_id
def add_with_name(self, cls, name):
- try:
- with Session(self.engine) as session:
- new = cls(name=name)
- session.add(chore)
- session.commit()
- return f'Created {name} [{str(cls)}].'
- except:
- return f'Unable to create {name} [{str(cls)}]. May be duplicate.'
-
+ id = -1
+ with Session(self.engine) as session:
+ new = cls(name=name)
+ session.add(new)
+ session.commit()
+ id = new.id
+ return id
def add_vision(self, name):
return self.add_with_name(Vision, name)
def add_goal(self, name):
return self.add_with_name(Goal, name)
- def add_task(self, name, goal_id):
+ def add_task(self, goal_id, name):
result = self.add_with_name(Task, name)
with Session(self.engine) as session:
- new = select(Task).where(Task.name == name).first()
+ cmd = select(Task).where(Task.name == name)
+ new = session.exec(cmd).first()
new.goal_id = goal_id
session.commit()
return result
def assign_vision(self, goal_id, vision_id):
'''Goals may move between visions'''
with Session(self.engine) as session:
- new = select(Goal).where(Goal.id == goal_id).first()
- new.vision_id = vision_id
+ cmd = select(Goal).where(Goal.id == goal_id)
+ goal = session.exec(cmd).first()
+ goal.vision_id = vision_id
session.commit()
+ return 'Success'
def add_note(self, note, goal):
'''Notes print with the goal.'''
# Someday: Notes on each
- def check(self, chore_id):
- with Session(self.engine) as session:
- return Data.check_chore(session, chore_id)
-
def list(self):
results = []
with Session(self.engine) as session:
results.append(Data.check_chore(session, chore.id))
return results
- def add(self, name):
- try:
- with Session(self.engine) as session:
- new_chore = Chore(name=name)
- session.add(new_chore)
- session.commit()
- return f'Created chore {name}.'
- except:
- return f'Unable to create {name}. May be duplicate.'
-
-
-
- def done(self, chore_id):
- with Session(self.engine) as session:
- chore = get_chore(session, chore_id)
- if(not chore):
- return f'Chore not found: {chore_id}'
-
- try:
- session.add(Record(chore_id=chore.id))
- session.commit()
- return f'Marked chore as done: {chore.name}.'
- except Exception as ex:
- return f'Unable to mark chore as done: {ex}'
-
-
- def undo(self, chore_id):
- with Session(self.engine) as session:
- cmd = select(Chore).where(Chore.id == chore_id)
- chore = session.exec(cmd).first()
-
- if(not chore):
- return f'Not found: {chore_id}'
-
- try:
- cmd = select(Record).where(Record.chore_id == chore_id).where(Record.complete_date >= today())
- results = session.exec(cmd).all()
- for result in results:
- session.delete(result)
- session.commit()
- return f'Marked chore as *not* done: {chore.name}.'
- except Exception as ex:
- return f'Unable to mark chore as done: {ex}'
-
- def remove(self, chore_id):
- try:
- with Session(self.engine) as session:
- chore = get_chore(session, chore_id)
- if(not chore):
- return f'Chore not found: {chore_id}'
-
- sql = text('''
- DELETE FROM Record
- WHERE chore_id = :chore_id
- ''')
- params = {}
- params['chore_id'] = chore_id
- results = session.execute(sql, params)
- logger.debug(results)
- sql = text('''
- DELETE FROM Chore
- WHERE id = :chore_id
- ''')
- results = session.execute(sql, params)
- logger.debug(results)
- session.commit()
- return f'Removed chore {chore_id}'
- except Exception as ex:
- return f'Unable to remove chore: {ex}'
projects / matrix-goal-bot.git / blobdiff