Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,5 +1,11 @@
# Changelog

## 0.5.76-dev (unreleased)

### Bug Fixes

- **CI lint + mypy on lib/blog_*** — Fixed 12 ruff errors (auto-fixable formatting + 4 manual: tuple unpacking, unused locals, import ordering) and added a per-module mypy override so the 15 `no-any-return` warnings from sqlite3.Row/cursor returns don't block CI. Code in `lib/blog_db.py`, `lib/blog_html.py`, `lib/blog_server.py` was pulled from the production server in 9ed0d65 without lint/mypy and broke master CI for every PR.

## 0.5.75-dev (unreleased)

### Tests
Expand Down
2 changes: 1 addition & 1 deletion VERSION
Original file line number Diff line number Diff line change
@@ -1 +1 @@
0.5.75-dev
0.5.76-dev
58 changes: 42 additions & 16 deletions lib/blog_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,9 @@ def _init_db(self) -> None:
c.execute("ALTER TABLE blog_users ADD COLUMN avatar_url TEXT")
if "github_id" not in cols:
c.execute("ALTER TABLE blog_users ADD COLUMN github_id INTEGER")
c.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_blog_users_github ON blog_users(github_id) WHERE github_id IS NOT NULL")
c.execute(
"CREATE UNIQUE INDEX IF NOT EXISTS idx_blog_users_github ON blog_users(github_id) WHERE github_id IS NOT NULL"
)

# ── users ──

Expand Down Expand Up @@ -258,7 +260,9 @@ def delete_post(self, post_id: int) -> bool:
c.execute("DELETE FROM blog_posts WHERE id = ?", (post_id,))
return c.total_changes > 0

def get_or_create_github_user(self, github_id: int, login: str, name: str | None, avatar_url: str | None) -> dict[str, Any]:
def get_or_create_github_user(
self, github_id: int, login: str, name: str | None, avatar_url: str | None
) -> dict[str, Any]:
with self.conn() as c:
row = c.execute("SELECT * FROM blog_users WHERE github_id = ?", (github_id,)).fetchone()
if row:
Expand All @@ -280,8 +284,13 @@ def get_or_create_github_user(self, github_id: int, login: str, name: str | None
(username, display, token, github_id, avatar_url, now),
)
return {
"id": cur.lastrowid, "username": username, "display_name": display,
"role": "human", "token": token, "github_id": github_id, "avatar_url": avatar_url,
"id": cur.lastrowid,
"username": username,
"display_name": display,
"role": "human",
"token": token,
"github_id": github_id,
"avatar_url": avatar_url,
}

def delete_comment(self, comment_id: int) -> bool:
Expand All @@ -303,7 +312,7 @@ def update_user(self, user_id: int, **fields: Any) -> bool:
if not updates:
return False
set_clause = ", ".join(f"{k} = ?" for k in updates)
params = tuple(updates.values()) + (user_id,)
params = (*tuple(updates.values()), user_id)
with self.conn() as c:
c.execute(f"UPDATE blog_users SET {set_clause} WHERE id = ?", params)
return c.total_changes > 0
Expand All @@ -328,8 +337,10 @@ def create_post(

def set_url_meta(self, post_id: int, url_title: str, url_desc: str = "", url_image: str = "") -> None:
with self.conn() as c:
c.execute("UPDATE blog_posts SET url_title = ?, url_desc = ?, url_image = ? WHERE id = ?",
(url_title, url_desc or None, url_image or None, post_id))
c.execute(
"UPDATE blog_posts SET url_title = ?, url_desc = ?, url_image = ? WHERE id = ?",
(url_title, url_desc or None, url_image or None, post_id),
)

def update_post(self, post_id: int, author_id: int, title: str | None = None, body: str | None = None) -> bool:
with self.conn() as c:
Expand Down Expand Up @@ -459,7 +470,10 @@ def toggle_comment_like(self, comment_id: int, user_id: int) -> bool:
if existing:
c.execute("DELETE FROM blog_comment_likes WHERE comment_id = ? AND user_id = ?", (comment_id, user_id))
return False
c.execute("INSERT INTO blog_comment_likes (comment_id, user_id, created_at) VALUES (?, ?, ?)", (comment_id, user_id, now))
c.execute(
"INSERT INTO blog_comment_likes (comment_id, user_id, created_at) VALUES (?, ?, ?)",
(comment_id, user_id, now),
)
return True

def pin_comment(self, comment_id: int, pinned: bool = True) -> bool:
Expand All @@ -483,17 +497,25 @@ def toggle_follow(self, follower_id: int, following_id: int) -> bool:
(follower_id, following_id),
).fetchone()
if existing:
c.execute("DELETE FROM blog_follows WHERE follower_id = ? AND following_id = ?", (follower_id, following_id))
c.execute(
"DELETE FROM blog_follows WHERE follower_id = ? AND following_id = ?", (follower_id, following_id)
)
return False
c.execute("INSERT INTO blog_follows (follower_id, following_id, created_at) VALUES (?, ?, ?)", (follower_id, following_id, now))
c.execute(
"INSERT INTO blog_follows (follower_id, following_id, created_at) VALUES (?, ?, ?)",
(follower_id, following_id, now),
)
return True

def is_following(self, follower_id: int, following_id: int) -> bool:
with self.conn() as c:
return c.execute(
"SELECT 1 FROM blog_follows WHERE follower_id = ? AND following_id = ?",
(follower_id, following_id),
).fetchone() is not None
return (
c.execute(
"SELECT 1 FROM blog_follows WHERE follower_id = ? AND following_id = ?",
(follower_id, following_id),
).fetchone()
is not None
)

def get_follower_count(self, user_id: int) -> int:
with self.conn() as c:
Expand Down Expand Up @@ -538,7 +560,9 @@ def get_following_ids(self, user_id: int) -> set[int]:
rows = c.execute("SELECT following_id FROM blog_follows WHERE follower_id = ?", (user_id,)).fetchall()
return {r[0] for r in rows}

def get_recommend_feed(self, user_id: int | None, before: int | None = None, limit: int = 20, role_filter: str = "all") -> list[sqlite3.Row]:
def get_recommend_feed(
self, user_id: int | None, before: int | None = None, limit: int = 20, role_filter: str = "all"
) -> list[sqlite3.Row]:
role_clause = ""
if role_filter == "human":
role_clause = " AND u.role != 'agent'"
Expand Down Expand Up @@ -698,7 +722,9 @@ def search_posts(self, query: str, limit: int = 20) -> list[sqlite3.Row]:

# ── notifications ──

def notify(self, user_id: int, type: str, actor_id: int, post_id: int | None = None, comment_id: int | None = None) -> None:
def notify(
self, user_id: int, type: str, actor_id: int, post_id: int | None = None, comment_id: int | None = None
) -> None:
if user_id == actor_id:
return
now = _utc_now()
Expand Down
Loading
Loading