add get_collection_by_id endpoint

collection_models.py

@@ -146,3 +146,7 @@ class CollectionsContainingNameRequest(BaseCollectionSearchLimitOffsetSort):
 
 class CollectionsContainingNameResponse(BaseCollectionQueryResponse):
     collections: list[Collection] = Field(title='list of public collections the provided label is a member of')
+
+
+class GetCollectionByIdRequest(BaseCollectionRequest):
+    collection_id: str = Field(title='id of the collection to fetch', examples=['ri2QqxnAqZT7'])

tests/test_collections_api.py

@@ -597,6 +597,32 @@ def test_fetch_collection_members_high_offset(self, test_test_client):
         # Should return empty suggestions when offset is beyond collection size
         assert len(response_json['suggestions']) == 0
 
+    @mark.integration_test
+    def test_get_collection_by_id(self, test_test_client):
+        # Test successful retrieval
+        response = test_test_client.post("/get_collection_by_id", json={
+            "collection_id": "ri2QqxnAqZT7"  # Known collection ID from other tests
+        })
+        assert response.status_code == 200
+        collection = response.json()
+        assert collection['collection_id'] == "ri2QqxnAqZT7"
+        assert 'title' in collection
+        assert 'owner' in collection
+        assert 'number_of_names' in collection
+        assert 'last_updated_timestamp' in collection
+        assert 'top_names' in collection
+        assert 'types' in collection
+        assert 'avatar_emoji' in collection
+        assert 'avatar_image' in collection
+
+        # Test non-existent collection
+        response = test_test_client.post("/get_collection_by_id", json={
+            "collection_id": "nonexistent_id"
+        })
+        assert response.status_code == 404
+        assert "Collection with id=nonexistent_id not found" in response.text
+
+
 @mark.usefixtures("unavailable_configuration")
 class TestCollectionApiUnavailable:
     @mark.integration_test
@@ -641,3 +667,11 @@ def test_collection_api_unavailability_count_collections_by_member(self, test_te
             "label": "australia"
         })
         assert response.status_code == 503
+
+    @mark.integration_test
+    def test_get_collection_by_id_unavailable(self, test_test_client):
+        response = test_test_client.post("/get_collection_by_id", json={
+            "collection_id": "ri2QqxnAqZT7"
+        })
+        assert response.status_code == 503
+        assert "Elasticsearch Unavailable" in response.text

web_api.py

@@ -105,6 +105,7 @@ def seed_all(seed: int | str):
     ScrambleCollectionTokens,
     CollectionCategory,
     FetchCollectionMembersRequest,
+    GetCollectionByIdRequest,
 )
 
 from collection_models import (
@@ -116,6 +117,7 @@ def seed_all(seed: int | str):
     CollectionsContainingNameRequest,
     CollectionsContainingNameResponse,
     CollectionCountByStringRequest,
+    Collection as CollectionModel,
 )
 
 
@@ -651,4 +653,25 @@ async def fetch_collection_members(fetch_command: FetchCollectionMembersRequest)
     return response[0]
 
 
+@app.post("/get_collection_by_id", response_model=CollectionModel, tags=['collections'])
+async def get_collection_by_id(request: GetCollectionByIdRequest):
+    """
+    Get information about a single collection by its ID.
+    Returns 404 if collection is not found.
+    Returns 503 if Elasticsearch is unavailable.
+    """
+
+    if not collections_matcher.active:
+        return Response(status_code=503, content='Elasticsearch Unavailable')
+
+    collections = collections_matcher.get_collections_by_id_list([request.collection_id])
+
+    if not collections:
+        return Response(status_code=404, content=f'Collection with id={request.collection_id} not found')
+
+    collection = convert_to_collection_format(collections)[0]
+
+    return collection
+
+
 #TODO gc.freeze() ?