perf: run blocking route work in the thread pool

.claude/rules/routes.md

@@ -9,7 +9,8 @@ paths:
 
 FastAPI runs `def` handlers and dependencies in a thread pool and `async def` ones on the event loop, so a synchronous database call inside `async def` stalls every request on that process (#881).
 
-- New routes, dependencies and helpers are `def`.
-- Use `async def` only for real awaits (ARQ enqueue, async HTTP), and run the synchronous database work in the thread pool.
-- Start worker pipelines only through `enqueue_pipeline_for_score_set` (`lib/workflow/kickoff.py`).
+- Routes, dependencies and helpers are `def`. A sync HTTP call in a `def` route blocks only its own thread.
+- Use `async def` only where the code can't work without an await: reading a multipart form, or a route that waits only on an HTTP call and never touches the database. Keep database work out of it by running it in the thread pool, and say in a comment why the route is `async`.
+- `tests/routers/test_async_routes.py` fails when an `async` endpoint or dependency takes a session from `get_db`. Allowlist a route there only if it reads the form and runs the rest in the thread pool.
+- Start worker pipelines only through `lib/workflow/kickoff.py`: `enqueue_pipeline_for_score_set` from async code, or `enqueue_pipeline_for_score_set_from_thread` from code running in the thread pool.
 - Route tests go in `tests/routers/`.

AGENTS.md

@@ -32,7 +32,7 @@ The MaveDB API is a FastAPI app plus an ARQ/Redis worker over PostgreSQL. The we
 
 - Private records return 404, never 403, so a response doesn't reveal that they exist.
 - A route's permission check covers only the entity it names. Anything a read fans out to needs its own check through a `Viewer` (see `.claude/rules/permissions.md`).
-- Never write an `async def` that has no real await (#881).
+- Write `def` by default. Use `async def` only where the code can't work without an await, keep database work out of it, and say why in a comment; `tests/routers/test_async_routes.py` enforces the database part (#881, #950).
 - Changes reachable from a route or a serving `lib/` module need tests, and so do scripts that emit a recurring public artifact such as the data dump. One-shot operator scripts don't.
 - Never add `slow` or `network` markers to get green; those markers drop a test from PR CI.
 - Run operator scripts with `--commit` only against a local database unless asked. Never commit to production or staging databases without explicit permission.

src/mavedb/lib/authentication.py

@@ -118,7 +118,7 @@ def verify_jwt(token: str) -> dict:
 access_token_cookie = APIKeyCookie(name=ACCESS_TOKEN_NAME, auto_error=False)
 
 
-async def get_access_token(
+def get_access_token(
     # access_token_query: str = Security(access_token_query),
     access_token_header: Optional[str] = Security(access_token_header),
     access_token_cookie: Optional[str] = Security(access_token_cookie),

src/mavedb/lib/authorization.py

@@ -17,7 +17,7 @@
 ####################################################################################################
 
 
-async def require_current_user(
+def require_current_user(
     user_data: Optional[UserData] = Depends(get_current_user),
 ) -> UserData:
     if user_data is None:
@@ -27,7 +27,7 @@ async def require_current_user(
     return user_data
 
 
-async def get_principal(
+def get_principal(
     user_data: Optional[UserData] = Depends(get_current_user),
 ) -> Principal:
     """The principal for this request, for handlers that fan out to permission-checked sibling entities.
@@ -38,7 +38,7 @@ async def get_principal(
     return Principal(user_data)
 
 
-async def require_current_user_with_email(
+def require_current_user_with_email(
     user_data: UserData = Depends(require_current_user),
 ) -> UserData:
     # Both empty strings and NoneType values should raise an exception.
@@ -67,7 +67,7 @@ async def __call__(self, user_data: UserData = Depends(require_current_user)) ->
         return user_data
 
 
-async def require_role(roles: list[UserRole], user_data: UserData = Depends(require_current_user)) -> UserData:
+def require_role(roles: list[UserRole], user_data: UserData = Depends(require_current_user)) -> UserData:
     save_to_logging_context({"required_roles": [role.name for role in roles]})
     if not any(role.name in roles for role in user_data.active_roles):
         logger.info(

src/mavedb/lib/contributors.py

@@ -10,7 +10,7 @@
 logger = logging.getLogger(__name__)
 
 
-async def find_or_create_contributor(db: Session, orcid_id: str):
+def find_or_create_contributor(db: Session, orcid_id: str):
     """
     Find an existing contributor record with the specified ORCID ID, or create a new one and fetch the contributor name
     from ORCID.

src/mavedb/lib/identifiers.py

@@ -174,7 +174,7 @@ def reference_html(self) -> str:
         )
 
 
-async def find_or_create_doi_identifier(db: Session, identifier: str):
+def find_or_create_doi_identifier(db: Session, identifier: str):
     """
     Find an existing DOI identifier record with the specified identifier string, or create a new one.
 
@@ -188,7 +188,7 @@ async def find_or_create_doi_identifier(db: Session, identifier: str):
     return doi_identifier
 
 
-async def fetch_pubmed_article(identifier: str) -> Optional[PubmedArticle]:
+def fetch_pubmed_article(identifier: str) -> Optional[PubmedArticle]:
     """
     Fetch an existing PubMed article from NCBI
     """
@@ -216,7 +216,7 @@ async def fetch_pubmed_article(identifier: str) -> Optional[PubmedArticle]:
 #
 # NOTE: The most up to date version of a preprint will be the last element in the
 #       content detail list.
-async def fetch_biorxiv_article(identifier: str) -> Optional[RxivContentDetail]:
+def fetch_biorxiv_article(identifier: str) -> Optional[RxivContentDetail]:
     """
     Fetch an existing bioRxiv article from Rxiv
     """
@@ -228,7 +228,7 @@ async def fetch_biorxiv_article(identifier: str) -> Optional[RxivContentDetail]:
         return None
 
 
-async def fetch_medrxiv_article(identifier: str) -> Optional[RxivContentDetail]:
+def fetch_medrxiv_article(identifier: str) -> Optional[RxivContentDetail]:
     """
     Fetch an existing medRxiv article from Rxiv
     """
@@ -240,12 +240,12 @@ async def fetch_medrxiv_article(identifier: str) -> Optional[RxivContentDetail]:
         return None
 
 
-async def fetch_crossref_work(identifier: str) -> Optional[CrossrefWork]:
+def fetch_crossref_work(identifier: str) -> Optional[CrossrefWork]:
     fetch = Crossref(endpoint="works")
     return fetch.doi(identifier)
 
 
-async def find_generic_article(
+def find_generic_article(
     db: Session, identifier: str, db_name: Optional[str] = None
 ) -> Mapping[str, Union[PublicationIdentifier, ExternalPublication, None]]:
     """
@@ -282,7 +282,7 @@ async def find_generic_article(
         ).scalar_one_or_none()
 
         if not existing_publication:
-            external_publication = await db_specific_fetches[db_name](identifier)
+            external_publication = db_specific_fetches[db_name](identifier)
             db_specific_match = {
                 db_name: (
                     ExternalPublication(identifier, db_name, external_publication) if external_publication else None
@@ -304,7 +304,7 @@ async def find_generic_article(
         ).scalar_one_or_none()
 
         if not existing_publication:
-            external_publication = await fetch_crossref_work(identifier)
+            external_publication = fetch_crossref_work(identifier)
             found_articles["Crossref"] = (
                 ExternalPublication(identifier, "Crossref", external_publication) if external_publication else None
             )
@@ -329,7 +329,7 @@ async def find_generic_article(
 
             if not existing_publication:
                 try:
-                    external_publication = await db_specific_fetches[publication_db](identifier)
+                    external_publication = db_specific_fetches[publication_db](identifier)
                 except (json.JSONDecodeError, requests.exceptions.RequestException) as exc:
                     logger.warning(
                         "Failed to fetch identifier %r from %s while fanning out over candidate databases.",
@@ -374,7 +374,7 @@ def create_generic_article(article: ExternalPublication) -> PublicationIdentifie
     )
 
 
-async def find_or_create_publication_identifier(
+def find_or_create_publication_identifier(
     db: Session, identifier: str, db_name: Optional[str] = None
 ) -> PublicationIdentifier:
     """
@@ -384,7 +384,7 @@ async def find_or_create_publication_identifier(
     :param identifier: A valid publication identifier
     :return: An existing PublicationIdentifier containing the specified identifier string, or a new, unsaved PublicationIdentifier
     """
-    article_matches = await find_generic_article(db, identifier, db_name)
+    article_matches = find_generic_article(db, identifier, db_name)
 
     if not any(article_matches.values()):
         raise NonexistentIdentifierError(
@@ -408,7 +408,7 @@ async def find_or_create_publication_identifier(
     return create_generic_article(matched_article)
 
 
-async def find_or_create_raw_read_identifier(db: Session, identifier: str):
+def find_or_create_raw_read_identifier(db: Session, identifier: str):
     raw_read_identifier = db.query(RawReadIdentifier).filter(RawReadIdentifier.identifier == identifier).one_or_none()
     if not raw_read_identifier:
         raw_read_identifier = RawReadIdentifier(
@@ -417,7 +417,7 @@ async def find_or_create_raw_read_identifier(db: Session, identifier: str):
     return raw_read_identifier
 
 
-async def find_or_create_external_gene_identifier(db: Session, db_name: str, identifier: str):
+def find_or_create_external_gene_identifier(db: Session, db_name: str, identifier: str):
     """
     Find an existing gene identifier record with the specified gene database name and identifier string, or create a new
     one.
@@ -446,10 +446,10 @@ async def find_or_create_external_gene_identifier(db: Session, db_name: str, ide
     return external_gene_identifier
 
 
-async def create_external_gene_identifier_offset(
+def create_external_gene_identifier_offset(
     db: Session, target_gene: TargetGene, db_name: str, identifier: str, offset: int
 ):
-    external_gene_identifier = await find_or_create_external_gene_identifier(db, db_name, identifier)
+    external_gene_identifier = find_or_create_external_gene_identifier(db, db_name, identifier)
 
     # TODO Handle key errors.
     offset_class = EXTERNAL_GENE_IDENTIFIER_OFFSET_CLASSES[external_gene_identifier.db_name]

src/mavedb/lib/mondo_ols.py

@@ -32,7 +32,7 @@
 DEFAULT_SEARCH_LIMIT = 20
 
 
-async def search_mondo(query: str, limit: int = DEFAULT_SEARCH_LIMIT) -> list[MondoSuggestion]:
+def search_mondo(query: str, limit: int = DEFAULT_SEARCH_LIMIT) -> list[MondoSuggestion]:
     """Search MONDO disease classes by free text via OLS, for the disease typeahead."""
     if not query or not query.strip():
         return []
@@ -47,8 +47,8 @@ async def search_mondo(query: str, limit: int = DEFAULT_SEARCH_LIMIT) -> list[Mo
     }
     save_to_logging_context({"mondo_search_query": query, "mondo_search_limit": limit})
     try:
-        async with httpx.AsyncClient(timeout=OLS_TIMEOUT) as client:
-            response = await client.get(f"{OLS_BASE_URL}/select", params=params)
+        with httpx.Client(timeout=OLS_TIMEOUT) as client:
+            response = client.get(f"{OLS_BASE_URL}/select", params=params)
             response.raise_for_status()
             docs = response.json().get("response", {}).get("docs", [])
     except httpx.HTTPError as exc:
@@ -63,7 +63,7 @@ async def search_mondo(query: str, limit: int = DEFAULT_SEARCH_LIMIT) -> list[Mo
     ]
 
 
-async def fetch_mondo_term(code: str) -> Optional[MondoSuggestion]:
+def fetch_mondo_term(code: str) -> Optional[MondoSuggestion]:
     """Resolve one MONDO CURIE to its canonical label + IRI via OLS, or None if it does not exist."""
     params: dict[str, Any] = {
         "q": code,
@@ -74,8 +74,8 @@ async def fetch_mondo_term(code: str) -> Optional[MondoSuggestion]:
         "fieldList": "iri,label,obo_id",
     }
     try:
-        async with httpx.AsyncClient(timeout=OLS_TIMEOUT) as client:
-            response = await client.get(f"{OLS_BASE_URL}/search", params=params)
+        with httpx.Client(timeout=OLS_TIMEOUT) as client:
+            response = client.get(f"{OLS_BASE_URL}/search", params=params)
             response.raise_for_status()
             docs = response.json().get("response", {}).get("docs", [])
     except httpx.HTTPError as exc:
@@ -90,7 +90,7 @@ async def fetch_mondo_term(code: str) -> Optional[MondoSuggestion]:
     return MondoSuggestion(code=doc["obo_id"], label=doc["label"], iri=doc.get("iri", mondo_iri(code)))
 
 
-async def find_or_create_mondo_term(db: Session, code: str) -> MondoTerm:
+def find_or_create_mondo_term(db: Session, code: str) -> MondoTerm:
     """Find an existing MONDO term by code, or create one after validating it against OLS.
 
     A new code is validated against OLS and its canonical label stored. An unknown code raises
@@ -105,7 +105,7 @@ async def find_or_create_mondo_term(db: Session, code: str) -> MondoTerm:
     if term is not None:
         return term
 
-    resolved = await fetch_mondo_term(code)
+    resolved = fetch_mondo_term(code)
     if resolved is None:
         raise ValidationError(f"'{code}' is not a valid MONDO disease term.", custom_loc=["body", "disease"])
 
@@ -125,7 +125,7 @@ async def find_or_create_mondo_term(db: Session, code: str) -> MondoTerm:
     return term
 
 
-async def resolve_disease_term(db: Session, code: Optional[str]) -> MondoTerm:
+def resolve_disease_term(db: Session, code: Optional[str]) -> MondoTerm:
     """Resolve a calibration's submitted disease code to a stored MONDO term.
 
     ``None`` (no disease specified) resolves to the generic "disease or disorder" term; any other code
@@ -135,4 +135,4 @@ async def resolve_disease_term(db: Session, code: Optional[str]) -> MondoTerm:
     if code is None:
         return get_generic_disease_term(db)
 
-    return await find_or_create_mondo_term(db, code)
+    return find_or_create_mondo_term(db, code)