Merge pull request #177 from lsst-sqre:tickets/DM-50016

DM-50016: Employ pagination in the Ook Links API

Author: jonathansick

CHANGELOG.md

@@ -2,6 +2,17 @@
 
 <!-- scriv-insert-here -->
 
+<a id='changelog-0.12.0'></a>
+## 0.12.0 (2025-04-16)
+
+### New features
+
+- The Links API collection endpoints now use pagination for improved performance and usability. Ook uses keyset pagination, so look for a Links header with `next`, `prev`, and `first` links. Use these URLs to advance to the next page. The `X-Total-Count` header indicates the total number of items in the collection. Pagination applies to the following endpoints:
+
+  - `GET /ook/links/domains/sdm/schemas`
+  - `GET /ook/links/domains/sdm/schemas/:schema/tables`
+  - `GET /ook/links/domains/sdm/schemas/:schema/tables/:table/columns`
+
 <a id='changelog-0.11.0'></a>
 
 ## 0.11.0 (2025-04-04)

src/ook/dependencies/context.py

@@ -9,7 +9,7 @@
 from dataclasses import dataclass
 from typing import Annotated, Any
 
-from fastapi import Depends, Request
+from fastapi import Depends, Request, Response
 from safir.dependencies.db_session import db_session_dependency
 from safir.dependencies.logger import logger_dependency
 from sqlalchemy.ext.asyncio import async_scoped_session
@@ -37,6 +37,9 @@ class RequestContext:
     request: Request
     """The incoming request."""
 
+    response: Response
+    """The response to the request."""
+
     logger: BoundLogger
     """The request logger, rebound with discovered context."""
 
@@ -73,6 +76,7 @@ def __init__(self) -> None:
     async def __call__(
         self,
         request: Request,
+        response: Response,
         session: Annotated[
             async_scoped_session, Depends(db_session_dependency)
         ],
@@ -81,6 +85,7 @@ async def __call__(
         """Create a per-request context and return it."""
         return RequestContext(
             request=request,
+            response=response,
             logger=logger,
             session=session,
             factory=Factory(

src/ook/domain/links.py

@@ -2,106 +2,146 @@
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, Field, RootModel
 
 __all__ = [
     "Link",
     "LinksCollection",
     "SdmColumnLink",
     "SdmColumnLinksCollection",
+    "SdmLinksCollection",
     "SdmSchemaLink",
     "SdmSchemaLinksCollection",
     "SdmTableLink",
     "SdmTableLinksCollection",
 ]
 
 
-@dataclass(slots=True, kw_only=True)
-class Link:
-    """A link to documentation."""
+class Link(BaseModel):
+    """A link to documentation.
+
+    This is a Pydantic model to facilitate the parsing of SQLAlchemy queries
+    directly into the domain.
+    """
 
-    html_url: str
-    """The URL to the documentation page."""
+    html_url: str = Field(description="The URL to the documentation page.")
 
-    title: str
-    """The title of the documentation."""
+    title: str = Field(description="The title of the documentation.")
 
-    type: str
-    """The type of documentation."""
+    type: str = Field(description="The type of documentation.")
 
-    collection_title: str | None = None
-    """The title of the collection of documentation this link refers to."""
+    collection_title: str | None = Field(
+        None,
+        description=(
+            "The title of the collection of documentation this link refers to."
+        ),
+    )
 
 
-@dataclass(slots=True, kw_only=True)
 class SdmSchemaLink(Link):
     """A link to an SDM schema's documentation."""
 
-    name: str
-    """The name of the schema."""
+    schema_name: str = Field(description="The name of the schema.")
 
 
-@dataclass(slots=True, kw_only=True)
 class SdmTableLink(Link):
     """A link to an SDM table's documentation."""
 
-    schema_name: str
-    """The name of the schema."""
+    schema_name: str = Field(description="The name of the schema.")
 
-    name: str
-    """The name of the table."""
+    table_name: str = Field(description="The name of the table.")
 
 
-@dataclass(slots=True, kw_only=True)
 class SdmColumnLink(Link):
     """A link to an SDM column's documentation."""
 
-    schema_name: str
-    """The name of the schema."""
+    schema_name: str = Field(description="The name of the schema.")
 
-    table_name: str
-    """The name of the table."""
+    table_name: str = Field(description="The name of the table.")
 
-    name: str
-    """The name of the column."""
+    column_name: str = Field(description="The name of the column.")
 
 
-@dataclass(slots=True, kw_only=True)
-class LinksCollection[T: Link]:
-    """A collection of links to documentation of a specific entity."""
+class LinksCollection[T: Link](BaseModel):
+    """A collection of links to documentation of a specific entity.
 
-    links: list[T]
-    """The documentation links."""
+    This is a Pydantic model to facilitate the parsing of SQLAlchemy queries
+    directly into the domain.
+    """
+
+    links: list[T] = Field(description="The documentation links.")
 
 
-@dataclass(slots=True, kw_only=True)
 class SdmSchemaLinksCollection(LinksCollection[SdmSchemaLink]):
     """A collection of links to an SDM schema."""
 
-    schema_name: str
-    """The name of the schema."""
+    domain: Literal["sdm"] = Field(
+        description="The links domain of the entity."
+    )
+
+    entity_type: Literal["schema"] = Field(
+        description="The type of the entity in the domain."
+    )
+
+    schema_name: str = Field(description="The name of the schema.")
 
 
-@dataclass(slots=True, kw_only=True)
 class SdmTableLinksCollection(LinksCollection[SdmTableLink]):
     """A collection of links to an SDM table."""
 
-    schema_name: str
-    """The name of the schema."""
+    domain: Literal["sdm"] = Field(
+        description="The links domain of the entity."
+    )
+
+    entity_type: Literal["table"] = Field(
+        description="The type of the entity in the domain."
+    )
+
+    schema_name: str = Field(description="The name of the schema.")
+
+    table_name: str = Field(description="The name of the table.")
 
-    table_name: str
-    """The name of the table."""
+    tap_table_index: int | None = Field(
+        description="The sorting index of the table in the TAP schema."
+    )
 
 
-@dataclass(slots=True, kw_only=True)
 class SdmColumnLinksCollection(LinksCollection[SdmColumnLink]):
     """A collection of links to SDM columns."""
 
-    schema_name: str
-    """The name of the schema."""
+    domain: Literal["sdm"] = Field(
+        description="The links domain of the entity."
+    )
+
+    entity_type: Literal["column"] = Field(
+        description="The type of the entity in the domain."
+    )
+
+    schema_name: str = Field(description="The name of the schema.")
+
+    table_name: str = Field(description="The name of the table.")
+
+    column_name: str = Field(description="The name of the column.")
+
+    tap_column_index: int | None = Field(
+        description="The sorting index of the column in the TAP schema."
+    )
+
+
+SdmLinksCollection = RootModel[
+    Annotated[
+        SdmSchemaLinksCollection
+        | SdmTableLinksCollection
+        | SdmColumnLinksCollection,
+        Field(discriminator="entity_type"),
+    ]
+]
+"""A generic collection of SDM links for any SDM entity type.
 
-    table_name: str
-    """The name of the table."""
+Use this root model to parse SQLAlchemy query results when the domain type
+can be any of the three SDM entity types: schema, table, or column.
 
-    column_name: str
-    """The name of the column."""
+Access the underlying collection using the ``root`` attribute.
+"""