Add field projection to document listing (#408)

Add an optional `fields` argument to list_documents() (sync, async, folder, and user
scopes) so callers fetch only the document fields they need. The projection is pushed
into the SQL SELECT, so listing metadata never reads the full document text stored in
system_metadata.content.

- external_id and content_type are always included so projected responses parse into
  Document objects; metadata_types is included when a metadata field is requested so
  typed values (datetime/date/decimal) are reconstructed instead of returned as strings.
- Nested fields supported (e.g. metadata.client).
- DB-level column projection via labeled SELECT in list_documents_flexible.
- Bump SDK to 1.2.3; add unit tests for the SDK and server projection paths.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: Adityav369

core/database/postgres_database.py

@@ -30,6 +30,42 @@
     "summary_updated_at",
 }
 
+# Maps a public Document field name to the underlying table column. Used to SELECT
+# only the columns a projection needs, so listing metadata never reads the heavy
+# `system_metadata.content` (the full document text).
+DOCUMENT_PROJECTION_COLUMN_MAP = {
+    "external_id": DocumentModel.external_id,
+    "content_type": DocumentModel.content_type,
+    "filename": DocumentModel.filename,
+    "metadata": DocumentModel.doc_metadata,
+    "metadata_types": DocumentModel.metadata_types,
+    "storage_info": DocumentModel.storage_info,
+    "system_metadata": DocumentModel.system_metadata,
+    "additional_metadata": DocumentModel.additional_metadata,
+    "chunk_ids": DocumentModel.chunk_ids,
+    "folder_name": DocumentModel.folder_name,
+    "folder_path": DocumentModel.folder_path,
+    "folder_id": DocumentModel.folder_id,
+    "app_id": DocumentModel.app_id,
+    "end_user_id": DocumentModel.end_user_id,
+}
+DOCUMENT_PROJECTION_ORDER = [
+    "external_id",
+    "content_type",
+    "filename",
+    "metadata",
+    "metadata_types",
+    "storage_info",
+    "system_metadata",
+    "additional_metadata",
+    "chunk_ids",
+    "folder_name",
+    "folder_path",
+    "folder_id",
+    "app_id",
+    "end_user_id",
+]
+
 
 class PostgresDatabase:
     """PostgreSQL implementation for document metadata storage."""
@@ -403,10 +439,16 @@ async def list_documents_flexible(
         include_status_counts: bool = False,
         include_folder_counts: bool = False,
         return_documents: bool = True,
+        fields: Optional[List[str]] = None,
         sort_by: Optional[str] = None,
         sort_direction: str = "desc",
     ) -> Dict[str, Any]:
-        """List documents with optional aggregate metadata. Field projection is handled at application layer."""
+        """List documents with optional aggregate metadata and projected document fields.
+
+        When ``fields`` is provided, only the underlying columns required to serve those
+        fields are selected from Postgres, so listing metadata never reads the full
+        document text stored in ``system_metadata.content``.
+        """
         limit = max(limit, 0) if limit is not None else None
         skip = max(skip, 0)
 
@@ -440,16 +482,24 @@ async def list_documents_flexible(
 
                 final_where_clause = " AND ".join(where_clauses) if where_clauses else "TRUE"
 
-                documents: List[Document] = []
+                documents: List[Any] = []
                 returned_count = 0
                 has_more = False
 
                 fetch_documents = return_documents and (limit is None or limit > 0)
 
                 if fetch_documents:
-                    # Note: We always select all columns from the database
-                    # Field projection is handled at the application layer for simplicity
-                    base_query = select(DocumentModel).where(text(final_where_clause).bindparams(**filter_params))
+                    projection_fields = self._resolve_document_projection_fields(fields)
+                    if projection_fields:
+                        # Select only the columns the projection needs (skips the heavy
+                        # system_metadata/content read entirely).
+                        selected_columns = self._document_projection_columns(projection_fields)
+                        base_query = select(*selected_columns).where(
+                            text(final_where_clause).bindparams(**filter_params)
+                        )
+                    else:
+                        base_query = select(DocumentModel).where(text(final_where_clause).bindparams(**filter_params))
+
                     order_clause = self._resolve_document_sort_clause(sort_by, sort_direction)
                     if order_clause is not None:
                         base_query = base_query.order_by(order_clause, DocumentModel.external_id.asc())
@@ -462,13 +512,20 @@ async def list_documents_flexible(
                         base_query = base_query.limit(fetch_limit)
 
                     result = await session.execute(base_query)
-                    doc_models = result.scalars().all()
-
-                    if fetch_limit is not None and len(doc_models) > limit:
-                        has_more = True
-                        doc_models = doc_models[:limit]
 
-                    documents = [Document(**_document_model_to_dict(doc_model)) for doc_model in doc_models]
+                    if projection_fields:
+                        documents = [
+                            self._document_projection_row_to_dict(row, projection_fields) for row in result.mappings()
+                        ]
+                        if fetch_limit is not None and len(documents) > limit:
+                            has_more = True
+                            documents = documents[:limit]
+                    else:
+                        doc_models = result.scalars().all()
+                        if fetch_limit is not None and len(doc_models) > limit:
+                            has_more = True
+                            doc_models = doc_models[:limit]
+                        documents = [Document(**_document_model_to_dict(doc_model)) for doc_model in doc_models]
                     returned_count = len(documents)
 
                 total_count: Optional[int] = None
@@ -568,6 +625,60 @@ def _resolve_document_sort_clause(self, sort_by: Optional[str], sort_direction:
             f"{direction} NULLS LAST"
         )
 
+    @staticmethod
+    def _resolve_document_projection_fields(fields: Optional[List[str]]) -> Optional[set]:
+        """Resolve requested API fields to the document table columns needed to serve them.
+
+        Note: ``summary_*`` and ``page_count`` are derived from ``system_metadata``, so
+        projecting them reads that column (which also holds the full document text). Plain
+        ``metadata`` lives in its own column and stays lightweight.
+        """
+        if not fields:
+            return None
+
+        requested_roots = {field.strip().split(".", 1)[0] for field in fields if field and field.strip()}
+        if not requested_roots:
+            return None
+
+        # external_id is always needed to identify each document.
+        resolved_fields = {"external_id"}
+        for root in requested_roots:
+            if root in DOCUMENT_PROJECTION_COLUMN_MAP:
+                resolved_fields.add(root)
+            elif root in SUMMARY_METADATA_KEYS:
+                # summary_* values are derived from system_metadata.
+                resolved_fields.add("system_metadata")
+            elif root == "page_count":
+                resolved_fields.add("system_metadata")
+                resolved_fields.add("chunk_ids")
+
+        return resolved_fields
+
+    @staticmethod
+    def _document_projection_columns(fields: set):
+        """Return a stable list of labeled SQLAlchemy columns for a document projection."""
+        return [
+            DOCUMENT_PROJECTION_COLUMN_MAP[field].label(field) for field in DOCUMENT_PROJECTION_ORDER if field in fields
+        ]
+
+    @staticmethod
+    def _document_projection_row_to_dict(row: Any, fields: set) -> Dict[str, Any]:
+        """Convert a projected document row (a SQLAlchemy mapping) to the public document dict shape."""
+        document = dict(row)
+
+        for key in ("metadata", "metadata_types", "storage_info", "system_metadata", "additional_metadata"):
+            if key in document and document[key] is None:
+                document[key] = {}
+        if "chunk_ids" in document and document["chunk_ids"] is None:
+            document["chunk_ids"] = []
+
+        system_metadata = document.get("system_metadata") or {}
+        if "system_metadata" in fields and isinstance(system_metadata, dict):
+            for key in SUMMARY_METADATA_KEYS:
+                document[key] = system_metadata.get(key)
+
+        return document
+
     async def update_document(
         self,
         document_id: str,

core/routes/documents.py

@@ -102,6 +102,7 @@ async def list_docs(
             include_status_counts=request.include_status_counts,
             include_folder_counts=request.include_folder_counts,
             return_documents=request.return_documents,
+            fields=request.fields,
             sort_by=request.sort_by,
             sort_direction=request.sort_direction,
         )

core/tests/unit/test_document_projection.py

@@ -0,0 +1,89 @@
+"""Unit tests for document field projection (list_docs `fields`)."""
+
+from sqlalchemy import select
+
+from core.database.postgres_database import PostgresDatabase
+from core.routes.utils import project_document_fields
+
+
+class TestResolveProjectionFields:
+    """PostgresDatabase._resolve_document_projection_fields."""
+
+    def test_no_fields_returns_none(self):
+        assert PostgresDatabase._resolve_document_projection_fields(None) is None
+        assert PostgresDatabase._resolve_document_projection_fields([]) is None
+        assert PostgresDatabase._resolve_document_projection_fields(["  "]) is None
+
+    def test_always_includes_external_id(self):
+        assert PostgresDatabase._resolve_document_projection_fields(["metadata"]) == {
+            "external_id",
+            "metadata",
+        }
+
+    def test_nested_field_resolves_to_root_column(self):
+        # "metadata.client" only needs the doc_metadata column.
+        assert PostgresDatabase._resolve_document_projection_fields(["metadata.client"]) == {
+            "external_id",
+            "metadata",
+        }
+
+    def test_summary_key_requires_system_metadata(self):
+        assert PostgresDatabase._resolve_document_projection_fields(["summary_storage_key"]) == {
+            "external_id",
+            "system_metadata",
+        }
+
+    def test_page_count_requires_system_metadata_and_chunk_ids(self):
+        assert PostgresDatabase._resolve_document_projection_fields(["page_count"]) == {
+            "external_id",
+            "system_metadata",
+            "chunk_ids",
+        }
+
+
+class TestProjectionColumns:
+    """The generated SQL only selects the projected columns."""
+
+    def test_metadata_projection_does_not_read_content(self):
+        fields = PostgresDatabase._resolve_document_projection_fields(["metadata"])
+        sql = str(select(*PostgresDatabase._document_projection_columns(fields)))
+        assert "documents.external_id" in sql
+        assert "documents.doc_metadata" in sql
+        # The heavy column (system_metadata holds the full document text) is not selected.
+        assert "system_metadata" not in sql
+
+
+class TestProjectionRowToDict:
+    """PostgresDatabase._document_projection_row_to_dict."""
+
+    def test_none_jsonb_normalized(self):
+        row = {"external_id": "doc-1", "metadata": None, "chunk_ids": None}
+        out = PostgresDatabase._document_projection_row_to_dict(row, {"external_id", "metadata", "chunk_ids"})
+        assert out["metadata"] == {}
+        assert out["chunk_ids"] == []
+
+    def test_summary_keys_derived_when_system_metadata_present(self):
+        row = {"external_id": "doc-1", "system_metadata": {"summary_storage_key": "s3://x"}}
+        out = PostgresDatabase._document_projection_row_to_dict(row, {"external_id", "system_metadata"})
+        assert out["summary_storage_key"] == "s3://x"
+
+
+class TestProjectDocumentFields:
+    """core.routes.utils.project_document_fields (application-layer shaping)."""
+
+    def test_projects_requested_fields_only(self):
+        doc = {"external_id": "d1", "content_type": "text/plain", "metadata": {"a": 1}, "system_metadata": {"big": "x"}}
+        out = project_document_fields(doc, ["metadata"])
+        assert set(out) == {"external_id", "metadata"}
+        assert out["metadata"] == {"a": 1}
+
+    def test_nested_projection(self):
+        doc = {"external_id": "d1", "metadata": {"client": "ExampleCo", "doc_type": "invoice", "secret": "z"}}
+        out = project_document_fields(doc, ["metadata.client", "metadata.doc_type"])
+        assert out["metadata"] == {"client": "ExampleCo", "doc_type": "invoice"}
+
+    def test_no_fields_returns_all(self):
+        doc = {"external_id": "d1", "metadata": {"a": 1}}
+        out = project_document_fields(doc, None)
+        assert out["metadata"] == {"a": 1}
+        assert out["external_id"] == "d1"

sdks/python/CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.2.3] - 2026-06-18
+
+### Added
+- `list_documents(fields=[...])` on sync, async, folder, and user-scoped clients: request only
+  the document fields you need (e.g. `["metadata"]`). The server reads and returns only those
+  columns, so listing metadata never downloads the full document text. `external_id` and
+  `content_type` are always included; `metadata_types` is included automatically when a metadata
+  field is requested so typed values (datetime/date/decimal) are reconstructed rather than
+  returned as raw strings. Nested fields are supported (e.g. `["metadata.client"]`).
+
 ## [1.2.2] - 2026-02-09
 
 ### Added

sdks/python/README.md

@@ -93,6 +93,11 @@ renamed = moved.rename("specs-v2")
 # Scope queries to a path and include descendants with folder_depth=-1
 chunks = folder.retrieve_chunks(query="design notes", folder_depth=-1)
 docs = db.list_documents(folder_name="/projects/alpha", folder_depth=-1)
+
+# List only the fields you need. The server reads and returns just those columns, so
+# the full document text is never downloaded — fast for large corpora.
+for doc in db.list_documents(fields=["metadata"]).documents:
+    print(doc.external_id, doc.metadata)
 ```
 
 `Folder.full_path` is exposed on folder objects, and `Document.folder_path` mirrors server responses for tracing scope.

sdks/python/morphik/__init__.py

@@ -14,4 +14,4 @@
     "DocumentQueryResponse",
 ]
 
-__version__ = "1.2.2"
+__version__ = "1.2.3"

sdks/python/morphik/_internal.py

@@ -428,6 +428,7 @@ def _prepare_list_documents_request(
         completed_only: bool,
         sort_by: Optional[str],
         sort_direction: str,
+        fields: Optional[List[str]] = None,
     ) -> Tuple[Dict[str, Any], Dict[str, Any]]:
         """Prepare request for list_docs endpoint"""
         params = {}
@@ -450,6 +451,15 @@ def _prepare_list_documents_request(
             "sort_by": sort_by,
             "sort_direction": sort_direction,
         }
+        if fields:
+            # Always include the fields required to reconstruct a Document client-side, so
+            # projected responses still parse into Document objects. When any metadata field
+            # is requested, also pull metadata_types so typed values (datetime/date/decimal)
+            # are reconstructed instead of returned as raw strings.
+            projected = ["external_id", "content_type", *fields]
+            if any(field.split(".", 1)[0] == "metadata" for field in fields):
+                projected.append("metadata_types")
+            data["fields"] = list(dict.fromkeys(projected))
         return params, data
 
     def _prepare_batch_get_documents_request(

sdks/python/morphik/_scoped_ops.py

@@ -277,6 +277,7 @@ def _scoped_list_documents(
         completed_only: bool,
         sort_by: Optional[str],
         sort_direction: str,
+        fields: Optional[List[str]] = None,
     ):
         params, data = self._logic._prepare_list_documents_request(
             skip,
@@ -291,6 +292,7 @@ def _scoped_list_documents(
             completed_only,
             sort_by,
             sort_direction,
+            fields,
         )
 
         return self._execute_scoped_operation(