Add PDF split/merge support with live tests

- Introduced `split_pdf` and `merge_pdfs` methods in both sync and async
  clients.
- Added `PdfSplitPayload` and `PdfMergePayload` models for payload
  validation and serialization.
- Created associated types and serializers for page groupings and merge
  sources.
- Added comprehensive live and unit tests for splitting and merging
  functionality.

Assisted-by: Codex

Author: datalogics-kam

src/pdfrest/client.py

@@ -40,16 +40,20 @@
     GifPdfRestPayload,
     JpegPdfRestPayload,
     PdfInfoPayload,
+    PdfMergePayload,
     PdfRedactionApplyPayload,
     PdfRedactionPreviewPayload,
     PdfRestRawFileResponse,
+    PdfSplitPayload,
     PngPdfRestPayload,
     TiffPdfRestPayload,
     UploadURLs,
 )
 from .types import (
     ALL_PDF_INFO_QUERIES,
     PdfInfoQuery,
+    PdfMergeInput,
+    PdfPageSelection,
     PdfRedactionInstruction,
     PdfRGBColor,
 )
@@ -1591,6 +1595,61 @@ def apply_redactions(
             timeout=timeout,
         )
 
+    def split_pdf(
+        self,
+        file: PdfRestFile | Sequence[PdfRestFile],
+        *,
+        page_groups: Sequence[PdfPageSelection] | PdfPageSelection | None = None,
+        output_prefix: str | None = None,
+        extra_query: Query | None = None,
+        extra_headers: AnyMapping | None = None,
+        extra_body: Body | None = None,
+        timeout: TimeoutTypes | None = None,
+    ) -> PdfRestFileBasedResponse:
+        """Split a PDF into one or more PDF files based on the provided page groups."""
+
+        payload: dict[str, Any] = {"files": file}
+        if page_groups is not None:
+            payload["page_groups"] = page_groups
+        if output_prefix is not None:
+            payload["output_prefix"] = output_prefix
+
+        return self._post_file_operation(
+            endpoint="/split-pdf",
+            payload=payload,
+            payload_model=PdfSplitPayload,
+            extra_query=extra_query,
+            extra_headers=extra_headers,
+            extra_body=extra_body,
+            timeout=timeout,
+        )
+
+    def merge_pdfs(
+        self,
+        sources: Sequence[PdfMergeInput],
+        *,
+        output_prefix: str | None = None,
+        extra_query: Query | None = None,
+        extra_headers: AnyMapping | None = None,
+        extra_body: Body | None = None,
+        timeout: TimeoutTypes | None = None,
+    ) -> PdfRestFileBasedResponse:
+        """Merge multiple PDFs (or page subsets) into a single PDF file."""
+
+        payload: dict[str, Any] = {"sources": sources}
+        if output_prefix is not None:
+            payload["output_prefix"] = output_prefix
+
+        return self._post_file_operation(
+            endpoint="/merged-pdf",
+            payload=payload,
+            payload_model=PdfMergePayload,
+            extra_query=extra_query,
+            extra_headers=extra_headers,
+            extra_body=extra_body,
+            timeout=timeout,
+        )
+
     def convert_to_png(
         self,
         files: PdfRestFile | Sequence[PdfRestFile],
@@ -1963,6 +2022,61 @@ async def _convert_to_graphic(
             timeout=timeout,
         )
 
+    async def split_pdf(
+        self,
+        file: PdfRestFile | Sequence[PdfRestFile],
+        *,
+        page_groups: Sequence[PdfPageSelection] | PdfPageSelection | None = None,
+        output_prefix: str | None = None,
+        extra_query: Query | None = None,
+        extra_headers: AnyMapping | None = None,
+        extra_body: Body | None = None,
+        timeout: TimeoutTypes | None = None,
+    ) -> PdfRestFileBasedResponse:
+        """Asynchronously split a PDF into one or more PDF files."""
+
+        payload: dict[str, Any] = {"files": file}
+        if page_groups is not None:
+            payload["page_groups"] = page_groups
+        if output_prefix is not None:
+            payload["output_prefix"] = output_prefix
+
+        return await self._post_file_operation(
+            endpoint="/split-pdf",
+            payload=payload,
+            payload_model=PdfSplitPayload,
+            extra_query=extra_query,
+            extra_headers=extra_headers,
+            extra_body=extra_body,
+            timeout=timeout,
+        )
+
+    async def merge_pdfs(
+        self,
+        sources: Sequence[PdfMergeInput],
+        *,
+        output_prefix: str | None = None,
+        extra_query: Query | None = None,
+        extra_headers: AnyMapping | None = None,
+        extra_body: Body | None = None,
+        timeout: TimeoutTypes | None = None,
+    ) -> PdfRestFileBasedResponse:
+        """Asynchronously merge multiple PDFs (or page subsets) into a single PDF."""
+
+        payload: dict[str, Any] = {"sources": sources}
+        if output_prefix is not None:
+            payload["output_prefix"] = output_prefix
+
+        return await self._post_file_operation(
+            endpoint="/merged-pdf",
+            payload=payload,
+            payload_model=PdfMergePayload,
+            extra_query=extra_query,
+            extra_headers=extra_headers,
+            extra_body=extra_body,
+            timeout=timeout,
+        )
+
     async def convert_to_png(
         self,
         files: PdfRestFile | Sequence[PdfRestFile],

src/pdfrest/models/_internal.py

@@ -15,6 +15,7 @@
     Field,
     HttpUrl,
     PlainSerializer,
+    model_serializer,
     model_validator,
 )
 
@@ -122,6 +123,12 @@ def join_tuple(value: str | int | tuple[str | int, ...]) -> str:
     return ",".join(join_tuple(v) for v in value)
 
 
+def _serialize_grouped_page_ranges(
+    value: list[list[str | int | tuple[str | int, ...]]],
+) -> list[str]:
+    return [_serialize_page_ranges(v) for v in value]
+
+
 def _serialize_redactions(value: list[_PdfRedactionVariant]) -> str:
     payload = [entry.model_dump(mode="json", exclude_none=True) for entry in value]
     return json.dumps(payload, separators=(",", ":"))
@@ -182,6 +189,17 @@ def _ascending_page_range(
     return range
 
 
+_PageRangeTupleWithLast = Annotated[
+    tuple[PageNumber, PageNumber]
+    | tuple[Literal["last"], PageNumber]
+    | tuple[PageNumber, Literal["last"]],
+    BeforeValidator(_split_page_range_tuple),
+]
+
+SplitMergePageRange = (
+    Literal["even", "odd", "last"] | PageNumber | _PageRangeTupleWithLast
+)
+
 _AscendingPageRangeTuple = Annotated[
     tuple[PageNumber, PageNumber] | tuple[PageNumber, Literal["last"]],
     BeforeValidator(_split_page_range_tuple),
@@ -349,6 +367,121 @@ class PngPdfRestPayload(BasePdfRestGraphicPayload[Literal["rgb", "rgba", "gray"]
     color_model: Annotated[Literal["rgb", "rgba", "gray"], Field(default="rgb")]
 
 
+_DEFAULT_FULL_DOCUMENT_RANGE: list[str] = ["1-last"]
+
+
+class PdfSplitPayload(BaseModel):
+    """Adapt caller options into a pdfRest-ready split request payload."""
+
+    files: Annotated[
+        list[PdfRestFile],
+        Field(
+            min_length=1,
+            max_length=1,
+            validation_alias=AliasChoices("file", "files"),
+            serialization_alias="id",
+        ),
+        BeforeValidator(_ensure_list),
+        AfterValidator(
+            _allowed_mime_types("application/pdf", error_msg="Must be a PDF file")
+        ),
+        PlainSerializer(_serialize_as_first_file_id),
+    ]
+    page_groups: Annotated[
+        list[
+            Annotated[
+                list[SplitMergePageRange],
+                BeforeValidator(_ensure_list),
+                BeforeValidator(_split_comma_string),
+            ]
+        ]
+        | None,
+        Field(
+            default=None,
+            validation_alias=AliasChoices("pages", "page_groups"),
+            serialization_alias="pages",
+            min_length=1,
+        ),
+        BeforeValidator(_ensure_list),
+        BeforeValidator(_int_to_string),
+        PlainSerializer(_serialize_grouped_page_ranges),
+    ]
+    output_prefix: Annotated[
+        str | None,
+        Field(serialization_alias="output", min_length=1, default=None),
+        AfterValidator(_validate_output_prefix),
+    ] = None
+
+
+class _PdfMergeItem(BaseModel):
+    file: Annotated[
+        PdfRestFile,
+        AfterValidator(
+            _allowed_mime_types("application/pdf", error_msg="Must be a PDF file")
+        ),
+    ]
+    pages: Annotated[
+        list[SplitMergePageRange],
+        Field(
+            min_length=1,
+            default_factory=lambda: list(_DEFAULT_FULL_DOCUMENT_RANGE).copy(),
+        ),
+        BeforeValidator(_list_of_strings),
+        BeforeValidator(_ensure_list),
+        PlainSerializer(_serialize_page_ranges),
+    ]
+
+    @model_validator(mode="before")
+    @classmethod
+    def _transform_input(cls, data: Any) -> Any:
+        if isinstance(data, tuple):
+            if len(data) != 2:
+                msg = (
+                    "Tuple merge entries must contain exactly two items: (file, pages)."
+                )
+                raise ValueError(msg)
+            file_candidate, pages = data
+            return {"file": file_candidate, "pages": pages}
+        if isinstance(data, PdfRestFile):
+            return {"file": data}
+        return data
+
+
+class PdfMergePayload(BaseModel):
+    """Adapt caller options into a pdfRest-ready merge request payload."""
+
+    sources: Annotated[
+        list[_PdfMergeItem],
+        Field(
+            min_length=2,
+            validation_alias=AliasChoices("sources", "documents", "files"),
+        ),
+        BeforeValidator(_ensure_list),
+    ]
+    output_prefix: Annotated[
+        str | None,
+        Field(serialization_alias="output", min_length=1, default=None),
+        AfterValidator(_validate_output_prefix),
+    ] = None
+
+    @model_serializer(mode="wrap")
+    def _serialize_pdf_merge_payload(
+        self, handler: Callable[[PdfMergePayload], dict[str, Any]]
+    ) -> dict[str, Any]:
+        # Invoke all the serializers on the payload, which then properly serializes
+        # all the fields.
+        payload = handler(self)
+        # Reorganize the serialized data into the parallel arrays that pdfRest expects
+        payload["type"] = ["id"] * len(self.sources)
+        payload["pages"] = [
+            source.get("pages", _DEFAULT_FULL_DOCUMENT_RANGE[0])
+            for source in payload["sources"]
+        ]
+        payload["id"] = [source["file"]["id"] for source in payload["sources"]]
+        del payload["sources"]
+        return payload
+
+
 class BmpPdfRestPayload(BasePdfRestGraphicPayload[Literal["rgb", "gray"]]):
     """Adapt caller options into a pdfRest-ready BMP request payload."""
 

src/pdfrest/types/__init__.py

@@ -3,6 +3,9 @@
 from .public import (
     ALL_PDF_INFO_QUERIES,
     PdfInfoQuery,
+    PdfMergeInput,
+    PdfMergeSource,
+    PdfPageSelection,
     PdfRedactionInstruction,
     PdfRedactionPreset,
     PdfRedactionType,
@@ -12,6 +15,9 @@
 __all__ = [
     "ALL_PDF_INFO_QUERIES",
     "PdfInfoQuery",
+    "PdfMergeInput",
+    "PdfMergeSource",
+    "PdfPageSelection",
     "PdfRGBColor",
     "PdfRedactionInstruction",
     "PdfRedactionPreset",

src/pdfrest/types/public.py

@@ -2,13 +2,22 @@
 
 from __future__ import annotations
 
-from typing import Literal, cast, get_args
+from collections.abc import Sequence
+from typing import TYPE_CHECKING, Any, Literal, cast, get_args
 
-from typing_extensions import TypedDict
+from typing_extensions import Required, TypedDict
+
+if TYPE_CHECKING:
+    from pdfrest.models import PdfRestFile
+else:  # pragma: no cover - used only for typing at runtime
+    PdfRestFile = Any
 
 __all__ = (
     "ALL_PDF_INFO_QUERIES",
     "PdfInfoQuery",
+    "PdfMergeInput",
+    "PdfMergeSource",
+    "PdfPageSelection",
     "PdfRGBColor",
     "PdfRedactionInstruction",
     "PdfRedactionPreset",
@@ -77,3 +86,13 @@ class PdfRedactionInstruction(TypedDict):
 
 
 PdfRGBColor = tuple[int, int, int]
+
+PdfPageSelection = str | int | Sequence[str | int]
+
+
+class PdfMergeSource(TypedDict, total=False):
+    file: Required[PdfRestFile]
+    pages: PdfPageSelection | None
+
+
+PdfMergeInput = PdfRestFile | PdfMergeSource | tuple[PdfRestFile, PdfPageSelection]