software/src/orchestration/state.py

1"""SQLite-based job state persistence for fault-tolerant translation.23Provides ACID-compliant storage for job state, enabling crash recovery4and resume functionality. Uses SQLite for reliable atomic operations.5"""67from __future__ import annotations89import json10import sqlite311from contextlib import contextmanager12from datetime import datetime13from pathlib import Path14from typing import Any, Generator1516from src.orchestration.models import BibleJob, BookJob, JobStatus, VerseJob171819class JobStateStore:20    """SQLite-based storage for Bible translation job state.2122    Provides atomic operations for:23    - Creating and managing Bible translation jobs24    - Tracking verse-level progress25    - Claiming verses for worker processes26    - Recovering from crashes2728    All operations use SQLite transactions for ACID compliance.29    """3031    def __init__(self, db_path: Path | str) -> None:32        """Initialize the job state store.3334        Args:35            db_path: Path to the SQLite database file.36        """37        self.db_path = Path(db_path)38        self.db_path.parent.mkdir(parents=True, exist_ok=True)39        self._init_schema()4041    def _init_schema(self) -> None:42        """Initialize the database schema."""43        with self._connection() as conn:44            conn.executescript("""45                CREATE TABLE IF NOT EXISTS bible_jobs (46                    job_id TEXT PRIMARY KEY,47                    books TEXT NOT NULL,48                    status TEXT NOT NULL DEFAULT 'pending',49                    config_snapshot TEXT NOT NULL DEFAULT '{}',50                    created_at TEXT NOT NULL,51                    started_at TEXT,52                    completed_at TEXT,53                    total_verses INTEGER DEFAULT 0,54                    completed_verses INTEGER DEFAULT 0,55                    failed_verses INTEGER DEFAULT 056                );5758                CREATE TABLE IF NOT EXISTS book_jobs (59                    job_id TEXT NOT NULL,60                    book TEXT NOT NULL,61                    status TEXT NOT NULL DEFAULT 'pending',62                    total_verses INTEGER DEFAULT 0,63                    completed_verses INTEGER DEFAULT 0,64                    failed_verses INTEGER DEFAULT 0,65                    worker_pid INTEGER,66                    started_at TEXT,67                    completed_at TEXT,68                    error_message TEXT,69                    PRIMARY KEY (job_id, book),70                    FOREIGN KEY (job_id) REFERENCES bible_jobs(job_id)71                );7273                CREATE TABLE IF NOT EXISTS verse_jobs (74                    job_id TEXT NOT NULL,75                    book TEXT NOT NULL,76                    chapter INTEGER NOT NULL,77                    verse INTEGER NOT NULL,78                    status TEXT NOT NULL DEFAULT 'pending',79                    attempts INTEGER DEFAULT 0,80                    last_error TEXT,81                    worker_pid INTEGER,82                    started_at TEXT,83                    completed_at TEXT,84                    PRIMARY KEY (job_id, book, chapter, verse),85                    FOREIGN KEY (job_id) REFERENCES bible_jobs(job_id)86                );8788                CREATE INDEX IF NOT EXISTS idx_verse_pending89                ON verse_jobs(job_id, book, status)90                WHERE status = 'pending';9192                CREATE INDEX IF NOT EXISTS idx_bible_jobs_status93                ON bible_jobs(status);9495                CREATE INDEX IF NOT EXISTS idx_book_jobs_status96                ON book_jobs(job_id, status);97            """)9899    @contextmanager100    def _connection(self) -> Generator[sqlite3.Connection, None, None]:101        """Create a database connection with proper settings.102103        Yields:104            SQLite connection with row factory configured.105        """106        conn = sqlite3.connect(str(self.db_path), timeout=30.0)107        conn.row_factory = sqlite3.Row108        conn.execute("PRAGMA journal_mode=WAL")109        conn.execute("PRAGMA foreign_keys=ON")110        try:111            yield conn112            conn.commit()113        except Exception:114            conn.rollback()115            raise116        finally:117            conn.close()118119    # -------------------------------------------------------------------------120    # Bible Job Management121    # -------------------------------------------------------------------------122123    def create_bible_job(124        self,125        books: list[str],126        config: dict[str, Any],127        job_id: str | None = None,128    ) -> BibleJob:129        """Create a new Bible translation job.130131        Args:132            books: List of book names to translate.133            config: Configuration snapshot for this job.134            job_id: Optional specific job ID (generates UUID if not provided).135136        Returns:137            Created BibleJob instance.138        """139        import uuid140141        job = BibleJob(142            job_id=job_id or str(uuid.uuid4()),143            books=books,144            status=JobStatus.PENDING,145            config_snapshot=config,146            created_at=datetime.utcnow(),147        )148149        with self._connection() as conn:150            conn.execute(151                """152                INSERT INTO bible_jobs153                (job_id, books, status, config_snapshot, created_at,154                 total_verses, completed_verses, failed_verses)155                VALUES (?, ?, ?, ?, ?, ?, ?, ?)156                """,157                (158                    job.job_id,159                    json.dumps(job.books),160                    job.status.value,161                    json.dumps(job.config_snapshot),162                    job.created_at.isoformat(),163                    job.total_verses,164                    job.completed_verses,165                    job.failed_verses,166                ),167            )168169            # Create book job entries170            for book in books:171                conn.execute(172                    """173                    INSERT INTO book_jobs (job_id, book, status)174                    VALUES (?, ?, ?)175                    """,176                    (job.job_id, book, JobStatus.PENDING.value),177                )178179        return job180181    def get_bible_job(self, job_id: str) -> BibleJob | None:182        """Get a Bible job by ID.183184        Args:185            job_id: Job identifier.186187        Returns:188            BibleJob if found, None otherwise.189        """190        with self._connection() as conn:191            row = conn.execute(192                "SELECT * FROM bible_jobs WHERE job_id = ?",193                (job_id,),194            ).fetchone()195196            if not row:197                return None198199            return self._row_to_bible_job(row)200201    def get_latest_incomplete_job(self) -> BibleJob | None:202        """Get the most recent incomplete job.203204        Returns:205            Most recent BibleJob with status pending or in_progress.206        """207        with self._connection() as conn:208            row = conn.execute(209                """210                SELECT * FROM bible_jobs211                WHERE status IN (?, ?)212                ORDER BY created_at DESC213                LIMIT 1214                """,215                (JobStatus.PENDING.value, JobStatus.IN_PROGRESS.value),216            ).fetchone()217218            if not row:219                return None220221            return self._row_to_bible_job(row)222223    def update_job_status(224        self,225        job_id: str,226        status: JobStatus,227        started_at: datetime | None = None,228        completed_at: datetime | None = None,229    ) -> None:230        """Update the status of a Bible job.231232        Args:233            job_id: Job identifier.234            status: New status.235            started_at: When job started (if transitioning to in_progress).236            completed_at: When job completed (if transitioning to completed/failed).237        """238        with self._connection() as conn:239            updates = ["status = ?"]240            params: list[Any] = [status.value]241242            if started_at:243                updates.append("started_at = ?")244                params.append(started_at.isoformat())245246            if completed_at:247                updates.append("completed_at = ?")248                params.append(completed_at.isoformat())249250            params.append(job_id)251252            conn.execute(253                f"UPDATE bible_jobs SET {', '.join(updates)} WHERE job_id = ?",254                params,255            )256257    def update_job_verse_counts(258        self,259        job_id: str,260        total: int | None = None,261        completed: int | None = None,262        failed: int | None = None,263    ) -> None:264        """Update verse counts for a Bible job.265266        Args:267            job_id: Job identifier.268            total: New total verse count.269            completed: New completed verse count.270            failed: New failed verse count.271        """272        with self._connection() as conn:273            updates = []274            params: list[Any] = []275276            if total is not None:277                updates.append("total_verses = ?")278                params.append(total)279280            if completed is not None:281                updates.append("completed_verses = ?")282                params.append(completed)283284            if failed is not None:285                updates.append("failed_verses = ?")286                params.append(failed)287288            if not updates:289                return290291            params.append(job_id)292293            conn.execute(294                f"UPDATE bible_jobs SET {', '.join(updates)} WHERE job_id = ?",295                params,296            )297298    # -------------------------------------------------------------------------299    # Book Job Management300    # -------------------------------------------------------------------------301302    def get_book_job(self, job_id: str, book: str) -> BookJob | None:303        """Get a book job by job ID and book name.304305        Args:306            job_id: Parent job identifier.307            book: Book name.308309        Returns:310            BookJob if found, None otherwise.311        """312        with self._connection() as conn:313            row = conn.execute(314                "SELECT * FROM book_jobs WHERE job_id = ? AND book = ?",315                (job_id, book),316            ).fetchone()317318            if not row:319                return None320321            return self._row_to_book_job(row)322323    def get_book_jobs(self, job_id: str) -> list[BookJob]:324        """Get all book jobs for a Bible job.325326        Args:327            job_id: Parent job identifier.328329        Returns:330            List of BookJob instances.331        """332        with self._connection() as conn:333            rows = conn.execute(334                "SELECT * FROM book_jobs WHERE job_id = ? ORDER BY book",335                (job_id,),336            ).fetchall()337338            return [self._row_to_book_job(row) for row in rows]339340    def update_book_status(341        self,342        job_id: str,343        book: str,344        status: JobStatus,345        worker_pid: int | None = None,346        started_at: datetime | None = None,347        completed_at: datetime | None = None,348        error_message: str | None = None,349    ) -> None:350        """Update the status of a book job.351352        Args:353            job_id: Parent job identifier.354            book: Book name.355            status: New status.356            worker_pid: Worker process PID.357            started_at: When book started.358            completed_at: When book completed.359            error_message: Error message if failed.360        """361        with self._connection() as conn:362            updates = ["status = ?"]363            params: list[Any] = [status.value]364365            if worker_pid is not None:366                updates.append("worker_pid = ?")367                params.append(worker_pid)368369            if started_at:370                updates.append("started_at = ?")371                params.append(started_at.isoformat())372373            if completed_at:374                updates.append("completed_at = ?")375                params.append(completed_at.isoformat())376377            if error_message is not None:378                updates.append("error_message = ?")379                params.append(error_message)380381            params.extend([job_id, book])382383            conn.execute(384                f"UPDATE book_jobs SET {', '.join(updates)} "385                "WHERE job_id = ? AND book = ?",386                params,387            )388389    def update_book_verse_counts(390        self,391        job_id: str,392        book: str,393        total: int | None = None,394        completed: int | None = None,395        failed: int | None = None,396    ) -> None:397        """Update verse counts for a book job.398399        Args:400            job_id: Parent job identifier.401            book: Book name.402            total: New total verse count.403            completed: New completed verse count.404            failed: New failed verse count.405        """406        with self._connection() as conn:407            updates = []408            params: list[Any] = []409410            if total is not None:411                updates.append("total_verses = ?")412                params.append(total)413414            if completed is not None:415                updates.append("completed_verses = ?")416                params.append(completed)417418            if failed is not None:419                updates.append("failed_verses = ?")420                params.append(failed)421422            if not updates:423                return424425            params.extend([job_id, book])426427            conn.execute(428                f"UPDATE book_jobs SET {', '.join(updates)} "429                "WHERE job_id = ? AND book = ?",430                params,431            )432433    # -------------------------------------------------------------------------434    # Verse Job Management435    # -------------------------------------------------------------------------436437    def initialize_verse_jobs(438        self,439        job_id: str,440        book: str,441        verses: list[tuple[int, int]],442        completed_verses: list[tuple[int, int]] | None = None,443    ) -> None:444        """Initialize verse jobs for a book.445446        Args:447            job_id: Parent job identifier.448            book: Book name.449            verses: List of (chapter, verse) tuples to mark as pending.450            completed_verses: List of (chapter, verse) tuples already451                translated on disk, marked COMPLETED immediately.452        """453        completed_verses = completed_verses or []454        total = len(verses) + len(completed_verses)455456        with self._connection() as conn:457            # Delete any existing verse jobs for this book458            conn.execute(459                "DELETE FROM verse_jobs WHERE job_id = ? AND book = ?",460                (job_id, book),461            )462463            # Insert pending verse jobs464            if verses:465                conn.executemany(466                    """467                    INSERT INTO verse_jobs (job_id, book, chapter, verse, status)468                    VALUES (?, ?, ?, ?, ?)469                    """,470                    [471                        (job_id, book, chapter, verse, JobStatus.PENDING.value)472                        for chapter, verse in verses473                    ],474                )475476            # Insert already-completed verse jobs477            if completed_verses:478                now = datetime.utcnow().isoformat()479                conn.executemany(480                    """481                    INSERT INTO verse_jobs482                        (job_id, book, chapter, verse, status, completed_at)483                    VALUES (?, ?, ?, ?, ?, ?)484                    """,485                    [486                        (job_id, book, chapter, verse,487                         JobStatus.COMPLETED.value, now)488                        for chapter, verse in completed_verses489                    ],490                )491492            # Update book totals493            conn.execute(494                """495                UPDATE book_jobs496                SET total_verses = ?, completed_verses = ?497                WHERE job_id = ? AND book = ?498                """,499                (total, len(completed_verses), job_id, book),500            )501502    def atomic_claim_verse(503        self,504        job_id: str,505        book: str,506        chapter: int,507        verse: int,508        worker_pid: int,509    ) -> bool:510        """Atomically claim a verse for translation.511512        Uses a conditional update to ensure only one worker can claim a verse.513514        Args:515            job_id: Parent job identifier.516            book: Book name.517            chapter: Chapter number.518            verse: Verse number.519            worker_pid: Worker process PID claiming the verse.520521        Returns:522            True if claim succeeded, False if verse was already claimed.523        """524        with self._connection() as conn:525            cursor = conn.execute(526                """527                UPDATE verse_jobs528                SET status = ?, worker_pid = ?, started_at = ?, attempts = attempts + 1529                WHERE job_id = ? AND book = ? AND chapter = ? AND verse = ?530                AND status = ?531                """,532                (533                    JobStatus.IN_PROGRESS.value,534                    worker_pid,535                    datetime.utcnow().isoformat(),536                    job_id,537                    book,538                    chapter,539                    verse,540                    JobStatus.PENDING.value,541                ),542            )543            return cursor.rowcount > 0544545    def update_verse_status(546        self,547        job_id: str,548        book: str,549        chapter: int,550        verse: int,551        status: JobStatus,552        error: str | None = None,553    ) -> None:554        """Update the status of a verse job.555556        Args:557            job_id: Parent job identifier.558            book: Book name.559            chapter: Chapter number.560            verse: Verse number.561            status: New status.562            error: Error message if failed.563        """564        with self._connection() as conn:565            updates = ["status = ?"]566            params: list[Any] = [status.value]567568            if status in (JobStatus.COMPLETED, JobStatus.FAILED):569                updates.append("completed_at = ?")570                params.append(datetime.utcnow().isoformat())571572            if error is not None:573                updates.append("last_error = ?")574                params.append(error)575576            params.extend([job_id, book, chapter, verse])577578            conn.execute(579                f"UPDATE verse_jobs SET {', '.join(updates)} "580                "WHERE job_id = ? AND book = ? AND chapter = ? AND verse = ?",581                params,582            )583584            # Update book counts585            if status == JobStatus.COMPLETED:586                conn.execute(587                    """588                    UPDATE book_jobs589                    SET completed_verses = completed_verses + 1590                    WHERE job_id = ? AND book = ?591                    """,592                    (job_id, book),593                )594            elif status == JobStatus.FAILED:595                conn.execute(596                    """597                    UPDATE book_jobs598                    SET failed_verses = failed_verses + 1599                    WHERE job_id = ? AND book = ?600                    """,601                    (job_id, book),602                )603604    def get_pending_verses(self, job_id: str, book: str) -> list[VerseJob]:605        """Get all pending verses for a book.606607        Args:608            job_id: Parent job identifier.609            book: Book name.610611        Returns:612            List of pending VerseJob instances.613        """614        with self._connection() as conn:615            rows = conn.execute(616                """617                SELECT * FROM verse_jobs618                WHERE job_id = ? AND book = ? AND status = ?619                ORDER BY chapter, verse620                """,621                (job_id, book, JobStatus.PENDING.value),622            ).fetchall()623624            return [self._row_to_verse_job(row) for row in rows]625626    def get_failed_verses(self, job_id: str, book: str) -> list[VerseJob]:627        """Get all failed verses for a book.628629        Args:630            job_id: Parent job identifier.631            book: Book name.632633        Returns:634            List of failed VerseJob instances.635        """636        with self._connection() as conn:637            rows = conn.execute(638                """639                SELECT * FROM verse_jobs640                WHERE job_id = ? AND book = ? AND status = ?641                ORDER BY chapter, verse642                """,643                (job_id, book, JobStatus.FAILED.value),644            ).fetchall()645646            return [self._row_to_verse_job(row) for row in rows]647648    def get_book_progress(self, job_id: str, book: str) -> dict[str, Any]:649        """Get progress statistics for a book.650651        Args:652            job_id: Parent job identifier.653            book: Book name.654655        Returns:656            Dictionary with progress statistics.657        """658        with self._connection() as conn:659            row = conn.execute(660                """661                SELECT662                    total_verses,663                    completed_verses,664                    failed_verses,665                    status666                FROM book_jobs667                WHERE job_id = ? AND book = ?668                """,669                (job_id, book),670            ).fetchone()671672            if not row:673                return {}674675            total = row["total_verses"]676            completed = row["completed_verses"]677            failed = row["failed_verses"]678            pending = total - completed - failed679680            return {681                "book": book,682                "status": row["status"],683                "total_verses": total,684                "completed_verses": completed,685                "failed_verses": failed,686                "pending_verses": pending,687                "progress_percentage": (completed / total * 100) if total > 0 else 0,688            }689690    def reset_orphaned_verses(self, job_id: str) -> int:691        """Reset verses that were in_progress but not completed (orphaned by crash).692693        Args:694            job_id: Parent job identifier.695696        Returns:697            Number of verses reset.698        """699        with self._connection() as conn:700            cursor = conn.execute(701                """702                UPDATE verse_jobs703                SET status = ?, worker_pid = NULL704                WHERE job_id = ? AND status = ?705                """,706                (JobStatus.PENDING.value, job_id, JobStatus.IN_PROGRESS.value),707            )708            return cursor.rowcount709710    def reset_failed_verses(self, job_id: str, book: str) -> int:711        """Reset failed verses to pending for retry.712713        Args:714            job_id: Parent job identifier.715            book: Book name.716717        Returns:718            Number of verses reset.719        """720        with self._connection() as conn:721            cursor = conn.execute(722                """723                UPDATE verse_jobs724                SET status = ?, worker_pid = NULL, last_error = NULL725                WHERE job_id = ? AND book = ? AND status = ?726                """,727                (JobStatus.PENDING.value, job_id, book, JobStatus.FAILED.value),728            )729730            # Update book counts731            reset_count = cursor.rowcount732            if reset_count > 0:733                conn.execute(734                    """735                    UPDATE book_jobs736                    SET failed_verses = failed_verses - ?737                    WHERE job_id = ? AND book = ?738                    """,739                    (reset_count, job_id, book),740                )741742            return reset_count743744    # -------------------------------------------------------------------------745    # Helper Methods746    # -------------------------------------------------------------------------747748    def _row_to_bible_job(self, row: sqlite3.Row) -> BibleJob:749        """Convert a database row to a BibleJob."""750        return BibleJob(751            job_id=row["job_id"],752            books=json.loads(row["books"]),753            status=JobStatus(row["status"]),754            config_snapshot=json.loads(row["config_snapshot"]),755            created_at=datetime.fromisoformat(row["created_at"]),756            started_at=(757                datetime.fromisoformat(row["started_at"])758                if row["started_at"]759                else None760            ),761            completed_at=(762                datetime.fromisoformat(row["completed_at"])763                if row["completed_at"]764                else None765            ),766            total_verses=row["total_verses"],767            completed_verses=row["completed_verses"],768            failed_verses=row["failed_verses"],769        )770771    def _row_to_book_job(self, row: sqlite3.Row) -> BookJob:772        """Convert a database row to a BookJob."""773        return BookJob(774            book=row["book"],775            status=JobStatus(row["status"]),776            total_verses=row["total_verses"],777            completed_verses=row["completed_verses"],778            failed_verses=row["failed_verses"],779            worker_pid=row["worker_pid"],780            started_at=(781                datetime.fromisoformat(row["started_at"])782                if row["started_at"]783                else None784            ),785            completed_at=(786                datetime.fromisoformat(row["completed_at"])787                if row["completed_at"]788                else None789            ),790            error_message=row["error_message"],791        )792793    def _row_to_verse_job(self, row: sqlite3.Row) -> VerseJob:794        """Convert a database row to a VerseJob."""795        return VerseJob(796            book=row["book"],797            chapter=row["chapter"],798            verse=row["verse"],799            status=JobStatus(row["status"]),800            attempts=row["attempts"],801            last_error=row["last_error"],802            worker_pid=row["worker_pid"],803            started_at=(804                datetime.fromisoformat(row["started_at"])805                if row["started_at"]806                else None807            ),808            completed_at=(809                datetime.fromisoformat(row["completed_at"])810                if row["completed_at"]811                else None812            ),813        )814815    # -------------------------------------------------------------------------816    # Cleanup Methods817    # -------------------------------------------------------------------------818819    def delete_job(self, job_id: str) -> bool:820        """Delete a job and all its associated data.821822        Args:823            job_id: Job identifier to delete.824825        Returns:826            True if job was deleted, False if job not found.827        """828        with self._connection() as conn:829            # Check if job exists830            existing = conn.execute(831                "SELECT 1 FROM bible_jobs WHERE job_id = ?",832                (job_id,),833            ).fetchone()834835            if not existing:836                return False837838            # Delete in order: verse_jobs -> book_jobs -> bible_jobs839            conn.execute("DELETE FROM verse_jobs WHERE job_id = ?", (job_id,))840            conn.execute("DELETE FROM book_jobs WHERE job_id = ?", (job_id,))841            conn.execute("DELETE FROM bible_jobs WHERE job_id = ?", (job_id,))842843            return True844845    def delete_completed_jobs(self, keep_latest: int = 1) -> int:846        """Delete completed jobs, optionally keeping the most recent ones.847848        Args:849            keep_latest: Number of most recent completed jobs to keep.850851        Returns:852            Number of jobs deleted.853        """854        with self._connection() as conn:855            # Get completed job IDs ordered by completion time856            rows = conn.execute(857                """858                SELECT job_id FROM bible_jobs859                WHERE status = ?860                ORDER BY completed_at DESC861                """,862                (JobStatus.COMPLETED.value,),863            ).fetchall()864865            # Skip the ones we want to keep866            jobs_to_delete = [row["job_id"] for row in rows[keep_latest:]]867868            if not jobs_to_delete:869                return 0870871            # Delete each job872            placeholders = ",".join("?" * len(jobs_to_delete))873            conn.execute(874                f"DELETE FROM verse_jobs WHERE job_id IN ({placeholders})",875                jobs_to_delete,876            )877            conn.execute(878                f"DELETE FROM book_jobs WHERE job_id IN ({placeholders})",879                jobs_to_delete,880            )881            conn.execute(882                f"DELETE FROM bible_jobs WHERE job_id IN ({placeholders})",883                jobs_to_delete,884            )885886            return len(jobs_to_delete)887888    def delete_jobs_older_than(self, days: int) -> int:889        """Delete jobs older than a specified number of days.890891        Args:892            days: Delete jobs created more than this many days ago.893894        Returns:895            Number of jobs deleted.896        """897        from datetime import timedelta898899        cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat()900901        with self._connection() as conn:902            # Get job IDs older than cutoff903            rows = conn.execute(904                """905                SELECT job_id FROM bible_jobs906                WHERE created_at < ?907                """,908                (cutoff,),909            ).fetchall()910911            jobs_to_delete = [row["job_id"] for row in rows]912913            if not jobs_to_delete:914                return 0915916            # Delete each job917            placeholders = ",".join("?" * len(jobs_to_delete))918            conn.execute(919                f"DELETE FROM verse_jobs WHERE job_id IN ({placeholders})",920                jobs_to_delete,921            )922            conn.execute(923                f"DELETE FROM book_jobs WHERE job_id IN ({placeholders})",924                jobs_to_delete,925            )926            conn.execute(927                f"DELETE FROM bible_jobs WHERE job_id IN ({placeholders})",928                jobs_to_delete,929            )930931            return len(jobs_to_delete)932933    def list_jobs(self, limit: int = 20) -> list[BibleJob]:934        """List all jobs, most recent first.935936        Args:937            limit: Maximum number of jobs to return.938939        Returns:940            List of BibleJob instances.941        """942        with self._connection() as conn:943            rows = conn.execute(944                """945                SELECT * FROM bible_jobs946                ORDER BY created_at DESC947                LIMIT ?948                """,949                (limit,),950            ).fetchall()951952            return [self._row_to_bible_job(row) for row in rows]953954    def get_database_stats(self) -> dict[str, Any]:955        """Get database statistics.956957        Returns:958            Dictionary with job counts and database size.959        """960        with self._connection() as conn:961            total_jobs = conn.execute(962                "SELECT COUNT(*) as count FROM bible_jobs"963            ).fetchone()["count"]964965            pending_jobs = conn.execute(966                "SELECT COUNT(*) as count FROM bible_jobs WHERE status = ?",967                (JobStatus.PENDING.value,),968            ).fetchone()["count"]969970            in_progress_jobs = conn.execute(971                "SELECT COUNT(*) as count FROM bible_jobs WHERE status = ?",972                (JobStatus.IN_PROGRESS.value,),973            ).fetchone()["count"]974975            completed_jobs = conn.execute(976                "SELECT COUNT(*) as count FROM bible_jobs WHERE status = ?",977                (JobStatus.COMPLETED.value,),978            ).fetchone()["count"]979980            failed_jobs = conn.execute(981                "SELECT COUNT(*) as count FROM bible_jobs WHERE status = ?",982                (JobStatus.FAILED.value,),983            ).fetchone()["count"]984985            total_verses = conn.execute(986                "SELECT COUNT(*) as count FROM verse_jobs"987            ).fetchone()["count"]988989        # Get file size990        db_size_bytes = self.db_path.stat().st_size if self.db_path.exists() else 0991        db_size_mb = db_size_bytes / (1024 * 1024)992993        return {994            "total_jobs": total_jobs,995            "pending_jobs": pending_jobs,996            "in_progress_jobs": in_progress_jobs,997            "completed_jobs": completed_jobs,998            "failed_jobs": failed_jobs,999            "total_verse_records": total_verses,1000            "database_size_mb": round(db_size_mb, 2),1001        }1002