")
if len(f.entries) > 0:
for entry in f.entries:
title = html.escape(entry.title)
buffer.write(
- f'
• {title} ({entry.time_ago()}) '
+ f'
• {title} ({entry.time_ago()}) '
)
else:
buffer.write("
No entries...")
diff --git a/cry/database.py b/cry/database.py
index 4bd37f6..c63b8f4 100644
--- a/cry/database.py
+++ b/cry/database.py
@@ -102,20 +102,24 @@ class Database:
return db
def get_property(self, prop: str, default=None) -> typing.Any:
- cursor = self.db.execute("SELECT value FROM properties WHERE name=?", (prop,))
- result = cursor.fetchone()
- if result is None:
- return default
- return result[0]
+ with self.db:
+ cursor = self.db.execute(
+ "SELECT value FROM properties WHERE name=?", (prop,)
+ )
+ result = cursor.fetchone()
+ if result is None:
+ return default
+ return result[0]
def set_property(self, prop: str, value):
- self.db.execute(
- """
- INSERT INTO properties (name, value) VALUES (?, ?)
- ON CONFLICT DO UPDATE SET value=excluded.value
- """,
- (prop, value),
- )
+ with self.db:
+ self.db.execute(
+ """
+ INSERT INTO properties (name, value) VALUES (?, ?)
+ ON CONFLICT DO UPDATE SET value=excluded.value
+ """,
+ (prop, value),
+ )
def ensure_database_schema(self):
with self.db:
@@ -138,67 +142,135 @@ class Database:
self.set_property("origin", self.origin)
def load_all_meta(self) -> list[feed.FeedMeta]:
- cursor = self.db.execute(
- """
- SELECT
- url,
- last_fetched_ts,
- retry_after_ts,
- status,
- etag,
- modified
- FROM feeds
- """
- )
- rows = cursor.fetchall()
- return [
- feed.FeedMeta(
- url=url,
- last_fetched_ts=int(last_fetched_ts),
- retry_after_ts=int(retry_after_ts),
- status=int(status),
- etag=etag,
- modified=modified,
- origin=self.origin,
+ with self.db:
+ cursor = self.db.execute(
+ """
+ SELECT
+ url,
+ last_fetched_ts,
+ retry_after_ts,
+ status,
+ etag,
+ modified
+ FROM feeds
+ """
)
- for url, last_fetched_ts, retry_after_ts, status, etag, modified in rows
- ]
+ rows = cursor.fetchall()
+ return [
+ feed.FeedMeta(
+ url=url,
+ last_fetched_ts=int(last_fetched_ts),
+ retry_after_ts=int(retry_after_ts),
+ status=int(status),
+ etag=etag,
+ modified=modified,
+ origin=self.origin,
+ )
+ for url, last_fetched_ts, retry_after_ts, status, etag, modified in rows
+ ]
def load_all(self, feed_limit: int = 20, pattern: str = "") -> list[feed.Feed]:
- pattern = pattern.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
- sql_pattern = f"%{pattern}%"
- cursor = self.db.execute(
- """
- SELECT
- url,
- last_fetched_ts,
- retry_after_ts,
- status,
- etag,
- modified,
- title,
- link
- FROM feeds
- WHERE (title LIKE :sql_pattern ESCAPE '\\'
- OR link LIKE :sql_pattern ESCAPE '\\')
- AND status != 2 -- UNSUBSCRIBED
- """,
- {"sql_pattern": sql_pattern},
- )
- rows = cursor.fetchall()
+ with self.db:
+ pattern = (
+ pattern.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+ )
+ sql_pattern = f"%{pattern}%"
+ cursor = self.db.execute(
+ """
+ SELECT
+ url,
+ last_fetched_ts,
+ retry_after_ts,
+ status,
+ etag,
+ modified,
+ title,
+ link
+ FROM feeds
+ WHERE (title LIKE :sql_pattern ESCAPE '\\'
+ OR link LIKE :sql_pattern ESCAPE '\\')
+ AND status != 2 -- UNSUBSCRIBED
+ """,
+ {"sql_pattern": sql_pattern},
+ )
+ rows = cursor.fetchall()
- almost_feeds = []
- for row in rows:
- (
- url,
- last_fetched_ts,
- retry_after_ts,
- status,
- etag,
- modified,
- title,
- link,
- ) = row
+ almost_feeds = []
+ for row in rows:
+ (
+ url,
+ last_fetched_ts,
+ retry_after_ts,
+ status,
+ etag,
+ modified,
+ title,
+ link,
+ ) = row
+ meta = feed.FeedMeta(
+ url=url,
+ last_fetched_ts=last_fetched_ts,
+ retry_after_ts=retry_after_ts,
+ status=status,
+ etag=etag,
+ modified=modified,
+ origin=self.origin,
+ )
+ almost_feeds.append((meta, title, link))
+
+ feeds = []
+ for meta, title, link in almost_feeds:
+ if feed_limit > 0:
+ cursor = self.db.execute(
+ """
+ SELECT
+ id,
+ inserted_at,
+ title,
+ link
+ FROM entries
+ WHERE feed_url=?
+ ORDER BY inserted_at DESC
+ LIMIT ?
+ """,
+ [meta.url, feed_limit],
+ )
+ rows = cursor.fetchall()
+ else:
+ rows = []
+
+ entries = [
+ feed.Entry(id=id, inserted_at=inserted_at, title=title, link=link)
+ for id, inserted_at, title, link in rows
+ ]
+ f = feed.Feed(meta=meta, title=title, link=link, entries=entries)
+ feeds.append(f)
+
+ return feeds
+
+ def load_feed(self, url: str) -> feed.Feed | None:
+ with self.db:
+ cursor = self.db.execute(
+ """
+ SELECT
+ last_fetched_ts,
+ retry_after_ts,
+ status,
+ etag,
+ modified,
+ title,
+ link
+ FROM feeds
+ WHERE url=?
+ """,
+ [url],
+ )
+
+ row = cursor.fetchone()
+ if row is None:
+ return None
+
+ last_fetched_ts, retry_after_ts, status, etag, modified, title, link = row
meta = feed.FeedMeta(
url=url,
last_fetched_ts=last_fetched_ts,
@@ -208,88 +280,25 @@ class Database:
modified=modified,
origin=self.origin,
)
- almost_feeds.append((meta, title, link))
- feeds = []
- for meta, title, link in almost_feeds:
- if feed_limit > 0:
- cursor = self.db.execute(
- """
- SELECT
- id,
- inserted_at,
- title,
- link
- FROM entries
- WHERE feed_url=?
- ORDER BY inserted_at DESC
- LIMIT ?
- """,
- [meta.url, feed_limit],
- )
- rows = cursor.fetchall()
- else:
- rows = []
+ cursor = self.db.execute(
+ """
+ SELECT
+ id,
+ inserted_at,
+ title,
+ link
+ FROM entries
+ WHERE feed_url=?
+ """,
+ [url],
+ )
+ rows = cursor.fetchall()
entries = [
feed.Entry(id=id, inserted_at=inserted_at, title=title, link=link)
for id, inserted_at, title, link in rows
]
- f = feed.Feed(meta=meta, title=title, link=link, entries=entries)
- feeds.append(f)
-
- return feeds
-
- def load_feed(self, url: str) -> feed.Feed | None:
- cursor = self.db.execute(
- """
- SELECT
- last_fetched_ts,
- retry_after_ts,
- status,
- etag,
- modified,
- title,
- link
- FROM feeds
- WHERE url=?
- """,
- [url],
- )
-
- row = cursor.fetchone()
- if row is None:
- return None
-
- last_fetched_ts, retry_after_ts, status, etag, modified, title, link = row
- meta = feed.FeedMeta(
- url=url,
- last_fetched_ts=last_fetched_ts,
- retry_after_ts=retry_after_ts,
- status=status,
- etag=etag,
- modified=modified,
- origin=self.origin,
- )
-
- cursor = self.db.execute(
- """
- SELECT
- id,
- inserted_at,
- title,
- link
- FROM entries
- WHERE feed_url=?
- """,
- [url],
- )
-
- rows = cursor.fetchall()
- entries = [
- feed.Entry(id=id, inserted_at=inserted_at, title=title, link=link)
- for id, inserted_at, title, link in rows
- ]
return feed.Feed(meta=meta, title=title, link=link, entries=entries)