fastware v0.6.0 /src.fastware.error_log
Updated
On this page

SQLite-backed, thread-safe append-only error log that records 5xx responses with request context and exposes a recent() query.

#src.fastware.error_log

#src.fastware.error_log

SQLite-backed error log for recording and querying 5xx server responses with request context, tracebacks, and timestamps for post-mortem analysis.

Provides a simple append-only store that the request timing middleware can write to on server errors. Each entry captures enough context for dashboard display and post-mortem investigation.

Usage::

from fastware.error_log import ErrorLog

error_log = ErrorLog("errors.db") error_log.append( method="POST", path="/api/deploy", status_code=500, detail="Docker timeout", request_id="abc-123", )

#ErrorLog

Append-only SQLite error log with non-blocking writes.

append() never touches SQLite on the calling thread. Instead it enqueues the entry and a single dedicated background worker thread performs the connect/INSERT/commit off the event loop. This keeps the ASGI event loop responsive even during 5xx bursts, when the request timing middleware calls append() on every failing request.

Ordering and durability:

  • A single FIFO queue and a single writer thread preserve insertion

order. The timestamp is captured at append() time so it reflects the true event order regardless of write latency.

  • recent() (and the explicit flush()) drain the queue before

reading, so reads always observe every preceding append().

  • Errors raised by the writer (e.g. a bad path) are surfaced -- not

swallowed -- the next time flush()/recent() is called.

Args:

  • path: Filesystem path for the SQLite database. Created on

first write if it does not exist.

#_run_worker

python
def _run_worker(self) -> None

Drain the queue, writing each entry with one long-lived connection.

#append

python
def append(self, *, method: str, path: str, status_code: int, detail: str='', request_id: str='', user: str='', traceback: str='') -> None

Enqueue an error entry for non-blocking, off-thread persistence.

Returns immediately; the actual SQLite write happens on the background worker thread. The timestamp is captured here so ordering reflects the true event order regardless of write latency.

#flush

python
def flush(self) -> None

Block until all queued writes have been committed.

Raises any error the background writer encountered while persisting entries, so failures are surfaced rather than silently dropped.

#recent

python
def recent(self, limit: int=50) -> list[dict]

Return the most recent limit error entries, newest first.

Pending queued writes are flushed first, so the result reflects every preceding append().

#close

python
def close(self) -> None

Gracefully stop the background writer after draining pending writes.

Search