Skip to content

digest_store

digest_store

DigestStore — SQLite-backed storage for pre-computed digest artifacts.

Classes

DigestArtifact dataclass 

DigestArtifact(text: str, audio_path: Path, sections: Dict[str, str], sources_used: List[str], generated_at: datetime, model_used: str, voice_used: str, quality_score: float = 0.0, evaluator_feedback: str = '')

A pre-computed morning digest ready for delivery.

DigestStore 

DigestStore(db_path: str = '')

SQLite store for digest artifacts.

Source code in src/openjarvis/agents/digest_store.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(self, db_path: str = "") -> None:
    if not db_path:
        db_path = str(Path.home() / ".openjarvis" / "digest.db")
    self._db_path = db_path
    self._conn = sqlite3.connect(db_path, check_same_thread=False)
    self._conn.execute("PRAGMA journal_mode=WAL")
    self._conn.execute(
        """
        CREATE TABLE IF NOT EXISTS digests (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            text TEXT NOT NULL,
            audio_path TEXT NOT NULL,
            sections TEXT NOT NULL,
            sources_used TEXT NOT NULL,
            generated_at TEXT NOT NULL,
            model_used TEXT NOT NULL,
            voice_used TEXT NOT NULL,
            quality_score REAL NOT NULL DEFAULT 0.0,
            evaluator_feedback TEXT NOT NULL DEFAULT ''
        )
        """
    )
    self._migrate()
    self._conn.commit()
Functions
save 
save(artifact: DigestArtifact) -> None

Save a digest artifact.

Source code in src/openjarvis/agents/digest_store.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def save(self, artifact: DigestArtifact) -> None:
    """Save a digest artifact."""
    self._conn.execute(
        """
        INSERT INTO digests
            (text, audio_path, sections, sources_used,
             generated_at, model_used, voice_used,
             quality_score, evaluator_feedback)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """,
        (
            artifact.text,
            str(artifact.audio_path),
            json.dumps(artifact.sections),
            json.dumps(artifact.sources_used),
            artifact.generated_at.isoformat(),
            artifact.model_used,
            artifact.voice_used,
            artifact.quality_score,
            artifact.evaluator_feedback,
        ),
    )
    self._conn.commit()
get_latest 
get_latest() -> Optional[DigestArtifact]

Return the most recent digest, or None.

Source code in src/openjarvis/agents/digest_store.py 
109
110
111
112
113
114
115
116
117
118
119
def get_latest(self) -> Optional[DigestArtifact]:
    """Return the most recent digest, or None."""
    row = self._conn.execute(
        "SELECT text, audio_path, sections, sources_used,"
        " generated_at, model_used, voice_used,"
        " quality_score, evaluator_feedback"
        " FROM digests ORDER BY id DESC LIMIT 1"
    ).fetchone()
    if row is None:
        return None
    return self._row_to_artifact(row)
get_today 
get_today(timezone_name: str = 'UTC') -> Optional[DigestArtifact]

Return today's digest if it exists, or None.

Source code in src/openjarvis/agents/digest_store.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def get_today(self, timezone_name: str = "UTC") -> Optional[DigestArtifact]:
    """Return today's digest if it exists, or None."""
    try:
        from zoneinfo import ZoneInfo

        today = datetime.now(ZoneInfo(timezone_name)).strftime("%Y-%m-%d")
    except ImportError:
        today = datetime.now().strftime("%Y-%m-%d")

    row = self._conn.execute(
        "SELECT text, audio_path, sections, sources_used,"
        " generated_at, model_used, voice_used,"
        " quality_score, evaluator_feedback"
        " FROM digests WHERE generated_at LIKE ? ORDER BY id DESC LIMIT 1",
        (f"{today}%",),
    ).fetchone()
    if row is None:
        return None
    return self._row_to_artifact(row)
history 
history(limit: int = 10) -> List[DigestArtifact]

Return the N most recent digests.

Source code in src/openjarvis/agents/digest_store.py 
141
142
143
144
145
146
147
148
149
150
def history(self, limit: int = 10) -> List[DigestArtifact]:
    """Return the N most recent digests."""
    rows = self._conn.execute(
        "SELECT text, audio_path, sections, sources_used,"
        " generated_at, model_used, voice_used,"
        " quality_score, evaluator_feedback"
        " FROM digests ORDER BY id DESC LIMIT ?",
        (limit,),
    ).fetchall()
    return [self._row_to_artifact(r) for r in rows]