news-api / db.py
radames's picture
replace entries every hour,
d7974b7
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