db.py

import sqlite3
from pathlib import Path
from typing import List, Tuple
import json


class Database:
    def __init__(self, db_path=None):
        if db_path is None:
            raise ValueError("db_path must be provided")
        self.db_path = db_path
        self.db_file = self.db_path / "cache.db"
        if not self.db_file.exists():
            print("Creating database")
            print("DB_FILE", self.db_file)
            db = sqlite3.connect(self.db_file)
            with open(Path("schema.sql"), "r") as f:
                db.executescript(f.read())
            db.commit()
            db.close()

    def get_db(self):
        db = sqlite3.connect(self.db_file, check_same_thread=False)
        db.row_factory = sqlite3.Row
        return db

    def __enter__(self):
        self.db = self.get_db()
        return self.db

    def __exit__(self, exc_type, exc_value, traceback):
        self.db.close()

    def __call__(self):
        return self

    def insert(self, data: List[Tuple[str, str, str]]):
        with self() as db:
            cursor = db.cursor()
            try:
                for entry in data:
                    url, title, entries = entry
                    cursor.execute(
                        "INSERT OR REPLACE INTO cache (url, title, entries) VALUES (?, ?, ?)",
                        (url, title, entries),
                    )
            except Exception as e:
                print(e)
            db.commit()

    def filter(self, category: str):
        with self() as db:
            entries = db.execute("SELECT url, title, entries FROM cache").fetchall()
            out = []
            for row in entries:
                # parse json
                data = json.loads(row["entries"])
                try:
                    data = [
                        {
                            "title": entry["title"],
                            "link": entry["link"],
                            "published": entry["published"]
                            if "published" in entry
                            else entry["pubDate"]
                            if "pubDate" in entry
                            else "",
                            "summary": entry["summary"] if "summary" in entry else "",
                        }
                        for entry in data["entries"]
                    ]
                    if len(data) > 0:
                        out.append(
                            {
                                "entries": data,
                                "url": row["url"],
                                "title": row["title"],
                            }
                        )
                except Exception as e:
                    print(f"Errro on {row['url']}: {e}")
            return out