refactor(multiple): unified crud grammar for domains

mpcontribs-api/src/mpcontribs_api/domains/_shared/components.py

@@ -36,7 +36,7 @@ async def _existing_by_md5(
         ).to_list()
         return {doc.md5: doc for doc in existing_docs}
 
-    async def insert_components(
+    async def insert_many(
         self,
         components: list[TIn],
         session: AsyncClientSession | None = None,
@@ -79,21 +79,18 @@ async def insert_components(
         resolved = existing_by_md5 | new_by_md5
         return [resolved[md5] for md5 in unique_md5s]
 
-    async def insert_component(self, component: TIn, *, session: AsyncClientSession | None = None) -> TDoc:
-        """Insert a single component.
+    async def insert_one(self, component: TIn, *, session: AsyncClientSession | None = None) -> TDoc:  # pyright: ignore[reportIncompatibleMethodOverride]
+        """Insert a single component, deduplicated by content hash.
 
         Args:
-            component (TIn): the table to insert
+            component (TIn): the component to insert
 
         Returns:
             TDoc: the component actually in the database
-
-        Raises:
-            AppError: If insert_one returns None, raises
         """
-        return (await self.insert_components(components=[component], session=session))[0]
+        return (await self.insert_many(components=[component], session=session))[0]
 
-    async def delete_components(
+    async def delete_many(
         self,
         filter: TFilter,
         session: AsyncClientSession | None = None,

mpcontribs-api/src/mpcontribs_api/domains/_shared/filters.py

@@ -5,11 +5,11 @@
 from fastapi_filter.contrib.beanie.filter import _odm_operator_transformer
 from pydantic import ValidationInfo, field_validator
 
+from mpcontribs_api.domains._shared.types import nfc_normalize
+
 # Register a custom __contains filter suffix to search where lists are a superset of a provided list
 _odm_operator_transformer.setdefault("contains", lambda value: {"$all": value})
 
-from mpcontribs_api.domains._shared.types import nfc_normalize
-
 
 def _normalize_query_values(value: Any) -> Any:
     """Recursively NFC-normalize every string in a built query condition value.

mpcontribs-api/src/mpcontribs_api/domains/_shared/repository.py

@@ -185,7 +185,7 @@ async def insert_one(self, in_resource: TIn) -> TDoc:
             ) from exc
         return document
 
-    async def delete(self, filter: TFilter, session: AsyncClientSession | None = None) -> DeleteResponse:
+    async def delete_many(self, filter: TFilter, session: AsyncClientSession | None = None) -> DeleteResponse:
         """Delete every scoped document matching an arbitrary ``filter``.
 
         This is the bulk path (e.g. "delete every ProjectGroup with owner == X"). It does not raise
@@ -217,25 +217,6 @@ async def delete_one(
             raise NotFoundError(f"{self.document_model.__name__} not found", identifiers=identifiers)
         return DeleteResponse.from_delete_result(result)
 
-    async def delete_by_ids(self, ids: list[Any], session: AsyncClientSession | None = None) -> DeleteResponse:
-        """Delete multiple scoped documents by id.
-
-        The user scope is injected so callers cannot delete documents they are not permitted to
-        see; out-of-scope ids simply match nothing and are reported as zero deletions.
-
-        Args:
-            ids (list[Any]): list of ids to delete
-            session: the session to perform the deletes within
-
-        Returns:
-            DeleteResponse: the result of the deletion
-        """
-        docs = self.document_model.find(self._scope, In(self.document_model.id, ids), session=session)
-        delete_result = await docs.delete_many(session=session)
-        if not delete_result:
-            raise ValidationError("DeleteResult not returned internally")
-        return DeleteResponse.from_delete_result(delete_result)
-
     def _patch_update_fields(self, update: TPatch) -> dict[str, Any]:
         """Map a patch model to the MongoDB ``$set`` field dict.
 

mpcontribs-api/src/mpcontribs_api/domains/_shared/service.py

@@ -91,13 +91,13 @@ async def get_one(self, identifiers: dict[str, Any], fields: frozenset[str] | No
             return None
         return await self._components.get_one(identifiers, fields)
 
-    async def insert(
+    async def insert_many(
         self,
         components: list[TIn],
         session: AsyncClientSession | None = None,
     ) -> list[TDoc]:
-        """Bulk-insert components, deduplicated by content hash. See ``insert_components``."""
-        return await self._components.insert_components(components=components, session=session)
+        """Bulk-insert components, deduplicated by content hash. See repository ``insert_many``."""
+        return await self._components.insert_many(components=components, session=session)
 
     async def patch_one(self, identifiers: dict[str, Any], update: TPatch) -> TDoc:
         """Partially update a component matching ``identifiers``, gated by contribution reachability.
@@ -136,7 +136,7 @@ async def download(
             restrict_ids=allowed,
         )
 
-    async def delete(self, filter: TFilter) -> ComponentDeleteResponse:
+    async def delete_many(self, filter: TFilter) -> ComponentDeleteResponse:
         """Delete components matching ``filter`` that are reachable and globally unreferenced.
 
         Args:
@@ -152,7 +152,11 @@ async def delete(self, filter: TFilter) -> ComponentDeleteResponse:
             return ComponentDeleteResponse(num_deleted=0)
         referenced = await self._contributions.referenced_component_ids(self._ref_field, list(reachable), scoped=False)
         deletable = [cid for cid in reachable if cid not in referenced]
-        num_deleted = (await self._components.delete_by_ids(deletable)).num_deleted if deletable else 0
+        num_deleted = (
+            (await self._components.delete_many(type(filter)(id__in=deletable))).num_deleted  # pyright: ignore[reportCallIssue]
+            if deletable
+            else 0
+        )
         return ComponentDeleteResponse(
             num_deleted=num_deleted,
             num_skipped=len(referenced),

mpcontribs-api/src/mpcontribs_api/domains/attachments/router.py

@@ -70,7 +70,7 @@ async def download_attachment(
 
 @router.delete("", response_model=ComponentDeleteResponse, dependencies=[Depends(require_user)])
 async def delete_attachments(service: AttachmentServiceDep, filter: AttachmentFilter = FilterDepends(AttachmentFilter)):
-    return await service.delete(filter=filter)
+    return await service.delete_many(filter=filter)
 
 
 @router.delete("/{id}", response_model=ComponentDeleteResponse, dependencies=[Depends(require_user)])

mpcontribs-api/src/mpcontribs_api/domains/contributions/repository.py

@@ -6,7 +6,6 @@
 from beanie.operators import Set
 from pymongo.asynchronous.client_session import AsyncClientSession
 from pymongo.errors import DuplicateKeyError
-from pymongo.results import DeleteResult
 from types_aiobotocore_s3 import S3Client
 
 from mpcontribs_api.authz import User
@@ -30,7 +29,6 @@
     merge_contribution_columns,
 )
 from mpcontribs_api.exceptions import ConflictError, NotFoundError, PermissionError
-from mpcontribs_api.pagination import CursorParams
 
 # Sentinel for "leave unique_value untouched" on patch (distinct from a real None value).
 _UNSET: Any = object()
@@ -92,15 +90,6 @@ async def count_contributions_for_project(self, project_name: str) -> int:
         """
         return await self.document_model.find(self.document_model.project == project_name).count()
 
-    async def get_contributions(
-        self,
-        filter: ContributionFilter,
-        pagination: CursorParams | None = None,
-        fields: frozenset[str] | None = None,
-    ):
-        """Query the Contribution collection, scoped to the current user. See ``get_many``."""
-        return await self.get_many(pagination=pagination, filter=filter, fields=fields)
-
     async def patch_one(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
         identifiers: dict[str, Any],
@@ -145,18 +134,7 @@ async def patch_one(  # pyright: ignore[reportIncompatibleMethodOverride]
                 identifiers=identifiers,
             ) from err
 
-    async def delete_contributions(
-        self,
-        filter: ContributionFilter,
-    ) -> DeleteResult | None:
-        """Bulk deletion of Contributions described by the filter.
-
-        Args:
-            filter (ContribtionFilter): the filter to use to identify contributions to delete
-        """
-        return await filter.filter(self.document_model.find(self._scope)).delete_many()
-
-    async def bulk_update(
+    async def patch_many(
         self,
         filter: ContributionFilter,
         fields: dict[str, Any],
@@ -187,9 +165,10 @@ async def bulk_update(
             matched=result.matched_count, modified=result.modified_count, projects=sorted(projects)
         )
 
-    async def get_contribution_ids(
+    async def list_ids(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
         filter: ContributionFilter,
+        session: AsyncClientSession | None = None,
     ) -> list[PydanticObjectId]:
         """Return the ids of scoped rows matching ``filter``.
 
@@ -199,6 +178,7 @@ async def get_contribution_ids(
 
         Args:
             filter: the caller-supplied query, applied on top of the user scope
+            session: unused; accepted to match the base ``list_ids`` signature
         """
         criteria: list[Any] = []
         if self._scope:
@@ -207,7 +187,7 @@ async def get_contribution_ids(
         collection = self.document_model.get_pymongo_collection()
         return [doc["_id"] async for doc in collection.find(query, {"_id": 1})]
 
-    async def insert_many_contributions(
+    async def insert_many(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
         docs: list[Contribution],
         session: AsyncClientSession | None = None,
@@ -220,7 +200,7 @@ async def insert_many_contributions(
         """
         return await self.document_model.insert_many(docs, ordered=False, session=session)
 
-    async def insert_contribution(
+    async def insert_one(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
         doc: Contribution,
         session: AsyncClientSession | None = None,
@@ -328,26 +308,23 @@ async def aggregate_project_stats(self, project_id: str) -> ProjectAggregate:
         agg.columns = finalize_columns(acc)
         return agg
 
-    async def upsert_one(
+    async def upsert_one(  # pyright: ignore[reportIncompatibleMethodOverride]
         self,
         identifiers: dict[str, Any],
         contribution: ContributionIn,
+        unique_value: Scalar | None = _UNSET,
         session: AsyncClientSession | None = None,
     ) -> Contribution:
-        """Atomically upsert a Contribution by its full identity.
-
-        Relies on the unique index over (project, material_id, chemical_system_id, formula,
-        unique_value, condition_key) so that concurrent requests targeting the same identity cannot both win the
-        insert branch. Fields the caller did not set are not touched (partial update). On insert a
-        fresh Contribution document is written with ``is_public=False``.
+        """Atomically upsert a single Contribution, keyed by the shape of ``identifiers``.
 
-        Args:
-            identifiers: the identity dict ContributionIn.identity_dict(unique_value) returns
-            contribution: the input payload to upsert
-
-        Returns:
-            Contribution: the document as it stands after the operation
+        Relies on the unique index over the identity fields as the concurrency tiebreaker.
+        On insert a fresh document is written with ``is_public=False``.
         """
+        if identifiers.keys() == {"id"}:
+            return await self._upsert_by_id(
+                identifiers["id"], contribution, None if unique_value is _UNSET else unique_value
+            )
+
         project = str(identifiers["project"])
         # Make sure the user is allowed to upsert a contribution under the provided project
         if not self._user.can_write(project):
@@ -373,31 +350,13 @@ async def upsert_one(
         result = await query  # pyright: ignore[reportGeneralTypeIssues] # beanie UpdateQuery is awaitable, but pyright doesn't see it
         return cast(Contribution, result)  # upsert always returns the resulting document
 
-    async def upsert_contribution_by_id(
+    async def _upsert_by_id(
         self,
         id: str,
         contribution: ContributionIn,
         unique_value: Scalar | None = None,
-    ):
-        """Upserts a single Contribution by its Mongo ``_id``.
-
-        If a Contribution with this id exists it is updated, otherwise inserted. ``unique_value`` is
-        server-resolved by the service from the project's ``unique_column`` and stamped on the doc so
-        the identity index stays correct. Because it is server-owned it is forced into the ``$set``
-        (bypassing ``exclude_none``), so re-resolving to ``None`` clears a previously-stored value on
-        update rather than leaving it stale.
-
-        Args:
-            id (str): the id of the Contribution to upsert
-            contribution (ContributionIn): the Contribution to be upserted
-            unique_value: the resolved identity value to stamp on the document
-
-        Returns:
-            Contribution: the upserted document
-
-        Raises:
-            PermissionError: if the caller is not authorized to write to ``contribution.project``
-        """
+    ) -> Contribution:
+        """Upsert a single Contribution keyed on its Mongo ``_id`` (see :meth:`upsert_one`)."""
         if not self._user.can_write(contribution.project):
             raise PermissionError(f"not authorized to write to project '{contribution.project}'")
 
@@ -412,7 +371,7 @@ async def upsert_contribution_by_id(
         try:
             query = self.document_model.find_one(
                 self._scope,
-                self.document_model.id == self._convert_object_id(id),
+                self.document_model.id == oid,
             ).upsert(
                 Set(update_data),
                 on_insert=doc,

mpcontribs-api/src/mpcontribs_api/domains/contributions/router.py

@@ -58,18 +58,15 @@ async def get_contributions(
     fields: FieldSelector = None,
 ):
     selected = ContributionOut.parse_fields(fields)
-    return await repo.get_contributions(pagination=pagination, filter=filter, fields=selected)
+    return await repo.get_many(pagination=pagination, filter=filter, fields=selected)
 
 
 @router.delete("", response_model=DeleteResponse, dependencies=[Depends(require_user)])
 async def delete_contributions(
     repo: ContributionDep,
     filter: ContributionFilter = FilterDepends(ContributionFilter),
 ) -> DeleteResponse:
-    # The repository returns a raw pymongo DeleteResult (or None when the filter matched nothing);
-    # convert it to the typed DeleteResponse so the endpoint has a stable, serializable contract.
-    result = await repo.delete_contributions(filter=filter)
-    return DeleteResponse.from_delete_result(result) if result is not None else DeleteResponse(num_deleted=0)
+    return await repo.delete_many(filter=filter)
 
 
 @router.patch("", dependencies=[Depends(require_user)])
@@ -87,7 +84,7 @@ async def patch_contributions(
     ``data`` deep-merges into each row's stored ``data`` by default
     Pass ``?replace_data=true`` to overwrite the whole ``data`` dict instead.
     """
-    return await service.bulk_update(filter=filter, update=body, replace_data=replace_data)
+    return await service.patch_many(filter=filter, update=body, replace_data=replace_data)
 
 
 # TODO: Might want to take contributions in from request body and run model_validate_json on it (much faster)
@@ -97,7 +94,7 @@ async def insert_contributions(
     contributions: list[ContributionIn],
 ):
     _enforce_bulk_limit(contributions)
-    return await service.insert_contributions(contributions=contributions)
+    return await service.insert_many(contributions=contributions)
 
 
 @router.put("", response_model=BulkWriteSummary[Contribution], dependencies=[Depends(require_user)])
@@ -106,7 +103,7 @@ async def upsert_contributions(
     contributions: list[ContributionIn],
 ):
     _enforce_bulk_limit(contributions)
-    return await service.upsert_contributions(contributions=contributions)
+    return await service.upsert_many(contributions=contributions)
 
 
 @router.get("/download/{short_mime}")
@@ -158,13 +155,13 @@ async def get_one(
 @router.put("/{id}", dependencies=[Depends(require_user)])
 async def upsert_one(service: ContributionServiceDep, id: str, contribution: ContributionIn):
     # The by-id upsert resolves the server-owned ``unique_value`` and enforces the unapproved quota
-    # (see ``ContributionService.upsert_contribution_by_id``), which the generic identity upsert does not.
-    return await service.upsert_contribution_by_id(id, contribution)
+    # (see ``ContributionService.upsert_one``), which the generic identity upsert does not.
+    return await service.upsert_one({"id": id}, contribution)
 
 
 @router.patch("/{id}", dependencies=[Depends(require_user)])
 async def patch_one(service: ContributionServiceDep, id: str, update: ContributionPatch, replace_data: bool = False):
     # The by-id patch re-resolves ``unique_value`` and validates the identifier hierarchy against the
-    # merged state (see ``ContributionService.patch_contribution_by_id``); ``?replace_data=true``
+    # merged state (see ``ContributionService.patch_one``); ``?replace_data=true``
     # overwrites the whole ``data`` dict instead of deep-merging.
-    return await service.patch_contribution_by_id(id, update=update, replace_data=replace_data)
+    return await service.patch_one({"id": id}, update=update, replace_data=replace_data)