refactor: tighten exception handling in database.py and add BLE lint rule

- Wrap duckdb.connect() in try/except with specific duckdb.Error
- Replace bare  with  in _init_database
- Replace broad  with  for ALTER TABLE
- Add ruff BLE (blind except) lint rule to prevent regressions
- Add tests verifying graceful error handling for connect, insert, query

P2-002: Fix broad exception handling
main
Sven Geboers 4 weeks ago
parent c85a367a8e
commit 04cc62ea06
  1. 20
      database.py
  2. 2
      pyproject.toml
  3. 38
      tests/test_database_exceptions.py

@ -39,12 +39,17 @@ class MotionDatabase:
fh.write("[]") fh.write("[]")
return return
conn = duckdb.connect(self.db_path) try:
conn = duckdb.connect(self.db_path)
except (duckdb.Error, OSError) as e:
_logger.warning("Could not connect to DuckDB at %s: %s. Operating in file mode.", self.db_path, e)
self._file_mode = True
return
# Create sequence for auto-incrementing IDs # Create sequence for auto-incrementing IDs
try: try:
conn.execute("CREATE SEQUENCE IF NOT EXISTS motions_id_seq START 1") conn.execute("CREATE SEQUENCE IF NOT EXISTS motions_id_seq START 1")
except: except duckdb.Error:
pass pass
# Create tables with proper ID handling # Create tables with proper ID handling
@ -72,7 +77,7 @@ class MotionDatabase:
"ALTER TABLE motions ADD COLUMN IF NOT EXISTS externe_identifier TEXT" "ALTER TABLE motions ADD COLUMN IF NOT EXISTS externe_identifier TEXT"
) )
conn.execute("ALTER TABLE motions ADD COLUMN IF NOT EXISTS body_text TEXT") conn.execute("ALTER TABLE motions ADD COLUMN IF NOT EXISTS body_text TEXT")
except Exception: except duckdb.Error:
# Best-effort: if ALTER fails for any reason, continue without stopping app startup # Best-effort: if ALTER fails for any reason, continue without stopping app startup
_logger.debug( _logger.debug(
"Could not ALTER motions table to add new columns (may already exist or unsupported)." "Could not ALTER motions table to add new columns (may already exist or unsupported)."
@ -188,7 +193,10 @@ class MotionDatabase:
) )
""") """)
conn.close() try:
conn.close()
except duckdb.Error:
pass
def reset_database(self): def reset_database(self):
"""Development helper: drop known tables and re-run initialization. """Development helper: drop known tables and re-run initialization.
@ -201,7 +209,7 @@ class MotionDatabase:
for t in ("party_results", "user_sessions", "motions"): for t in ("party_results", "user_sessions", "motions"):
try: try:
conn.execute(f"DROP TABLE IF EXISTS {t}") conn.execute(f"DROP TABLE IF EXISTS {t}")
except Exception: except duckdb.Error:
pass pass
# Recreate schema # Recreate schema
conn.close() conn.close()
@ -209,7 +217,7 @@ class MotionDatabase:
finally: finally:
try: try:
conn.close() conn.close()
except Exception: except duckdb.Error:
pass pass
def append_audit_event( def append_audit_event(

@ -28,7 +28,7 @@ dev = [
] ]
[tool.ruff.lint] [tool.ruff.lint]
select = ["T20"] select = ["T20", "BLE"]
ignore = [] ignore = []
[tool.ruff.lint.per-file-ignores] [tool.ruff.lint.per-file-ignores]

@ -0,0 +1,38 @@
import logging
from unittest.mock import MagicMock, patch
import pytest
from database import MotionDatabase
class TestMotionDatabaseExceptionHandling:
@patch("database.duckdb")
def test_init_database_catches_duckdb_connect_errors(self, mock_duckdb, tmp_path):
mock_duckdb.Error = Exception
mock_duckdb.connect.side_effect = mock_duckdb.Error("db locked")
db = MotionDatabase(db_path=str(tmp_path / "test.db"))
assert db._file_mode is True
@patch("database.duckdb")
def test_insert_motion_catches_errors_and_returns_false(self, mock_duckdb, tmp_path):
mock_duckdb.Error = Exception
mock_conn = MagicMock()
mock_duckdb.connect.return_value = mock_conn
db = MotionDatabase(db_path=str(tmp_path / "test.db"))
# Now make insert_motion fail
mock_conn.execute.side_effect = mock_duckdb.Error("constraint violation")
result = db.insert_motion({"title": "Test", "date": "2024-01-01", "url": "http://test"})
assert result is False
@patch("database.duckdb")
def test_query_motions_catches_errors_and_returns_empty_list(self, mock_duckdb, tmp_path):
mock_duckdb.Error = Exception
mock_conn = MagicMock()
mock_duckdb.connect.return_value = mock_conn
db = MotionDatabase(db_path=str(tmp_path / "test.db"))
mock_conn.execute.side_effect = mock_duckdb.Error("syntax error")
result = db.get_filtered_motions()
assert result == []
Loading…
Cancel
Save