From f4626fe46b323457d3201d0a6bcae52864851f2e Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Fri, 23 Oct 2026 14:37:56 -0700 Subject: [PATCH 1/3] perf: run blocking route work in the thread pool FastAPI runs async def handlers on the event loop, and most of ours ran synchronous SQLAlchemy queries there, so one slow query stalled every in-flight request on its process. The publication fetchers were async in name only: they made synchronous HTTP calls to NCBI, bioRxiv and Crossref on the loop, with no timeout. Routes that use the database are now def, so FastAPI runs them in its thread pool. The helpers they awaited, which never awaited anything real, are sync. MONDO and NCBI Taxonomy use a sync httpx client, and routes queue jobs through enqueue_pipeline_for_score_set_from_thread, which writes the pipeline records in the calling thread and hands only the ARQ call to the loop. The two upload routes stay async to read the multipart form and run everything after it, including the CSV rebuild and S3 upload, in the thread pool. Making them def would mean declaring every form field, which changes the OpenAPI schema. The schema is unchanged. Disease search could be async, since it only calls OLS, but it's def to keep one client style in lib/mondo_ols.py; its comment says when to revisit that. Tests that were async only because these helpers were are now sync, with the same assertions. --- src/mavedb/lib/authentication.py | 2 +- src/mavedb/lib/authorization.py | 8 +- src/mavedb/lib/contributors.py | 2 +- src/mavedb/lib/identifiers.py | 30 +- src/mavedb/lib/mondo_ols.py | 20 +- src/mavedb/lib/score_calibrations.py | 40 ++- src/mavedb/lib/taxonomies.py | 10 +- src/mavedb/lib/workflow/kickoff.py | 99 +++++-- src/mavedb/models/access_key.py | 6 +- src/mavedb/models/experiment.py | 12 +- src/mavedb/models/score_set.py | 6 +- src/mavedb/models/user.py | 6 +- src/mavedb/routers/access_keys.py | 4 +- src/mavedb/routers/alphafold.py | 2 + src/mavedb/routers/collections.py | 18 +- src/mavedb/routers/diseases.py | 9 +- src/mavedb/routers/experiments.py | 28 +- src/mavedb/routers/orcid.py | 2 + src/mavedb/routers/permissions.py | 2 +- src/mavedb/routers/publication_identifiers.py | 8 +- src/mavedb/routers/score_calibrations.py | 16 +- src/mavedb/routers/score_sets.py | 162 +++++++---- src/mavedb/routers/taxonomies.py | 4 +- src/mavedb/routers/users.py | 16 +- src/mavedb/scripts/load_calibration_csv.py | 5 +- .../scripts/load_excalibr_calibrations.py | 5 +- tests/helpers/util/score_calibration.py | 4 +- tests/lib/mondo/network/test_mondo.py | 21 +- tests/lib/mondo/test_mondo.py | 84 +++--- tests/lib/test_identifiers.py | 15 +- tests/lib/test_score_calibrations.py | 271 +++++++----------- tests/lib/workflow/test_kickoff.py | 62 +++- tests/routers/test_diseases.py | 4 +- tests/routers/test_score_calibrations.py | 6 +- tests/routers/test_score_set.py | 2 +- 35 files changed, 553 insertions(+), 438 deletions(-) diff --git a/src/mavedb/lib/authentication.py b/src/mavedb/lib/authentication.py index a66503165..a96162b22 100644 --- a/src/mavedb/lib/authentication.py +++ b/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), diff --git a/src/mavedb/lib/authorization.py b/src/mavedb/lib/authorization.py index 252febb4e..1810104b8 100644 --- a/src/mavedb/lib/authorization.py +++ b/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( diff --git a/src/mavedb/lib/contributors.py b/src/mavedb/lib/contributors.py index 4edfce976..ea2763f59 100644 --- a/src/mavedb/lib/contributors.py +++ b/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. diff --git a/src/mavedb/lib/identifiers.py b/src/mavedb/lib/identifiers.py index 7855b6af7..6c530d221 100644 --- a/src/mavedb/lib/identifiers.py +++ b/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] diff --git a/src/mavedb/lib/mondo_ols.py b/src/mavedb/lib/mondo_ols.py index c6a58bcec..d66f27033 100644 --- a/src/mavedb/lib/mondo_ols.py +++ b/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) diff --git a/src/mavedb/lib/score_calibrations.py b/src/mavedb/lib/score_calibrations.py index afa4723ac..551885007 100644 --- a/src/mavedb/lib/score_calibrations.py +++ b/src/mavedb/lib/score_calibrations.py @@ -326,7 +326,7 @@ def create_functional_classification( return functional_classification -async def _create_score_calibration( +def _create_score_calibration( db: Session, calibration_create: score_calibration.ScoreCalibrationCreate, user: User, @@ -384,8 +384,9 @@ async def _create_score_calibration( ----- - Duplicate identifiers across different source lists result in distinct association objects (no deduplication is performed here). - - The function is async because it awaits the underlying publication - identifier retrieval/creation calls. + - Resolving a new publication identifier calls the external publication + databases synchronously, so callers on a request path must run in the + thread pool. """ relation_sources = ( (ScoreCalibrationRelation.threshold, calibration_create.threshold_sources or []), @@ -396,7 +397,7 @@ async def _create_score_calibration( calibration_pub_assocs = [] for relation, sources in relation_sources: for identifier in sources: - pub = await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + pub = find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) calibration_pub_assocs.append( ScoreCalibrationPublicationIdentifierAssociation( publication=pub, @@ -427,7 +428,7 @@ async def _create_score_calibration( modified_by=user, ) # type: ignore[call-arg] - calibration.disease_term = await resolve_disease_term(db, getattr(calibration_create, "disease", None)) + calibration.disease_term = resolve_disease_term(db, getattr(calibration_create, "disease", None)) if containing_score_set: calibration.score_set = containing_score_set @@ -443,7 +444,7 @@ async def _create_score_calibration( return calibration -async def create_score_calibration_in_score_set( +def create_score_calibration_in_score_set( db: Session, calibration_create: score_calibration.ScoreCalibrationCreate, user: User, @@ -452,7 +453,7 @@ async def create_score_calibration_in_score_set( """ Create a new score calibration and associate it with an existing score set. - This coroutine ensures that the provided ScoreCalibrationCreate payload includes a + This function ensures that the provided ScoreCalibrationCreate payload includes a score_set_urn, loads the corresponding ScoreSet from the database, delegates creation of the ScoreCalibration to an internal helper, and then links the created calibration to the fetched score set. @@ -477,8 +478,6 @@ async def create_score_calibration_in_score_set( (should not occur if URNs are unique). Notes: - - This function is async because it awaits the internal _create_score_calibration - helper, which may perform asynchronous operations (e.g., I/O or async ORM tasks). - The passed Session is expected to be valid for the lifetime of this call; committing or flushing is assumed to be handled externally (depending on the surrounding transaction management strategy). @@ -490,7 +489,7 @@ async def create_score_calibration_in_score_set( require_phi_affirmation(bool(submitted_controls), getattr(calibration_create, "controls_not_phi", None)) containing_score_set = db.query(ScoreSet).where(ScoreSet.urn == calibration_create.score_set_urn).one() - calibration = await _create_score_calibration(db, calibration_create, user, variant_classes, containing_score_set) + calibration = _create_score_calibration(db, calibration_create, user, variant_classes, containing_score_set) if user.username in [contributor.orcid_id for contributor in containing_score_set.contributors] + [ containing_score_set.created_by.username, @@ -506,14 +505,14 @@ async def create_score_calibration_in_score_set( return calibration -async def create_score_calibration( +def create_score_calibration( db: Session, calibration_create: score_calibration.ScoreCalibrationCreate, user: User, variant_classes: Optional[ClassificationDict] = None, ) -> ScoreCalibration: """ - Asynchronously create and persist a new ScoreCalibration record. + Create and persist a new ScoreCalibration record. This is a thin wrapper that delegates to the internal _create_score_calibration implementation, allowing for separation of public API and internal logic. @@ -550,9 +549,8 @@ async def create_score_calibration( Notes ----- - - Because this function is asynchronous, callers must await it. Any transaction - management (commit / rollback) is expected to be handled by the session lifecycle - manager in the calling context. + - Any transaction management (commit / rollback) is expected to be handled by the + session lifecycle manager in the calling context. - Because the calibration database model enforces that a calibration must belong to a ScoreSet, callers should perform this association themselves after creation (e.g., by assigning the calibration's score_set attribute to an existing ScoreSet @@ -561,7 +559,7 @@ async def create_score_calibration( if calibration_create.score_set_urn: raise ValueError("score_set_urn must not be provided to create a score calibration outside a score set.") - created_calibration = await _create_score_calibration( + created_calibration = _create_score_calibration( db, calibration_create, user, variant_classes, containing_score_set=None ) @@ -569,7 +567,7 @@ async def create_score_calibration( return created_calibration -async def modify_score_calibration( +def modify_score_calibration( db: Session, calibration: ScoreCalibration, calibration_update: score_calibration.ScoreCalibrationModify, @@ -577,7 +575,7 @@ async def modify_score_calibration( variant_classes: Optional[ClassificationDict] = None, ) -> ScoreCalibration: """ - Asynchronously modify an existing ScoreCalibration record and its related publication + Modify an existing ScoreCalibration record and its related publication identifier associations. This function: @@ -602,7 +600,7 @@ async def modify_score_calibration( Parameters ---------- db : Session - An active SQLAlchemy session (synchronous engine session used within an async context). + An active SQLAlchemy session. calibration : ScoreCalibration The existing calibration ORM instance to be modified (must be persistent or pending). calibration_update : score_calibration.ScoreCalibrationModify @@ -687,7 +685,7 @@ async def modify_score_calibration( updated_assocs = [] for relation, sources in relation_sources: for identifier in sources: - pub = await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + pub = find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) assoc_key = (relation, pub.id) if assoc_key in existing_assocs_map: # Keep existing association @@ -734,7 +732,7 @@ async def modify_score_calibration( }: setattr(calibration, attr, value) - calibration.disease_term = await resolve_disease_term(db, getattr(calibration_update, "disease", None)) + calibration.disease_term = resolve_disease_term(db, getattr(calibration_update, "disease", None)) calibration.score_set = containing_score_set calibration.score_set_id = containing_score_set.id diff --git a/src/mavedb/lib/taxonomies.py b/src/mavedb/lib/taxonomies.py index 9dfe39cd3..072051158 100644 --- a/src/mavedb/lib/taxonomies.py +++ b/src/mavedb/lib/taxonomies.py @@ -8,7 +8,7 @@ from mavedb.view_models.taxonomy import TaxonomyCreate -async def find_or_create_taxonomy(db: Session, taxonomy: TaxonomyCreate): +def find_or_create_taxonomy(db: Session, taxonomy: TaxonomyCreate): """ Find an existing taxonomy ID record with the specified code, or create a new one. @@ -19,11 +19,11 @@ async def find_or_create_taxonomy(db: Session, taxonomy: TaxonomyCreate): """ taxonomy_record = db.query(Taxonomy).filter(Taxonomy.code == taxonomy.code).one_or_none() if not taxonomy_record: - taxonomy_record = await search_NCBI_taxonomy(db, str(taxonomy.code)) + taxonomy_record = search_NCBI_taxonomy(db, str(taxonomy.code)) return taxonomy_record -async def search_NCBI_taxonomy(db: Session, search: str) -> Any: +def search_NCBI_taxonomy(db: Session, search: str) -> Any: """ Check whether this taxonomy exists in NCBI website. If it exists, saving it in local database and return it to users. @@ -35,8 +35,8 @@ async def search_NCBI_taxonomy(db: Session, search: str) -> Any: search_text = str(search) url = f"https://api.ncbi.nlm.nih.gov/datasets/v2alpha/taxonomy/taxon/{search_text}" - async with httpx.AsyncClient() as client: - response = await client.get(url) + with httpx.Client() as client: + response = client.get(url) if response.status_code == 200: data = response.json() diff --git a/src/mavedb/lib/workflow/kickoff.py b/src/mavedb/lib/workflow/kickoff.py index 35f236bf2..c43e8e74f 100644 --- a/src/mavedb/lib/workflow/kickoff.py +++ b/src/mavedb/lib/workflow/kickoff.py @@ -4,20 +4,27 @@ :class:`PipelineFactory`, then enqueue the ``start_pipeline`` entrypoint in ARQ. Both the ad-hoc ``run_pipeline`` script and the batch ``enrich_backfilled_score_sets`` driver go through here so the enqueue contract (entrypoint id + :func:`arq_job_id` + :func:`arq_queue_name`) lives in exactly one place. + +API routes run in FastAPI's thread pool and call :func:`enqueue_pipeline_for_score_set_from_thread`, which +writes the records in the calling thread and hands only the ARQ call to the event loop. Async callers +use :func:`enqueue_pipeline_for_score_set`. """ from __future__ import annotations import datetime +import functools import logging from typing import Any, Optional +import anyio.from_thread from arq.connections import ArqRedis from arq.jobs import Job from sqlalchemy.orm import Session from mavedb.lib.logging.context import correlation_id_for_context, logging_context, save_to_logging_context from mavedb.lib.workflow.pipeline_factory import PipelineFactory +from mavedb.models.job_run import JobRun from mavedb.models.pipeline import Pipeline from mavedb.models.score_set import ScoreSet from mavedb.models.user import User @@ -44,6 +51,56 @@ async def enqueue_pipeline_for_score_set( pipeline records are discarded (via :meth:`PipelineFactory.discard_pipeline`) before the exception is re-raised, so a failed enqueue never leaves an orphaned pipeline behind. """ + factory, pipeline, entrypoint = _create_pipeline(db, pipeline_name, score_set, user, extra_params) + + try: + job = await redis.enqueue_job(*_enqueue_args(entrypoint), **_enqueue_kwargs(entrypoint)) + except Exception: + _discard_after_failed_enqueue(factory, pipeline, pipeline_name, score_set) + raise + + _log_enqueued(pipeline, entrypoint, job, pipeline_name, score_set) + return pipeline, job + + +def enqueue_pipeline_for_score_set_from_thread( + db: Session, + redis: ArqRedis, + *, + pipeline_name: str, + score_set: ScoreSet, + user: User, + extra_params: Optional[dict[str, Any]] = None, +) -> tuple[Pipeline, Optional[Job]]: + """Do what :func:`enqueue_pipeline_for_score_set` does, from a thread started by AnyIO. + + Writes the pipeline records in the calling thread and runs only the ARQ enqueue on the event loop, + so a route in FastAPI's thread pool never runs database work on the loop. Must be called from a + worker thread that AnyIO started, such as a ``def`` route or a function passed to + ``run_in_threadpool``; ``redis`` must belong to that loop. Returns and raises as + :func:`enqueue_pipeline_for_score_set` does. + """ + factory, pipeline, entrypoint = _create_pipeline(db, pipeline_name, score_set, user, extra_params) + + try: + job = anyio.from_thread.run( + functools.partial(redis.enqueue_job, *_enqueue_args(entrypoint), **_enqueue_kwargs(entrypoint)) + ) + except Exception: + _discard_after_failed_enqueue(factory, pipeline, pipeline_name, score_set) + raise + + _log_enqueued(pipeline, entrypoint, job, pipeline_name, score_set) + return pipeline, job + + +def _create_pipeline( + db: Session, + pipeline_name: str, + score_set: ScoreSet, + user: User, + extra_params: Optional[dict[str, Any]], +) -> tuple[PipelineFactory, Pipeline, JobRun]: correlation_id = ( correlation_id_for_context() or f"{pipeline_name}_{score_set.urn}_{user.id}_{datetime.datetime.now().isoformat()}" @@ -60,23 +117,34 @@ async def enqueue_pipeline_for_score_set( pipeline, entrypoint = factory.create_pipeline( pipeline_name=pipeline_name, creating_user=user, pipeline_params=params ) - job_id = arq_job_id(entrypoint) + return factory, pipeline, entrypoint - try: - job = await redis.enqueue_job( - entrypoint.job_function, entrypoint.id, _job_id=job_id, _queue_name=arq_queue_name(entrypoint) - ) - except Exception: - logger.error( - "Failed to enqueue pipeline %s for score set %s; discarding pipeline %s", - pipeline_name, - score_set.urn, - pipeline.id, - extra=logging_context(), - ) - factory.discard_pipeline(pipeline) - raise +def _enqueue_args(entrypoint: JobRun) -> tuple[Any, ...]: + return (entrypoint.job_function, entrypoint.id) + + +def _enqueue_kwargs(entrypoint: JobRun) -> dict[str, Any]: + return {"_job_id": arq_job_id(entrypoint), "_queue_name": arq_queue_name(entrypoint)} + + +def _discard_after_failed_enqueue( + factory: PipelineFactory, pipeline: Pipeline, pipeline_name: str, score_set: ScoreSet +) -> None: + logger.error( + "Failed to enqueue pipeline %s for score set %s; discarding pipeline %s", + pipeline_name, + score_set.urn, + pipeline.id, + extra=logging_context(), + ) + factory.discard_pipeline(pipeline) + + +def _log_enqueued( + pipeline: Pipeline, entrypoint: JobRun, job: Optional[Job], pipeline_name: str, score_set: ScoreSet +) -> None: + job_id = arq_job_id(entrypoint) save_to_logging_context({"pipeline_id": pipeline.id, "job_id": job_id}) if not job: @@ -95,4 +163,3 @@ async def enqueue_pipeline_for_score_set( job_id, extra=logging_context(), ) - return pipeline, job diff --git a/src/mavedb/models/access_key.py b/src/mavedb/models/access_key.py index 41092beb2..77d83226c 100644 --- a/src/mavedb/models/access_key.py +++ b/src/mavedb/models/access_key.py @@ -33,11 +33,11 @@ def role(self) -> Optional["UserRole"]: role_obj = self.role_obj or None return None if role_obj is None else role_obj.name - async def set_role(self, db, role: "UserRole"): - self.role_obj = await self._find_or_create_role(db, role) + def set_role(self, db, role: "UserRole"): + self.role_obj = self._find_or_create_role(db, role) @staticmethod - async def _find_or_create_role(db, role): + def _find_or_create_role(db, role): role_obj = db.query(Role).filter(Role.name == role).one_or_none() if not role_obj: role_obj = Role(name=role) diff --git a/src/mavedb/models/experiment.py b/src/mavedb/models/experiment.py index eb8bb2cf2..d5fbec2e7 100644 --- a/src/mavedb/models/experiment.py +++ b/src/mavedb/models/experiment.py @@ -177,18 +177,18 @@ def legacy_keywords(self) -> list[str]: legacy_keyword_objs = self.legacy_keyword_objs or [] return [keyword_obj.text for keyword_obj in legacy_keyword_objs if keyword_obj.text is not None] - async def set_legacy_keywords(self, db, keywords: Optional[list[str]]): + def set_legacy_keywords(self, db, keywords: Optional[list[str]]): if keywords: - self.keyword_objs = [await self._find_or_create_legacy_keyword(db, text) for text in keywords] + self.keyword_objs = [self._find_or_create_legacy_keyword(db, text) for text in keywords] else: self.keyword_objs = [] - async def set_keywords(self, db, keywords: list): + def set_keywords(self, db, keywords: list): if keywords: self.keyword_objs = [ ExperimentControlledKeywordAssociation( experiment=self, - controlled_keyword=await self._find_keyword( + controlled_keyword=self._find_keyword( db, keyword_obj.keyword.key, keyword_obj.keyword.label, @@ -206,13 +206,13 @@ async def set_keywords(self, db, keywords: list): # async def set_keywords(self, db, keywords: list[str]): # self._keyword_objs = [await self._find_or_create_keyword(text) for text in keywords] - async def _find_or_create_legacy_keyword(self, db, keyword_text): + def _find_or_create_legacy_keyword(self, db, keyword_text): keyword_obj = db.query(LegacyKeyword).filter(LegacyKeyword.text == keyword_text).one_or_none() if not keyword_obj: keyword_obj = LegacyKeyword(text=keyword_text) return keyword_obj - async def _find_keyword(self, db, key: str, label: str, code: Optional[str]): + def _find_keyword(self, db, key: str, label: str, code: Optional[str]): query = ( db.query(ControlledKeyword).filter(ControlledKeyword.key == key).filter(ControlledKeyword.label == label) ) diff --git a/src/mavedb/models/score_set.py b/src/mavedb/models/score_set.py index 1c71bf6b5..6b9adf4c0 100644 --- a/src/mavedb/models/score_set.py +++ b/src/mavedb/models/score_set.py @@ -227,15 +227,15 @@ def legacy_keywords(self) -> list[str]: legacy_keyword_obj.text for legacy_keyword_obj in legacy_keyword_objs if legacy_keyword_obj.text is not None ] - async def set_legacy_keywords(self, db, keywords: list[str]): - self.keyword_objs = [await self._find_or_create_legacy_keyword(db, text) for text in keywords] + def set_legacy_keywords(self, db, keywords: list[str]): + self.keyword_objs = [self._find_or_create_legacy_keyword(db, text) for text in keywords] # See https://gist.github.com/tachyondecay/e0fe90c074d6b6707d8f1b0b1dcc8e3a # @keywords.setter # async def set_keywords(self, db, keywords: list[str]): # self._keyword_objs = [await self._find_or_create_keyword(text) for text in keywords] - async def _find_or_create_legacy_keyword(self, db, keyword_text): + def _find_or_create_legacy_keyword(self, db, keyword_text): keyword_obj = db.query(LegacyKeyword).filter(LegacyKeyword.text == keyword_text).one_or_none() if not keyword_obj: keyword_obj = LegacyKeyword(text=keyword_text) diff --git a/src/mavedb/models/user.py b/src/mavedb/models/user.py index 1cff0effc..00282061b 100644 --- a/src/mavedb/models/user.py +++ b/src/mavedb/models/user.py @@ -42,11 +42,11 @@ def roles(self) -> list[UserRole]: role_objs = self.role_objs or [] return [role_obj.name for role_obj in role_objs if role_obj.name is not None] - async def set_roles(self, db, roles: list[UserRole]): - self.role_objs = [await self._find_or_create_role(db, name) for name in roles] + def set_roles(self, db, roles: list[UserRole]): + self.role_objs = [self._find_or_create_role(db, name) for name in roles] @staticmethod - async def _find_or_create_role(db, role_name): + def _find_or_create_role(db, role_name): role_obj = db.query(Role).filter(Role.name == role_name).one_or_none() if not role_obj: role_obj = Role(name=role_name) diff --git a/src/mavedb/routers/access_keys.py b/src/mavedb/routers/access_keys.py index d6762cbf2..582d77bbf 100644 --- a/src/mavedb/routers/access_keys.py +++ b/src/mavedb/routers/access_keys.py @@ -116,7 +116,7 @@ def create_my_access_key( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Create a new access key for myself with a specified role", ) -async def create_my_access_key_with_role( +def create_my_access_key_with_role( *, role: UserRole, db: Session = Depends(deps.get_db), @@ -139,7 +139,7 @@ async def create_my_access_key_with_role( ) key, private_key, item = new_access_key(user_data.user) - await item.set_role(db, role) + item.set_role(db, role) db.add(item) db.commit() db.refresh(item) diff --git a/src/mavedb/routers/alphafold.py b/src/mavedb/routers/alphafold.py index 96139c529..bc52f4310 100644 --- a/src/mavedb/routers/alphafold.py +++ b/src/mavedb/routers/alphafold.py @@ -17,6 +17,8 @@ @router.get("/alphafold-files/version") +# `async` because it only waits on AlphaFold and never touches the database, so awaiting the request holds +# no thread from the pool (#950). async def proxy_alphafold_index(): """ Proxy the AlphaFold files index (XML document). diff --git a/src/mavedb/routers/collections.py b/src/mavedb/routers/collections.py index 051d5bb3a..bff15673e 100644 --- a/src/mavedb/routers/collections.py +++ b/src/mavedb/routers/collections.py @@ -172,7 +172,7 @@ def fetch_collection( response_model_exclude_none=True, summary="Create a collection", ) -async def create_collection( +def create_collection( *, item_create: collection.CollectionCreate, db: Session = Depends(deps.get_db), @@ -298,7 +298,7 @@ async def create_collection( response_model_exclude_none=True, summary="Update a collection", ) -async def update_collection( +def update_collection( *, item_update: collection.CollectionModify, urn: str, @@ -443,7 +443,7 @@ async def update_collection( }, summary="Add a score set to a collection", ) -async def add_score_set_to_collection( +def add_score_set_to_collection( *, body: collection.AddScoreSetToCollectionRequest, collection_urn: str, @@ -517,7 +517,7 @@ async def add_score_set_to_collection( responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE}, summary="Remove a score set from a collection", ) -async def delete_score_set_from_collection( +def delete_score_set_from_collection( *, collection_urn: str, score_set_urn: str, @@ -602,7 +602,7 @@ async def delete_score_set_from_collection( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Add an experiment to a collection", ) -async def add_experiment_to_collection( +def add_experiment_to_collection( *, body: collection.AddExperimentToCollectionRequest, collection_urn: str, @@ -672,7 +672,7 @@ async def add_experiment_to_collection( responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE}, summary="Remove an experiment from a collection", ) -async def delete_experiment_from_collection( +def delete_experiment_from_collection( *, collection_urn: str, experiment_urn: str, @@ -753,7 +753,7 @@ async def delete_experiment_from_collection( responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE}, summary="Add a user to a collection role", ) -async def add_user_to_collection_role( +def add_user_to_collection_role( *, body: collection.AddUserToCollectionRoleRequest, urn: str, @@ -847,7 +847,7 @@ async def add_user_to_collection_role( responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE}, summary="Remove a user from a collection role", ) -async def remove_user_from_collection_role( +def remove_user_from_collection_role( *, urn: str, role: ContributionRole, @@ -938,7 +938,7 @@ async def remove_user_from_collection_role( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Delete a collection", ) -async def delete_collection( +def delete_collection( *, urn: str, db: Session = Depends(deps.get_db), diff --git a/src/mavedb/routers/diseases.py b/src/mavedb/routers/diseases.py index f0f2f9daa..6881e9a9d 100644 --- a/src/mavedb/routers/diseases.py +++ b/src/mavedb/routers/diseases.py @@ -31,7 +31,12 @@ summary="Search MONDO disease terms", responses={**GATEWAY_ERROR_RESPONSES}, ) -async def search_diseases( +# This route only calls OLS, so it could be `async` with an async client and hold no thread while OLS +# answers. As `def`, each search holds a thread for up to `OLS_TIMEOUT`. That's cheap while only the +# calibration editor calls it, debounced and cancelled on each keystroke, and it keeps `lib/mondo_ols.py` +# on one client style. Make it `async` if it moves onto a high-traffic page or slow OLS responses start +# exhausting the thread pool (#950). +def search_diseases( *, q: str = Query(..., description="Free-text query for a MONDO disease term."), limit: int = Query(DEFAULT_SEARCH_LIMIT, ge=1, le=100, description="Maximum number of results."), @@ -39,5 +44,5 @@ async def search_diseases( """Typeahead search for MONDO disease terms, returned as GA4GH ``MappableConcept`` suggestions.""" save_to_logging_context({"query": q, "limit": limit}) - suggestions = await search_mondo(q, limit) + suggestions = search_mondo(q, limit) return [mondo_suggestion_to_mappable_concept(suggestion) for suggestion in suggestions] diff --git a/src/mavedb/routers/experiments.py b/src/mavedb/routers/experiments.py index a58d127b9..cd2656e99 100644 --- a/src/mavedb/routers/experiments.py +++ b/src/mavedb/routers/experiments.py @@ -284,7 +284,7 @@ def get_experiment_score_sets( }, response_model_exclude_none=True, ) -async def create_experiment( +def create_experiment( *, item_create: experiment.ExperimentCreate, db: Session = Depends(deps.get_db), @@ -317,7 +317,7 @@ async def create_experiment( contributors: list[Contributor] = [] try: contributors = [ - await find_or_create_contributor(db, contributor.orcid_id) for contributor in item_create.contributors or [] + find_or_create_contributor(db, contributor.orcid_id) for contributor in item_create.contributors or [] ] except NonexistentOrcidUserError as e: logger.error(msg="Could not find ORCID user with the provided user ID.", extra=logging_context()) @@ -325,19 +325,18 @@ async def create_experiment( try: doi_identifiers = [ - await find_or_create_doi_identifier(db, identifier.identifier) - for identifier in item_create.doi_identifiers or [] + find_or_create_doi_identifier(db, identifier.identifier) for identifier in item_create.doi_identifiers or [] ] raw_read_identifiers = [ - await find_or_create_raw_read_identifier(db, identifier.identifier) + find_or_create_raw_read_identifier(db, identifier.identifier) for identifier in item_create.raw_read_identifiers or [] ] primary_publication_identifiers = [ - await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) for identifier in item_create.primary_publication_identifiers or [] ] publication_identifiers = [ - await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) for identifier in item_create.secondary_publication_identifiers or [] ] + primary_publication_identifiers @@ -427,7 +426,7 @@ async def create_experiment( }, response_model_exclude_none=True, ) -async def update_experiment( +def update_experiment( *, item_update: experiment.ExperimentUpdate, urn: str, @@ -468,7 +467,7 @@ async def update_experiment( try: item.contributors = [ - await find_or_create_contributor(db, contributor.orcid_id) for contributor in item_update.contributors or [] + find_or_create_contributor(db, contributor.orcid_id) for contributor in item_update.contributors or [] ] except NonexistentOrcidUserError as e: logger.error(msg="Could not find ORCID user with the provided user ID.", extra=logging_context()) @@ -476,20 +475,19 @@ async def update_experiment( try: doi_identifiers = [ - await find_or_create_doi_identifier(db, identifier.identifier) - for identifier in item_update.doi_identifiers or [] + find_or_create_doi_identifier(db, identifier.identifier) for identifier in item_update.doi_identifiers or [] ] raw_read_identifiers = [ - await find_or_create_raw_read_identifier(db, identifier.identifier) + find_or_create_raw_read_identifier(db, identifier.identifier) for identifier in item_update.raw_read_identifiers or [] ] primary_publication_identifiers = [ - await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) for identifier in item_update.primary_publication_identifiers or [] ] publication_identifiers = [ - await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) for identifier in item_update.secondary_publication_identifiers or [] ] + primary_publication_identifiers @@ -557,7 +555,7 @@ async def update_experiment( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Delete an experiment", ) -async def delete_experiment( +def delete_experiment( *, urn: str, db: Session = Depends(deps.get_db), diff --git a/src/mavedb/routers/orcid.py b/src/mavedb/routers/orcid.py index 8df3898b5..6c3575cb3 100644 --- a/src/mavedb/routers/orcid.py +++ b/src/mavedb/routers/orcid.py @@ -78,6 +78,8 @@ def lookup_orcid_user( summary="Exchange an ORCID authorization code for an access token", include_in_schema=False, ) +# `async` because it only waits on ORCID and never touches the database, so awaiting the request holds no +# thread from the pool (#950). async def get_token_from_code(*, request: orcid.OrcidAuthTokenRequest) -> Any: """ Perform the second step of three-legged Oauth2. The user has signed into ORCID in the browser, and the client has diff --git a/src/mavedb/routers/permissions.py b/src/mavedb/routers/permissions.py index cd6349f66..a31bc20ac 100644 --- a/src/mavedb/routers/permissions.py +++ b/src/mavedb/routers/permissions.py @@ -50,7 +50,7 @@ class ModelName(str, Enum): responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Check user permissions on a resource", ) -async def check_permission( +def check_permission( *, model_name: ModelName, urn: str, diff --git a/src/mavedb/routers/publication_identifiers.py b/src/mavedb/routers/publication_identifiers.py index 936f625eb..17be23a95 100644 --- a/src/mavedb/routers/publication_identifiers.py +++ b/src/mavedb/routers/publication_identifiers.py @@ -247,7 +247,7 @@ def search_publication_identifiers(search: TextSearch, db: Session = Depends(dep response_model=publication_identifier.PublicationIdentifier, summary="Search publication identifiers by their identifier", ) -async def search_publications_by_identifier(*, identifier: str, db: Session = Depends(deps.get_db)) -> Any: +def search_publications_by_identifier(*, identifier: str, db: Session = Depends(deps.get_db)) -> Any: """ Search saved publication identifiers via their identifier. """ @@ -264,7 +264,7 @@ async def search_publications_by_identifier(*, identifier: str, db: Session = De response_model=list[publication_identifier.PublicationIdentifier], summary="Search publication identifiers by their identifier and database", ) -async def search_publications_by_identifier_and_db( +def search_publications_by_identifier_and_db( *, identifier: str, db_name: str, @@ -297,7 +297,7 @@ async def search_publications_by_identifier_and_db( }, summary="Search external publication identifiers", ) -async def search_external_publication_identifiers(search: TextSearch, db: Session = Depends(deps.get_db)) -> Any: +def search_external_publication_identifiers(search: TextSearch, db: Session = Depends(deps.get_db)) -> Any: """ Search external publication identifiers via a TextSearch query. The provided text is searched against multiple external publication databases, and should be a valid identifier in at least one of those databases. @@ -305,7 +305,7 @@ async def search_external_publication_identifiers(search: TextSearch, db: Sessio if search.text and len(search.text.strip()) > 0: lower_search_text = search.text.strip().lower() - items = await find_generic_article(db, lower_search_text) + items = find_generic_article(db, lower_search_text) else: raise HTTPException(status_code=400, detail="Search text is required") diff --git a/src/mavedb/routers/score_calibrations.py b/src/mavedb/routers/score_calibrations.py index f8f43e4af..998260f67 100644 --- a/src/mavedb/routers/score_calibrations.py +++ b/src/mavedb/routers/score_calibrations.py @@ -130,7 +130,7 @@ def get_score_calibration( response_model=list[score_calibration.ScoreCalibrationWithScoreSetUrn], responses={404: {}}, ) -async def get_score_calibrations_for_score_set( +def get_score_calibrations_for_score_set( *, score_set_urn: str, db: Session = Depends(deps.get_db), @@ -170,7 +170,7 @@ async def get_score_calibrations_for_score_set( response_model=score_calibration.ScoreCalibrationDetailWithScoreSetUrn, responses={404: {}}, ) -async def get_primary_score_calibrations_for_score_set( +def get_primary_score_calibrations_for_score_set( *, score_set_urn: str, db: Session = Depends(deps.get_db), @@ -256,7 +256,7 @@ async def get_primary_score_calibrations_for_score_set( } }, ) -async def create_score_calibration_route( +def create_score_calibration_route( *, calibration: score_calibration.ScoreCalibrationCreate = Depends(calibration_create_loader), classes_file: Optional[UploadFile] = File( @@ -405,7 +405,7 @@ async def create_score_calibration_route( ) try: - created_calibration = await create_score_calibration_in_score_set( + created_calibration = create_score_calibration_in_score_set( db, calibration, user_data.user, variant_classes if classes_file else None ) except ValidationError as e: @@ -457,7 +457,7 @@ async def create_score_calibration_route( } }, ) -async def modify_score_calibration_route( +def modify_score_calibration_route( *, urn: str, calibration_update: score_calibration.ScoreCalibrationModify = Depends(calibration_modify_loader), @@ -633,7 +633,7 @@ async def modify_score_calibration_route( ) try: - updated_calibration = await modify_score_calibration( + updated_calibration = modify_score_calibration( db, item, calibration_update, user_data.user, variant_classes if classes_file else None ) except ValidationError as e: @@ -654,7 +654,7 @@ async def modify_score_calibration_route( responses={404: {}}, status_code=204, ) -async def delete_score_calibration_route( +def delete_score_calibration_route( *, urn: str, db: Session = Depends(deps.get_db), @@ -688,7 +688,7 @@ async def delete_score_calibration_route( response_model=score_calibration.ScoreCalibrationDetailWithScoreSetUrn, responses={404: {}}, ) -async def promote_score_calibration_to_primary_route( +def promote_score_calibration_to_primary_route( *, urn: str, demote_existing_primary: bool = Query( diff --git a/src/mavedb/routers/score_sets.py b/src/mavedb/routers/score_sets.py index 01ed631dc..4c8444080 100644 --- a/src/mavedb/routers/score_sets.py +++ b/src/mavedb/routers/score_sets.py @@ -11,6 +11,7 @@ import requests from arq import ArqRedis from fastapi import APIRouter, Depends, File, Query, Request, Response, UploadFile +from fastapi.concurrency import run_in_threadpool from fastapi.encoders import jsonable_encoder from fastapi.exceptions import HTTPException, RequestValidationError from fastapi.responses import StreamingResponse @@ -20,6 +21,7 @@ from sqlalchemy import or_, select from sqlalchemy.exc import IntegrityError, MultipleResultsFound from sqlalchemy.orm import Session +from starlette.datastructures import FormData from mavedb import deps from mavedb.data_providers.services import CSV_UPLOAD_S3_BUCKET_NAME, s3_client @@ -93,7 +95,7 @@ ) from mavedb.lib.validation.exceptions import ValidationError as MaveDBValidationError from mavedb.lib.variant_detail import load_variant_detail_inputs -from mavedb.lib.workflow.kickoff import enqueue_pipeline_for_score_set +from mavedb.lib.workflow.kickoff import enqueue_pipeline_for_score_set_from_thread from mavedb.models.contributor import Contributor from mavedb.models.enums.processing_state import ProcessingState from mavedb.models.experiment import Experiment @@ -131,7 +133,7 @@ SCORE_SET_SEARCH_MAX_PUBLICATION_IDENTIFIERS = 40 -async def enqueue_variant_creation( +def enqueue_variant_creation( *, item: ScoreSet, user_data: UserData, @@ -198,7 +200,7 @@ async def enqueue_variant_creation( ) try: - await enqueue_pipeline_for_score_set( + enqueue_pipeline_for_score_set_from_thread( db=db, redis=worker, pipeline_name="validate_map_annotate_score_set", @@ -239,7 +241,7 @@ class ScoreSetUpdateResult(TypedDict): should_create_variants: bool -async def score_set_update( +def score_set_update( *, db: Session, urn: str, @@ -299,7 +301,7 @@ async def score_set_update( DoiIdentifierCreate(**identifier) for identifier in item_update_dict.get("doi_identifiers") or [] ] item.doi_identifiers = [ - await find_or_create_doi_identifier(db, identifier.identifier) for identifier in doi_identifiers_list + find_or_create_doi_identifier(db, identifier.identifier) for identifier in doi_identifiers_list ] if any(key in item_update_dict for key in ["primary_publication_identifiers", "secondary_publication_identifiers"]): @@ -310,7 +312,7 @@ async def score_set_update( ] try: primary_publication_identifiers = [ - await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) for identifier in primary_publication_identifiers_list ] except requests.exceptions.ConnectTimeout: @@ -339,7 +341,7 @@ async def score_set_update( ] try: secondary_publication_identifiers = [ - await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) for identifier in secondary_publication_identifiers_list ] except requests.exceptions.ConnectTimeout: @@ -379,9 +381,7 @@ async def score_set_update( contributors = [ ContributorCreate(**contributor) for contributor in item_update_dict.get("contributors") or [] ] - item.contributors = [ - await find_or_create_contributor(db, contributor.orcid_id) for contributor in contributors - ] + item.contributors = [find_or_create_contributor(db, contributor.orcid_id) for contributor in contributors] except NonexistentOrcidUserError as e: logger.error(msg="Could not find ORCID user with the provided user ID.", extra=logging_context()) raise HTTPException(status_code=404, detail=str(e)) @@ -411,7 +411,7 @@ async def score_set_update( upload_taxonomy = gene.target_sequence.taxonomy save_to_logging_context({"requested_taxonomy": gene.target_sequence.taxonomy.code}) - taxonomy = await find_or_create_taxonomy(db, upload_taxonomy) + taxonomy = find_or_create_taxonomy(db, upload_taxonomy) if not taxonomy: logger.info( @@ -480,7 +480,7 @@ async def score_set_update( for external_gene_identifier_offset_create in gene.external_identifiers: offset = external_gene_identifier_offset_create.offset identifier_create = external_gene_identifier_offset_create.identifier - await create_external_gene_identifier_offset( + create_external_gene_identifier_offset( db, target_gene, identifier_create.db_name, @@ -515,7 +515,7 @@ class ParseScoreSetUpdate(TypedDict): counts_df: Optional[pd.DataFrame] -async def parse_score_set_variants_uploads( +def parse_score_set_variants_uploads( scores_file: Optional[UploadFile] = File(None), counts_file: Optional[UploadFile] = File(None), ) -> ParseScoreSetUpdate: @@ -555,7 +555,7 @@ async def parse_score_set_variants_uploads( } -async def fetch_score_set_by_urn( +def fetch_score_set_by_urn( db, urn: str, user: Optional[UserData], owner_or_contributor: Optional[UserData], only_published: bool ) -> ScoreSet: """ @@ -868,7 +868,7 @@ def list_recently_published_score_sets( response_model_exclude_none=True, summary="Fetch score sets by URN list", ) -async def show_score_sets( +def show_score_sets( *, urns: str = Query(..., description="Comma-separated list of score set URNs"), db: Session = Depends(deps.get_db), @@ -885,7 +885,7 @@ async def show_score_sets( save_to_logging_context({"requested_resource": urn_list}) response_items: list[score_set.ScoreSet] = [] for urn in urn_list: - item = await fetch_score_set_by_urn(db, urn, user_data, None, False) + item = fetch_score_set_by_urn(db, urn, user_data, None, False) response_items.append(_score_set_response(item, principal)) return response_items @@ -899,7 +899,7 @@ async def show_score_sets( response_model_exclude_none=True, summary="Fetch score set by URN", ) -async def show_score_set( +def show_score_set( *, urn: str, db: Session = Depends(deps.get_db), @@ -910,7 +910,7 @@ async def show_score_set( Fetch a single score set by URN. """ save_to_logging_context({"requested_resource": urn}) - item = await fetch_score_set_by_urn(db, urn, user_data, None, False) + item = fetch_score_set_by_urn(db, urn, user_data, None, False) return _score_set_response(item, principal) @@ -985,7 +985,7 @@ def get_score_set_csv_namespaces( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Get the lean whole-set variant view for a score set", ) -async def get_score_set_lean_variants( +def get_score_set_lean_variants( *, urn: str, response: Response, @@ -1013,7 +1013,7 @@ async def get_score_set_lean_variants( save_to_logging_context({"requested_resource": urn, "resource_property": "lean-variants", "as_of": as_of}) response.headers["X-As-Of"] = as_of.isoformat() if as_of is not None else "current" - score_set = await fetch_score_set_by_urn(db, urn, user_data, None, False) + score_set = fetch_score_set_by_urn(db, urn, user_data, None, False) return get_lean_score_set_variants(db, score_set, as_of=as_of) @@ -1062,7 +1062,7 @@ def _stream_score_set_variant_details( }, summary="Download a score set's variant details (VRS + Cat-VRS + annotations)", ) -async def get_score_set_variant_details( +def get_score_set_variant_details( *, urn: str, as_of: Optional[datetime] = Query( @@ -1093,7 +1093,7 @@ async def get_score_set_variant_details( """ save_to_logging_context({"requested_resource": urn, "resource_property": "variant-details", "as_of": as_of}) - score_set = await fetch_score_set_by_urn(db, urn, user_data, None, False) + score_set = fetch_score_set_by_urn(db, urn, user_data, None, False) # Resolved here rather than inherited: fetch_score_set_by_urn returns the score set as it actually # is, leaving narrowing to each response path. Mirrors the single-variant route, which must agree @@ -1944,7 +1944,7 @@ def get_score_set_annotated_variants_functional_study_result( responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE, **GATEWAY_ERROR_RESPONSES}, summary="Create a score set", ) -async def create_score_set( +def create_score_set( *, item_create: score_set.ScoreSetCreate, db: Session = Depends(deps.get_db), @@ -1993,7 +1993,7 @@ async def create_score_set( if item_create.superseded_score_set_urn is not None: # Passing user_data as owner_or_contributor is what authorizes the supersession: the fetch returns # only published score sets this user owns or contributes to. There is no Action for supersession yet. - superseded_score_set = await fetch_score_set_by_urn( + superseded_score_set = fetch_score_set_by_urn( db, item_create.superseded_score_set_urn, user_data, user_data, True ) @@ -2023,8 +2023,7 @@ async def create_score_set( meta_analyzes_score_sets = [ ss for ss in [ - await fetch_score_set_by_urn(db, urn, user_data, None, True) - for urn in distinct_meta_analyzes_score_set_urns + fetch_score_set_by_urn(db, urn, user_data, None, True) for urn in distinct_meta_analyzes_score_set_urns ] if ss is not None ] @@ -2109,7 +2108,7 @@ async def create_score_set( contributors: list[Contributor] = [] try: contributors = [ - await find_or_create_contributor(db, contributor.orcid_id) for contributor in item_create.contributors or [] + find_or_create_contributor(db, contributor.orcid_id) for contributor in item_create.contributors or [] ] except NonexistentOrcidUserError as e: logger.error(msg="Could not find ORCID user with the provided user ID.", extra=logging_context()) @@ -2117,15 +2116,14 @@ async def create_score_set( try: doi_identifiers = [ - await find_or_create_doi_identifier(db, identifier.identifier) - for identifier in item_create.doi_identifiers or [] + find_or_create_doi_identifier(db, identifier.identifier) for identifier in item_create.doi_identifiers or [] ] primary_publication_identifiers = [ - await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) for identifier in item_create.primary_publication_identifiers or [] ] publication_identifiers = [ - await find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) + find_or_create_publication_identifier(db, identifier.identifier, identifier.db_name) for identifier in item_create.secondary_publication_identifiers or [] ] + primary_publication_identifiers @@ -2175,7 +2173,7 @@ async def create_score_set( ) try: - created_calibration_item = await create_score_calibration( + created_calibration_item = create_score_calibration( db, calibration_create, user_data.user, variant_classes=None ) except MaveDBValidationError as e: @@ -2208,7 +2206,7 @@ async def create_score_set( ) upload_taxonomy = gene.target_sequence.taxonomy save_to_logging_context({"requested_taxonomy": gene.target_sequence.taxonomy.code}) - taxonomy = await find_or_create_taxonomy(db, upload_taxonomy) + taxonomy = find_or_create_taxonomy(db, upload_taxonomy) if not taxonomy: logger.info( @@ -2271,7 +2269,7 @@ async def create_score_set( for external_gene_identifier_offset_create in gene.external_identifiers: offset = external_gene_identifier_offset_create.offset identifier_create = external_gene_identifier_offset_create.identifier - await create_external_gene_identifier_offset( + create_external_gene_identifier_offset( db, target_gene, identifier_create.db_name, @@ -2391,12 +2389,43 @@ async def upload_score_set_variant_data( Upload scores and variant count files for a score set, and initiate processing these files to create variants. """ + # `async` only to read the multipart form, the one real await. Parsing, the database work and the + # enqueue run in the thread pool so they don't stall the event loop. Making this `def` would mean + # declaring every form field, which changes the OpenAPI schema; revisit with #629 (#950). + try: + form_data = await data.form() + except Exception as e: + raise HTTPException(status_code=422, detail=str(e)) + + return await run_in_threadpool( + _upload_score_set_variant_data, + urn=urn, + form_data=form_data, + counts_file=counts_file, + scores_file=scores_file, + db=db, + user_data=user_data, + worker=worker, + principal=principal, + ) + + +def _upload_score_set_variant_data( + *, + urn: str, + form_data: FormData, + counts_file: Optional[UploadFile], + scores_file: Optional[UploadFile], + db: Session, + user_data: UserData, + worker: ArqRedis, + principal: Principal, +) -> Any: save_to_logging_context({"requested_resource": urn, "resource_property": "variants"}) try: - score_set_variants_data = await parse_score_set_variants_uploads(scores_file, counts_file) + score_set_variants_data = parse_score_set_variants_uploads(scores_file, counts_file) - form_data = await data.form() # Parse variants dataset column metadata JSON strings dataset_column_metadata = { key: json.loads(str(value)) @@ -2423,7 +2452,7 @@ async def upload_score_set_variant_data( logger.info(msg="Enqueuing variant creation job.", extra=logging_context()) try: - await enqueue_variant_creation( + enqueue_variant_creation( item=item, user_data=user_data, new_scores_df=score_set_variants_data["scores_df"], @@ -2522,6 +2551,38 @@ async def update_score_set_with_variants( """ Update a score set and variants. """ + # `async` only to read the multipart form, the one real await. Parsing, the database work and the + # enqueue run in the thread pool so they don't stall the event loop. Making this `def` would mean + # declaring every form field, which changes the OpenAPI schema; revisit with #629 (#950). + try: + form_data = await request.form() + except Exception as e: + raise HTTPException(status_code=422, detail=str(e)) + + return await run_in_threadpool( + _update_score_set_with_variants, + urn=urn, + form_data=form_data, + counts_file=counts_file, + scores_file=scores_file, + db=db, + user_data=user_data, + worker=worker, + principal=principal, + ) + + +def _update_score_set_with_variants( + *, + urn: str, + form_data: FormData, + counts_file: Optional[UploadFile], + scores_file: Optional[UploadFile], + db: Session, + user_data: UserData, + worker: ArqRedis, + principal: Principal, +) -> Any: logger.info(msg="Began score set with variants update.", extra=logging_context()) # TODO#629: Use `flexible_model_loader` utility here to support both form data and JSON body. @@ -2532,9 +2593,6 @@ async def update_score_set_with_variants( # set update fields as separate form fields and parsing them into an object. Doing so will also # simplify the OpenAPI schema for this endpoint. try: - # Get all form data from the request - form_data = await request.form() - # Convert form data to dictionary, excluding file and associated column metadata fields form_dict = { key: value @@ -2545,7 +2603,7 @@ async def update_score_set_with_variants( item_update_partial = score_set.ScoreSetUpdateAllOptional.as_form(**form_dict) # parse uploaded CSV files - score_set_variants_data = await parse_score_set_variants_uploads( + score_set_variants_data = parse_score_set_variants_uploads( scores_file, counts_file, ) @@ -2600,7 +2658,7 @@ async def update_score_set_with_variants( if request_sets_scores: assert_permission(user_data, existing_item, Action.SET_SCORES) - itemUpdateResult = await score_set_update( + itemUpdateResult = score_set_update( db=db, urn=urn, item_update=item_update_partial, @@ -2620,7 +2678,7 @@ async def update_score_set_with_variants( logger.info(msg="Enqueuing variant creation job.", extra=logging_context()) try: - await enqueue_variant_creation( + enqueue_variant_creation( item=updatedItem, user_data=user_data, worker=worker, @@ -2673,7 +2731,7 @@ async def update_score_set_with_variants( responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE, **GATEWAY_ERROR_RESPONSES}, summary="Update a score set", ) -async def update_score_set( +def update_score_set( *, urn: str, item_update: score_set.ScoreSetUpdate, @@ -2691,7 +2749,7 @@ async def update_score_set( # this object will contain all required fields because item_update type is ScoreSetUpdate, but # is converted to instance of ScoreSetUpdateAllOptional to match expected input of score_set_update function score_set_update_item = score_set.ScoreSetUpdateAllOptional.model_validate(item_update.model_dump()) - itemUpdateResult = await score_set_update( + itemUpdateResult = score_set_update( db=db, urn=urn, item_update=score_set_update_item, exclude_unset=False, user_data=user_data ) updatedItem = itemUpdateResult["item"] @@ -2710,7 +2768,7 @@ async def update_score_set( logger.info(msg="Enqueuing variant creation job.", extra=logging_context()) try: - await enqueue_variant_creation( + enqueue_variant_creation( item=updatedItem, user_data=user_data, worker=worker, @@ -2754,7 +2812,7 @@ async def update_score_set( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Delete a score set", ) -async def delete_score_set( +def delete_score_set( *, urn: str, db: Session = Depends(deps.get_db), @@ -2793,7 +2851,7 @@ async def delete_score_set( response_model_exclude_none=True, responses={**ACCESS_CONTROL_ERROR_RESPONSES, **BASE_409_RESPONSE}, ) -async def publish_score_set( +def publish_score_set( *, urn: str, db: Session = Depends(deps.get_db), @@ -2908,7 +2966,7 @@ async def publish_score_set( db.refresh(item) try: - await enqueue_pipeline_for_score_set( + enqueue_pipeline_for_score_set_from_thread( db=db, redis=worker, pipeline_name="publish_score_set", @@ -2933,7 +2991,7 @@ async def publish_score_set( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Get clinical controls for a score set", ) -async def get_clinical_controls_for_score_set( +def get_clinical_controls_for_score_set( *, urn: str, response: Response, @@ -3021,7 +3079,7 @@ async def get_clinical_controls_for_score_set( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Get clinical control options for a score set", ) -async def get_clinical_controls_options_for_score_set( +def get_clinical_controls_options_for_score_set( *, urn: str, as_of: Optional[datetime] = Query( @@ -3085,7 +3143,7 @@ async def get_clinical_controls_options_for_score_set( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Get gnomad variants for a score set", ) -async def get_gnomad_variants_for_score_set( +def get_gnomad_variants_for_score_set( *, urn: str, response: Response, diff --git a/src/mavedb/routers/taxonomies.py b/src/mavedb/routers/taxonomies.py index 0d6801661..fb6d945f1 100644 --- a/src/mavedb/routers/taxonomies.py +++ b/src/mavedb/routers/taxonomies.py @@ -96,7 +96,7 @@ def fetch_taxonomy_by_code( @router.post("/search", status_code=200, response_model=List[taxonomy.Taxonomy], summary="Search taxonomies") -async def search_taxonomies(search: TextSearch, db: Session = Depends(deps.get_db)) -> Any: +def search_taxonomies(search: TextSearch, db: Session = Depends(deps.get_db)) -> Any: """ Search Taxonomy. If no search text, return the whole taxonomy list so that front end Taxonomy component can get data to show in dropdown button. @@ -117,7 +117,7 @@ async def search_taxonomies(search: TextSearch, db: Session = Depends(deps.get_d items = query.order_by(Taxonomy.organism_name).all() if not items and search.text: - search_taxonomy = await search_NCBI_taxonomy(db, search.text) + search_taxonomy = search_NCBI_taxonomy(db, search.text) if search_taxonomy: items = [search_taxonomy] else: diff --git a/src/mavedb/routers/users.py b/src/mavedb/routers/users.py index 7aafc0ab0..68c15072a 100644 --- a/src/mavedb/routers/users.py +++ b/src/mavedb/routers/users.py @@ -55,7 +55,7 @@ def to_string(self, value: str) -> str: responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="List users", ) -async def list_users( +def list_users( *, db: Session = Depends(deps.get_db), _: UserData = Depends(RoleRequirer([UserRole.admin])), @@ -74,7 +74,7 @@ async def list_users( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Show my user", ) -async def show_me(*, user_data: UserData = Depends(require_current_user)) -> User: +def show_me(*, user_data: UserData = Depends(require_current_user)) -> User: """ Return the current user. """ @@ -88,7 +88,7 @@ async def show_me(*, user_data: UserData = Depends(require_current_user)) -> Use responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Show user by ID", ) -async def show_user_admin( +def show_user_admin( *, id: int, user_data: UserData = Depends(RoleRequirer([UserRole.admin])), @@ -118,7 +118,7 @@ async def show_user_admin( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Show user by Orcid ID", ) -async def show_user( +def show_user( *, orcid_id: str, user_data: UserData = Depends(require_current_user), @@ -149,7 +149,7 @@ async def show_user( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Update my user", ) -async def update_me( +def update_me( *, user_update: user.CurrentUserUpdate, db: Session = Depends(deps.get_db), @@ -175,7 +175,7 @@ async def update_me( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Mark that the current user has logged in", ) -async def user_has_logged_in( +def user_has_logged_in( *, db: Session = Depends(deps.get_db), user_data: UserData = Depends(require_current_user), @@ -200,7 +200,7 @@ async def user_has_logged_in( responses={**ACCESS_CONTROL_ERROR_RESPONSES}, summary="Update user by ID", ) -async def update_user( +def update_user( *, id: int, item_update: user.AdminUserUpdate, @@ -229,7 +229,7 @@ async def update_user( if item_update.email: item.email = item_update.email if item_update.roles: - await item.set_roles(db, item_update.roles) + item.set_roles(db, item_update.roles) db.add(item) db.commit() diff --git a/src/mavedb/scripts/load_calibration_csv.py b/src/mavedb/scripts/load_calibration_csv.py index 066509e39..aa0579dc3 100644 --- a/src/mavedb/scripts/load_calibration_csv.py +++ b/src/mavedb/scripts/load_calibration_csv.py @@ -88,7 +88,6 @@ """ -import asyncio import csv import re from pathlib import Path @@ -414,8 +413,8 @@ def main(db: Session, csv_path: str, delimiter: str, overwrite: bool, purge_publ system_user = db.query(User).filter(User.id == 1).one() calibration_user = score_set.created_by if calibration_is_investigator_provided else system_user - new_calibration_object = asyncio.run( - create_score_calibration_in_score_set(db, created_score_calibration, calibration_user) + new_calibration_object = create_score_calibration_in_score_set( + db, created_score_calibration, calibration_user ) new_calibration_object.primary = primary new_calibration_object.private = False diff --git a/src/mavedb/scripts/load_excalibr_calibrations.py b/src/mavedb/scripts/load_excalibr_calibrations.py index 72f7ed2cd..1ff28a48a 100644 --- a/src/mavedb/scripts/load_excalibr_calibrations.py +++ b/src/mavedb/scripts/load_excalibr_calibrations.py @@ -40,7 +40,6 @@ python load_excalibr_calibrations_csv.py /path/to/calibrations.csv /path/to/dataset_mapping.json --overwrite """ -import asyncio import csv import json from typing import Dict, List, Optional, Tuple @@ -238,8 +237,8 @@ def main(db: Session, csv_path: str, dataset_map: str, overwrite: bool, remove: method_sources=[EXCALIBR_CALIBRATION_CITATION], ) - new_calibration_object = asyncio.run( - create_score_calibration_in_score_set(db, score_calibration_create, system_user) + new_calibration_object = create_score_calibration_in_score_set( + db, score_calibration_create, system_user ) new_calibration_object.primary = False new_calibration_object.private = False diff --git a/tests/helpers/util/score_calibration.py b/tests/helpers/util/score_calibration.py index ab3d0b9c0..e6f48b525 100644 --- a/tests/helpers/util/score_calibration.py +++ b/tests/helpers/util/score_calibration.py @@ -13,13 +13,13 @@ from sqlalchemy.orm import Session -async def create_test_range_based_score_calibration_in_score_set( +def create_test_range_based_score_calibration_in_score_set( db: "Session", score_set_urn: str, user: User ) -> ScoreCalibration: calibration_create = ScoreCalibrationCreate( **TEST_BRNICH_SCORE_CALIBRATION_RANGE_BASED, score_set_urn=score_set_urn ) - created_score_calibration = await create_score_calibration_in_score_set(db, calibration_create, user) + created_score_calibration = create_score_calibration_in_score_set(db, calibration_create, user) assert created_score_calibration is not None db.commit() diff --git a/tests/lib/mondo/network/test_mondo.py b/tests/lib/mondo/network/test_mondo.py index 8e056927f..146e2e6eb 100644 --- a/tests/lib/mondo/network/test_mondo.py +++ b/tests/lib/mondo/network/test_mondo.py @@ -20,13 +20,12 @@ @pytest.mark.network -@pytest.mark.asyncio class TestMondoOlsNetwork: """Canary tests exercising the real OLS4 service.""" - async def test_search_returns_known_term_with_expected_shape(self): + def test_search_returns_known_term_with_expected_shape(self): # /select endpoint: a known query surfaces the expected code, in the {code, label, iri} shape. - results = await search_mondo("brugada syndrome") + results = search_mondo("brugada syndrome") assert results, "OLS returned no results for a known disease query." assert BRUGADA_CODE in {result["code"] for result in results} first = results[0] @@ -34,22 +33,22 @@ async def test_search_returns_known_term_with_expected_shape(self): assert first["label"] assert first["iri"].startswith("http") - async def test_search_broad_prefix_returns_results(self): + def test_search_broad_prefix_returns_results(self): # A two-letter prefix must still return via /select — /search timed out on these (the #754 switch). - assert await search_mondo("ca") + assert search_mondo("ca") - async def test_fetch_known_code_returns_canonical_term(self): + def test_fetch_known_code_returns_canonical_term(self): # /search exact-obo_id lookup: a known code resolves to its canonical label. - resolved = await fetch_mondo_term(BRUGADA_CODE) + resolved = fetch_mondo_term(BRUGADA_CODE) assert resolved is not None assert resolved["code"] == BRUGADA_CODE assert "brugada" in resolved["label"].lower() - async def test_fetch_generic_root_resolves(self): - resolved = await fetch_mondo_term(MONDO_GENERIC_CODE) + def test_fetch_generic_root_resolves(self): + resolved = fetch_mondo_term(MONDO_GENERIC_CODE) assert resolved is not None assert resolved["code"] == MONDO_GENERIC_CODE - async def test_fetch_unknown_code_returns_none(self): + def test_fetch_unknown_code_returns_none(self): # The not-found path must stay a clean None so validation can reject unknown codes. - assert await fetch_mondo_term("MONDO:9999999") is None + assert fetch_mondo_term("MONDO:9999999") is None diff --git a/tests/lib/mondo/test_mondo.py b/tests/lib/mondo/test_mondo.py index 409fe8102..9023d8d6a 100644 --- a/tests/lib/mondo/test_mondo.py +++ b/tests/lib/mondo/test_mondo.py @@ -36,32 +36,32 @@ def json(self): return self._payload -class _FakeAsyncClient: - """Minimal stand-in for httpx.AsyncClient returning a canned OLS payload.""" +class _FakeClient: + """Minimal stand-in for httpx.Client returning a canned OLS payload.""" def __init__(self, payload): self._payload = payload - async def __aenter__(self): + def __enter__(self): return self - async def __aexit__(self, *exc): + def __exit__(self, *exc): return False - async def get(self, url, params=None): + def get(self, url, params=None): return _FakeResponse(self._payload) -class _FailingAsyncClient: - """Stand-in for httpx.AsyncClient whose request fails, as when OLS is unreachable.""" +class _FailingClient: + """Stand-in for httpx.Client whose request fails, as when OLS is unreachable.""" - async def __aenter__(self): + def __enter__(self): return self - async def __aexit__(self, *exc): + def __exit__(self, *exc): return False - async def get(self, url, params=None): + def get(self, url, params=None): raise httpx.ConnectError("ols down") @@ -98,9 +98,8 @@ def test_suggestion_to_mappable_concept(self): assert concept.primaryCoding.iris[0].root == "https://example.org/MONDO_0015263" -@pytest.mark.asyncio class TestMondoOlsSearch: - async def test_search_mondo_parses_ols_docs(self, monkeypatch): + def test_search_mondo_parses_ols_docs(self, monkeypatch): payload = _ols_payload( [ { @@ -111,9 +110,9 @@ async def test_search_mondo_parses_ols_docs(self, monkeypatch): {"obo_id": "MONDO:0000001", "label": "disease or disorder"}, # missing iri → derived ] ) - monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: _FakeAsyncClient(payload)) + monkeypatch.setattr(httpx, "Client", lambda *a, **k: _FakeClient(payload)) - results = await search_mondo("brugada") + results = search_mondo("brugada") assert results[0] == { "code": "MONDO:0015263", "label": "Brugada syndrome", @@ -121,86 +120,85 @@ async def test_search_mondo_parses_ols_docs(self, monkeypatch): } assert results[1]["iri"] == mondo_iri("MONDO:0000001") - async def test_search_mondo_empty_query_skips_request(self, monkeypatch): + def test_search_mondo_empty_query_skips_request(self, monkeypatch): # No network expected; if a request were made the fake would still be needed, so assert on the shortcut. - assert await search_mondo(" ") == [] + assert search_mondo(" ") == [] - async def test_fetch_mondo_term_returns_none_when_absent(self, monkeypatch): - monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: _FakeAsyncClient(_ols_payload([]))) - assert await fetch_mondo_term("MONDO:9999999") is None + def test_fetch_mondo_term_returns_none_when_absent(self, monkeypatch): + monkeypatch.setattr(httpx, "Client", lambda *a, **k: _FakeClient(_ols_payload([]))) + assert fetch_mondo_term("MONDO:9999999") is None - async def test_search_mondo_wraps_http_errors_in_service_error(self, monkeypatch): - monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: _FailingAsyncClient()) + def test_search_mondo_wraps_http_errors_in_service_error(self, monkeypatch): + monkeypatch.setattr(httpx, "Client", lambda *a, **k: _FailingClient()) with pytest.raises(MondoServiceError): - await search_mondo("brugada") + search_mondo("brugada") - async def test_fetch_mondo_term_wraps_http_errors_in_service_error(self, monkeypatch): - monkeypatch.setattr(httpx, "AsyncClient", lambda *a, **k: _FailingAsyncClient()) + def test_fetch_mondo_term_wraps_http_errors_in_service_error(self, monkeypatch): + monkeypatch.setattr(httpx, "Client", lambda *a, **k: _FailingClient()) with pytest.raises(MondoServiceError): - await fetch_mondo_term("MONDO:0015263") + fetch_mondo_term("MONDO:0015263") -@pytest.mark.asyncio class TestFindOrCreateMondoTerm: - async def test_creates_from_ols_and_is_idempotent(self, session, monkeypatch): - async def fake_fetch(code): + def test_creates_from_ols_and_is_idempotent(self, session, monkeypatch): + def fake_fetch(code): return {"code": code, "label": "Brugada syndrome", "iri": mondo_iri(code)} monkeypatch.setattr("mavedb.lib.mondo_ols.fetch_mondo_term", fake_fetch) - term = await find_or_create_mondo_term(session, "MONDO:0015263") + term = find_or_create_mondo_term(session, "MONDO:0015263") assert term.code == "MONDO:0015263" assert term.label == "Brugada syndrome" assert term.system == MONDO_SYSTEM # A second call reuses the row rather than creating a duplicate. - again = await find_or_create_mondo_term(session, "MONDO:0015263") + again = find_or_create_mondo_term(session, "MONDO:0015263") assert again.id == term.id assert session.query(MondoTerm).filter(MondoTerm.code == "MONDO:0015263").count() == 1 - async def test_non_canonical_code_reuses_existing_canonical_row(self, session, monkeypatch): + def test_non_canonical_code_reuses_existing_canonical_row(self, session, monkeypatch): # OLS normalizes both spellings to the same canonical code; the lookup must key on the resolved # code so a second submission reuses the row instead of tripping UNIQUE(system, code). - async def fake_fetch(code): + def fake_fetch(code): return {"code": "MONDO:0015263", "label": "Brugada syndrome", "iri": mondo_iri("MONDO:0015263")} monkeypatch.setattr("mavedb.lib.mondo_ols.fetch_mondo_term", fake_fetch) - canonical = await find_or_create_mondo_term(session, "MONDO:0015263") + canonical = find_or_create_mondo_term(session, "MONDO:0015263") # A differently-spelled submission (here lowercase) resolves to the same canonical term. - aliased = await find_or_create_mondo_term(session, "mondo:0015263") + aliased = find_or_create_mondo_term(session, "mondo:0015263") assert aliased.id == canonical.id assert aliased.code == "MONDO:0015263" assert session.query(MondoTerm).filter(MondoTerm.code == "MONDO:0015263").count() == 1 - async def test_generic_code_resolves_without_ols(self, session, monkeypatch): - async def fail_fetch(code): + def test_generic_code_resolves_without_ols(self, session, monkeypatch): + def fail_fetch(code): raise AssertionError("OLS should not be consulted for the generic term") monkeypatch.setattr("mavedb.lib.mondo_ols.fetch_mondo_term", fail_fetch) - term = await find_or_create_mondo_term(session, MONDO_GENERIC_CODE) + term = find_or_create_mondo_term(session, MONDO_GENERIC_CODE) assert term.code == MONDO_GENERIC_CODE assert term.label == MONDO_GENERIC_LABEL - async def test_rejects_unknown_code(self, session, monkeypatch): - async def fake_fetch(code): + def test_rejects_unknown_code(self, session, monkeypatch): + def fake_fetch(code): return None monkeypatch.setattr("mavedb.lib.mondo_ols.fetch_mondo_term", fake_fetch) with pytest.raises(ValidationError): - await find_or_create_mondo_term(session, "MONDO:9999999") + find_or_create_mondo_term(session, "MONDO:9999999") - async def test_ols_outage_propagates_service_error(self, session, monkeypatch): + def test_ols_outage_propagates_service_error(self, session, monkeypatch): # A controlled term is never persisted unvalidated: an OLS outage surfaces as MondoServiceError. - async def fake_fetch(code): + def fake_fetch(code): raise MondoServiceError("ols down") monkeypatch.setattr("mavedb.lib.mondo_ols.fetch_mondo_term", fake_fetch) with pytest.raises(MondoServiceError): - await find_or_create_mondo_term(session, "MONDO:0015263") + find_or_create_mondo_term(session, "MONDO:0015263") @pytest.mark.unit diff --git a/tests/lib/test_identifiers.py b/tests/lib/test_identifiers.py index cc378223b..ffe248c69 100644 --- a/tests/lib/test_identifiers.py +++ b/tests/lib/test_identifiers.py @@ -53,8 +53,7 @@ def biorxiv_url(identifier): return f"https://api.biorxiv.org/details/biorxiv/10.1101/{identifier}/na/json" -@pytest.mark.asyncio -async def test_find_generic_article_drops_candidate_with_malformed_response(session, requests_mock): +def test_find_generic_article_drops_candidate_with_malformed_response(session, requests_mock): """ A PMID like TEST_PUBMED_IDENTIFIER is also a plausible (8 digit) legacy medRxiv identifier, so find_generic_article fans out to both databases. If medRxiv responds @@ -64,14 +63,13 @@ async def test_find_generic_article_drops_candidate_with_malformed_response(sess mock_pubmed_success(requests_mock, TEST_PUBMED_IDENTIFIER) requests_mock.get(medrxiv_url(TEST_PUBMED_IDENTIFIER), text="not json") - matches = await find_generic_article(session, TEST_PUBMED_IDENTIFIER) + matches = find_generic_article(session, TEST_PUBMED_IDENTIFIER) assert matches["PubMed"] is not None assert matches.get("medRxiv") is None -@pytest.mark.asyncio -async def test_find_generic_article_drops_candidate_on_request_exception(session, requests_mock): +def test_find_generic_article_drops_candidate_on_request_exception(session, requests_mock): """ The same fan-out should also tolerate a candidate database being entirely unreachable (timeouts, connection errors, bad status codes), not just malformed JSON bodies. @@ -79,14 +77,13 @@ async def test_find_generic_article_drops_candidate_on_request_exception(session mock_pubmed_success(requests_mock, TEST_PUBMED_IDENTIFIER) requests_mock.get(medrxiv_url(TEST_PUBMED_IDENTIFIER), exc=requests.exceptions.ConnectTimeout) - matches = await find_generic_article(session, TEST_PUBMED_IDENTIFIER) + matches = find_generic_article(session, TEST_PUBMED_IDENTIFIER) assert matches["PubMed"] is not None assert matches.get("medRxiv") is None -@pytest.mark.asyncio -async def test_find_generic_article_reraises_error_when_no_candidate_succeeds(session, requests_mock): +def test_find_generic_article_reraises_error_when_no_candidate_succeeds(session, requests_mock): """ When every candidate database fails and none produces a match, the underlying error should propagate instead of being silently reported as "not found", so callers can @@ -95,4 +92,4 @@ async def test_find_generic_article_reraises_error_when_no_candidate_succeeds(se requests_mock.get(biorxiv_url(TEST_BIORXIV_IDENTIFIER), exc=requests.exceptions.ConnectTimeout) with pytest.raises(requests.exceptions.ConnectTimeout): - await find_generic_article(session, TEST_BIORXIV_IDENTIFIER) + find_generic_article(session, TEST_BIORXIV_IDENTIFIER) diff --git a/tests/lib/test_score_calibrations.py b/tests/lib/test_score_calibrations.py index b941cc089..1cc3971c3 100644 --- a/tests/lib/test_score_calibrations.py +++ b/tests/lib/test_score_calibrations.py @@ -312,8 +312,7 @@ def test_create_functional_classification_does_not_commit_transaction(setup_lib_ ### create_score_calibration_in_score_set -@pytest.mark.asyncio -async def test_create_score_set_in_score_set_raises_value_error_when_score_set_urn_is_missing( +def test_create_score_set_in_score_set_raises_value_error_when_score_set_urn_is_missing( setup_lib_db, session, mock_user ): MockCalibrationCreate = create_model("MockCalibrationCreate", score_set_urn=(str | None, None)) @@ -321,11 +320,10 @@ async def test_create_score_set_in_score_set_raises_value_error_when_score_set_u ValueError, match="score_set_urn must be provided to create a score calibration.", ): - await create_score_calibration_in_score_set(session, MockCalibrationCreate(), mock_user) + create_score_calibration_in_score_set(session, MockCalibrationCreate(), mock_user) -@pytest.mark.asyncio -async def test_create_score_set_in_score_set_raises_no_result_found_error_when_score_set_does_not_exist( +def test_create_score_set_in_score_set_raises_no_result_found_error_when_score_set_does_not_exist( setup_lib_db, session, mock_user ): MockCalibrationCreate = create_model("MockCalibrationCreate", score_set_urn=(str | None, "urn:invalid")) @@ -333,11 +331,10 @@ async def test_create_score_set_in_score_set_raises_no_result_found_error_when_s NoResultFound, match="No row was found when one was required", ): - await create_score_calibration_in_score_set(session, MockCalibrationCreate(), mock_user) + create_score_calibration_in_score_set(session, MockCalibrationCreate(), mock_user) -@pytest.mark.asyncio -async def test_create_score_calibration_in_score_set_creates_score_calibration_when_score_set_exists( +def test_create_score_calibration_in_score_set_creates_score_calibration_when_score_set_exists( setup_lib_db_with_score_set, session ): test_user = session.execute(select(User)).scalars().first() @@ -351,13 +348,12 @@ async def test_create_score_calibration_in_score_set_creates_score_calibration_w functional_classifications=(list, []), ) - calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), test_user) + calibration = create_score_calibration_in_score_set(session, MockCalibrationCreate(), test_user) assert calibration is not None assert calibration.score_set == setup_lib_db_with_score_set -@pytest.mark.asyncio -async def test_create_score_calibration_in_score_set_investigator_provided_set_when_creator_is_owner( +def test_create_score_calibration_in_score_set_investigator_provided_set_when_creator_is_owner( setup_lib_db_with_score_set, session, mock_user ): test_user = session.execute(select(User)).scalars().first() @@ -371,7 +367,7 @@ async def test_create_score_calibration_in_score_set_investigator_provided_set_w functional_classifications=(list, []), ) - calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), test_user) + calibration = create_score_calibration_in_score_set(session, MockCalibrationCreate(), test_user) assert calibration is not None assert calibration.score_set == setup_lib_db_with_score_set assert calibration.created_by == test_user @@ -379,8 +375,7 @@ async def test_create_score_calibration_in_score_set_investigator_provided_set_w assert calibration.investigator_provided is True -@pytest.mark.asyncio -async def test_create_score_calibration_in_score_set_investigator_provided_set_when_creator_is_contributor( +def test_create_score_calibration_in_score_set_investigator_provided_set_when_creator_is_contributor( setup_lib_db_with_score_set, session ): extra_user = session.execute(select(User).where(User.username == EXTRA_USER["username"])).scalars().first() @@ -403,7 +398,7 @@ async def test_create_score_calibration_in_score_set_investigator_provided_set_w functional_classifications=(list, []), ) - calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user) + calibration = create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user) assert calibration is not None assert calibration.score_set == setup_lib_db_with_score_set assert calibration.created_by == extra_user @@ -411,8 +406,7 @@ async def test_create_score_calibration_in_score_set_investigator_provided_set_w assert calibration.investigator_provided is True -@pytest.mark.asyncio -async def test_create_score_calibration_in_score_set_investigator_provided_not_set_when_creator_not_owner( +def test_create_score_calibration_in_score_set_investigator_provided_not_set_when_creator_not_owner( setup_lib_db_with_score_set, session ): MockCalibrationCreate = create_model( @@ -427,7 +421,7 @@ async def test_create_score_calibration_in_score_set_investigator_provided_not_s # invoke from a different user context extra_user = session.execute(select(User).where(User.username == EXTRA_USER["username"])).scalars().first() - calibration = await create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user) + calibration = create_score_calibration_in_score_set(session, MockCalibrationCreate(), extra_user) assert calibration is not None assert calibration.score_set == setup_lib_db_with_score_set assert calibration.created_by == extra_user @@ -438,20 +432,16 @@ async def test_create_score_calibration_in_score_set_investigator_provided_not_s ### create_score_calibration -@pytest.mark.asyncio -async def test_create_score_calibration_raises_value_error_when_score_set_urn_is_provided( - setup_lib_db, session, mock_user -): +def test_create_score_calibration_raises_value_error_when_score_set_urn_is_provided(setup_lib_db, session, mock_user): MockCalibrationCreate = create_model("MockCalibrationCreate", score_set_urn=(str | None, "urn:provided")) with pytest.raises( ValueError, match="score_set_urn must not be provided to create a score calibration outside a score set.", ): - await create_score_calibration(session, MockCalibrationCreate(), mock_user) + create_score_calibration(session, MockCalibrationCreate(), mock_user) -@pytest.mark.asyncio -async def test_create_score_calibration_creates_score_calibration_when_score_set_urn_is_absent(setup_lib_db, session): +def test_create_score_calibration_creates_score_calibration_when_score_set_urn_is_absent(setup_lib_db, session): test_user = session.execute(select(User)).scalars().first() MockCalibrationCreate = create_model( @@ -463,7 +453,7 @@ async def test_create_score_calibration_creates_score_calibration_when_score_set functional_classifications=(list, []), ) - calibration = await create_score_calibration(session, MockCalibrationCreate(), test_user) + calibration = create_score_calibration(session, MockCalibrationCreate(), test_user) assert calibration is not None assert calibration.score_set is None @@ -471,7 +461,6 @@ async def test_create_score_calibration_creates_score_calibration_when_score_set ### Shared tests for create_score_calibration_in_score_set and create_score_calibration -@pytest.mark.asyncio @pytest.mark.parametrize( "create_function_to_call,score_set_urn", [ @@ -479,7 +468,7 @@ async def test_create_score_calibration_creates_score_calibration_when_score_set (create_score_calibration, None), ], ) -async def test_create_score_calibration_propagates_errors_from_publication_find_create( +def test_create_score_calibration_propagates_errors_from_publication_find_create( setup_lib_db_with_score_set, session, mock_user, create_function_to_call, score_set_urn ): MockCalibrationCreate = create_model( @@ -507,10 +496,9 @@ async def test_create_score_calibration_propagates_errors_from_publication_find_ side_effect=ValueError("Propagated error"), ), ): - await create_function_to_call(session, MockCalibrationCreate(), mock_user) + create_function_to_call(session, MockCalibrationCreate(), mock_user) -@pytest.mark.asyncio @pytest.mark.parametrize( "create_function_to_call,score_set_urn", [ @@ -533,7 +521,7 @@ async def test_create_score_calibration_propagates_errors_from_publication_find_ ], indirect=["mock_publication_fetch"], ) -async def test_create_score_calibration_publication_identifier_associations_created_with_appropriate_relation( +def test_create_score_calibration_publication_identifier_associations_created_with_appropriate_relation( setup_lib_db_with_score_set, session, mock_publication_fetch, @@ -560,14 +548,13 @@ async def test_create_score_calibration_publication_identifier_associations_crea [create_model("MockPublicationCreate", db_name=(str, "PubMed"), identifier=(str, TEST_PUBMED_IDENTIFIER))()], ) - calibration = await create_function_to_call(session, mocked_calibration, test_user) + calibration = create_function_to_call(session, mocked_calibration, test_user) assert calibration.publication_identifier_associations[0].publication.db_name == "PubMed" assert calibration.publication_identifier_associations[0].publication.identifier == TEST_PUBMED_IDENTIFIER assert calibration.publication_identifier_associations[0].relation == expected_relation assert len(calibration.publication_identifier_associations) == 1 -@pytest.mark.asyncio @pytest.mark.parametrize( "create_function_to_call,score_set_urn", [ @@ -575,7 +562,7 @@ async def test_create_score_calibration_publication_identifier_associations_crea (create_score_calibration, None), ], ) -async def test_create_score_calibration_user_is_set_as_creator_and_modifier( +def test_create_score_calibration_user_is_set_as_creator_and_modifier( setup_lib_db_with_score_set, session, create_function_to_call, score_set_urn ): MockCalibrationCreate = create_model( @@ -589,12 +576,11 @@ async def test_create_score_calibration_user_is_set_as_creator_and_modifier( test_user = session.execute(select(User)).scalars().first() - calibration = await create_function_to_call(session, MockCalibrationCreate(), test_user) + calibration = create_function_to_call(session, MockCalibrationCreate(), test_user) assert calibration.created_by == test_user assert calibration.modified_by == test_user -@pytest.mark.asyncio @pytest.mark.parametrize( "create_function_to_call,score_set_urn", [ @@ -619,7 +605,7 @@ async def test_create_score_calibration_user_is_set_as_creator_and_modifier( TEST_BRNICH_SCORE_CALIBRATION_CLASS_BASED, ], ) -async def test_create_score_calibration_fully_valid_calibration( +def test_create_score_calibration_fully_valid_calibration( setup_lib_db_with_score_set, session, create_function_to_call, @@ -631,7 +617,7 @@ async def test_create_score_calibration_fully_valid_calibration( test_user = session.execute(select(User)).scalars().first() - calibration = await create_function_to_call(session, calibration_create, test_user) + calibration = create_function_to_call(session, calibration_create, test_user) for field in valid_score_calibration_data: # Sources are tested elsewhere. @@ -646,7 +632,6 @@ async def test_create_score_calibration_fully_valid_calibration( ) -@pytest.mark.asyncio @pytest.mark.parametrize( "create_function_to_call,score_set_urn", [ @@ -664,7 +649,7 @@ async def test_create_score_calibration_fully_valid_calibration( ], indirect=["mock_publication_fetch"], ) -async def test_create_score_calibration_does_not_commit_transaction( +def test_create_score_calibration_does_not_commit_transaction( setup_lib_db_with_score_set, session, mock_user, create_function_to_call, score_set_urn, mock_publication_fetch ): calibration_create = ScoreCalibrationCreate( @@ -673,7 +658,7 @@ async def test_create_score_calibration_does_not_commit_transaction( test_user = session.execute(select(User)).scalars().first() with mock.patch.object(session, "commit") as mock_commit: - await create_function_to_call(session, calibration_create, test_user) + create_function_to_call(session, calibration_create, test_user) mock_commit.assert_not_called() @@ -682,8 +667,7 @@ async def test_create_score_calibration_does_not_commit_transaction( ################################################################################ -@pytest.mark.asyncio -async def test_modify_score_calibration_raises_value_error_when_score_set_urn_is_missing( +def test_modify_score_calibration_raises_value_error_when_score_set_urn_is_missing( setup_lib_db_with_score_set, session, mock_user, mock_functional_calibration ): MockCalibrationModify = create_model("MockCalibrationModify", score_set_urn=(str | None, None)) @@ -691,11 +675,10 @@ async def test_modify_score_calibration_raises_value_error_when_score_set_urn_is ValueError, match="score_set_urn must be provided to modify a score calibration.", ): - await modify_score_calibration(session, mock_functional_calibration, MockCalibrationModify(), mock_user) + modify_score_calibration(session, mock_functional_calibration, MockCalibrationModify(), mock_user) -@pytest.mark.asyncio -async def test_modify_score_calibration_raises_no_result_found_error_when_score_set_does_not_exist( +def test_modify_score_calibration_raises_no_result_found_error_when_score_set_does_not_exist( setup_lib_db, session, mock_user, mock_functional_calibration ): MockCalibrationModify = create_model("MockCalibrationModify", score_set_urn=(str | None, "urn:invalid")) @@ -703,10 +686,9 @@ async def test_modify_score_calibration_raises_no_result_found_error_when_score_ NoResultFound, match="No row was found when one was required", ): - await modify_score_calibration(session, mock_functional_calibration, MockCalibrationModify(), mock_user) + modify_score_calibration(session, mock_functional_calibration, MockCalibrationModify(), mock_user) -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -717,12 +699,12 @@ async def test_modify_score_calibration_raises_no_result_found_error_when_score_ ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_modifies_score_calibration_when_score_set_exists( +def test_modify_score_calibration_modifies_score_calibration_when_score_set_exists( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) @@ -736,15 +718,12 @@ async def test_modify_score_calibration_modifies_score_calibration_when_score_se functional_classifications=(list, []), ) - modified_calibration = await modify_score_calibration( - session, existing_calibration, MockCalibrationModify(), test_user - ) + modified_calibration = modify_score_calibration(session, existing_calibration, MockCalibrationModify(), test_user) assert modified_calibration is not None assert modified_calibration.description == "Modified description" assert modified_calibration.score_set == setup_lib_db_with_score_set -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -755,12 +734,12 @@ async def test_modify_score_calibration_modifies_score_calibration_when_score_se ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_clears_existing_publication_identifier_associations( +def test_modify_score_calibration_clears_existing_publication_identifier_associations( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) @@ -775,11 +754,10 @@ async def test_modify_score_calibration_clears_existing_publication_identifier_a mocked_calibration = MockCalibrationModify() - calibration = await modify_score_calibration(session, existing_calibration, mocked_calibration, test_user) + calibration = modify_score_calibration(session, existing_calibration, mocked_calibration, test_user) assert len(calibration.publication_identifier_associations) == 0 -@pytest.mark.asyncio @pytest.mark.parametrize( "relation,expected_relation", [ @@ -798,7 +776,7 @@ async def test_modify_score_calibration_clears_existing_publication_identifier_a ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_publication_identifier_associations_created_with_appropriate_relation( +def test_modify_score_calibration_publication_identifier_associations_created_with_appropriate_relation( setup_lib_db_with_score_set, session, mock_publication_fetch, @@ -807,7 +785,7 @@ async def test_modify_score_calibration_publication_identifier_associations_crea ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) @@ -827,14 +805,13 @@ async def test_modify_score_calibration_publication_identifier_associations_crea [create_model("MockPublicationCreate", db_name=(str, "PubMed"), identifier=(str, TEST_PUBMED_IDENTIFIER))()], ) - calibration = await modify_score_calibration(session, existing_calibration, mocked_calibration, test_user) + calibration = modify_score_calibration(session, existing_calibration, mocked_calibration, test_user) assert calibration.publication_identifier_associations[0].publication.db_name == "PubMed" assert calibration.publication_identifier_associations[0].publication.identifier == TEST_PUBMED_IDENTIFIER assert calibration.publication_identifier_associations[0].relation == expected_relation assert len(calibration.publication_identifier_associations) == 1 -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -845,12 +822,12 @@ async def test_modify_score_calibration_publication_identifier_associations_crea ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_retains_existing_publication_relationships_when_not_modified( +def test_modify_score_calibration_retains_existing_publication_relationships_when_not_modified( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) calibration_publication_relations = existing_calibration.publication_identifier_associations.copy() @@ -894,14 +871,11 @@ async def test_modify_score_calibration_retains_existing_publication_relationshi functional_classifications=(list, []), ) - modified_calibration = await modify_score_calibration( - session, existing_calibration, MockCalibrationModify(), test_user - ) + modified_calibration = modify_score_calibration(session, existing_calibration, MockCalibrationModify(), test_user) assert modified_calibration is not None assert modified_calibration.publication_identifier_associations == calibration_publication_relations -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -913,12 +887,12 @@ async def test_modify_score_calibration_retains_existing_publication_relationshi ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_adds_new_publication_association( +def test_modify_score_calibration_adds_new_publication_association( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) @@ -940,9 +914,7 @@ async def test_modify_score_calibration_adds_new_publication_association( functional_classifications=(list, []), ) - modified_calibration = await modify_score_calibration( - session, existing_calibration, MockCalibrationModify(), test_user - ) + modified_calibration = modify_score_calibration(session, existing_calibration, MockCalibrationModify(), test_user) assert modified_calibration is not None assert modified_calibration.publication_identifier_associations[0].publication.db_name == "Crossref" assert ( @@ -952,7 +924,6 @@ async def test_modify_score_calibration_adds_new_publication_association( assert len(modified_calibration.publication_identifier_associations) == 1 -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -963,12 +934,10 @@ async def test_modify_score_calibration_adds_new_publication_association( ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_user_is_set_as_modifier( - setup_lib_db_with_score_set, session, mock_publication_fetch -): +def test_modify_score_calibration_user_is_set_as_modifier(setup_lib_db_with_score_set, session, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) @@ -982,15 +951,12 @@ async def test_modify_score_calibration_user_is_set_as_modifier( ) modify_user = session.execute(select(User).where(User.id != test_user.id)).scalars().first() - modified_calibration = await modify_score_calibration( - session, existing_calibration, MockCalibrationModify(), modify_user - ) + modified_calibration = modify_score_calibration(session, existing_calibration, MockCalibrationModify(), modify_user) assert modified_calibration is not None assert modified_calibration.modified_by == modify_user assert modified_calibration.created_by == test_user -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1001,7 +967,7 @@ async def test_modify_score_calibration_user_is_set_as_modifier( ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_new_score_set(setup_lib_db_with_score_set, session, mock_publication_fetch): +def test_modify_score_calibration_new_score_set(setup_lib_db_with_score_set, session, mock_publication_fetch): existing_experiment = setup_lib_db_with_score_set.experiment score_set_scaffold = TEST_SEQ_SCORESET.copy() score_set_scaffold.pop("target_genes") @@ -1018,7 +984,7 @@ async def test_modify_score_calibration_new_score_set(setup_lib_db_with_score_se session.refresh(new_containing_score_set) test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, new_containing_score_set.urn, test_user ) @@ -1031,14 +997,11 @@ async def test_modify_score_calibration_new_score_set(setup_lib_db_with_score_se functional_classifications=(list, []), ) - modified_calibration = await modify_score_calibration( - session, existing_calibration, MockCalibrationModify(), test_user - ) + modified_calibration = modify_score_calibration(session, existing_calibration, MockCalibrationModify(), test_user) assert modified_calibration is not None assert modified_calibration.score_set == new_containing_score_set -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1049,12 +1012,12 @@ async def test_modify_score_calibration_new_score_set(setup_lib_db_with_score_se ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_clears_functional_classifications( +def test_modify_score_calibration_clears_functional_classifications( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) @@ -1067,14 +1030,11 @@ async def test_modify_score_calibration_clears_functional_classifications( functional_classifications=(list, []), ) - modified_calibration = await modify_score_calibration( - session, existing_calibration, MockCalibrationModify(), test_user - ) + modified_calibration = modify_score_calibration(session, existing_calibration, MockCalibrationModify(), test_user) assert modified_calibration is not None assert len(modified_calibration.functional_classifications) == 0 -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1085,19 +1045,17 @@ async def test_modify_score_calibration_clears_functional_classifications( ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_fully_valid_calibration( - setup_lib_db_with_score_set, session, mock_publication_fetch -): +def test_modify_score_calibration_fully_valid_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) modify_calibration = ScoreCalibrationModify( **TEST_PATHOGENICITY_SCORE_CALIBRATION, score_set_urn=setup_lib_db_with_score_set.urn ) - modified_calibration = await modify_score_calibration(session, existing_calibration, modify_calibration, test_user) + modified_calibration = modify_score_calibration(session, existing_calibration, modify_calibration, test_user) for field in TEST_PATHOGENICITY_SCORE_CALIBRATION: # Sources are tested elsewhere. @@ -1112,7 +1070,6 @@ async def test_modify_score_calibration_fully_valid_calibration( ) -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1123,12 +1080,12 @@ async def test_modify_score_calibration_fully_valid_calibration( ], indirect=["mock_publication_fetch"], ) -async def test_modify_score_calibration_does_not_commit_transaction( +def test_modify_score_calibration_does_not_commit_transaction( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) @@ -1146,7 +1103,6 @@ async def test_modify_score_calibration_does_not_commit_transaction( ################################################################################ -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1157,12 +1113,10 @@ async def test_modify_score_calibration_does_not_commit_transaction( ], indirect=["mock_publication_fetch"], ) -async def test_cannot_publish_already_published_calibration( - setup_lib_db_with_score_set, session, mock_publication_fetch -): +def test_cannot_publish_already_published_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.private = False @@ -1174,7 +1128,6 @@ async def test_cannot_publish_already_published_calibration( publish_score_calibration(session, existing_calibration, test_user) -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1185,12 +1138,12 @@ async def test_cannot_publish_already_published_calibration( ], indirect=["mock_publication_fetch"], ) -async def test_publish_score_calibration_marks_calibration_public( +def test_publish_score_calibration_marks_calibration_public( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.score_set.private = False @@ -1200,7 +1153,6 @@ async def test_publish_score_calibration_marks_calibration_public( assert published_calibration.private is False -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1211,12 +1163,12 @@ async def test_publish_score_calibration_marks_calibration_public( ], indirect=["mock_publication_fetch"], ) -async def test_cannot_publish_calibration_when_score_set_is_private( +def test_cannot_publish_calibration_when_score_set_is_private( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) assert existing_calibration.score_set.private is True @@ -1226,7 +1178,6 @@ async def test_cannot_publish_calibration_when_score_set_is_private( assert existing_calibration.private is True -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1237,12 +1188,12 @@ async def test_cannot_publish_calibration_when_score_set_is_private( ], indirect=["mock_publication_fetch"], ) -async def test_publish_score_calibration_user_is_set_as_modifier( +def test_publish_score_calibration_user_is_set_as_modifier( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.score_set.private = False @@ -1259,7 +1210,6 @@ async def test_publish_score_calibration_user_is_set_as_modifier( ################################################################################ -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1270,10 +1220,10 @@ async def test_publish_score_calibration_user_is_set_as_modifier( ], indirect=["mock_publication_fetch"], ) -async def test_cannot_promote_already_primary_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): +def test_cannot_promote_already_primary_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.primary = True @@ -1285,7 +1235,6 @@ async def test_cannot_promote_already_primary_calibration(setup_lib_db_with_scor promote_score_calibration_to_primary(session, existing_calibration, test_user, force=False) -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1296,12 +1245,12 @@ async def test_cannot_promote_already_primary_calibration(setup_lib_db_with_scor ], indirect=["mock_publication_fetch"], ) -async def test_cannot_promote_calibration_when_calibration_is_research_use_only( +def test_cannot_promote_calibration_when_calibration_is_research_use_only( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.research_use_only = True @@ -1313,7 +1262,6 @@ async def test_cannot_promote_calibration_when_calibration_is_research_use_only( promote_score_calibration_to_primary(session, existing_calibration, test_user, force=False) -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1324,12 +1272,12 @@ async def test_cannot_promote_calibration_when_calibration_is_research_use_only( ], indirect=["mock_publication_fetch"], ) -async def test_cannot_promote_calibration_when_calibration_is_private( +def test_cannot_promote_calibration_when_calibration_is_private( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.private = True @@ -1341,7 +1289,6 @@ async def test_cannot_promote_calibration_when_calibration_is_private( promote_score_calibration_to_primary(session, existing_calibration, test_user, force=False) -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1352,15 +1299,15 @@ async def test_cannot_promote_calibration_when_calibration_is_private( ], indirect=["mock_publication_fetch"], ) -async def test_cannot_promote_calibration_when_another_primary_exists( +def test_cannot_promote_calibration_when_another_primary_exists( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_primary_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_primary_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_primary_calibration.private = False @@ -1378,7 +1325,6 @@ async def test_cannot_promote_calibration_when_another_primary_exists( promote_score_calibration_to_primary(session, existing_calibration, test_user, force=False) -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1389,12 +1335,12 @@ async def test_cannot_promote_calibration_when_another_primary_exists( ], indirect=["mock_publication_fetch"], ) -async def test_promote_score_calibration_to_primary_marks_calibration_primary( +def test_promote_score_calibration_to_primary_marks_calibration_primary( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.private = False @@ -1407,7 +1353,6 @@ async def test_promote_score_calibration_to_primary_marks_calibration_primary( assert promoted_calibration.primary is True -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1418,15 +1363,15 @@ async def test_promote_score_calibration_to_primary_marks_calibration_primary( ], indirect=["mock_publication_fetch"], ) -async def test_promote_score_calibration_to_primary_demotes_existing_primary_when_forced( +def test_promote_score_calibration_to_primary_demotes_existing_primary_when_forced( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_primary_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_primary_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_primary_calibration.private = False @@ -1450,7 +1395,6 @@ async def test_promote_score_calibration_to_primary_demotes_existing_primary_whe assert existing_primary_calibration.primary is False -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1461,12 +1405,12 @@ async def test_promote_score_calibration_to_primary_demotes_existing_primary_whe ], indirect=["mock_publication_fetch"], ) -async def test_promote_score_calibration_to_primary_user_is_set_as_modifier( +def test_promote_score_calibration_to_primary_user_is_set_as_modifier( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.private = False @@ -1484,7 +1428,6 @@ async def test_promote_score_calibration_to_primary_user_is_set_as_modifier( assert promoted_calibration.created_by == test_user -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1495,15 +1438,15 @@ async def test_promote_score_calibration_to_primary_user_is_set_as_modifier( ], indirect=["mock_publication_fetch"], ) -async def test_promote_score_calibration_to_primary_demoted_existing_primary_user_is_set_as_modifier( +def test_promote_score_calibration_to_primary_demoted_existing_primary_user_is_set_as_modifier( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_primary_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_primary_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_primary_calibration.private = False @@ -1530,7 +1473,6 @@ async def test_promote_score_calibration_to_primary_demoted_existing_primary_use assert promoted_calibration.created_by == test_user -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1541,12 +1483,12 @@ async def test_promote_score_calibration_to_primary_demoted_existing_primary_use ], indirect=["mock_publication_fetch"], ) -async def test_promote_score_calibration_to_primary_does_not_commit_transaction( +def test_promote_score_calibration_to_primary_does_not_commit_transaction( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.private = False @@ -1565,7 +1507,6 @@ async def test_promote_score_calibration_to_primary_does_not_commit_transaction( ################################################################################ -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1576,10 +1517,10 @@ async def test_promote_score_calibration_to_primary_does_not_commit_transaction( ], indirect=["mock_publication_fetch"], ) -async def test_cannot_demote_non_primary_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): +def test_cannot_demote_non_primary_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.primary = False @@ -1591,7 +1532,6 @@ async def test_cannot_demote_non_primary_calibration(setup_lib_db_with_score_set demote_score_calibration_from_primary(session, existing_calibration, test_user) -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1602,12 +1542,12 @@ async def test_cannot_demote_non_primary_calibration(setup_lib_db_with_score_set ], indirect=["mock_publication_fetch"], ) -async def test_demote_score_calibration_from_primary_marks_calibration_non_primary( +def test_demote_score_calibration_from_primary_marks_calibration_non_primary( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.primary = True @@ -1620,7 +1560,6 @@ async def test_demote_score_calibration_from_primary_marks_calibration_non_prima assert demoted_calibration.primary is False -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1631,12 +1570,12 @@ async def test_demote_score_calibration_from_primary_marks_calibration_non_prima ], indirect=["mock_publication_fetch"], ) -async def test_demote_score_calibration_from_primary_user_is_set_as_modifier( +def test_demote_score_calibration_from_primary_user_is_set_as_modifier( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.primary = True @@ -1651,7 +1590,6 @@ async def test_demote_score_calibration_from_primary_user_is_set_as_modifier( assert demoted_calibration.created_by == test_user -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1662,12 +1600,12 @@ async def test_demote_score_calibration_from_primary_user_is_set_as_modifier( ], indirect=["mock_publication_fetch"], ) -async def test_demote_score_calibration_from_primary_does_not_commit_transaction( +def test_demote_score_calibration_from_primary_does_not_commit_transaction( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.primary = True @@ -1685,7 +1623,6 @@ async def test_demote_score_calibration_from_primary_does_not_commit_transaction ################################################################################ -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1696,10 +1633,10 @@ async def test_demote_score_calibration_from_primary_does_not_commit_transaction ], indirect=["mock_publication_fetch"], ) -async def test_cannot_delete_primary_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): +def test_cannot_delete_primary_calibration(setup_lib_db_with_score_set, session, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) existing_calibration.primary = True @@ -1711,7 +1648,6 @@ async def test_cannot_delete_primary_calibration(setup_lib_db_with_score_set, se delete_score_calibration(session, existing_calibration) -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1722,12 +1658,10 @@ async def test_cannot_delete_primary_calibration(setup_lib_db_with_score_set, se ], indirect=["mock_publication_fetch"], ) -async def test_delete_score_calibration_deletes_calibration( - session, setup_lib_db_with_score_set, mock_publication_fetch -): +def test_delete_score_calibration_deletes_calibration(session, setup_lib_db_with_score_set, mock_publication_fetch): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) calibration_id = existing_calibration.id @@ -1739,7 +1673,6 @@ async def test_delete_score_calibration_deletes_calibration( session.execute(select(ScoreCalibration).where(ScoreCalibration.id == calibration_id)).scalars().one() -@pytest.mark.asyncio @pytest.mark.parametrize( "mock_publication_fetch", [ @@ -1750,12 +1683,12 @@ async def test_delete_score_calibration_deletes_calibration( ], indirect=["mock_publication_fetch"], ) -async def test_delete_score_calibration_does_not_commit_transaction( +def test_delete_score_calibration_does_not_commit_transaction( setup_lib_db_with_score_set, session, mock_publication_fetch ): test_user = session.execute(select(User)).scalars().first() - existing_calibration = await create_test_range_based_score_calibration_in_score_set( + existing_calibration = create_test_range_based_score_calibration_in_score_set( session, setup_lib_db_with_score_set.urn, test_user ) diff --git a/tests/lib/workflow/test_kickoff.py b/tests/lib/workflow/test_kickoff.py index f81879b12..1e87a2d32 100644 --- a/tests/lib/workflow/test_kickoff.py +++ b/tests/lib/workflow/test_kickoff.py @@ -3,12 +3,14 @@ pytest.importorskip("arq") +import functools from unittest.mock import AsyncMock, Mock, patch +import anyio.to_thread from arq.jobs import Job as ArqJob from sqlalchemy import select -from mavedb.lib.workflow.kickoff import enqueue_pipeline_for_score_set +from mavedb.lib.workflow.kickoff import enqueue_pipeline_for_score_set, enqueue_pipeline_for_score_set_from_thread from mavedb.models.job_run import JobRun from mavedb.models.pipeline import Pipeline from mavedb.models.user import User @@ -258,3 +260,61 @@ async def test_discards_pipeline_records_when_enqueue_raises( # The pipeline (and its job runs) created before the failed enqueue must not be left behind. assert session.execute(select(Pipeline)).scalars().all() == [] assert session.execute(select(JobRun)).scalars().all() == [] + + +@pytest.mark.integration +@pytest.mark.asyncio +class TestEnqueuePipelineForScoreSetFromThreadIntegration: + """The thread-pool variant, run in an AnyIO worker thread as FastAPI runs a ``def`` route.""" + + @pytest.fixture(autouse=True) + def setup(self, session, setup_lib_db_with_score_set): + self.score_set = setup_lib_db_with_score_set + self.user = session.query(User).filter(User.username == TEST_USER["username"]).first() + + def _kickoff(self, session, arq_redis, pipeline_name): + return functools.partial( + enqueue_pipeline_for_score_set_from_thread, + session, + arq_redis, + pipeline_name=pipeline_name, + score_set=self.score_set, + user=self.user, + extra_params={"required_param": "some_value"}, + ) + + async def test_enqueue_pipeline_for_score_set_from_thread_in_worker_thread_creates_records_and_enqueues_job( + self, session, arq_redis, with_test_pipeline_definition_ctx, sample_independent_pipeline_definition + ): + pipeline, job = await anyio.to_thread.run_sync( + self._kickoff(session, arq_redis, sample_independent_pipeline_definition["name"]) + ) + + assert job is not None + job_run = session.execute( + select(JobRun).where(JobRun.pipeline_id == pipeline.id, JobRun.job_function == "process_data") + ).scalar_one() + assert job_run.job_params["required_param"] == "some_value" + + async def test_enqueue_pipeline_for_score_set_from_thread_enqueue_raises_discards_records( + self, session, arq_redis, with_test_pipeline_definition_ctx, sample_independent_pipeline_definition + ): + with ( + patch.object(arq_redis, "enqueue_job", AsyncMock(side_effect=ConnectionError("redis is unreachable"))), + pytest.raises(ConnectionError), + ): + await anyio.to_thread.run_sync( + self._kickoff(session, arq_redis, sample_independent_pipeline_definition["name"]) + ) + + assert session.execute(select(Pipeline)).scalars().all() == [] + assert session.execute(select(JobRun)).scalars().all() == [] + + async def test_enqueue_pipeline_for_score_set_from_thread_outside_worker_thread_raises_and_discards_records( + self, session, arq_redis, with_test_pipeline_definition_ctx, sample_independent_pipeline_definition + ): + with pytest.raises(RuntimeError): + self._kickoff(session, arq_redis, sample_independent_pipeline_definition["name"])() + + assert session.execute(select(Pipeline)).scalars().all() == [] + assert session.execute(select(JobRun)).scalars().all() == [] diff --git a/tests/routers/test_diseases.py b/tests/routers/test_diseases.py index 19c3c2bf9..1000fae9c 100644 --- a/tests/routers/test_diseases.py +++ b/tests/routers/test_diseases.py @@ -13,7 +13,7 @@ @pytest.fixture def mock_mondo_search(monkeypatch): - async def fake_search(query, limit=20): + def fake_search(query, limit=20): return [ { "code": "MONDO:0015263", @@ -41,7 +41,7 @@ def test_search_diseases_requires_a_query(client): def test_search_diseases_surfaces_ontology_outage(client, monkeypatch): - async def failing_search(query, limit=20): + def failing_search(query, limit=20): raise MondoServiceError("Disease ontology service temporarily unavailable") monkeypatch.setattr("mavedb.routers.diseases.search_mondo", failing_search) diff --git a/tests/routers/test_score_calibrations.py b/tests/routers/test_score_calibrations.py index c9045ef72..b19cb63b4 100644 --- a/tests/routers/test_score_calibrations.py +++ b/tests/routers/test_score_calibrations.py @@ -190,7 +190,7 @@ def test_create_score_calibration_without_disease_defaults_to_generic( def test_create_score_calibration_with_disease_resolves_mondo_concept( client, setup_router_db, mock_publication_fetch, session, data_provider, data_files, monkeypatch ): - async def fake_fetch(code): + def fake_fetch(code): return { "code": code, "label": "Brugada syndrome", @@ -230,7 +230,7 @@ async def fake_fetch(code): def test_cannot_create_score_calibration_with_unknown_disease( client, setup_router_db, mock_publication_fetch, session, data_provider, data_files, monkeypatch ): - async def fake_fetch(code): + def fake_fetch(code): return None monkeypatch.setattr("mavedb.lib.mondo_ols.fetch_mondo_term", fake_fetch) @@ -265,7 +265,7 @@ async def fake_fetch(code): def test_cannot_modify_score_calibration_with_unknown_disease( client, setup_router_db, mock_publication_fetch, session, data_provider, data_files, monkeypatch ): - async def fake_fetch(code): + def fake_fetch(code): return None monkeypatch.setattr("mavedb.lib.mondo_ols.fetch_mondo_term", fake_fetch) diff --git a/tests/routers/test_score_set.py b/tests/routers/test_score_set.py index b17e29fd9..b26d641b7 100644 --- a/tests/routers/test_score_set.py +++ b/tests/routers/test_score_set.py @@ -231,7 +231,7 @@ def test_create_score_set_with_contributor(client, mock_publication_fetch, setup def test_cannot_create_score_set_with_score_calibration_with_unknown_disease( client, mock_publication_fetch, setup_router_db, monkeypatch ): - async def fake_fetch(code): + def fake_fetch(code): return None monkeypatch.setattr("mavedb.lib.mondo_ols.fetch_mondo_term", fake_fetch) From 5223b481774d60b3fd82efbaef24b09058acf09e Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Fri, 23 Oct 2026 14:37:58 -0700 Subject: [PATCH 2/3] test: fail when async code takes a database session Walks the app's routes and fails when an async endpoint or dependency takes a session from get_db, so sync database work can't drift back onto the event loop. A def dependency that uses the session passes, since it runs in the thread pool. FastAPI keeps each included router behind a wrapper rather than copying its routes into app.routes, so the walk follows original_router and asserts it finds at least 150 routes; without that, a walk that saw nothing would pass. It fails on the code before the conversion and on a route or dependency made async again. --- tests/routers/test_async_routes.py | 107 +++++++++++++++++++++++++++++ 1 file changed, 107 insertions(+) create mode 100644 tests/routers/test_async_routes.py diff --git a/tests/routers/test_async_routes.py b/tests/routers/test_async_routes.py new file mode 100644 index 000000000..f3064bbc3 --- /dev/null +++ b/tests/routers/test_async_routes.py @@ -0,0 +1,107 @@ +# ruff: noqa: E402 +"""Keep synchronous database work off the event loop. + +FastAPI runs ``async def`` endpoints and dependencies on the event loop, so one that uses the synchronous +database session stalls every request on its process (#881, #950). These tests walk the app's routes and +fail when an async endpoint or dependency takes a session from ``get_db``. A sync dependency that uses the +session is fine: FastAPI runs it in the thread pool. +""" + +import inspect +from typing import Any, Callable + +import pytest + +pytest.importorskip("fastapi") + +from fastapi.dependencies.models import Dependant +from fastapi.routing import APIRoute + +from mavedb import deps +from mavedb.routers import score_sets +from mavedb.server_main import app + +# These read the multipart form, which needs an await, and run everything after it in the thread pool. +ASYNC_ENDPOINTS_ALLOWED_TO_USE_DB: set[Callable[..., Any]] = { + score_sets.upload_score_set_variant_data, + score_sets.update_score_set_with_variants, +} + + +def _is_async(call: Any) -> bool: + if inspect.isroutine(call): + return inspect.iscoroutinefunction(call) + # Callable class instances, such as the JWT bearer dependency. + return inspect.iscoroutinefunction(getattr(call, "__call__", None)) + + +# The app has 161 routes; a walk that finds far fewer has stopped seeing them. +MIN_EXPECTED_ROUTES = 150 + + +def _takes_db(dependant: Dependant) -> bool: + return any(sub.call is deps.get_db for sub in dependant.dependencies) + + +def _async_dependencies_taking_db(dependant: Dependant) -> list[str]: + found = [] + for sub in dependant.dependencies: + if _is_async(sub.call) and _takes_db(sub): + found.append(getattr(sub.call, "__qualname__", type(sub.call).__qualname__)) + found.extend(_async_dependencies_taking_db(sub)) + return found + + +def _api_routes(routes: list[Any]) -> list[APIRoute]: + # FastAPI keeps each included router behind a wrapper instead of copying its routes into the app. + found = [] + for route in routes: + included = getattr(route, "original_router", None) + if included is not None: + found.extend(_api_routes(included.routes)) + elif isinstance(route, APIRoute): + found.append(route) + return found + + +@pytest.mark.unit +def test_app_routes_using_db_are_not_async(): + routes = _api_routes(app.routes) + assert len(routes) >= MIN_EXPECTED_ROUTES, f"Found only {len(routes)} routes; the walk no longer sees them all" + + offenders = [] + for route in routes: + label = f"{sorted(route.methods)} {route.path}" + if ( + _is_async(route.endpoint) + and _takes_db(route.dependant) + and route.endpoint not in ASYNC_ENDPOINTS_ALLOWED_TO_USE_DB + ): + offenders.append(f"{label}: async endpoint {route.endpoint.__qualname__} uses get_db") + offenders.extend( + f"{label}: async dependency {name} uses get_db" for name in _async_dependencies_taking_db(route.dependant) + ) + offenders.extend( + f"app-wide async dependency {dependency.dependency.__qualname__} uses get_db" + for dependency in app.router.dependencies + if _is_async(dependency.dependency) + and any( + getattr(param.default, "dependency", None) is deps.get_db + for param in inspect.signature(dependency.dependency).parameters.values() + ) + ) + + assert offenders == [], "Make these `def`, or move their database work into the thread pool:\n" + "\n".join( + offenders + ) + + +@pytest.mark.unit +def test_async_endpoints_allowed_to_use_db_are_registered_async_and_use_db(): + routes_by_endpoint = {route.endpoint: route for route in _api_routes(app.routes)} + + for endpoint in ASYNC_ENDPOINTS_ALLOWED_TO_USE_DB: + route = routes_by_endpoint.get(endpoint) + assert route is not None, f"{endpoint.__qualname__} is no longer a route; drop it from the allowlist" + assert _is_async(endpoint), f"{endpoint.__qualname__} is no longer async; drop it from the allowlist" + assert _takes_db(route.dependant), f"{endpoint.__qualname__} no longer uses get_db" From bd8f03f853574f0d77ce4c3723ffcec12b84d6d2 Mon Sep 17 00:00:00 2001 From: Benjamin Capodanno Date: Fri, 23 Oct 2026 14:37:59 -0700 Subject: [PATCH 3/3] build(claude): say when code may be async Replaces "never write an async def with no real await" with the rule the conversion settled on: def by default, async only where the code can't work without an await, database work kept out of it, and a comment saying why. The routes rule names the guard test, what may go on its allowlist, and the thread-pool variant of the pipeline kickoff. Expect: agents write new routes and helpers as def, and add a comment giving the reason whenever they write an async route. --- .claude/rules/routes.md | 7 ++++--- AGENTS.md | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/.claude/rules/routes.md b/.claude/rules/routes.md index 5ae884830..2a0537f8e 100644 --- a/.claude/rules/routes.md +++ b/.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/`. diff --git a/AGENTS.md b/AGENTS.md index eaf39e5bc..edfdd6adc 100644 --- a/AGENTS.md +++ b/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.