✨ metadata for api response and cursor navigation

app/api/routes/v1/affiliation_api.py

@@ -7,6 +7,7 @@
 from services.work import work_service
 from utils.encoder import JsonEncoder
 from schemas.general import QueryBase
+from core.config import settings
 
 router = Blueprint("affiliation_api_v1", __name__)
 
@@ -17,7 +18,7 @@ def affiliation(
     idx: str | None = None,
     section: str | None = "info",
     tab: str | None = None,
-    typ: str | None = None
+    typ: str | None = None,
 ) -> dict[str, Any] | None:
     result = None
     if section == "info":
@@ -36,9 +37,21 @@ def affiliation(
                     limit=params.max,
                     sort=params.sort,
                 )
+                total = work_service.count_papers(
+                    affiliation_id=idx, affiliation_type=typ
+                )
     else:
         result = None
-    return {"data": result, "count": len(result)}
+    return {
+        "data": result,
+        "info": {
+            "total_products": total,
+            "count": len(result),
+            "cursor": params.get_cursor(
+                path=f"{settings.API_V1_STR}/affiliation/{idx}/research/products"
+            ),
+        },
+    }
 
 
 @router.route("/<typ>/<id>", methods=["GET"])

app/api/routes/v1/person_api.py

@@ -6,6 +6,7 @@
 from services.work import work_service
 from utils.encoder import JsonEncoder
 from schemas.general import QueryBase
+from core.config import settings
 
 router = Blueprint("person_api_v1", __name__)
 
@@ -20,7 +21,17 @@ def get_person(id: str | None, section: str | None, tab: str | None):
         works = work_service.get_research_products_by_author_csv(
             author_id=id, sort=params.sort, skip=params.skip, limit=params.max
         )
-        result = {"data": works, "count": len(works)}
+        total = work_service.count_papers(author_id=id)
+        result = {
+            "data": works,
+            "info": {
+                "total_products": total,
+                "count": len(works),
+                "cursor": params.get_cursor(
+                    path=f"{settings.API_V1_STR}/person/{id}/research/products"
+                ),
+            },
+        }
     else:
         result = None
 

app/core/config.py

@@ -34,6 +34,8 @@ class Settings(BaseSettings):
 
     APP_PORT: str | int = 8010
 
+    APP_DOMAIN: str = "localhost:8010"
+
     @validator("MONGO_URI", pre=True)
     def validate_mongo_uri(cls, v: Optional[str], values: Dict[str, Any]) -> str:
         return MongoDsn.build(

app/schemas/general.py

@@ -2,9 +2,11 @@
 
 from pydantic import BaseModel, validator, Field, model_validator
 from odmantic.bson import BSON_TYPES_ENCODERS
+from core.config import settings
 
 DBSchemaType = TypeVar("DBSchemaType", bound=BaseModel)
 
+
 class Type(BaseModel):
     source: str
     type: str | None
@@ -20,6 +22,7 @@ class ExternalId(BaseModel):
     source: str | None
     provenance: str | None = None
 
+
 class ExternalURL(BaseModel):
     url: str | int | None
     source: str | None
@@ -37,7 +40,7 @@ class Status(BaseModel):
 
 class QueryBase(BaseModel):
     max: int = Field(10, gt=0)
-    page: int = 1
+    page: int = Field(1, gt=0)
     keywords: str | None = ""
     sort: str = ""
 
@@ -50,16 +53,37 @@ def limit_validator(cls, v, values):
     @validator("skip", always=True)
     def skip_validator(cls, v, values):
         return v if v else (values["page"] - 1) * values["max"]
-    
+
     @property
     def get_search(self) -> dict[str, Any]:
         return {}
 
+    def next_query(self) -> str:
+        return (
+            f"max={self.max}&page={self.page + 1}&"
+            f"sort={self.sort}&keywords={self.keywords}"
+        )
+
+    def previous_query(self) -> str | None:
+        return (
+            f"max={self.max}&page={self.page - 1}&"
+            f"sort={self.sort}&keywords={self.keywords}"
+        )
+
+    def get_cursor(self, path: str) -> dict[str, str]:
+        """
+        Compute the cursor for the given path and query.
+        """
+        return {
+            "next": f"{settings.APP_DOMAIN}{path}?{self.next_query()}",
+            "previous": f"{settings.APP_DOMAIN}{path}?{self.previous_query()}" if self.page > 1 else None,
+        }
+
 
 class GeneralMultiResponse(BaseModel, Generic[DBSchemaType]):
     total_results: int | None = None
     data: list[DBSchemaType] | None = Field(default_factory=list)
     count: int | None = None
     page: int | None = None
-    
+
     model_config = {"json_encoders": BSON_TYPES_ENCODERS}

app/schemas/person.py

@@ -91,7 +91,7 @@ def remove_sensitive_ids(cls, v: list[ExternalId]) -> list[ExternalId]:
             )
         )
     products_count: int | None = None
-    citations_count: int | None = None
+    citations_count: list | None = None
 
 
 class PersonQueryParams(QueryBase):

app/services/work.py

@@ -30,6 +30,21 @@ def update_source(work: WorkProccessed):
                 serials[serial.source] = serial.id
             work.source.serials = serials
 
+    def count_papers(
+        self,
+        *,
+        affiliation_id: str | None = None,
+        affiliation_type: str | None = None,
+        author_id: str | None = None,
+    ) -> int:
+        if affiliation_id and affiliation_type:
+            return WorkRepository.count_papers(
+                affiliation_id=affiliation_id, affiliation_type=affiliation_type
+            )
+        if author_id:
+            return WorkRepository.count_papers_by_author(author_id=author_id)
+        return 0
+
     def get_info(self, *, id: str) -> dict[str, Any]:
         work = super().get_by_id(id=id)
         self.update_authors_external_ids(work)