()
| 389 | |
| 390 | |
| 391 | def connect() -> sqlite3.Connection: |
| 392 | path = database_path() |
| 393 | path.parent.mkdir(parents=True, exist_ok=True) |
| 394 | for attempt in range(SQLITE_RETRY_ATTEMPTS): |
| 395 | connection = sqlite3.connect(path, timeout=5) |
| 396 | try: |
| 397 | connection.row_factory = sqlite3.Row |
| 398 | connection.execute("PRAGMA foreign_keys = ON") |
| 399 | connection.execute("PRAGMA busy_timeout = 5000") |
| 400 | apply_migrations(connection) |
| 401 | connection.execute("PRAGMA journal_mode = WAL") |
| 402 | path.chmod(0o600) |
| 403 | return connection |
| 404 | except sqlite3.OperationalError as exc: |
| 405 | connection.close() |
| 406 | if attempt == SQLITE_RETRY_ATTEMPTS - 1 or not sqlite_busy(exc): |
| 407 | raise |
| 408 | time.sleep(0.05 * (2**attempt)) |
| 409 | raise AssertionError("SQLite retry loop exhausted unexpectedly.") |
| 410 | |
| 411 | |
| 412 | def sqlite_busy(error: sqlite3.OperationalError) -> bool: |
no test coverage detected