Add API endpoint to GET single public collection

Author: tw4l

backend/btrixcloud/colls.py

@@ -2,6 +2,8 @@
 Collections API
 """
 
+# pylint: disable=too-many-lines
+
 from collections import Counter
 from uuid import UUID, uuid4
 from typing import Optional, List, TYPE_CHECKING, cast, Dict, Tuple, Any, Union
@@ -210,11 +212,11 @@ async def remove_crawls_from_collection(
         return await self.get_collection_out(coll_id, org)
 
     async def get_collection_raw(
-        self, coll_id: UUID, public_only: bool = False
+        self, coll_id: UUID, public_or_unlisted_only: bool = False
     ) -> Dict[str, Any]:
         """Get collection by id as dict from database"""
         query: dict[str, object] = {"_id": coll_id}
-        if public_only:
+        if public_or_unlisted_only:
             query["access"] = {"$in": ["public", "unlisted"]}
 
         result = await self.collections.find_one(query)
@@ -224,17 +226,21 @@ async def get_collection_raw(
         return result
 
     async def get_collection(
-        self, coll_id: UUID, public_only: bool = False
+        self, coll_id: UUID, public_or_unlisted_only: bool = False
     ) -> Collection:
         """Get collection by id"""
-        result = await self.get_collection_raw(coll_id, public_only)
+        result = await self.get_collection_raw(coll_id, public_or_unlisted_only)
         return Collection.from_dict(result)
 
     async def get_collection_out(
-        self, coll_id: UUID, org: Organization, resources=False, public_only=False
+        self,
+        coll_id: UUID,
+        org: Organization,
+        resources=False,
+        public_or_unlisted_only=False,
     ) -> CollOut:
         """Get CollOut by id"""
-        result = await self.get_collection_raw(coll_id, public_only)
+        result = await self.get_collection_raw(coll_id, public_or_unlisted_only)
 
         if resources:
             result["resources"] = await self.get_collection_crawl_resources(coll_id)
@@ -248,6 +254,26 @@ async def get_collection_out(
 
         return CollOut.from_dict(result)
 
+    async def get_public_collection_out(
+        self, coll_id: UUID, org: Organization
+    ) -> PublicCollOut:
+        """Get PublicCollOut by id"""
+        result = await self.get_collection_raw(coll_id)
+
+        if result.get("access") != "public":
+            raise HTTPException(status_code=404, detail="collection_not_found")
+
+        result["resources"] = await self.get_collection_crawl_resources(coll_id)
+
+        thumbnail = result.get("thumbnail")
+        if thumbnail:
+            image_file = ImageFile(**thumbnail)
+            result["thumbnail"] = await image_file.get_public_image_file_out(
+                org, self.storage_ops
+            )
+
+        return PublicCollOut.from_dict(result)
+
     async def list_collections(
         self,
         org: Organization,
@@ -825,7 +851,7 @@ async def get_collection_public_replay(
         org: Organization = Depends(org_public),
     ):
         coll = await colls.get_collection_out(
-            coll_id, org, resources=True, public_only=True
+            coll_id, org, resources=True, public_or_unlisted_only=True
         )
         response.headers["Access-Control-Allow-Origin"] = "*"
         response.headers["Access-Control-Allow-Headers"] = "*"
@@ -920,6 +946,24 @@ async def get_org_public_collections(
             sort_direction=sortDirection,
         )
 
+    @app.get(
+        "/public-collections/{org_slug}/collections/{coll_id}",
+        tags=["collections"],
+        response_model=PublicCollOut,
+    )
+    async def get_public_collection(
+        org_slug: str,
+        coll_id: UUID,
+    ):
+        try:
+            org = await colls.orgs.get_org_by_slug(org_slug)
+        # pylint: disable=broad-exception-caught
+        except Exception:
+            # pylint: disable=raise-missing-from
+            raise HTTPException(status_code=404, detail="collection_not_found")
+
+        return await colls.get_public_collection_out(coll_id, org)
+
     @app.get(
         "/orgs/{oid}/collections/{coll_id}/urls",
         tags=["collections"],

backend/test/test_collections.py

@@ -15,6 +15,19 @@
 CAPTION = "Short caption"
 UPDATED_CAPTION = "Updated caption"
 
+NON_PUBLIC_COLL_FIELDS = (
+    "oid",
+    "modified",
+    "crawlCount",
+    "pageCount",
+    "totalSize",
+    "tags",
+    "access",
+    "homeUrlPageId",
+)
+NON_PUBLIC_IMAGE_FIELDS = ("originalFilename", "userid", "userName", "created")
+
+
 _coll_id = None
 _second_coll_id = None
 _public_coll_id = None
@@ -1024,7 +1037,7 @@ def test_list_public_colls_home_url_thumbnail():
         assert coll["dateEarliest"]
         assert coll["dateLatest"]
 
-        for field in non_public_fields:
+        for field in NON_PUBLIC_COLL_FIELDS:
             assert field not in coll
 
         if coll["id"] == _public_coll_id:
@@ -1042,13 +1055,69 @@ def test_list_public_colls_home_url_thumbnail():
             assert thumbnail["size"]
             assert thumbnail["mime"]
 
-            for field in non_public_image_fields:
+            for field in NON_PUBLIC_IMAGE_FIELDS:
                 assert field not in thumbnail
 
         if coll["id"] == _second_public_coll_id:
             assert coll["description"]
 
 
+def test_get_public_collection():
+    r = requests.get(
+        f"{API_PREFIX}/public-collections/{default_org_slug}/collections/{_public_coll_id}"
+    )
+    assert r.status_code == 200
+    coll = r.json()
+
+    assert coll["id"] == _public_coll_id
+    assert coll["name"]
+    assert coll["resources"]
+    assert coll["dateEarliest"]
+    assert coll["dateLatest"]
+
+    for field in NON_PUBLIC_COLL_FIELDS:
+        assert field not in coll
+
+    assert coll["caption"] == CAPTION
+
+    assert coll["homeUrl"]
+    assert coll["homeUrlTs"]
+
+    thumbnail = coll["thumbnail"]
+    assert thumbnail
+
+    assert thumbnail["name"]
+    assert thumbnail["path"]
+    assert thumbnail["hash"]
+    assert thumbnail["size"]
+    assert thumbnail["mime"]
+
+    for field in NON_PUBLIC_IMAGE_FIELDS:
+        assert field not in thumbnail
+
+    # Invalid org slug - don't reveal whether org exists or not, use
+    # same exception as if collection doesn't exist
+    r = requests.get(
+        f"{API_PREFIX}/public-collections/doesntexist/collections/{_public_coll_id}"
+    )
+    assert r.status_code == 404
+    assert r.json()["detail"] == "collection_not_found"
+
+    # Invalid collection id
+    r = requests.get(
+        f"{API_PREFIX}/public-collections/{default_org_slug}/collections/doesntexist"
+    )
+    assert r.status_code == 404
+    assert r.json()["detail"] == "collection_not_found"
+
+    # Collection isn't public
+    r = requests.get(
+        f"{API_PREFIX}/public-collections/{default_org_slug}/collections/{ _coll_id}"
+    )
+    assert r.status_code == 404
+    assert r.json()["detail"] == "collection_not_found"
+
+
 def test_delete_thumbnail(crawler_auth_headers, default_org_id):
     r = requests.delete(
         f"{API_PREFIX}/orgs/{default_org_id}/collections/{_public_coll_id}/thumbnail",