Fix db.py connection handling and test isolation
db.py: - Add conn.isolation_level = None to get_conn() — fixes row_factory + autocommit conflict. row_factory disables implicit transactions, so we need explicit autocommit mode. - Remove all conn.commit() calls (unnecessary with autocommit) pyproject.toml: - Move pytest + pytest-asyncio to main dependencies (uv run pytest uses ephemeral env with main deps only) tests/test_db.py: - Fix test_upsert_user_updates_username to not chain upsert_user() calls in assert expressions (test isolation issue)
This commit is contained in:
@@ -9,8 +9,13 @@ DB_PATH = Path(__file__).parent / "jigaido.db"
|
||||
|
||||
|
||||
def get_conn() -> sqlite3.Connection:
|
||||
# isolation_level=None enables autocommit mode.
|
||||
# row_factory disables SQLite Python's implicit transaction management,
|
||||
# so we need explicit autocommit to make writes work correctly.
|
||||
conn = sqlite3.connect(DB_PATH, detect_types=sqlite3.PARSE_DECLTYPES)
|
||||
conn.isolation_level = None
|
||||
conn.execute("PRAGMA foreign_keys = ON")
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
|
||||
@@ -35,9 +40,7 @@ def upsert_user(telegram_user_id: int, username: str | None) -> int:
|
||||
RETURNING id""",
|
||||
(telegram_user_id, username),
|
||||
)
|
||||
result = cur.fetchone()
|
||||
conn.commit()
|
||||
return result[0]
|
||||
return cur.fetchone()["id"]
|
||||
|
||||
|
||||
def get_user_by_telegram_id(telegram_user_id: int) -> Optional[dict]:
|
||||
@@ -62,9 +65,7 @@ def upsert_group(telegram_chat_id: int, creator_user_id: int) -> int:
|
||||
RETURNING id""",
|
||||
(telegram_chat_id, creator_user_id),
|
||||
)
|
||||
result = cur.fetchone()
|
||||
conn.commit()
|
||||
return result[0]
|
||||
return cur.fetchone()["id"]
|
||||
|
||||
|
||||
def get_group(telegram_chat_id: int) -> Optional[dict]:
|
||||
@@ -82,7 +83,7 @@ def get_group_creator_user_id(group_id: int) -> Optional[int]:
|
||||
"SELECT creator_user_id FROM groups WHERE id = ?",
|
||||
(group_id,),
|
||||
).fetchone()
|
||||
return row[0] if row else None
|
||||
return row["creator_user_id"] if row else None
|
||||
|
||||
|
||||
# ── Group Admins ────────────────────────────────────────────────────────────
|
||||
@@ -95,7 +96,6 @@ def add_group_admin(group_id: int, user_id: int) -> bool:
|
||||
"INSERT INTO group_admins (group_id, user_id) VALUES (?, ?)",
|
||||
(group_id, user_id),
|
||||
)
|
||||
conn.commit()
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
return False
|
||||
@@ -108,7 +108,6 @@ def remove_group_admin(group_id: int, user_id: int) -> bool:
|
||||
"DELETE FROM group_admins WHERE group_id = ? AND user_id = ?",
|
||||
(group_id, user_id),
|
||||
)
|
||||
conn.commit()
|
||||
return cur.rowcount > 0
|
||||
|
||||
|
||||
@@ -155,9 +154,7 @@ def add_bounty(
|
||||
RETURNING id""",
|
||||
(group_id, created_by_user_id, informed_by_username, text, link, due_date_ts),
|
||||
)
|
||||
result = cur.fetchone()
|
||||
conn.commit()
|
||||
return result[0]
|
||||
return cur.fetchone()["id"]
|
||||
except sqlite3.IntegrityError as e:
|
||||
if "UNIQUE" in str(e) and "link" in str(e):
|
||||
raise ValueError(f"Link already exists in this group: {link}")
|
||||
@@ -204,7 +201,6 @@ def update_bounty(
|
||||
WHERE id = ?""",
|
||||
(text, link, due_date_ts, bounty_id),
|
||||
)
|
||||
conn.commit()
|
||||
return cur.rowcount > 0
|
||||
except sqlite3.IntegrityError as e:
|
||||
if "UNIQUE" in str(e) and "link" in str(e):
|
||||
@@ -215,7 +211,6 @@ def update_bounty(
|
||||
def delete_bounty(bounty_id: int) -> bool:
|
||||
with get_conn() as conn:
|
||||
cur = conn.execute("DELETE FROM bounties WHERE id = ?", (bounty_id,))
|
||||
conn.commit()
|
||||
return cur.rowcount > 0
|
||||
|
||||
|
||||
@@ -229,7 +224,6 @@ def track_bounty(user_id: int, bounty_id: int) -> bool:
|
||||
"INSERT INTO user_bounty_tracking (user_id, bounty_id) VALUES (?, ?)",
|
||||
(user_id, bounty_id),
|
||||
)
|
||||
conn.commit()
|
||||
return True
|
||||
except sqlite3.IntegrityError:
|
||||
return False
|
||||
@@ -241,7 +235,6 @@ def untrack_bounty(user_id: int, bounty_id: int) -> bool:
|
||||
"DELETE FROM user_bounty_tracking WHERE user_id = ? AND bounty_id = ?",
|
||||
(user_id, bounty_id),
|
||||
)
|
||||
conn.commit()
|
||||
return cur.rowcount > 0
|
||||
|
||||
|
||||
@@ -306,9 +299,8 @@ def log_reminder(user_id: int, bounty_id: int) -> None:
|
||||
"INSERT OR IGNORE INTO reminder_log (user_id, bounty_id) VALUES (?, ?)",
|
||||
(user_id, bounty_id),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def get_all_user_ids() -> list[int]:
|
||||
with get_conn() as conn:
|
||||
return [row[0] for row in conn.execute("SELECT telegram_user_id FROM users")]
|
||||
return [row["telegram_user_id"] for row in conn.execute("SELECT telegram_user_id FROM users")]
|
||||
|
||||
Reference in New Issue
Block a user