PROD-2352 helper: get_urn_list method

CHANGELOG.md

@@ -23,6 +23,7 @@ The types of changes are:
 - Adds user_read scope to approver role so that they can update their own password [#5178](https://github.com/ethyca/fides/pull/5178)
 - Added PATCH endpoint for partially updating connection secrets [#5172](https://github.com/ethyca/fides/pull/5172)
 - Add success toast on confirming classification in data discovery tables [#5182](https://github.com/ethyca/fides/pull/5182)
+- Add function to return list of StagedResource objs according to list of URNs [#5192](https://github.com/ethyca/fides/pull/5192)
 
 ### Fixed
 - Fixed the OAuth2 configuration for the Snap integration [#5158](https://github.com/ethyca/fides/pull/5158)

src/fides/api/models/detection_discovery.py

@@ -4,7 +4,7 @@
 from enum import Enum
 from typing import Any, Dict, Iterable, Optional, Type
 
-from sqlalchemy import ARRAY, Boolean, Column, DateTime, ForeignKey, String
+from sqlalchemy import ARRAY, Boolean, Column, DateTime, ForeignKey, String, select
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.mutable import MutableDict
 from sqlalchemy.orm import Session, relationship
@@ -280,6 +280,15 @@ class StagedResource(Base):
     def get_urn(cls, db: Session, urn: str) -> Optional[StagedResource]:
         """Utility to retrieve the staged resource with the given URN"""
         return cls.get_by(db=db, field="urn", value=urn)
+
+    @classmethod
+    def get_urn_list(cls, db: Session, urns: Iterable[str]) -> Iterable[StagedResource]:
+        """
+        Utility to retrieve all staged resources with the given URNs
+        """
+
+        results = db.execute(select(StagedResource).where(StagedResource.urn.in_(urns)))
+        return results.scalars().all()
 
     def add_child_diff_status(self, diff_status: DiffStatus) -> None:
         """Increments the specified child diff status"""

tests/ops/models/test_detection_discovery.py

@@ -60,6 +60,11 @@ def create_staged_resource(self, db: Session):
             },
         )
         return resource
+
+    def test_get_urn_list(self, db: Session, create_staged_resource) -> None:
+        urn_list = [create_staged_resource.urn]
+        from_db = StagedResource.get_urn_list(db, urn_list)
+        assert len(from_db) == len(urn_list)
 
     def test_create_staged_resource(self, db: Session, create_staged_resource) -> None:
         """