initial: blog-app snapshot

repositories/release_repo.py

@@ -0,0 +1,94 @@
+"""Release queries. No Flask imports; take an explicit sqlite3 connection.
+
+Queries filter on `releases.project_id`, which is backed by the
+`idx_releases_project_id` index (Stage 4.2; docs/analysis/optional-indexes.sql).
+"""
+
+import config
+
+RELEASE_COLUMNS = (
+    "id, project_id, tag_name, name, body, published_at, html_url"
+)
+
+# LLM context rows never need more than LLM_BODY_LIMIT body chars, so the
+# body is truncated in SQL and the full body never leaves the database.
+SLIM_RELEASE_COLUMNS = (
+    f"id, project_id, tag_name, name, "
+    f"substr(body, 1, {config.LLM_BODY_LIMIT}) AS body, published_at, html_url"
+)
+
+
+def _rows(cur):
+    return [dict(row) for row in cur.fetchall()]
+
+
+def list_recent_releases(db, limit=5):
+    return _rows(
+        db.execute(
+            f"SELECT {RELEASE_COLUMNS} FROM releases "
+            "ORDER BY published_at DESC LIMIT ?",
+            (limit,),
+        )
+    )
+
+
+def list_recent_releases_slim(db, limit=5):
+    return _rows(
+        db.execute(
+            f"SELECT {SLIM_RELEASE_COLUMNS} FROM releases "
+            "ORDER BY published_at DESC LIMIT ?",
+            (limit,),
+        )
+    )
+
+
+def list_releases_for_project(db, project_id, limit=None):
+    sql = (
+        f"SELECT {RELEASE_COLUMNS} FROM releases "
+        "WHERE project_id = ? ORDER BY published_at DESC"
+    )
+    params = (project_id,)
+    if limit is not None:
+        sql += " LIMIT ?"
+        params = (project_id, limit)
+    return _rows(db.execute(sql, params))
+
+
+def list_releases_for_project_slim(db, project_id, limit=None):
+    sql = (
+        f"SELECT {SLIM_RELEASE_COLUMNS} FROM releases "
+        "WHERE project_id = ? ORDER BY published_at DESC"
+    )
+    params = (project_id,)
+    if limit is not None:
+        sql += " LIMIT ?"
+        params = (project_id, limit)
+    return _rows(db.execute(sql, params))
+
+
+def search_releases(db, keyword, limit=None):
+    like = f"%{keyword}%"
+    sql = (
+        f"SELECT {RELEASE_COLUMNS} FROM releases "
+        "WHERE tag_name LIKE ? OR name LIKE ? "
+        "ORDER BY published_at DESC"
+    )
+    params = (like, like)
+    if limit is not None:
+        sql += " LIMIT ?"
+        params = (like, like, limit)
+    return _rows(db.execute(sql, params))
+
+
+def search_releases_slim(db, keyword, limit=None):
+    like = f"%{keyword}%"
+    sql = (
+        f"SELECT {SLIM_RELEASE_COLUMNS} FROM releases "
+        "WHERE tag_name LIKE ? OR name LIKE ? "
+        "ORDER BY published_at DESC"
+    )
+    params = (like, like)
+    if limit is not None:
+        sql += " LIMIT ?"
+        params = (like, like, limit)
+    return _rows(db.execute(sql, params))