fenen/database.py

import sqlite3
from hashlib import sha256


class Database:
    db = None
    c = None

    def __init__(self, dbfile):
        self.db = sqlite3.connect(dbfile, timeout=30)
        self.db.row_factory = sqlite3.Row
        self.c = self.db.cursor()

    def init_db(self):
        with self.db:
            self.c.execute(
                "CREATE TABLE IF NOT EXISTS feeds "
                "(url TEXT PRIMARY KEY, name TEXT COLLATE NOCASE, "
                "custom_name INTEGER, site_url TEXT, "
                "category TEXT COLLATE NOCASE, table_name TEXT)"
            )

    def insert_feed(self, data, from_opml=False):
        with self.db:
            table_name = "table" + sha256(data[0].encode()).hexdigest()
            data.append(table_name)
            self.c.execute(
                f"CREATE TABLE IF NOT EXISTS {table_name} "
                "(id TEXT PRIMARY KEY, date DATE, title TEXT, "
                "url TEXT, content TEXT, unread INTEGER)"
            )
            if from_opml:
                self.c.execute(
                    "INSERT OR IGNORE INTO feeds VALUES (?, ?, ?, ?, ?, ?)",
                    tuple(data),
                )
            else:
                self.c.execute(
                    "INSERT OR IGNORE INTO feeds VALUES (?, NULL, NULL, NULL, NULL, ?)",
                    tuple(data),
                )

    def update_feed(self, name, category, url):
        with self.db:
            if name:
                self.c.execute(
                    "UPDATE feeds SET name = ?, custom_name = 1 WHERE url = ?",
                    (name, url),
                )
            if category:
                self.c.execute(
                    "UPDATE feeds SET category = ? WHERE url = ?",
                    (category, url),
                )

    def get_all_feeds(self):
        return self.c.execute(
            "SELECT * FROM feeds ORDER BY category, name, url"
        ).fetchall()

    def get_feed_by_url(self, url):
        return self.c.execute("SELECT * FROM feeds WHERE url = ?", (url,)).fetchone()

    def get_feed_by_index(self, index):
        return self.c.execute(
            f"SELECT * FROM feeds ORDER BY category, name, url LIMIT 1 OFFSET {index}"
        ).fetchone()

    def get_all_entries(self, table):
        return self.c.execute(f"SELECT * FROM {table} ORDER BY date").fetchall()

    def get_entry_by_index(self, table, index):
        return self.c.execute(
            f"SELECT * FROM {table} ORDER BY date LIMIT 1 OFFSET {index}"
        ).fetchone()

    def get_entry_by_id(self, table, id):
        return self.c.execute(f"SELECT * FROM {table} WHERE id = ?", (id,)).fetchone()

    def get_unread_count(self, table):
        return self.c.execute(
            f"SELECT COUNT(unread) FROM {table} WHERE unread = 1"
        ).fetchone()[0]

    def get_unread_entries(self, table):
        return self.c.execute(f"SELECT * FROM {table} WHERE unread = 1").fetchall()

    def get_num_feeds(self):
        return self.c.execute("SELECT COUNT(1) FROM feeds").fetchone()[0]

    def get_num_entries(self, table):
        return self.c.execute(f"SELECT COUNT(1) FROM {table}").fetchone()[0]

    def search_feeds(self, query):
        return self.c.execute(
            "SELECT * FROM feeds WHERE url LIKE ? OR name LIKE ? ORDER BY category, name, url",
            (query, query),
        ).fetchall()

    def search_entries(self, table, query):
        return self.c.execute(
            f"SELECT * FROM {table} WHERE "
            "date LIKE ? OR title LIKE ? OR url LIKE ? ORDER BY date ASC",
            (query, query, query),
        ).fetchall()

    def feed_missing_name(self, url):
        if self.c.execute(
            "SELECT url FROM feeds WHERE url = ? AND name IS NULL", (url,)
        ).fetchone():
            return True
        return False

    def feed_missing_site_url(self, url):
        if self.c.execute(
            "SELECT url FROM feeds WHERE url = ? AND site_url IS NULL", (url,)
        ).fetchone():
            return True
        return False

    def populate_feed_name(self, url, name):
        with self.db:
            self.c.execute(
                "UPDATE feeds SET name = ?, custom_name = 0 WHERE url = ?", (name, url)
            )

    def populate_feed_site_url(self, url, site_url):
        with self.db:
            self.c.execute(
                "UPDATE feeds SET site_url = ? WHERE url = ?", (site_url, url)
            )

    def remove_feeds(self, urls, tables):
        urls = [(url,) for url in urls]
        with self.db:
            self.c.executemany("DELETE FROM feeds WHERE url = ?", urls)
            for table in tables:
                self.c.execute(f"DROP TABLE {table}")

    def change_unread_status_of_feeds(self, feeds, unread_status):
        with self.db:
            for feed in feeds:
                self.c.execute(
                    f'UPDATE {feed["table_name"]} SET unread = ?', (unread_status,)
                )

    def change_unread_status_of_entries(self, entries, unread_status):
        with self.db:
            for entry in entries:
                self.c.execute(
                    f'UPDATE {entry["table"]} SET unread = ? WHERE id = ?',
                    (unread_status, entry["id"]),
                )

    def insert_entries(self, table, entries):
        with self.db:
            try:
                self.c.executemany(
                    f"INSERT INTO {table} VALUES (?, ?, ?, ?, ?, 1)", entries
                )
            except sqlite3.IntegrityError:
                pass  # This is likely just a duplicate entry i.e. unique
                # constraint failed on table<n>.id which doesn't matter

    def update_entries(self, table, entries):
        with self.db:
            self.c.executemany(
                f"UPDATE {table} SET title = ?, url = ?, content = ? WHERE id = ?",
                entries,
            )

    def vacuum(self):
        with self.db:
            self.c.execute("VACUUM")