store

store

Transactional SQLite persistence for Succès.

The schema is local-first but sync-ready: every mutation appends an immutable operation, deletes are tombstones, and updates use millisecond LWW clocks. No remote service is implied or reported until one is configured.

Classes

SuccesError

Bases: ValueError

Domain validation or conflict error safe to show to the user.

SuccesNotFound

Bases: SuccesError

Requested Succès entity was not found.

SuccesStore

SuccesStore(db_path: str | Path | None = None)

Thread-safe facade using one short-lived SQLite connection per action.

Source code in src/diapason/succes/store.py

def __init__(self, db_path: str | Path | None = None) -> None:
    self.db_path = Path(db_path or (get_data_dir() / "succes.db"))
    self.db_path.parent.mkdir(parents=True, exist_ok=True)
    self._lock = threading.RLock()
    with self._connect() as conn:
        conn.executescript(_SCHEMA)
        operation_columns = {
            row["name"]
            for row in conn.execute(
                "PRAGMA table_info(succes_operations)"
            ).fetchall()
        }
        if "request_json" not in operation_columns:
            conn.execute(
                "ALTER TABLE succes_operations ADD COLUMN "
                "request_json TEXT NOT NULL DEFAULT '{}'"
            )
        self._ensure_task_columns(conn)
        self._ensure_project_columns(conn)
        conn.execute(
            "INSERT OR IGNORE INTO succes_meta(key, value) VALUES('device_id', ?)",
            (f"mac-{secrets.token_hex(8)}",),
        )
        conn.commit()

Methods:

reschedule_series

reschedule_series(
    task_id: str,
    scheduled_date: str,
    *,
    op_id: str | None = None,
) -> dict[str, Any]

Shift every occurrence of a recurrence by the same day offset.

The dragged occurrence defines the delta between its current date and the drop target; all sibling occurrences (same template_id) move by that delta. Occurrences without a date are left untouched, and the recurrence rule itself is edited separately on the Récurrences page.

Source code in src/diapason/succes/store.py

def reschedule_series(
    self, task_id: str, scheduled_date: str, *, op_id: str | None = None
) -> dict[str, Any]:
    """Shift every occurrence of a recurrence by the same day offset.

    The dragged occurrence defines the delta between its current date and
    the drop target; all sibling occurrences (same ``template_id``) move by
    that delta. Occurrences without a date are left untouched, and the
    recurrence rule itself is edited separately on the Récurrences page.
    """
    scheduled = _validate_iso_date(scheduled_date)
    if not scheduled:
        raise SuccesError("Une date valide est requise pour reporter la série.")
    ts = now_ms()
    with self._transaction() as conn:
        anchor = self._load_task(conn, task_id)
        if anchor is None:
            raise SuccesNotFound("Cette tâche n'existe pas ou a été supprimée.")
        template_id = str(anchor.get("templateId") or "")
        if not template_id:
            raise SuccesError("Cette tâche ne provient pas d'une récurrence.")
        current = str(anchor.get("date") or "")
        if not current:
            raise SuccesError("La tâche de référence n'a pas encore de date.")
        delta = (date.fromisoformat(scheduled) - date.fromisoformat(current)).days
        rows = conn.execute(
            "SELECT id, scheduled_date FROM succes_tasks "
            "WHERE template_id=? AND deleted_at_ms IS NULL",
            (template_id,),
        ).fetchall()
        updated = 0
        for row in rows:
            old = str(row["scheduled_date"] or "")
            if not old:
                continue
            new_date = (date.fromisoformat(old) + timedelta(days=delta)).isoformat()
            if delta != 0:
                conn.execute(
                    "UPDATE succes_tasks SET scheduled_date=?, updated_at_ms=? "
                    "WHERE id=?",
                    (new_date, ts, row["id"]),
                )
                task = self._load_task(conn, row["id"])
                assert task is not None
                self._record_op(
                    conn,
                    entity="tasks",
                    entity_id=row["id"],
                    kind="upsert",
                    payload=task,
                    request={
                        "action": "reschedule_series",
                        "taskId": row["id"],
                        "templateId": template_id,
                        "date": new_date,
                    },
                    timestamp_ms=ts,
                )
            updated += 1
    return {"templateId": template_id, "deltaDays": delta, "updated": updated}

Functions: