Add public API endpoint for public collections

backend/btrixcloud/colls.py

@@ -30,6 +30,8 @@
     UpdatedResponse,
     SuccessResponse,
     CollectionSearchValuesResponse,
+    OrgPublicCollections,
+    CollAccessType,
 )
 from .utils import dt_now
 
@@ -395,6 +397,18 @@ async def add_successful_crawl_to_collections(self, crawl_id: str, cid: UUID):
             )
             await self.update_crawl_collections(crawl_id)
 
+    async def get_org_public_collections(self, org_slug: str):
+        """List public collections for org"""
+        org = await self.orgs.get_org_by_slug(org_slug)
+        collections, _ = await self.list_collections(
+            org.id,
+            access=CollAccessType.PUBLIC
+        )
+        return OrgPublicCollections(
+            name=org.name,
+            collections=collections
+        )
+
 
 # ============================================================================
 # pylint: disable=too-many-locals
@@ -582,4 +596,12 @@ async def download_collection(
     ):
         return await colls.download_collection(coll_id, org)
 
+    @app.get(
+        "/orgs/{org_slug}/collections/public",
+        tags=["collections"],
+        response_model=OrgPublicCollections,
+    )
+    async def get_org_public_collections(org_slug: str):
+        return await colls.get_org_public_collections(org_slug)
+
     return colls

backend/btrixcloud/models.py

@@ -1144,6 +1144,15 @@ class RenameOrg(BaseModel):
     slug: Optional[str] = None
 
 
+# ============================================================================
+class OrgPublicCollections(BaseModel):
+    """Model for listing public collections in org"""
+
+    orgName: str
+
+    collections: List[CollOut]
+
+
 # ============================================================================
 class OrgStorageRefs(BaseModel):
     """Input model for setting primary storage + optional replicas"""

backend/btrixcloud/orgs.py

@@ -284,6 +284,14 @@ async def get_org_by_id(self, oid: UUID) -> Organization:
 
         return Organization.from_dict(res)
 
+    async def get_org_by_slug(self, slug: str) -> Organization:
+        """Get an org by id"""
+        res = await self.orgs.find_one({"slug": slug})
+        if not res:
+            raise HTTPException(status_code=400, detail="invalid_org_slug")
+
+        return Organization.from_dict(res)
+
     async def get_default_org(self) -> Organization:
         """Get default organization"""
         res = await self.orgs.find_one({"default": True})

backend/test/test_collections.py

@@ -15,6 +15,7 @@
 
 _coll_id = None
 _second_coll_id = None
+_public_coll_id = None
 upload_id = None
 modified = None
 
@@ -66,6 +67,7 @@ def test_create_public_collection(
     assert data["added"]
     assert data["name"] == PUBLIC_COLLECTION_NAME
 
+    global _public_coll_id
     _public_coll_id = data["id"]
 
     # Verify that it is public
@@ -725,6 +727,50 @@ def test_filter_sort_collections(
     assert r.json()["detail"] == "invalid_sort_direction"
 
 
+def test_list_public_collections(
+    crawler_auth_headers, default_org_id, crawler_crawl_id, admin_crawl_id
+):
+    SECOND_PUBLIC_COLL_NAME = "Second public collection"
+
+    # Create new public collection
+    r = requests.post(
+        f"{API_PREFIX}/orgs/{default_org_id}/collections",
+        headers=crawler_auth_headers,
+        json={
+            "crawlIds": [crawler_crawl_id],
+            "name": SECOND_COLL_NAME,
+        },
+    )
+    assert r.status_code == 200
+    second_public_coll_id = r.json()["id"]
+
+    # Get default org slug
+    r = requests.get(
+        f"{API_PREFIX}/orgs/{default_org_id}",
+        headers=crawler_auth_headers,
+    )
+    assert r.status_code == 200
+    data = r.json()
+    org_slug = data["slug"]
+    org_name = data["name"]
+
+    # List public collections with no auth
+    r = requests.get(f"{API_PREFIX}/orgs/{org_slug}/collections/public")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["name"] == org_name
+
+    collections = data["collections"]
+    assert len(collections) == 2
+    for collection in collections:
+        assert collection["id"] in (_public_coll_id, second_public_coll_id)
+        assert collection["access"] == "public"
+
+    # Test non-existing slug
+    r = requests.get(f"{API_PREFIX}/orgs/nonexistentslug/collections/public")
+    assert r.status_code == 404
+
+
 def test_delete_collection(crawler_auth_headers, default_org_id, crawler_crawl_id):
     # Delete second collection
     r = requests.delete(