db: one request-scoped database handle per API request (#773)

* db: one request-scoped database handle per API request

Every API request now gets `request.ctx.db` (`wrolpi.db.RequestDB`): a Session plus raw cursors on that
Session's connection, so a request needs at most one SQLite connection.  Creating it does no I/O; requests
that never touch the database never connect.  `request.ctx.session` remains the handle's Session.

A `RequestSession` binds itself to one lazily-opened Connection.  An engine-bound Session releases its
connection at every commit, and NullPool closes it, so a request that read and then wrote reconnected.

Handlers write in `request.ctx.db.write()`: BEGIN IMMEDIATE, committed when the block exits, so the
write lock is not held for the rest of the handler or by background work it starts.  This replaces the
handler-level `get_db_session(commit=True)` blocks (download create, collection item add/remove/reorder,
channel create).  The zim search handlers use the request's Session in their worker thread.

Tests count the connections a request opens with production's per-call sessions
(`count_request_connections`).  Endpoints whose library functions still open their own connection are
strict xfails, to be removed as those functions take the request's session.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* tests: name the real extra connections in the request connection xfails

Channel detail: get_channel already uses the request session; the extra connection is
Channel.get_statistics opening a raw cursor (get_db_curs).  Search: count_file_groups opens a
cursor for the total, then handle_file_group_search_results opens a cursor and a session.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

modules/videos/channel/api.py

@@ -9,7 +9,6 @@ from sanic_ext.extensions.openapi import openapi
 
 from wrolpi.api_utils import json_response
 from wrolpi.common import logger, wrol_mode_check, get_media_directory, get_relative_to_media_directory
-from wrolpi.db import get_db_session
 from wrolpi.events import Events
 from wrolpi.schema import JSONErrorResponse
 from wrolpi.vars import PYTEST
@@ -54,12 +53,12 @@ def channel_get(request: Request, channel_id: int = None):
 def channel_post(request: Request, body: schema.ChannelPostRequest):
     body.directory = get_media_directory() / body.directory
 
-    # Not `request.ctx.session`: creating a Channel reads (the URL/name/directory conflict check)
-    # before it INSERTs, and a deferred transaction fails that lock upgrade *instantly* with
-    # "database is locked" if anything else wrote in between -- which is what a user creating a
-    # Channel while background workers saved configs hit.  This session also commits before the
-    # refresh below is fired, so the refresh does not wait on a lock this request still holds.
-    with get_db_session(commit=True) as session:
+    # A write block: creating a Channel reads (the URL/name/directory conflict check) before it
+    # INSERTs, and a deferred transaction fails that lock upgrade *instantly* with "database is
+    # locked" if anything else wrote in between -- which is what a user creating a Channel while
+    # background workers saved configs hit.  The block also commits before the refresh below is
+    # fired, so the refresh does not wait on a lock this request still holds.
+    with request.ctx.db.write() as session:
         channel = lib.create_channel(session, data=body, return_dict=False)
         channel_id, channel_name = channel.id, channel.name
 

modules/zim/api.py

@@ -9,7 +9,6 @@ from sanic_ext.extensions.openapi import openapi
 from wrolpi import lang
 from wrolpi.api_utils import json_response
 from wrolpi.common import logger
-from wrolpi.db import get_db_session
 from wrolpi.downloader import download_manager
 from wrolpi.events import Events
 from . import lib, schema
@@ -65,13 +64,12 @@ async def post_set_zim_auto_search(request: Request, zim_id: int, body: schema.Z
     body=schema.ZimSearchRequest,
 )
 @validate(schema.ZimSearchRequest)
-async def search_all_zims(_: Request, body: schema.ZimSearchRequest):
-    # libzim searches are synchronous and slow; run off the event loop.  The session is opened in the
-    # worker thread: a SQLAlchemy Session must not be shared across threads.
+async def search_all_zims(request: Request, body: schema.ZimSearchRequest):
+    # libzim searches are synchronous and slow; run off the event loop.  The request's Session is used
+    # in the worker thread; the handler awaits the thread, so the Session has one user at a time.
     def search():
-        with get_db_session() as session:
-            return lib.search_all_zims(session, body.search_str, tag_names=body.tag_names, offset=body.offset,
-                                       limit=body.limit)
+        return lib.search_all_zims(request.ctx.session, body.search_str, tag_names=body.tag_names,
+                                   offset=body.offset, limit=body.limit)
 
     results = await asyncio.to_thread(search)
     return json_response({'zims': results})
@@ -84,11 +82,10 @@ async def search_all_zims(_: Request, body: schema.ZimSearchRequest):
     response=schema.ZimSearchResponse,
 )
 @validate(schema.ZimSearchRequest)
-async def search_zim(_: Request, zim_id: int, body: schema.ZimSearchRequest):
+async def search_zim(request: Request, zim_id: int, body: schema.ZimSearchRequest):
     def search():
-        with get_db_session() as session:
-            return lib.headline_zim(session, body.search_str, zim_id, tag_names=body.tag_names, offset=body.offset,
-                                    limit=body.limit)
+        return lib.headline_zim(request.ctx.session, body.search_str, zim_id, tag_names=body.tag_names,
+                                offset=body.offset, limit=body.limit)
 
     headlines = await asyncio.to_thread(search)
     return json_response({'zim': headlines})
@@ -219,17 +216,14 @@ async def post_search_estimates(request: Request, body: schema.SearchEstimateReq
         return response.empty(HTTPStatus.BAD_REQUEST)
 
     def estimate():
-        zims_estimates = list()
-        with get_db_session() as session:
-            if body.tag_names:
-                # Get actual count of entries tagged with the tag names.
-                counts = Zims.entries_with_tags(session, body.tag_names)
-            else:
-                # Get estimates using libzim.
-                counts = Zims.estimate(session, body.search_str)
-            for zim, count in counts.items():
-                zims_estimates.append(dict(estimate=count, **zim.__json__()))
-        return zims_estimates
+        session = request.ctx.session
+        if body.tag_names:
+            # Get actual count of entries tagged with the tag names.
+            counts = Zims.entries_with_tags(session, body.tag_names)
+        else:
+            # Get estimates using libzim.
+            counts = Zims.estimate(session, body.search_str)
+        return [dict(estimate=count, **zim.__json__()) for zim, count in counts.items()]
 
     zims_estimates = await asyncio.to_thread(estimate)