import json from datetime import date, timedelta from ..db import db, ThingGenerator from .things import add_thing def get_generator(item_id: int) -> ThingGenerator: return ThingGenerator.get_by_id(item_id) def get_generators() -> list[ThingGenerator]: query = ThingGenerator.select().where(~ThingGenerator.deleted) return query.order_by("type", "template") def add_generator( template: str, type: str, val: str, ) -> ThingGenerator: # JSON for future expansion config = json.dumps({"val": val}) with db.atomic(): task = ThingGenerator.create( template=template, type=type, config=config, ) return task def generate_needed_tasks(): to_create = [] for g in get_generators(): next = g.next_at() if not next: continue # TODO: make configurable if date.today() - next > timedelta(days=14): to_create.append( { "text": g.template.format(next=next), "project": "recurring", "due": next, } ) for c in to_create: add_thing(**c) return to_create def update_generator( item_id: int, **kwargs, ) -> ThingGenerator: # replace "val" with JSON if "val" in kwargs: config = {"val": kwargs.pop("val")} kwargs["config"] = json.dumps(config) with db.atomic(): query = ThingGenerator.update(kwargs).where(ThingGenerator.id == item_id) query.execute() task = ThingGenerator.get_by_id(item_id) return task