diff --git a/.claude/rules/routes.md b/.claude/rules/routes.md index 5ae88483..2a0537f8 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 eaf39e5b..edfdd6ad 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. diff --git a/src/mavedb/lib/authentication.py b/src/mavedb/lib/authentication.py index a6650316..a96162b2 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 252febb4..1810104b 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 4edfce97..ea2763f5 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 7855b6af..6c530d22 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 c6a58bce..d66f2703 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 afa4723a..55188500 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 9dfe39cd..07205115 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 35f236bf..c43e8e74 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 41092beb..77d83226 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 eb8bb2cf..d5fbec2e 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 1c71bf6b..6b9adf4c 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 1cff0eff..00282061 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 d6762cbf..582d77bb 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 96139c52..bc52f431 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 051d5bb3..bff15673 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 f0f2f9da..6881e9a9 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 a58d127b..cd2656e9 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 8df3898b..6c3575cb 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 cd6349f6..a31bc20a 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 936f625e..17be23a9 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 f8f43e4a..998260f6 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 01ed631d..4c844408 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 0d680166..fb6d945f 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 7aafc0ab..68c15072 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 066509e3..aa0579dc 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 72f7ed2c..1ff28a48 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 ab3d0b9c..e6f48b52 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 8e056927..146e2e6e 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 409fe810..9023d8d6 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 cc378223..ffe248c6 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 b941cc08..1cc3971c 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 f81879b1..1e87a2d3 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_async_routes.py b/tests/routers/test_async_routes.py new file mode 100644 index 00000000..f3064bbc --- /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" diff --git a/tests/routers/test_diseases.py b/tests/routers/test_diseases.py index 19c3c2bf..1000fae9 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 c9045ef7..b19cb63b 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 b17e29fd..b26d641b 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)