client.py: Add compress PDF support with comprehensive validation

- Introduced `compress_pdf` method for both synchronous and asynchronous
  `PdfRestClient` instances.
- Added `PdfCompressPayload` to model validation with constraints for files,
  compression levels, profiles, and output settings.
- Included live and unit tests to validate presets, custom profiles, request
  customization, and error handling.
- Added new live test resources and sample compression profile.

Assisted-by: Codex

Author: datalogics-kam

src/pdfrest/client.py

@@ -73,6 +73,7 @@
     BmpPdfRestPayload,
     GifPdfRestPayload,
     JpegPdfRestPayload,
+    PdfCompressPayload,
     PdfFlattenFormsPayload,
     PdfInfoPayload,
     PdfMergePayload,
@@ -2197,6 +2198,39 @@ def flatten_pdf_forms(
             timeout=timeout,
         )
 
+    def compress_pdf(
+        self,
+        file: PdfRestFile | Sequence[PdfRestFile],
+        *,
+        compression_level: Literal["low", "medium", "high", "custom"],
+        profile: PdfRestFile | Sequence[PdfRestFile] | None = None,
+        output: str | None = None,
+        extra_query: Query | None = None,
+        extra_headers: AnyMapping | None = None,
+        extra_body: Body | None = None,
+        timeout: TimeoutTypes | None = None,
+    ) -> PdfRestFileBasedResponse:
+        """Compress a PDF using preset or custom compression profiles."""
+
+        payload: dict[str, Any] = {
+            "files": file,
+            "compression_level": compression_level,
+        }
+        if profile is not None:
+            payload["profile"] = profile
+        if output is not None:
+            payload["output"] = output
+
+        return self._post_file_operation(
+            endpoint="/compressed-pdf",
+            payload=payload,
+            payload_model=PdfCompressPayload,
+            extra_query=extra_query,
+            extra_headers=extra_headers,
+            extra_body=extra_body,
+            timeout=timeout,
+        )
+
     def convert_to_pdfx(
         self,
         file: PdfRestFile | Sequence[PdfRestFile],
@@ -2707,6 +2741,39 @@ async def flatten_pdf_forms(
             timeout=timeout,
         )
 
+    async def compress_pdf(
+        self,
+        file: PdfRestFile | Sequence[PdfRestFile],
+        *,
+        compression_level: Literal["low", "medium", "high", "custom"],
+        profile: PdfRestFile | Sequence[PdfRestFile] | None = None,
+        output: str | None = None,
+        extra_query: Query | None = None,
+        extra_headers: AnyMapping | None = None,
+        extra_body: Body | None = None,
+        timeout: TimeoutTypes | None = None,
+    ) -> PdfRestFileBasedResponse:
+        """Asynchronously compress a PDF."""
+
+        payload: dict[str, Any] = {
+            "files": file,
+            "compression_level": compression_level,
+        }
+        if profile is not None:
+            payload["profile"] = profile
+        if output is not None:
+            payload["output"] = output
+
+        return await self._post_file_operation(
+            endpoint="/compressed-pdf",
+            payload=payload,
+            payload_model=PdfCompressPayload,
+            extra_query=extra_query,
+            extra_headers=extra_headers,
+            extra_body=extra_body,
+            timeout=timeout,
+        )
+
     async def convert_to_pdfx(
         self,
         file: PdfRestFile | Sequence[PdfRestFile],

src/pdfrest/models/_internal.py

@@ -549,6 +549,64 @@ class PdfFlattenFormsPayload(BaseModel):
     ] = None
 
 
+class PdfCompressPayload(BaseModel):
+    """Adapt caller options into a pdfRest-ready compress 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),
+    ]
+    compression_level: Annotated[
+        Literal["low", "medium", "high", "custom"],
+        Field(serialization_alias="compression_level"),
+    ]
+    profile: Annotated[
+        list[PdfRestFile] | None,
+        Field(
+            default=None,
+            min_length=1,
+            max_length=1,
+            validation_alias=AliasChoices("profile", "profiles"),
+            serialization_alias="profile_id",
+        ),
+        BeforeValidator(_ensure_list),
+        BeforeValidator(
+            _allowed_mime_types(
+                "application/json",
+                "text/json",
+                error_msg="Profile must be a JSON file",
+            )
+        ),
+        PlainSerializer(_serialize_as_first_file_id),
+    ] = None
+    output: Annotated[
+        str | None,
+        Field(serialization_alias="output", min_length=1, default=None),
+        AfterValidator(_validate_output_prefix),
+    ] = None
+
+    @model_validator(mode="after")
+    def _validate_profile_dependency(self) -> PdfCompressPayload:
+        if self.compression_level == "custom":
+            if not self.profile:
+                msg = "compression_level 'custom' requires a profile to be provided."
+                raise ValueError(msg)
+        elif self.profile:
+            msg = "A profile can only be provided when compression_level is 'custom'."
+            raise ValueError(msg)
+        return self
+
+
 class BmpPdfRestPayload(BasePdfRestGraphicPayload[Literal["rgb", "gray"]]):
     """Adapt caller options into a pdfRest-ready BMP request payload."""
 

tests/live/test_live_compress_pdf.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+from typing import Literal
+
+import pytest
+
+from pdfrest import AsyncPdfRestClient, PdfRestApiError, PdfRestClient
+from pdfrest.models import PdfRestFile
+
+from ..resources import get_test_resource_path
+
+
+@pytest.fixture(scope="module")
+def uploaded_pdf_for_compression(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+) -> PdfRestFile:
+    resource = get_test_resource_path("report.pdf")
+    with PdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        return client.files.create_from_paths([resource])[0]
+
+
+@pytest.fixture(scope="module")
+def uploaded_compression_profile(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+) -> PdfRestFile:
+    resource = get_test_resource_path("compression_profile.json")
+    with PdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        return client.files.create_from_paths([resource])[0]
+
+
+@pytest.mark.parametrize(
+    "compression_level",
+    [
+        pytest.param("low", id="low"),
+        pytest.param("medium", id="medium"),
+        pytest.param("high", id="high"),
+    ],
+)
+def test_live_compress_pdf_presets(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_pdf_for_compression: PdfRestFile,
+    compression_level: Literal["low", "medium", "high"],
+) -> None:
+    with PdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        response = client.compress_pdf(
+            uploaded_pdf_for_compression,
+            compression_level=compression_level,
+        )
+
+    assert response.output_files
+    output_file = response.output_file
+    assert output_file.type == "application/pdf"
+    assert str(response.input_id) == str(uploaded_pdf_for_compression.id)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "compression_level",
+    [
+        pytest.param("low", id="low"),
+        pytest.param("medium", id="medium"),
+        pytest.param("high", id="high"),
+    ],
+)
+async def test_live_async_compress_pdf_presets(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_pdf_for_compression: PdfRestFile,
+    compression_level: Literal["low", "medium", "high"],
+) -> None:
+    async with AsyncPdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        response = await client.compress_pdf(
+            uploaded_pdf_for_compression,
+            compression_level=compression_level,
+        )
+
+    assert response.output_files
+    output_file = response.output_file
+    assert output_file.type == "application/pdf"
+    assert str(response.input_id) == str(uploaded_pdf_for_compression.id)
+
+
+def test_live_compress_pdf_custom(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_pdf_for_compression: PdfRestFile,
+    uploaded_compression_profile: PdfRestFile,
+) -> None:
+    with PdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        response = client.compress_pdf(
+            uploaded_pdf_for_compression,
+            compression_level="custom",
+            profile=uploaded_compression_profile,
+            output="compressed-custom",
+        )
+
+    assert (
+        response.warning
+        == "The document could not be made smaller. No output was produced."
+    )
+    assert len(response.input_ids) == 2
+    assert uploaded_pdf_for_compression.id in response.input_ids
+    assert uploaded_compression_profile.id in response.input_ids
+
+
+@pytest.mark.asyncio
+async def test_live_async_compress_pdf_custom(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_pdf_for_compression: PdfRestFile,
+    uploaded_compression_profile: PdfRestFile,
+) -> None:
+    async with AsyncPdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        response = await client.compress_pdf(
+            uploaded_pdf_for_compression,
+            compression_level="custom",
+            profile=uploaded_compression_profile,
+            output="compressed-custom",
+        )
+
+    assert (
+        response.warning
+        == "The document could not be made smaller. No output was produced."
+    )
+    assert len(response.input_ids) == 2
+    assert uploaded_pdf_for_compression.id in response.input_ids
+    assert uploaded_compression_profile.id in response.input_ids
+
+
+def test_live_compress_pdf_invalid_level(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_pdf_for_compression: PdfRestFile,
+) -> None:
+    with (
+        PdfRestClient(
+            api_key=pdfrest_api_key,
+            base_url=pdfrest_live_base_url,
+        ) as client,
+        pytest.raises(PdfRestApiError),
+    ):
+        client.compress_pdf(
+            uploaded_pdf_for_compression,
+            compression_level="low",
+            extra_body={"compression_level": "extreme"},
+        )
+
+
+@pytest.mark.asyncio
+async def test_live_async_compress_pdf_invalid_level(
+    pdfrest_api_key: str,
+    pdfrest_live_base_url: str,
+    uploaded_pdf_for_compression: PdfRestFile,
+) -> None:
+    async with AsyncPdfRestClient(
+        api_key=pdfrest_api_key,
+        base_url=pdfrest_live_base_url,
+    ) as client:
+        with pytest.raises(PdfRestApiError):
+            await client.compress_pdf(
+                uploaded_pdf_for_compression,
+                compression_level="low",
+                extra_body={"compression_level": "extreme"},
+            )