Merge pull request #45 from artpods56/chore/update-openapi-spec

chore: update OpenAPI spec to KSeF API 2.6.0 and regenerate models

Author: artpods56

openapi.json

@@ -1 +1 @@
-version = "2.5.0"
+version = "2.6.0"

src/ksef2/__openapi_version__.py

@@ -8,6 +8,7 @@
 )
 from ksef2.core.async_protocols import AsyncMiddleware
 from ksef2.core.crypto import encrypt_symmetric_key, generate_session_key
+from ksef2.domain.models.compression import CompressionType, normalize_compression_type
 from ksef2.domain.models.invoices import (
     ExportHandle,
     ExportInvoicesPayload,
@@ -94,6 +95,7 @@ async def schedule_export(
         encryption_certificate: str,
         encryption_public_key_id: str | None = None,
         only_metadata: bool = False,
+        compression_type: CompressionType | str | None = None,
     ) -> ExportHandle:
         aes_key, iv = generate_session_key()
         encrypted_key = encrypt_symmetric_key(
@@ -107,6 +109,11 @@ async def schedule_export(
                 initialization_vector=base64.b64encode(iv).decode(),
                 public_key_id=encryption_public_key_id,
                 only_metadata=only_metadata,
+                compression_type=(
+                    normalize_compression_type(compression_type)
+                    if compression_type is not None
+                    else None
+                ),
             )
         )
         spec_resp = await self._endpoints.export(body=spec_request)

src/ksef2/clients/async_invoices.py

@@ -8,6 +8,7 @@
 )
 from ksef2.core.crypto import encrypt_symmetric_key, generate_session_key
 from ksef2.core.protocols import Middleware
+from ksef2.domain.models.compression import CompressionType, normalize_compression_type
 from ksef2.domain.models.invoices import (
     ExportHandle,
     ExportInvoicesPayload,
@@ -97,6 +98,7 @@ def schedule_export(
         encryption_certificate: str,
         encryption_public_key_id: str | None = None,
         only_metadata: bool = False,
+        compression_type: CompressionType | str | None = None,
     ) -> ExportHandle:
         """Schedule an export and return the handle needed to decrypt it later."""
         aes_key, iv = generate_session_key()
@@ -111,6 +113,11 @@ def schedule_export(
                 initialization_vector=base64.b64encode(iv).decode(),
                 public_key_id=encryption_public_key_id,
                 only_metadata=only_metadata,
+                compression_type=(
+                    normalize_compression_type(compression_type)
+                    if compression_type is not None
+                    else None
+                ),
             )
         )
         spec_resp = self._endpoints.export(body=spec_request)

src/ksef2/clients/invoices.py

@@ -1,4 +1,5 @@
 from ksef2.domain.models.base import KSeFBaseModel, KSeFBaseParams
+from ksef2.domain.models.compression import CompressionType, CompressionTypeEnum
 from ksef2.domain.models.session import (
     BaseSessionState,
     FormSchema,
@@ -177,6 +178,8 @@
     # base
     "KSeFBaseModel",
     "KSeFBaseParams",
+    "CompressionType",
+    "CompressionTypeEnum",
     # session
     "BaseSessionState",
     "FormSchema",

src/ksef2/domain/models/__init__.py

@@ -5,7 +5,13 @@
 import base64
 from typing import Self
 
+from pydantic import field_validator
+
 from ksef2.domain.models.base import KSeFBaseModel
+from ksef2.domain.models.compression import (
+    CompressionType,
+    normalize_compression_type,
+)
 from ksef2.domain.models.session import BaseSessionState, FormSchema
 
 
@@ -45,9 +51,21 @@ class BatchFileInfo(KSeFBaseModel):
     file_hash: str
     """SHA-256 hash of the batch file, Base64 encoded."""
 
+    compression_type: CompressionType | None = None
+    """Compression used for the batch file. Defaults to KSeF's ZIP behavior."""
+
     parts: list[BatchFilePart]
     """List of file parts. Max 50 parts, each max 100MB before encryption."""
 
+    @field_validator("compression_type", mode="before")
+    @classmethod
+    def _normalize_compression(cls, value: object) -> object:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return normalize_compression_type(value)
+        return value
+
 
 class BatchEncryptionData(KSeFBaseModel):
     """Encryption material used for the prepared batch payload."""

src/ksef2/domain/models/batch.py

@@ -0,0 +1,44 @@
+from enum import StrEnum
+from typing import Literal
+
+type CompressionType = Literal["zip", "tar_gz"]
+type CompressionTypeSpecValue = Literal["Zip", "TarGz"]
+
+
+class CompressionTypeEnum(StrEnum):
+    ZIP = "Zip"
+    TAR_GZ = "TarGz"
+
+
+_COMPRESSION_TYPE_TO_SPEC: dict[CompressionType, CompressionTypeSpecValue] = {
+    "zip": "Zip",
+    "tar_gz": "TarGz",
+}
+_COMPRESSION_TYPE_FROM_SPEC: dict[CompressionTypeSpecValue, CompressionType] = {
+    value: key for key, value in _COMPRESSION_TYPE_TO_SPEC.items()
+}
+
+
+def normalize_compression_type(
+    value: CompressionType | CompressionTypeEnum | str,
+) -> CompressionType:
+    if isinstance(value, CompressionTypeEnum):
+        return _COMPRESSION_TYPE_FROM_SPEC[value.value]
+
+    lowered_value = value.strip().lower()
+    if lowered_value in _COMPRESSION_TYPE_TO_SPEC:
+        return lowered_value  # pyright: ignore[reportReturnType]
+
+    if value in _COMPRESSION_TYPE_FROM_SPEC:
+        return _COMPRESSION_TYPE_FROM_SPEC[value]  # pyright: ignore[index]
+
+    raise ValueError(
+        f"Invalid compression type: {value}. Valid compression types are: "
+        f"{', '.join(_COMPRESSION_TYPE_TO_SPEC)}"
+    )
+
+
+def compression_type_to_spec(
+    value: CompressionType | CompressionTypeEnum | str,
+) -> CompressionTypeSpecValue:
+    return _COMPRESSION_TYPE_TO_SPEC[normalize_compression_type(value)]

src/ksef2/domain/models/compression.py

@@ -6,6 +6,10 @@
 from pydantic import field_validator
 
 from ksef2.domain.models.base import KSeFBaseModel
+from ksef2.domain.models.compression import (
+    CompressionType,
+    normalize_compression_type,
+)
 from ksef2.domain.models.session import FormSchema
 from ksef2.domain.types import CurrencyCodes, KsefInvoiceTypes
 
@@ -330,6 +334,16 @@ class ExportInvoicesPayload(KSeFBaseModel):
     initialization_vector: str
     public_key_id: str | None = None
     only_metadata: bool = False
+    compression_type: CompressionType | None = None
+
+    @field_validator("compression_type", mode="before")
+    @classmethod
+    def _normalize_compression_type(cls, value: object) -> object:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return normalize_compression_type(value)
+        return value
 
 
 class SendInvoicePayload(KSeFBaseModel):

src/ksef2/domain/models/invoices.py

@@ -7,6 +7,7 @@
 from pydantic import BaseModel
 
 from ksef2.core.crypto import sha256_b64
+from ksef2.domain.models.compression import compression_type_to_spec
 from ksef2.domain.models import invoices
 from ksef2.infra.mappers.helpers import to_aware_datetime
 from ksef2.infra.schema.api import spec
@@ -248,6 +249,11 @@ def _(request: invoices.ExportInvoicesPayload) -> spec.InvoiceExportRequest:
         ),
         onlyMetadata=request.only_metadata,
         filters=to_spec(request.filter),
+        compressionType=(
+            spec.CompressionType(compression_type_to_spec(request.compression_type))
+            if request.compression_type is not None
+            else None
+        ),
     )
 
 

src/ksef2/infra/mappers/invoices/requests.py

@@ -4,6 +4,7 @@
 
 from pydantic import BaseModel
 
+from ksef2.domain.models.compression import compression_type_to_spec
 from ksef2.domain.models.batch import (
     BatchFileInfo,
     BatchFilePart,
@@ -80,6 +81,11 @@ def _(request: BatchFileInfo) -> spec.BatchFileInfo:
     return spec.BatchFileInfo(
         fileSize=request.file_size,
         fileHash=request.file_hash,
+        compressionType=(
+            spec.CompressionType(compression_type_to_spec(request.compression_type))
+            if request.compression_type is not None
+            else None
+        ),
         fileParts=[to_spec(part) for part in request.parts],
     )