feat(print): add print module

README.md

@@ -24,6 +24,7 @@ The Python SDK offers a clean, type-safe API following Python best practices whi
 - **Secret Resolver**
 - **Telemetry & Observability**
 - **Data Anonymization Service**
+- **Print Service**
 
 ## Requirements and Setup
 
@@ -77,6 +78,7 @@ Each module has comprehensive usage guides:
 - [ObjectStore](src/sap_cloud_sdk/objectstore/user-guide.md)
 - [Secret Resolver](src/sap_cloud_sdk/core/secret_resolver/user-guide.md)
 - [Telemetry](src/sap_cloud_sdk/core/telemetry/user-guide.md)
+- [Print](src/sap_cloud_sdk/print/user-guide.md)
 - [Data Anonymization](src/sap_cloud_sdk/core/data_anonymization/user-guide.md)
 
 ## Support, Feedback, Contributing

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "sap-cloud-sdk"
-version = "0.23.2"
+version = "0.24.1"
 description = "SAP Cloud SDK for Python"
 readme = "README.md"
 license = "Apache-2.0"

src/sap_cloud_sdk/core/telemetry/module.py

@@ -16,6 +16,7 @@ class Module(str, Enum):
     DMS = "dms"
     EXTENSIBILITY = "extensibility"
     OBJECTSTORE = "objectstore"
+    PRINT = "print"
     TELEMETRY = "telemetry"
 
     def __str__(self) -> str:

src/sap_cloud_sdk/core/telemetry/operation.py

@@ -81,6 +81,14 @@ class Operation(str, Enum):
     AICORE_SET_CONFIG = "set_aicore_config"
     AICORE_AUTO_INSTRUMENT = "auto_instrument"
 
+    # Print Operations
+    PRINT_LIST_QUEUES = "list_queues"
+    PRINT_CREATE_QUEUE = "create_queue"
+    PRINT_GET_PROFILES = "get_print_profiles"
+    PRINT_UPLOAD_DOCUMENT = "upload_document"
+    PRINT_CREATE_TASK = "create_print_task"
+    PRINT_CREATE_CLIENT = "create_client"
+
     # DMS Operations
     DMS_ONBOARD_REPOSITORY = "onboard_repository"
     DMS_GET_REPOSITORY = "get_repository"

src/sap_cloud_sdk/print/__init__.py

@@ -0,0 +1,106 @@
+"""SAP Cloud SDK for Python - Print module
+
+The create_client() function loads credentials from mounts/env vars and
+returns a configured PrintClient.
+
+Usage:
+    from sap_cloud_sdk.print import create_client, PrintQueue, PrintContent, PrintTask
+
+    client = create_client()
+
+    # List queues
+    queues = client.list_queues()
+
+    # Upload a document and print it
+    with open("invoice.pdf", "rb") as f:
+        doc_id = client.upload_document(f, filename="invoice.pdf")
+
+    task = PrintTask(
+        item_id=doc_id,
+        qname="my-queue",
+        print_contents=[PrintContent(object_key=doc_id, document_name="invoice.pdf")],
+    )
+    client.create_print_task(task)
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from sap_cloud_sdk.print._models import (
+    PrintContent,
+    PrintProfile,
+    PrintQueue,
+    PrintTask,
+    PrintTaskMetadata,
+)
+from sap_cloud_sdk.print.config import load_from_env_or_mount, PrintConfig
+from sap_cloud_sdk.print._http import PrintHttp, TokenProvider
+from sap_cloud_sdk.print.client import PrintClient
+from sap_cloud_sdk.print.exceptions import (
+    PrintError,
+    ClientCreationError,
+    ConfigError,
+    HttpError,
+    PrintOperationError,
+)
+
+from sap_cloud_sdk.core.telemetry import (
+    Module,
+    Operation,
+    record_error_metric as _record_error_metric,
+)
+
+
+def create_client(
+    *,
+    instance: Optional[str] = None,
+    config: Optional[PrintConfig] = None,
+    _telemetry_source: Optional[Module] = None,
+) -> PrintClient:
+    """Create a PrintClient with secret resolution and OAuth setup.
+
+    Args:
+        instance: Instance name used for secret resolution. Defaults to "default".
+        config: Optional explicit PrintConfig, bypasses secret resolution.
+        _telemetry_source: Internal parameter for telemetry. Not for external use.
+
+    Returns:
+        Configured PrintClient.
+
+    Raises:
+        ClientCreationError: If client creation fails.
+    """
+    try:
+        binding = config or load_from_env_or_mount(instance)
+        tp = TokenProvider(binding)
+        http = PrintHttp(config=binding, token_provider=tp)
+        return PrintClient(http, _telemetry_source=_telemetry_source)
+    except Exception as e:
+        _record_error_metric(
+            Module.PRINT,
+            _telemetry_source,
+            Operation.PRINT_CREATE_CLIENT,
+        )
+        raise ClientCreationError(f"failed to create print client: {e}") from e
+
+
+__all__ = [
+    # Models
+    "PrintQueue",
+    "PrintProfile",
+    "PrintContent",
+    "PrintTask",
+    "PrintTaskMetadata",
+    "PrintConfig",
+    # Factory
+    "create_client",
+    # Client
+    "PrintClient",
+    # Exceptions
+    "PrintError",
+    "ClientCreationError",
+    "ConfigError",
+    "HttpError",
+    "PrintOperationError",
+]

src/sap_cloud_sdk/print/_http.py

@@ -0,0 +1,186 @@
+"""HTTP transport and OAuth utilities for SAP Print Service."""
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+from typing import Any, Dict, Optional, Protocol
+
+import requests
+from requests import Response
+from requests.exceptions import RequestException
+from oauthlib.oauth2 import BackendApplicationClient
+from requests_oauthlib import OAuth2Session
+
+from sap_cloud_sdk.print.config import PrintConfig
+from sap_cloud_sdk.print.exceptions import HttpError
+
+logger = logging.getLogger(__name__)
+
+
+class AbstractTokenProvider(Protocol):
+    """Protocol for token providers — allows injection of mock providers in tests."""
+
+    def get_token(self) -> str: ...
+    def resolve_username(self) -> str: ...
+
+
+class TokenProvider:
+    """Provides OAuth2 access tokens via client credentials flow."""
+
+    def __init__(self, config: PrintConfig) -> None:
+        self._config = config
+        client = BackendApplicationClient(client_id=config.client_id)
+        self._session = OAuth2Session(client=client)
+        self._cached_token: Optional[str] = None
+
+    def get_token(self) -> str:
+        """Return a valid bearer token for the Print Service.
+
+        Returns:
+            A non-empty OAuth2 access token string.
+
+        Raises:
+            HttpError: If the token response is missing an access_token or
+                token acquisition fails.
+        """
+
+        try:
+            token: Dict[str, Any] = self._session.fetch_token(
+                token_url=self._config.token_url,
+                client_id=self._config.client_id,
+                client_secret=self._config.client_secret,
+                include_client_id=True,
+            )
+        except Exception as e:
+            logger.error("failed to acquire token: %s", e)
+            raise HttpError(f"failed to acquire token: {e}") from e
+        access_token = token.get("access_token")
+        if not access_token:
+            raise HttpError("token response missing access_token")
+        self._cached_token = str(access_token)
+        return self._cached_token
+
+    def resolve_username(self) -> str:
+        """Resolve a username from the current access token claims.
+
+        Returns the ``user_name`` JWT claim when present (interactive user
+        flows), otherwise falls back to ``client_id`` (client-credentials /
+        technical-user flows).
+        """
+        token = self._cached_token or self.get_token()
+        try:
+            payload_b64 = token.split(".")[1]
+            # JWT base64 uses URL-safe alphabet without padding
+            padding = 4 - len(payload_b64) % 4
+            if padding != 4:
+                payload_b64 += "=" * padding
+            claims = json.loads(base64.urlsafe_b64decode(payload_b64))
+            return str(
+                claims.get("user_name")
+                or claims.get("client_id")
+                or self._config.client_id
+            )
+        except Exception:
+            logger.debug("could not decode JWT claims, falling back to client_id")
+            return self._config.client_id
+
+
+class PrintHttp:
+    """HTTP client for SAP Print Service."""
+
+    def __init__(
+        self,
+        config: PrintConfig,
+        token_provider: AbstractTokenProvider,
+        session: Optional[requests.Session] = None,
+    ) -> None:
+        self._config = config
+        self._token_provider = token_provider
+        self._session = session or requests.Session()
+        self._base_url = config.url.rstrip("/")
+
+    def get_username(self) -> str:
+        """Resolve the username from the current OAuth token (or fall back to client_id)."""
+        return self._token_provider.resolve_username()
+
+    def _auth_headers(self) -> Dict[str, str]:
+        token = self._token_provider.get_token()
+        return {"Authorization": f"Bearer {token}"}
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        params: Optional[Dict[str, Any]] = None,
+        json: Optional[Any] = None,
+        data: Optional[Any] = None,
+        files: Optional[Any] = None,
+        extra_headers: Optional[Dict[str, str]] = None,
+    ) -> Response:
+        url = f"{self._base_url}/{path.lstrip('/')}"
+        headers = self._auth_headers()
+        if extra_headers:
+            headers.update(extra_headers)
+
+        try:
+            resp = self._session.request(
+                method=method,
+                url=url,
+                headers=headers,
+                params=params,
+                json=json,
+                data=data,
+                files=files,
+            )
+        except RequestException as e:
+            logger.error("request failed [%s %s]: %s", method, url, e)
+            raise HttpError(f"request failed: {e}") from e
+
+        if 200 <= resp.status_code < 300:
+            return resp
+
+        text: str = ""
+        try:
+            text = resp.text
+        except Exception:
+            text = "<failed to read response body>"
+
+        raise HttpError(
+            f"HTTP {resp.status_code} for {method} {url}",
+            status_code=resp.status_code,
+            response_text=text,
+        )
+
+    def get(
+        self,
+        path: str,
+        *,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Response:
+        return self._request("GET", path, params=params, extra_headers=headers)
+
+    def put(
+        self,
+        path: str,
+        *,
+        json: Optional[Any] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Response:
+        return self._request("PUT", path, json=json, extra_headers=headers)
+
+    def post(
+        self,
+        path: str,
+        *,
+        json: Optional[Any] = None,
+        data: Optional[Any] = None,
+        files: Optional[Any] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Response:
+        return self._request(
+            "POST", path, json=json, data=data, files=files, extra_headers=headers
+        )