ksef2.clients · ksef2 SDK docs

Public sync and async client entry points.

Classes

`AsyncAuthClient`

Import: from ksef2.clients import AsyncAuthClient

class AsyncAuthClient

python

AsyncAuthClient(
    transport: AsyncMiddleware,
    certificate_store: CertificateStoreProtocol,
    environment: Environment = Environment.PRODUCTION,
    transfer_transport: AsyncMiddleware | None = None,
    auto_refresh_tokens: bool = True,
)

python

Async entry point for creating authenticated SDK clients.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncAuthClient.resume`

def resume(state: AuthenticationResumeState | str) -> AsyncAuthenticatedClient

python

Rehydrate an authenticated client from saved authentication state.

The client refreshes its access token automatically unless TransportConfig.auto_refresh_tokens is False, so a state whose access token has expired still works while its refresh token is valid.

Parameters

state (AuthenticationResumeState | str) — State previously exported from an authenticated client, or its JSON string.

Returns

AsyncAuthenticatedClient — An authenticated client bound to the saved tokens.

`AsyncAuthClient.with_token`

async def with_token(
    *,
    ksef_token: str,
    nip: str,
    context_type: ContextIdentifierType = 'nip',
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> AsyncAuthenticatedClient

python

Authenticate with a KSeF token and return an authenticated client.

Parameters

ksef_token (str) — Token generated in KSeF for the target context.
nip (str) — Context identifier value used for authentication.
context_type (ContextIdentifierType) — Type of identifier represented by nip.
timeout (float) — Maximum number of seconds to wait for authentication.
poll_interval (float) — Delay in seconds between authentication status checks.

Returns

AsyncAuthenticatedClient — An authenticated client with redeemed access and refresh tokens.

Raises

NoCertificateAvailableError — If no valid token-encryption certificate is available.
KSeFEncryptionError — If token encryption fails.
KSeFAuthError — If authentication fails.
KSeFAuthPollingTimeoutError — If polling exceeds timeout.

Example

client = AsyncClient(Environment.TEST)
auth = await client.authentication.with_token(
    ksef_token="your-ksef-token",
    nip="5261040828",
)

python

`AsyncAuthClient.with_xades`

async def with_xades(
    *,
    nip: str,
    cert: Certificate,
    private_key: XAdESPrivateKey,
    verify_chain: bool = False,
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> AsyncAuthenticatedClient

python

Authenticate with an XAdES-signed challenge response.

Parameters

nip (str) — Taxpayer identifier embedded in the authentication request.
cert (Certificate) — Signing certificate accepted by the target environment.
private_key (XAdESPrivateKey) — RSA or EC private key used to sign the XAdES payload.
verify_chain (bool) — Whether KSeF should verify the certificate chain.
timeout (float) — Maximum number of seconds to wait for authentication.
poll_interval (float) — Delay in seconds between authentication status checks.

Returns

AsyncAuthenticatedClient — An authenticated client with redeemed access and refresh tokens.

Raises

KSeFAuthError — If authentication fails.
KSeFAuthPollingTimeoutError — If polling exceeds timeout.

Example

from ksef2.xades import load_certificate_from_pem, load_private_key_from_pem

client = AsyncClient(Environment.DEMO)
auth = await client.authentication.with_xades(
    nip="5261040828",
    cert=load_certificate_from_pem("company.pem"),
    private_key=load_private_key_from_pem("company.key"),
)

python

`AsyncAuthClient.with_test_certificate`

async def with_test_certificate(
    *,
    nip: str,
    verify_chain: bool = False,
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> AsyncAuthenticatedClient

python

Authenticate in the TEST environment using an SDK-generated certificate.

Parameters

nip (str) — NIP to authenticate as; the generated certificate is issued for it.
verify_chain (bool) — Whether KSeF should verify the certificate chain.
timeout (float) — Maximum number of seconds to wait for authentication.
poll_interval (float) — Delay in seconds between authentication status checks.

Returns

AsyncAuthenticatedClient — An authenticated client for the NIP's context.

Raises

KSeFUnsupportedEnvironmentError — If the client is not configured for TEST.
KSeFAuthError — If authentication fails.
KSeFAuthPollingTimeoutError — If polling exceeds timeout.

Example

client = AsyncClient(Environment.TEST)
auth = await client.authentication.with_test_certificate(nip="5261040828")

python

`AsyncAuthClient.with_profile`

async def with_profile(
    name: str | None = None,
    *,
    config_path: str | Path | None = None,
    timeout: float | None = None,
    poll_interval: float | None = None,
    verify_chain: bool = False,
) -> AsyncAuthenticatedClient

python

Authenticate with a local ksef2-cli profile.

Profile selection follows the CLI order: explicit name, KSEF2_PROFILE, then active_profile from the local CLI config. The profile environment must match the root client environment.

Parameters

name (str | None) — Profile name; when None, KSEF2_PROFILE and then the CLI's active profile are used.
config_path (str | Path | None) — Path of the ksef2-cli config file; the default location is used when None.
timeout (float | None) — Maximum number of seconds to wait for authentication; defaults to the profile's polling settings, else 60.
poll_interval (float | None) — Delay in seconds between authentication status checks; defaults to the profile's value, else 1.
verify_chain (bool) — Whether KSeF should verify the certificate chain, for certificate-based profiles.

Returns

AsyncAuthenticatedClient — An authenticated client for the profile's context.

Raises

KSeFValidationError — If the profile's environment differs from the client's, or the profile is missing a required secret.
KSeFAuthError — If authentication fails.
KSeFAuthPollingTimeoutError — If polling exceeds timeout.

Example

client = AsyncClient(Environment.TEST)
auth = await client.authentication.with_profile("test-company")

python

`AsyncAuthClient.refresh`

async def refresh(*, refresh_token: str) -> RefreshedToken

python

Exchange a refresh token for a new access token.

Parameters

refresh_token (str) — Refresh token obtained during authentication.

Returns

RefreshedToken — A new access token with its expiry time.

`AsyncAuthenticatedClient`

Import: from ksef2.clients import AsyncAuthenticatedClient

class AsyncAuthenticatedClient

python

AsyncAuthenticatedClient(
    transport: AsyncMiddleware,
    auth_tokens: AuthTokens,
    certificate_store: CertificateStoreProtocol,
    environment: Environment = Environment.PRODUCTION,
    transfer_transport: AsyncMiddleware | None = None,
    refresh_access_token: AsyncRefresh | None = None,
)

python

Authenticated async entry point for KSeF operations.

Catch KSeFException for SDK-classified failures raised by authenticated branches, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncAuthenticatedClient.auth_tokens` (property)

auth_tokens: AuthTokens

python

Return the authenticated token pair used by this client branch.

Reflects automatic refreshes: after the access token is renewed, this holds the new access token alongside the original refresh token.

Returns

AuthTokens — The current authenticated token pair of this client branch.

`AsyncAuthenticatedClient.access_token` (property)

access_token: str

python

Return the bearer access token string used for authenticated calls.

Returns

str — The bearer access token string used for authenticated calls.

`AsyncAuthenticatedClient.refresh_token` (property)

refresh_token: str

python

Return the refresh token string paired with the access token.

Returns

str — The refresh token string paired with the access token.

`AsyncAuthenticatedClient.resume_state`

def resume_state() -> AuthenticationResumeState

python

Return the authentication state needed to rehydrate this branch later.

The state holds the current tokens, so it includes any access token obtained through automatic refresh.

Returns

AuthenticationResumeState — The authentication state needed to rehydrate this branch later.

`AsyncAuthenticatedClient.get_encryption_key`

async def get_encryption_key() -> tuple[bytes, bytes, bytes]

python

Generate a session AES key, IV, and encrypted symmetric key payload.

Returns

tuple[bytes, bytes, bytes] — A tuple of the raw AES key, the initialization vector and the AES key encrypted with the KSeF public key.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If symmetric-key encryption fails.

`AsyncAuthenticatedClient.online_session`

def online_session(
    *,
    form_code: FormSchema | None = None,
    state: OnlineSessionResumeState | str | None = None,
) -> _AwaitableSession[AsyncOnlineSessionClient]

python

Open a new online invoice session, or resume one from saved state.

Pass exactly one of form_code (open a new session) or state (resume). When resuming, the form code, keys and expiry all come from the state. Leaving the with block closes the session in both cases; closing an already-closed session is a no-op.

Parameters

form_code (FormSchema | None) — Invoice schema the new session accepts, for example FormSchema.FA3.
state (OnlineSessionResumeState | str | None) — State from session.resume_state(), or its JSON string, to resume an open session.

Returns

_AwaitableSession[AsyncOnlineSessionClient] — A session client that can be used as a context manager and closes the session on exit.

Raises

KSeFArgumentError — If neither or both of form_code and state are given.
NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If symmetric-key encryption fails.
KSeFValidationError — If state is not valid online session state.

Example

from ksef2.models import FormSchema

async with auth.online_session(form_code=FormSchema.FA3) as session:
    submission = await session.send_invoice(xml_bytes)
    saved = session.resume_state().to_json()

async with auth.online_session(state=saved) as session:
    result = await session.submission(reference_number).wait()

python

`AsyncAuthenticatedClient.resume_online_session`

def resume_online_session(state: OnlineSessionResumeState) -> AsyncOnlineSessionClient

python

Deprecated: rebind an existing serialized online session state to this client.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use online_session(state=...) instead.

Parameters

state (OnlineSessionResumeState) — State previously exported from an online session client.

Returns

AsyncOnlineSessionClient — An online session client bound to the saved state.

`AsyncAuthenticatedClient.batch_session`

def batch_session(
    *,
    prepared_batch: PreparedBatch | None = None,
    batch_file: BatchFileInfo | None = None,
    form_code: FormSchema | None = None,
    offline_mode: bool | None = None,
    state: BatchSessionResumeState | str | None = None,
) -> _AwaitableSession[AsyncBatchSessionClient]

python

Open a batch session for upload work, or resume one from saved state.

Pass exactly one of prepared_batch, batch_file (open a new session) or state (resume). When resuming, everything comes from the state, including the form code, keys and part upload requests. Leaving the with block closes the session in every case; closing an already-closed session is a no-op.

Parameters

prepared_batch (PreparedBatch | None) — Prepared batch payload created by auth.batch.prepare().
batch_file (BatchFileInfo | None) — Declared ZIP package metadata and encrypted part metadata.
form_code (FormSchema | None) — Invoice schema declared for the batch session when batch_file is provided directly; defaults to FormSchema.FA3.
offline_mode (bool | None) — Whether to declare offline invoicing mode for the batch when batch_file is provided directly; defaults to False.
state (BatchSessionResumeState | str | None) — State from session.resume_state(), or its JSON string, to resume a batch session.

Returns

_AwaitableSession[AsyncBatchSessionClient] — A bound batch session client exposing presigned upload instructions.

Raises

KSeFArgumentError — If not exactly one of prepared_batch, batch_file and state is given, or form_code or offline_mode is combined with prepared_batch or state.
NoCertificateAvailableError — If certificate-backed encryption material is needed but no valid certificate is available.
KSeFEncryptionError — If symmetric-key encryption fails.
KSeFValidationError — If state is not valid batch session state.

Example

from ksef2.models import BatchInvoice

prepared = await auth.batch.prepare(
    [BatchInvoice(file_name="invoice-1.xml", content=xml_bytes)],
)
async with auth.batch_session(prepared_batch=prepared) as session:
    await session.upload_parts()
    saved = session.resume_state().to_json()

async with auth.batch_session(state=saved) as session:
    final = await session.wait()

python

`AsyncAuthenticatedClient.open_batch_session`

def open_batch_session(
    *,
    batch_file: BatchFileInfo,
    aes_key: bytes,
    iv: bytes,
    encrypted_key: bytes,
    public_key_id: str | None = None,
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    prepared_batch: PreparedBatch | None = None,
) -> _AwaitableSession[AsyncBatchSessionClient]

python

Deprecated: open a batch session using caller-prepared encryption metadata.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use auth.batch_session() for the standard flow, or raw for caller-supplied encryption material.

Parameters

batch_file (BatchFileInfo) — Declared ZIP package metadata and encrypted part metadata.
aes_key (bytes) — Raw symmetric key used for part encryption.
iv (bytes) — Initialization vector paired with aes_key.
encrypted_key (bytes) — RSA-encrypted symmetric key sent to KSeF.
public_key_id (str | None) — Identifier of the KSeF public key used for encryption.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
prepared_batch (PreparedBatch | None) — Optional prepared batch payload to attach to the returned session so session.upload_parts() can operate without extra args.

Returns

_AwaitableSession[AsyncBatchSessionClient] — A bound batch session client exposing presigned upload instructions.

Raises

KSeFValidationError — If the batch session request is invalid.

`AsyncAuthenticatedClient.resume_batch_session`

def resume_batch_session(state: BatchSessionResumeState) -> AsyncBatchSessionClient

python

Deprecated: rebind an existing serialized batch session state to this client.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use batch_session(state=...) instead.

Parameters

state (BatchSessionResumeState) — State previously exported from a batch session client.

Returns

AsyncBatchSessionClient — A batch session client bound to the saved state.

`AsyncAuthenticatedClient.invoices` (property)

invoices: AsyncInvoicesService

python

Return the invoices service with encryption support configured.

Returns

AsyncInvoicesService — The invoices service with encryption support configured.

`AsyncAuthenticatedClient.batch` (property)

batch: AsyncBatchService

python

Return the high-level batch upload workflow service.

The service orchestrates package preparation, session opening, presigned part uploads, session closing, and status polling.

Returns

AsyncBatchService — The batch upload service.

`AsyncAuthenticatedClient.limits` (property)

limits: AsyncLimitsClient

python

Return the authenticated rate-limit branch.

Returns

AsyncLimitsClient — The authenticated rate-limit branch.

`AsyncAuthenticatedClient.collective_identifiers` (property)

collective_identifiers: AsyncCollectiveIdentifiersClient

python

Return the collective invoice identifier branch.

Returns

AsyncCollectiveIdentifiersClient — The collective invoice identifier branch.

`AsyncAuthenticatedClient.tokens` (property)

tokens: AsyncTokensClient

python

Return the authenticated token lifecycle branch.

Returns

AsyncTokensClient — The authenticated token lifecycle branch.

`AsyncAuthenticatedClient.certificates` (property)

certificates: AsyncCertificatesClient

python

Return the authenticated certificate enrollment branch.

Returns

AsyncCertificatesClient — The authenticated certificate enrollment branch.

`AsyncAuthenticatedClient.sessions` (property)

sessions: AsyncSessionManagementClient

python

Return the authenticated session-management branch.

Returns

AsyncSessionManagementClient — The authenticated session-management branch.

`AsyncAuthenticatedClient.invoice_sessions` (property)

invoice_sessions: AsyncInvoiceSessionsClient

python

Return the authenticated invoice-session history branch.

Returns

AsyncInvoiceSessionsClient — The authenticated invoice-session history branch.

`AsyncAuthenticatedClient.permissions` (property)

permissions: AsyncPermissionsClient

python

Return the authenticated permissions branch.

Returns

AsyncPermissionsClient — The authenticated permissions branch.

`AsyncAuthenticatedClient.testdata` (property)

testdata: AsyncTestDataClient

python

Return authenticated TEST-only data mutation helpers.

Returns

AsyncTestDataClient — Authenticated TEST-only data mutation helpers.

`AsyncAuthenticatedClient.raw` (property)

raw: AsyncRawAuthenticatedClient

python

Return raw authenticated endpoints for advanced async integrations.

Returns

AsyncRawAuthenticatedClient — Raw authenticated endpoints for advanced async integrations.

`AsyncBatchService`

Import: from ksef2.clients import AsyncBatchService

class AsyncBatchService

python

AsyncBatchService(
    *,
    authed_transport: AsyncMiddleware,
    upload_transport: AsyncMiddleware,
    get_encryption_key: Callable[[], Awaitable[SessionEncryptionMaterial]],
    open_batch_session: AsyncBatchSessionOpener,
)

python

Async high-level workflow for preparing and sending invoice batches.

Catch KSeFException for SDK-classified failures raised by this service, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncBatchService.prepare`

async def prepare(
    invoices: Iterable[bytes | str | Path | BatchInvoice],
    *,
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> PreparedBatch

python

Build a ZIP package, split it, and encrypt each upload part.

Parameters

invoices (Iterable[bytes | str | Path | BatchInvoice]) — Invoices to include. Each item is invoice XML as bytes or str (named invoice-<position>.xml), a Path to an XML file (keeps its file name) or a BatchInvoice with an explicit file name.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
max_part_size (int) — Maximum size of each ZIP part before encryption.

Returns

PreparedBatch — A prepared batch with encrypted part payloads and the metadata required
PreparedBatch — to open a batch session.

Raises

FileNotFoundError — If an invoice XML path does not exist.
NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If the invoice list or part size is invalid.

Example

batch = await auth.batch.prepare([xml_1, xml_2])
session = await auth.batch.submit(batch)

python

`AsyncBatchService.submit`

async def submit(
    batch: PreparedBatch | Iterable[bytes | str | Path | BatchInvoice],
    *,
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> AsyncBatchSessionClient

python

Open a batch session, upload the parts and close the session.

Parameters

batch (PreparedBatch | Iterable[bytes | str | Path | BatchInvoice]) — A prepared batch from prepare(), or the invoices themselves, which are prepared first (same item types as prepare()).
form_code (FormSchema) — Invoice schema declared for the batch session. Ignored for a prepared batch, which carries its own.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch. Ignored for a prepared batch.
max_part_size (int) — Maximum size of each ZIP part before encryption. Ignored for a prepared batch.

Returns

AsyncBatchSessionClient — The closed batch session client. Call its wait() for the terminal status, list_failed_invoices() for rejected invoices and download_upo() for the UPO pages.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If preparation, session opening, or upload validation fails.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

Example

session = await auth.batch.submit([xml_1, xml_2])
final = await session.wait()
upos = await session.download_upo()

python

`AsyncBatchService.prepare_batch`

def prepare_batch(
    *,
    invoices: Iterable[BatchInvoice],
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> Coroutine[None, None, PreparedBatch]

python

Deprecated: build a ZIP package, split it, and encrypt each upload part.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use prepare() instead.

Parameters

invoices (Iterable[BatchInvoice]) — Invoice XML payloads to include in the batch package.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
max_part_size (int) — Maximum size of each ZIP part before encryption.

Returns

Coroutine[None, None, PreparedBatch] — A prepared batch with encrypted part payloads and the metadata required
Coroutine[None, None, PreparedBatch] — to open a batch session.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If the invoice list or part size is invalid.

`AsyncBatchService.prepare_batch_from_paths`

def prepare_batch_from_paths(
    *,
    invoice_paths: Iterable[Path | str],
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> Coroutine[None, None, PreparedBatch]

python

Deprecated: load invoice XML files from disk and prepare a batch package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use prepare() with Path items instead.

Parameters

invoice_paths (Iterable[Path | str]) — Paths to XML files that should be added to the batch.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
max_part_size (int) — Maximum size of each ZIP part before encryption.

Returns

Coroutine[None, None, PreparedBatch] — A prepared batch with encrypted parts ready to be uploaded.

Raises

FileNotFoundError — If an invoice XML path does not exist.
NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If the invoice list or part size is invalid.

`AsyncBatchService.open_session`

def open_session(
    *,
    prepared_batch: PreparedBatch,
) -> _AwaitableSession[AsyncBatchSessionClient]

python

Deprecated: open a batch session for an already prepared package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use auth.batch_session(prepared_batch=...) instead.

Parameters

prepared_batch (PreparedBatch) — Prepared batch payload returned by prepare().

Returns

_AwaitableSession[AsyncBatchSessionClient] — A session client exposing the upload instructions returned by KSeF.

Raises

KSeFValidationError — If the prepared batch cannot be opened.

`AsyncBatchService.upload_parts`

async def upload_parts(
    *,
    session: AsyncBatchSessionClient,
    prepared_batch: PreparedBatch,
) -> None

python

Upload all prepared batch parts using the session's presigned URLs.

Parameters

session (AsyncBatchSessionClient) — Open batch session that already contains upload instructions.
prepared_batch (PreparedBatch) — Prepared batch whose part ordinals match the session.

Raises

KSeFClientClosedError — If the session client is closed.
KSeFValidationError — If prepared part ordinals do not match session upload instructions.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

`AsyncBatchService.submit_prepared_batch`

def submit_prepared_batch(
    *,
    prepared_batch: PreparedBatch,
) -> Coroutine[None, None, BatchSessionResumeState]

python

Deprecated: open, upload, and close a batch session for a prepared package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use submit() instead.

Parameters

prepared_batch (PreparedBatch) — Prepared batch payload returned by prepare().

Returns

Coroutine[None, None, BatchSessionResumeState] — Serializable state of the submitted batch session.

Raises

KSeFClientClosedError — If the session client closes before upload.
KSeFValidationError — If the prepared batch cannot be opened or uploaded.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

`AsyncBatchService.submit_batch`

def submit_batch(
    *,
    invoices: Iterable[BatchInvoice],
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> Coroutine[None, None, BatchSessionResumeState]

python

Deprecated: prepare and submit a batch in one call.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use submit() instead.

Parameters

invoices (Iterable[BatchInvoice]) — Invoice XML payloads to include in the batch package.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
max_part_size (int) — Maximum size of each ZIP part before encryption.

Returns

Coroutine[None, None, BatchSessionResumeState] — Serializable state of the submitted batch session.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If preparation, session opening, or upload validation fails.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

`AsyncBatchService.get_status`

def get_status(
    *,
    session: str | BatchSessionResumeState | AsyncBatchSessionClient,
) -> Coroutine[None, None, SessionStatusResponse]

python

Deprecated: fetch the current status of a batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.get_status() instead.

Parameters

session (str | BatchSessionResumeState | AsyncBatchSessionClient) — Session reference number, persisted state, or open session client.

Returns

Coroutine[None, None, SessionStatusResponse] — Current batch session status as reported by KSeF.

`AsyncBatchService.list_invoices`

def list_invoices(
    *,
    session: str | BatchSessionResumeState | AsyncBatchSessionClient,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> Coroutine[None, None, SessionInvoicesResponse]

python

Deprecated: fetch one page of accepted invoices from a batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.list_invoices() instead.

Parameters

session (str | BatchSessionResumeState | AsyncBatchSessionClient) — Session reference number, persisted state, or open session client.
page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

Coroutine[None, None, SessionInvoicesResponse] — One page of invoices accepted in the session, with a continuation token when more exist.

`AsyncBatchService.list_failed_invoices`

def list_failed_invoices(
    *,
    session: str | BatchSessionResumeState | AsyncBatchSessionClient,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> Coroutine[None, None, SessionInvoicesResponse]

python

Deprecated: fetch one page of failed invoices from a batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.list_failed_invoices() instead.

Parameters

session (str | BatchSessionResumeState | AsyncBatchSessionClient) — Session reference number, persisted state, or open session client.
page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

Coroutine[None, None, SessionInvoicesResponse] — One page of invoices that failed processing, with a continuation token when more exist.

`AsyncBatchService.get_upo`

def get_upo(
    *,
    session: str | BatchSessionResumeState | AsyncBatchSessionClient,
    upo_reference_number: str,
) -> Coroutine[None, None, bytes]

python

Deprecated: download the collective UPO for a batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.download_upo() instead.

Parameters

session (str | BatchSessionResumeState | AsyncBatchSessionClient) — Session reference number, persisted state, or open session client.
upo_reference_number (str) — UPO page reference returned in the session status.

Returns

Coroutine[None, None, bytes] — Raw XML bytes of the requested UPO page.

`AsyncBatchService.wait_for_completion`

def wait_for_completion(
    *,
    session: str | BatchSessionResumeState | AsyncBatchSessionClient,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> Coroutine[None, None, SessionStatusResponse]

python

Deprecated: poll a batch session until KSeF reports a terminal status.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.wait() instead.

Parameters

session (str | BatchSessionResumeState | AsyncBatchSessionClient) — Session reference number, persisted state, or open session client.
timeout (float) — Maximum number of seconds to wait for completion.
poll_interval (float) — Delay between status checks.

Returns

Coroutine[None, None, SessionStatusResponse] — Final successful batch session status.

Raises

KSeFSessionError — If batch processing reaches a failed terminal status.
KSeFBatchSessionTimeoutError — If polling exceeds timeout.

`AsyncBatchSessionClient`

Import: from ksef2.clients import AsyncBatchSessionClient

class AsyncBatchSessionClient

python

AsyncBatchSessionClient(
    transport: AsyncMiddleware,
    state: BatchSessionResumeState,
    *,
    upload_transport: AsyncMiddleware | None = None,
    prepared_batch: PreparedBatch | None = None,
    access_token: str | None = None,
    resumed: bool = False,
)

python

Async client for managing an active batch session.

Catch KSeFException for SDK-classified failures raised by this session branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
KSeFClientClosedError — If an operation is attempted after closing the session.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncBatchSessionClient.reference_number` (property)

reference_number: str

python

Get the batch session reference number.

Returns

str — The KSeF reference number of the session.

`AsyncBatchSessionClient.access_token` (property)

access_token: str

python

Deprecated compatibility accessor for the current bearer token.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use AuthenticatedClient.access_token instead.

Returns

str — The bearer access token the session was opened with.

Raises

KSeFValidationError — If the session state carries no bearer token.

`AsyncBatchSessionClient.aes_key` (property)

aes_key: bytes

python

Get the AES key for encrypting batch files.

Returns

bytes — The raw AES key.

`AsyncBatchSessionClient.iv` (property)

iv: bytes

python

Get the initialization vector for AES encryption.

Returns

bytes — The raw initialization vector.

`AsyncBatchSessionClient.part_upload_requests` (property)

part_upload_requests: list[PartUploadRequest]

python

Get the upload instructions for each file part.

Returns

list[PartUploadRequest] — One upload request, with presigned URL and headers, per part.

`AsyncBatchSessionClient.resume_state`

def resume_state() -> BatchSessionResumeState

python

Return the sensitive session state needed to resume later.

Returns

BatchSessionResumeState — The sensitive session state needed to resume later.

`AsyncBatchSessionClient.get_state`

def get_state() -> BatchSessionResumeState

python

Deprecated compatibility wrapper for resume_state().

Deprecated

May be removed no earlier than ksef2 2.0.0. Use resume_state() instead.

Returns

BatchSessionResumeState — The same state as resume_state().

`AsyncBatchSessionClient.get_status`

async def get_status() -> SessionStatusResponse

python

Fetch the current processing state of the batch session.

Returns

SessionStatusResponse — The session status, including invoice counters.

`AsyncBatchSessionClient.list_invoices`

async def list_invoices(
    *,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Fetch one page of invoices submitted in this batch session.

Parameters

page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices submitted in the session, with a continuation token when more exist.

`AsyncBatchSessionClient.list_failed_invoices`

async def list_failed_invoices(
    *,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Fetch one page of failed invoices from this batch session.

Parameters

page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices that failed processing, with a continuation token when more exist.

`AsyncBatchSessionClient.get_upo`

def get_upo(*, upo_reference_number: str) -> Coroutine[None, None, bytes]

python

Deprecated: download one page of the collective UPO for the batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download_upo() instead.

Parameters

upo_reference_number (str) — Reference number of the UPO, taken from the session status.

Returns

Coroutine[None, None, bytes] — The UPO as XML bytes.

`AsyncBatchSessionClient.wait`

async def wait(
    *,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> SessionStatusResponse

python

Poll until KSeF reports a terminal status for the closed batch session.

KSeF processes the batch after the session is closed, so call this after auth.batch.submit() or after leaving the with block of a batch session.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between session status checks.

Returns

SessionStatusResponse — The final successful session status, including the session UPO references.

Raises

KSeFSessionError — If the session is still open, or if batch processing reaches a failed terminal status.
KSeFBatchSessionTimeoutError — If polling exceeds timeout.

`AsyncBatchSessionClient.download_upo`

async def download_upo() -> list[bytes]

python

Download every page of the collective UPO in one pass.

Reads the session status once, resolves the UPO page references from it and downloads each page. It never waits or polls: call wait() first.

Returns

list[bytes] — The XML bytes of each UPO page, in order; empty if KSeF issued no UPO because no invoice was accepted.

Raises

KSeFSessionError — If the session is still open.
KSeFNotReadyError — If KSeF has not finished processing the closed batch. Call wait() on the session, then ask again.

`AsyncBatchSessionClient.upload_parts`

async def upload_parts() -> None

python

Upload the prepared batch parts using the session's presigned URLs.

Raises

KSeFClientClosedError — If the upload window has already been closed.
KSeFValidationError — If the session has no prepared batch payload attached.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

`AsyncBatchSessionClient.aclose`

async def aclose() -> None

python

Close the batch session and start processing.

This triggers processing of the invoice batch and generation of UPO for valid invoices and a collective UPO for the session. Closing an already-closed session is a no-op; a client rebuilt from saved state first asks KSeF whether the session was already closed.

`AsyncCertificateEnrollment`

Import: from ksef2.clients import AsyncCertificateEnrollment

class AsyncCertificateEnrollment(AsyncOperationHandle[CertificateEnrollmentStatusResponse, CertificateEnrollmentStatusResponse])

python

AsyncCertificateEnrollment(
    client: AsyncCertificatesClient,
    response: CertificateEnrollmentResponse,
)

python

Handle to a certificate enrollment that KSeF issues asynchronously.

Returned by auth.certificates.enroll(). It exposes every field of the enrollment response, for example reference_number and timestamp, and wait() polls until KSeF has issued the certificate. get_enrollment_status() on the certificates client reads the same status for a reference number you stored earlier.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncCertificateEnrollment.timestamp` (property)

timestamp: datetime

python

Get when KSeF accepted the enrollment request.

Returns

datetime — The acceptance time reported by KSeF.

`AsyncCertificateEnrollment.response` (property)

response: CertificateEnrollmentResponse

python

Get the plain enrollment response.

Returns

CertificateEnrollmentResponse — The data model KSeF returned when it accepted the enrollment.

`AsyncCertificateEnrollment.get_status`

async def get_status() -> CertificateEnrollmentStatusResponse

python

Fetch the enrollment's current status without waiting.

Returns

CertificateEnrollmentStatusResponse — The enrollment status, including the certificate serial number once issued.

`AsyncCertificateEnrollment.wait`

async def wait(
    *,
    timeout: float = 60.0,
    poll_interval: float = 2.0,
) -> CertificateEnrollmentStatusResponse

python

Poll until KSeF has issued the certificate.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between status checks.

Returns

CertificateEnrollmentStatusResponse — The final status, with the issued certificate's serial number.

Raises

KSeFCertificateEnrollmentFailedError — If KSeF rejects, cancels or fails the enrollment.
KSeFCertificateEnrollmentTimeoutError — If polling exceeds timeout.

`AsyncCertificatesClient`

Import: from ksef2.clients import AsyncCertificatesClient

class AsyncCertificatesClient

python

AsyncCertificatesClient(transport: AsyncMiddleware)

python

Async high-level API for certificate enrollment, retrieval, and search.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncCertificatesClient.get_limits`

async def get_limits() -> CertificateLimitsResponse

python

Return the effective certificate enrollment and issuance quotas.

Returns

CertificateLimitsResponse — The effective certificate enrollment and issuance quotas.

`AsyncCertificatesClient.get_enrollment_data`

async def get_enrollment_data() -> CertificateEnrollmentData

python

Return subject data needed to prepare a CSR for enrollment.

Returns

CertificateEnrollmentData — Subject data needed to prepare a CSR for enrollment.

`AsyncCertificatesClient.enroll`

async def enroll(
    *,
    certificate_name: str,
    certificate_type: CertificateTypeValue,
    csr: str,
    valid_from: datetime | str | None = None,
) -> AsyncCertificateEnrollment

python

Request issuance of a certificate from a CSR.

Parameters

certificate_name (str) — Name to give the certificate.
certificate_type (CertificateTypeValue) — Certificate type to request, authentication or offline.
csr (str) — PKCS#10 certificate signing request, DER-encoded and Base64-encoded.
valid_from (datetime | str | None) — Requested start of validity as a datetime or ISO 8601 string; None for immediately.

Returns

AsyncCertificateEnrollment — A handle to the enrollment; call wait() to wait until KSeF has issued the certificate.

Example

enrollment = await auth.certificates.enroll(
    certificate_name="signing", certificate_type="offline", csr=csr
)
status = await enrollment.wait()
print(status.certificate_serial_number)

python

`AsyncCertificatesClient.get_enrollment_status`

async def get_enrollment_status(
    *,
    reference_number: str,
) -> CertificateEnrollmentStatusResponse

python

Fetch the current status of a certificate enrollment request.

Parameters

reference_number (str) — Reference number returned by enroll().

Returns

CertificateEnrollmentStatusResponse — The enrollment status, including the certificate serial number once issued.

`AsyncCertificatesClient.retrieve`

async def retrieve(
    *,
    certificate_serial_numbers: builtins.list[CertificateSerialNumber],
) -> RetrievedCertificatesList

python

Download issued certificates by serial number.

Parameters

certificate_serial_numbers (builtins.list[CertificateSerialNumber]) — Serial numbers of the certificates to download.

Returns

RetrievedCertificatesList — The requested certificates with their DER data.

`AsyncCertificatesClient.revoke`

async def revoke(
    *,
    certificate_serial_number: CertificateSerialNumber,
    reason: RevocationReason | None = None,
) -> None

python

Revoke a certificate, optionally providing a revocation reason.

Parameters

certificate_serial_number (CertificateSerialNumber) — Serial number of the certificate to revoke.
reason (RevocationReason | None) — Reason for the revocation; None to omit it.

`AsyncCertificatesClient.query`

def query(
    *,
    name: str | None = None,
    certificate_serial_number: CertificateSerialNumber | None = None,
    certificate_type: CertificateTypeValue | None = None,
    status: CertificateStatusValue | None = None,
    expires_after: datetime | str | None = None,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, CertificatesInfoList]

python

Deprecated: fetch one page of certificate search results.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

name (str | None) — Match this certificate name.
certificate_serial_number (CertificateSerialNumber | None) — Match this certificate serial number.
certificate_type (CertificateTypeValue | None) — Match this certificate type.
status (CertificateStatusValue | None) — Match this lifecycle status.
expires_after (datetime | str | None) — Match certificates that expire after this datetime or ISO 8601 string.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

Coroutine[None, None, CertificatesInfoList] — One page of certificate metadata.

`AsyncCertificatesClient.all`

async def all(
    *,
    certificate_serial_number: CertificateSerialNumber | None = None,
    name: str | None = None,
    certificate_type: CertificateTypeValue | None = None,
    status: CertificateStatusValue | None = None,
    expires_after: datetime | str | None = None,
    params: OffsetPaginationParams | None = None,
) -> AsyncIterator[CertificateInfo]

python

Deprecated: iterate over all certificates matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead.

Parameters

certificate_serial_number (CertificateSerialNumber | None) — Match this certificate serial number.
name (str | None) — Match this certificate name.
certificate_type (CertificateTypeValue | None) — Match this certificate type.
status (CertificateStatusValue | None) — Match this lifecycle status.
expires_after (datetime | str | None) — Match certificates that expire after this datetime or ISO 8601 string.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Yields

AsyncIterator[CertificateInfo] — Each matching certificate, across all pages.

`AsyncCertificatesClient.list`

def list(
    *,
    name: str | None = None,
    certificate_serial_number: CertificateSerialNumber | None = None,
    certificate_type: CertificateTypeValue | None = None,
    status: CertificateStatusValue | None = None,
    expires_after: datetime | str | None = None,
    params: OffsetPaginationParams | None = None,
) -> AsyncPager[CertificateInfo]

python

List the certificates of the authenticated context.

Nothing is requested until the result is consumed. Iterate it for every matching certificate, call pages() for page-sized lists or first_page() for one request only.

Parameters

name (str | None) — Match this certificate name.
certificate_serial_number (CertificateSerialNumber | None) — Match this certificate serial number.
certificate_type (CertificateTypeValue | None) — Match this certificate type.
status (CertificateStatusValue | None) — Match this lifecycle status.
expires_after (datetime | str | None) — Match certificates that expire after this datetime or ISO 8601 string.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[CertificateInfo] — A paging object over the matching certificates.

Example

async for certificate in auth.certificates.list(status="active"):
    print(certificate.certificate_serial_number)

python

`AsyncClient`

Import: from ksef2.clients import AsyncClient

class AsyncClient

python

AsyncClient(
    environment: Environment = Environment.PRODUCTION,
    *,
    transport_config: TransportConfig | None = None,
    http_client: httpx.AsyncClient | None = None,
    certificate_store: stores.CertificateStoreProtocol | None = None,
)

python

Root async KSeF SDK client responsible for transport and lifecycle management.

Branch properties only create child clients. Branch operations document their own API, validation, and transport failures.

Raises

KSeFClientClosedError — If a branch is accessed after the client is closed.
KSeFUnsupportedEnvironmentError — If a TEST-only branch is accessed outside Environment.TEST.

Example

from ksef2 import AsyncClient, Environment

async with AsyncClient(Environment.TEST) as client:
    auth = await client.authentication.with_test_certificate(nip="5261040828")
    async for invoice in auth.invoices.all_metadata(filters=filters):
        print(invoice.ksef_number)

python

Methods

`AsyncClient.authentication` (property)

authentication: AsyncAuthClient

python

Return the authentication entry point.

Returns

AsyncAuthClient — The authentication branch, used to log in and obtain an authenticated client.

Raises

KSeFClientClosedError — If the root client has been closed.

`AsyncClient.encryption` (property)

encryption: AsyncEncryptionClient

python

Return the public encryption-certificate client.

Returns

AsyncEncryptionClient — The client for downloading KSeF public-key certificates.

Raises

KSeFClientClosedError — If the root client has been closed.

`AsyncClient.peppol` (property)

peppol: AsyncPeppolClient

python

Return the public Peppol provider client.

Returns

AsyncPeppolClient — The client for listing registered Peppol providers.

Raises

KSeFClientClosedError — If the root client has been closed.

`AsyncClient.testdata` (property)

testdata: AsyncTestDataClient

python

Return the TEST-only data seeding client.

Returns

AsyncTestDataClient — The client for creating and removing TEST-environment data.

Raises

KSeFClientClosedError — If the root client has been closed.
KSeFUnsupportedEnvironmentError — If the client environment is not TEST.

`AsyncClient.raw` (property)

raw: AsyncRawClient

python

Return raw unauthenticated endpoints for advanced async integrations.

Returns

AsyncRawClient — Raw unauthenticated endpoints for advanced async integrations.

`AsyncClient.authenticated`

def authenticated(auth_tokens: AuthTokens) -> AsyncAuthenticatedClient

python

Deprecated compatibility wrapper for authentication.resume().

Deprecated

May be removed no earlier than ksef2 2.0.0. Use authentication.resume() with AuthenticationResumeState.from_tokens() instead.

Parameters

auth_tokens (AuthTokens) — Access and refresh tokens from a previous authentication.

Returns

AsyncAuthenticatedClient — An authenticated client bound to the tokens.

`AsyncClient.aclose`

async def aclose() -> None

python

Close owned resources and invalidate cached child clients.

`AsyncCollectiveIdentifiersClient`

Import: from ksef2.clients import AsyncCollectiveIdentifiersClient

class AsyncCollectiveIdentifiersClient

python

AsyncCollectiveIdentifiersClient(transport: AsyncMiddleware)

python

API for generating and querying collective invoice identifiers.

Methods

`AsyncCollectiveIdentifiersClient.generate`

async def generate(
    *,
    invoices: builtins.list[CollectiveIdentifierInvoice],
) -> GenerateCollectiveIdentifierResponse

python

Generate a collective identifier for the supplied invoices.

Parameters

invoices (builtins.list[CollectiveIdentifierInvoice]) — Invoices to group, with optional payment details.

Returns

GenerateCollectiveIdentifierResponse — The generated collective identifier number.

`AsyncCollectiveIdentifiersClient.query`

def query(
    *,
    filters: CollectiveIdentifiersQuery,
    continuation_token: str | None = None,
    params: CollectiveIdentifierParams | None = None,
) -> Coroutine[None, None, CollectiveIdentifiersPage]

python

Deprecated: fetch one page of collective identifiers visible in the context.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

filters (CollectiveIdentifiersQuery) — Criteria selecting collective identifiers.
continuation_token (str | None) — Token from the previous page's response; None requests the first page.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

Coroutine[None, None, CollectiveIdentifiersPage] — One page of matching collective identifiers.

`AsyncCollectiveIdentifiersClient.query_all`

async def query_all(
    *,
    filters: CollectiveIdentifiersQuery,
    params: CollectiveIdentifierParams | None = None,
) -> AsyncIterator[CollectiveIdentifiersPage]

python

Deprecated: iterate through every page matching a collective identifier query.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list().pages() instead.

Parameters

filters (CollectiveIdentifiersQuery) — Criteria selecting collective identifiers.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Yields

AsyncIterator[CollectiveIdentifiersPage] — Each page of matching collective identifiers, following continuation tokens.

`AsyncCollectiveIdentifiersClient.query_by_ksef_number`

def query_by_ksef_number(
    *,
    ksef_number: str,
    continuation_token: str | None = None,
    params: CollectiveIdentifierParams | None = None,
) -> Coroutine[None, None, CollectiveIdentifierReferencesPage]

python

Deprecated: fetch one page of identifiers associated with a KSeF invoice.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_for_invoice() instead; first_page() fetches one page.

Parameters

ksef_number (str) — KSeF number of the invoice.
continuation_token (str | None) — Token from the previous page's response; None requests the first page.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

Coroutine[None, None, CollectiveIdentifierReferencesPage] — One page of collective identifiers that reference the invoice.

`AsyncCollectiveIdentifiersClient.query_all_by_ksef_number`

async def query_all_by_ksef_number(
    *,
    ksef_number: str,
    params: CollectiveIdentifierParams | None = None,
) -> AsyncIterator[CollectiveIdentifierReferencesPage]

python

Deprecated: iterate through identifiers associated with one KSeF invoice.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_for_invoice().pages() instead.

Parameters

ksef_number (str) — KSeF number of the invoice.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Yields

AsyncIterator[CollectiveIdentifierReferencesPage] — Each page of identifiers that reference the invoice, following continuation tokens.

`AsyncCollectiveIdentifiersClient.list_all_invoices`

async def list_all_invoices(
    *,
    collective_identifier_numbers: builtins.list[str],
    params: CollectiveIdentifierParams | None = None,
) -> AsyncIterator[CollectiveIdentifierInvoicesPage]

python

Deprecated: iterate through every invoice page for the supplied collective identifiers.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_invoices().pages() instead.

Parameters

collective_identifier_numbers (builtins.list[str]) — Collective identifier numbers to expand (1–10).
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Yields

AsyncIterator[CollectiveIdentifierInvoicesPage] — Each page of invoices, following continuation tokens.

`AsyncCollectiveIdentifiersClient.list_for_invoice`

def list_for_invoice(
    ksef_number: str,
    *,
    params: CollectiveIdentifierParams | None = None,
) -> AsyncPager[CollectiveIdentifierReference]

python

List the collective identifiers that reference one KSeF invoice.

Nothing is requested until the result is consumed.

Parameters

ksef_number (str) — KSeF number of the invoice.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

AsyncPager[CollectiveIdentifierReference] — A paging object over the collective identifiers referencing the invoice.

Example

async for reference in auth.collective_identifiers.list_for_invoice(ksef_number):
    print(reference.collective_identifier_number)

python

`AsyncCollectiveIdentifiersClient.list_invoices`

def list_invoices(
    collective_identifier_numbers: builtins.list[str],
    *,
    params: CollectiveIdentifierParams | None = None,
) -> AsyncPager[CollectiveIdentifierInvoiceDetails]

python

List the invoices inside the supplied collective identifiers.

Nothing is requested until the result is consumed. Unlike the former list_invoices(), which fetched one page, this returns a paging object.

Parameters

collective_identifier_numbers (builtins.list[str]) — Collective identifier numbers to expand (1–10).
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

AsyncPager[CollectiveIdentifierInvoiceDetails] — A paging object over the invoices belonging to the identifiers.

Example

async for invoice in auth.collective_identifiers.list_invoices([number]):
    print(invoice.ksef_number)

python

`AsyncCollectiveIdentifiersClient.list`

def list(
    filters: CollectiveIdentifiersQuery,
    *,
    params: CollectiveIdentifierParams | None = None,
) -> AsyncPager[CollectiveIdentifierSummary]

python

List the collective identifiers visible in the context.

Nothing is requested until the result is consumed. Iterate it for every identifier, call pages() for page-sized lists or first_page() for one request only.

Parameters

filters (CollectiveIdentifiersQuery) — Criteria selecting collective identifiers.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

AsyncPager[CollectiveIdentifierSummary] — A paging object over the matching collective identifiers.

Example

async for identifier in auth.collective_identifiers.list(filters):
    print(identifier.collective_identifier_number)

python

`AsyncEncryptionClient`

Import: from ksef2.clients import AsyncEncryptionClient

class AsyncEncryptionClient

python

AsyncEncryptionClient(transport: AsyncMiddleware)

python

Async access to public encryption certificates published by KSeF.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncEncryptionClient.get_certificates`

async def get_certificates(
    *,
    usage: list[CertUsage] | None = None,
) -> list[PublicKeyCertificate]

python

Return public certificates, optionally filtered by supported usage.

Parameters

usage (list[CertUsage] | None) — Return only certificates valid for these usages; None for all.

Returns

list[PublicKeyCertificate] — The public-key certificates published by KSeF.

`AsyncExportJob`

Import: from ksef2.clients import AsyncExportJob

class AsyncExportJob(AsyncOperationHandle[InvoiceExportStatusResponse, ExportedInvoices])

python

AsyncExportJob(
    state: ExportResumeState,
    *,
    get_status: Callable[[], Awaitable[InvoiceExportStatusResponse]],
    download_parts: Callable[[InvoicePackage], Awaitable[list[bytes]]],
)

python

Handle to a scheduled invoice export.

Returned by auth.invoices.export(). wait() polls until KSeF has built the package, then downloads, decrypts and joins its parts into an ExportedInvoices result. The handle holds the keys needed to decrypt the package, so keep it in memory only.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncExportJob.resume_state`

def resume_state() -> ExportResumeState

python

Return the sensitive state needed to resume the export later.

Persist it with to_json() before waiting, and pass it to auth.invoices.export(state=...) after a restart to get the job back. It holds the key that decrypts the package, so store it as a credential.

Returns

ExportResumeState — The export's reference number and AES key material.

`AsyncExportJob.get_status`

async def get_status() -> InvoiceExportStatusResponse

python

Fetch the export's current status without waiting.

Returns

InvoiceExportStatusResponse — The export status, with package metadata once the export is ready.

`AsyncExportJob.wait`

async def wait(
    *,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> ExportedInvoices

python

Poll until the export is ready, then download and decrypt the package.

Parameters

timeout (float) — Maximum number of seconds to wait for the export to become ready.
poll_interval (float) — Delay in seconds between export status checks.

Returns

ExportedInvoices — The decrypted package: iterate invoices(), read metadata or save() it to disk.

Raises

KSeFExportFailedError — If KSeF reports that the export failed, was cancelled or expired.
KSeFExportTimeoutError — If polling exceeds timeout.
KSeFEncryptionError — If a downloaded package part cannot be decrypted.
KSeFExternalTransferError — If external storage rejects a part download or its outcome cannot be determined.

`AsyncGeneratedToken`

Import: from ksef2.clients import AsyncGeneratedToken

class AsyncGeneratedToken(AsyncOperationHandle[TokenStatusResponse, TokenStatusResponse])

python

AsyncGeneratedToken(client: AsyncTokensClient, response: GenerateTokenResponse)

python

Handle to a token KSeF is still activating.

Returned by auth.tokens.generate(). The one-time token is readable immediately, so persist it before calling wait(): a polling failure does not contain or recover that secret. The handle exposes every field of the generation response, for example token and reference_number, and never includes the token in its repr.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncGeneratedToken.token` (property)

token: str

python

Get the one-time token value.

Returns

str — The secret KSeF returned once at generation; store it securely.

`AsyncGeneratedToken.response` (property)

response: GenerateTokenResponse

python

Get the plain generation response.

Returns

GenerateTokenResponse — The data model KSeF returned when the token was generated.

`AsyncGeneratedToken.to_sensitive_dict`

def to_sensitive_dict() -> dict[str, str]

python

Export the one-time token for deliberately protected persistence.

Returns

dict[str, str] — A dictionary with the reference number and the plain token.

`AsyncGeneratedToken.get_status`

async def get_status() -> TokenStatusResponse

python

Fetch the token's current status without waiting.

Returns

TokenStatusResponse — The token's current lifecycle status.

`AsyncGeneratedToken.wait`

async def wait(
    *,
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> TokenStatusResponse

python

Poll until the token becomes active or reaches a terminal state.

Parameters

timeout (float) — Maximum number of seconds to wait for activation.
poll_interval (float) — Delay in seconds between status checks.

Returns

TokenStatusResponse — The first status response that reports the token as active.

Raises

KSeFApiError — If activation ends in a terminal failure state.
KSeFTokenStatusTimeoutError — If polling exceeds timeout.

`AsyncInvoiceSessionsClient`

Import: from ksef2.clients import AsyncInvoiceSessionsClient

class AsyncInvoiceSessionsClient

python

AsyncInvoiceSessionsClient(transport: AsyncMiddleware)

python

Async browse historical invoice sessions and download their UPO documents.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncInvoiceSessionsClient.download_upo`

async def download_upo(
    session_reference_number: str,
    *,
    invoice_reference_number: str | None = None,
    ksef_number: str | None = None,
    upo_reference_number: str | None = None,
) -> bytes

python

Download one UPO from stored references without opening a session.

Works for online and batch invoice sessions, including after a process restart. Uses the authenticated client's bearer token and makes one request. It never waits or polls, and needs no session crypto state.

Parameters

session_reference_number (str) — Reference number of the invoice session.
invoice_reference_number (str | None) — Invoice reference within that session. Pass exactly one of this, ksef_number or upo_reference_number.
ksef_number (str | None) — KSeF number of an accepted invoice in that session.
upo_reference_number (str | None) — Reference number of one session UPO page.

Returns

bytes — The invoice UPO or session UPO page as XML bytes.

Raises

KSeFValidationError — If no selector or multiple selectors are given.
KSeFNotReadyError — If KSeF has not issued the requested UPO yet. Check processing status before retrying. Rejected invoices may never receive a UPO.

`AsyncInvoiceSessionsClient.query`

def query(
    *,
    session_type: str,
    continuation_token: str | None = None,
    params: ListSessionsQuery | None = None,
    statuses: builtins.list[SessionStatus | SessionStatusEnum] | None = None,
) -> Coroutine[None, None, ListSessionsResponse]

python

Deprecated: fetch one page of invoice session history for the chosen session type.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

session_type (str) — Invoice session family to browse, such as "online" or "batch".
continuation_token (str | None) — Cursor returned by a previous page.
params (ListSessionsQuery | None) — Optional query object with pagination and filter settings.
statuses (builtins.list[SessionStatus | SessionStatusEnum] | None) — Optional list of status filters applied on top of params.

Returns

Coroutine[None, None, ListSessionsResponse] — One page of historical invoice sessions.

`AsyncInvoiceSessionsClient.all`

async def all(
    *,
    session_type: str,
    params: ListSessionsQuery | None = None,
) -> AsyncIterator[ListSessionsResponse]

python

Deprecated: iterate through all pages of invoice session history.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list().pages() instead.

Parameters

session_type (str) — Invoice session family to browse, such as "online" or "batch".
params (ListSessionsQuery | None) — Optional query object with pagination and filter settings.

Yields

AsyncIterator[ListSessionsResponse] — Successive pages of historical invoice sessions until KSeF stops
AsyncIterator[ListSessionsResponse] — returning a continuation token.

`AsyncInvoiceSessionsClient.list`

def list(
    session_type: str,
    *,
    params: ListSessionsQuery | None = None,
    statuses: builtins.list[SessionStatus | SessionStatusEnum] | None = None,
) -> AsyncPager[SessionSummary]

python

List the invoice sessions of the chosen type, newest history first as KSeF returns it.

Nothing is requested until the result is consumed. Iterate it for every session, call pages() for page-sized lists or first_page() for one request only.

Parameters

session_type (str) — Invoice session family to browse, such as "online" or "batch".
params (ListSessionsQuery | None) — Optional query object with page size and filter settings.
statuses (builtins.list[SessionStatus | SessionStatusEnum] | None) — Optional list of status filters applied on top of params.

Returns

AsyncPager[SessionSummary] — A paging object over the matching invoice sessions.

Example

async for session in auth.invoice_sessions.list("online"):
    print(session.reference_number, session.status)

python

`AsyncInvoicesService`

Import: from ksef2.clients import AsyncInvoicesService

class AsyncInvoicesService

python

AsyncInvoicesService(
    transport: AsyncMiddleware,
    download_transport: AsyncMiddleware,
    certificate_store: CertificateStoreProtocol,
    *,
    client: AsyncInvoicesClient | None = None,
    ensure_encryption_certificates_loaded: Callable[[], Awaitable[None]] | None = None,
)

python

Async invoice workflows with encryption, polling, and export helpers.

Catch KSeFException for SDK-classified failures raised by this service, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncInvoicesService.search`

def search(
    filters: InvoicesFilter,
    params: InvoiceMetadataParams | None = None,
) -> AsyncPager[InvoiceMetadata]

python

Search invoice metadata, following KSeF page and truncation mechanics.

Nothing is requested until the result is consumed. Iterate it for every matching invoice, call pages() for page-sized lists, first_page() for one request only, or wait() to poll until KSeF has indexed at least one matching invoice.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
params (InvoiceMetadataParams | None) — Page size and sort order; defaults are used when None.

Returns

AsyncPager[InvoiceMetadata] — A paging object over the metadata of matching invoices.

Raises

KSeFMetadataPaginationError — If KSeF returns inconsistent pagination boundaries while paging.

Example

from ksef2.models import InvoicesFilter

filters = InvoicesFilter.for_seller(date_from="2026-01-01T00:00:00+01:00")
async for invoice in auth.invoices.search(filters):
    print(invoice.ksef_number, invoice.gross_amount)

python

`AsyncInvoicesService.download`

async def download(
    ksef_number: str,
    *,
    timeout: float | None = None,
    poll_interval: float = 2.0,
) -> bytes

python

Download one processed invoice by KSeF number.

Parameters

ksef_number (str) — KSeF number of the invoice.
timeout (float | None) — Seconds to keep polling while KSeF has not made the invoice available yet; None tries once.
poll_interval (float) — Delay in seconds between download attempts when timeout is set.

Returns

bytes — The invoice XML as bytes.

Raises

KSeFApiError — If KSeF rejects the download, for example because the invoice is not processed yet and timeout is None.
KSeFInvoiceDownloadTimeoutError — If timeout is set and polling exceeds it.

Example

xml = await auth.invoices.download(ksef_number, timeout=60)

python

`AsyncInvoicesService.export`

async def export(
    filters: InvoicesFilter | None = None,
    *,
    state: ExportResumeState | str | None = None,
    only_metadata: bool = False,
    compression_type: CompressionType | str | None = None,
) -> AsyncExportJob

python

Schedule an encrypted invoice export, or resume one scheduled earlier.

Pass exactly one of filters (schedule a new export) or state (resume the export described by a saved ExportResumeState).

Parameters

filters (InvoicesFilter | None) — Criteria selecting the invoices to export.
state (ExportResumeState | str | None) — State from ExportJob.resume_state(), or its JSON string, to resume an export after a restart. Cannot be combined with the other arguments.
only_metadata (bool) — Export only invoice metadata instead of full invoice XML.
compression_type (CompressionType | str | None) — Compression applied to the package; None for the server default.

Returns

AsyncExportJob — A handle to the export. Call its wait() to download the decrypted package.

Raises

KSeFArgumentError — If neither or both of filters and state are given, or state is combined with scheduling options.
NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If export key encryption fails.
KSeFValidationError — If state is not valid export resume state.

Example

job = await auth.invoices.export(filters)
saved = job.resume_state().to_json()  # persist before waiting

job = await auth.invoices.export(state=saved)  # after a restart
package = await job.wait()
package.save("out/")

python

`AsyncInvoicesService.query_metadata`

def query_metadata(
    *,
    filters: InvoicesFilter,
    params: InvoiceMetadataParams | None = None,
) -> Coroutine[None, None, QueryInvoicesMetadataResponse]

python

Deprecated: fetch one invoice metadata page matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use search() instead; first_page() fetches one page.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
params (InvoiceMetadataParams | None) — Page size, page offset and sort order; defaults are used when None.

Returns

Coroutine[None, None, QueryInvoicesMetadataResponse] — One page of invoice metadata.

`AsyncInvoicesService.query_metadata_pages`

async def query_metadata_pages(
    *,
    filters: InvoicesFilter,
    params: InvoiceMetadataParams | None = None,
) -> AsyncIterator[QueryInvoicesMetadataResponse]

python

Deprecated: fetch metadata pages, following KSeF page and truncation mechanics.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use search().pages() instead.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
params (InvoiceMetadataParams | None) — Page size, page offset and sort order; defaults are used when None.

Yields

AsyncIterator[QueryInvoicesMetadataResponse] — Each page of invoice metadata in order.

Raises

KSeFMetadataPaginationError — If KSeF returns inconsistent pagination boundaries.

`AsyncInvoicesService.all_metadata`

async def all_metadata(
    *,
    filters: InvoicesFilter,
    params: InvoiceMetadataParams | None = None,
) -> AsyncIterator[InvoiceMetadata]

python

Deprecated: iterate over all invoice metadata items matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use search() instead.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
params (InvoiceMetadataParams | None) — Page size and sort order; defaults are used when None.

Yields

AsyncIterator[InvoiceMetadata] — Metadata of each matching invoice, across all pages.

Raises

KSeFMetadataPaginationError — If KSeF returns inconsistent pagination boundaries.

`AsyncInvoicesService.download_invoice`

def download_invoice(*, ksef_number: str) -> Coroutine[None, None, bytes]

python

Deprecated: download one processed invoice by KSeF number.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download() instead.

Parameters

ksef_number (str) — KSeF number of the invoice.

Returns

Coroutine[None, None, bytes] — The invoice XML as bytes.

`AsyncInvoicesService.wait_for_invoice_download`

def wait_for_invoice_download(
    *,
    ksef_number: str,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> Coroutine[None, None, bytes]

python

Deprecated: poll until KSeF makes a processed invoice available for download.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download(ksef_number, timeout=...) instead.

Parameters

ksef_number (str) — KSeF number of the invoice.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between download attempts.

Returns

Coroutine[None, None, bytes] — The invoice XML as bytes.

Raises

KSeFInvoiceDownloadTimeoutError — If polling exceeds timeout.

`AsyncInvoicesService.schedule_export`

def schedule_export(
    *,
    filters: InvoicesFilter,
    only_metadata: bool = False,
    compression_type: CompressionType | str | None = None,
) -> Coroutine[None, None, ExportHandle]

python

Deprecated: schedule an encrypted invoice export.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export() instead.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices to export.
only_metadata (bool) — Export only invoice metadata instead of full invoice XML.
compression_type (CompressionType | str | None) — Compression applied to the package; None for the server default.

Returns

Coroutine[None, None, ExportHandle] — A handle holding the export reference number and the keys needed to decrypt the package.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If export key encryption fails.

`AsyncInvoicesService.get_export_status`

def get_export_status(
    *,
    reference_number: str,
) -> Coroutine[None, None, InvoiceExportStatusResponse]

python

Deprecated: fetch the current status for a scheduled invoice export.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export() and ExportJob.get_status() instead.

Parameters

reference_number (str) — Reference number of the export, from ExportHandle.reference_number.

Returns

Coroutine[None, None, InvoiceExportStatusResponse] — The export status, with package metadata once the export is ready.

`AsyncInvoicesService.fetch_package`

def fetch_package(
    *,
    package: InvoicePackage,
    export: ExportHandle,
    target_directory: Path | str = '.',
) -> Coroutine[None, None, list[Path]]

python

Deprecated: download and decrypt all parts of an export package to disk.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export().wait() and ExportedInvoices.save() instead.

Parameters

package (InvoicePackage) — Package metadata from the export status.
export (ExportHandle) — Handle returned when the export was scheduled; supplies the decryption keys.
target_directory (Path | str) — Directory to write the decrypted parts to. Defaults to the current directory.

Returns

Coroutine[None, None, list[Path]] — The paths of the written part files.

Raises

KSeFEncryptionError — If a downloaded package part cannot be decrypted.
ValueError — If a package part name is unsafe for local filesystem output.
OSError — If the target directory or output file cannot be written.
KSeFExternalTransferError — If external storage rejects a part download or its outcome cannot be determined.

`AsyncInvoicesService.fetch_package_bytes`

def fetch_package_bytes(
    *,
    package: InvoicePackage,
    export: ExportHandle,
) -> Coroutine[None, None, list[bytes]]

python

Deprecated: download and decrypt all parts of an export package in memory.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export().wait() and ExportedInvoices.archive instead.

Parameters

package (InvoicePackage) — Package metadata from the export status.
export (ExportHandle) — Handle returned when the export was scheduled; supplies the decryption keys.

Returns

Coroutine[None, None, list[bytes]] — The decrypted package parts, in order.

Raises

KSeFEncryptionError — If a downloaded package part cannot be decrypted.
KSeFExternalTransferError — If external storage rejects a part download or its outcome cannot be determined.

`AsyncInvoicesService.wait_for_invoices`

def wait_for_invoices(
    *,
    filters: InvoicesFilter,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> Coroutine[None, None, QueryInvoicesMetadataResponse]

python

Deprecated: poll invoice metadata until at least one invoice matches the filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use search(filters).wait() instead.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between queries.

Returns

Coroutine[None, None, QueryInvoicesMetadataResponse] — The first metadata page that contains at least one invoice.

Raises

KSeFInvoiceQueryTimeoutError — If polling exceeds timeout.

`AsyncInvoicesService.wait_for_export_package`

def wait_for_export_package(
    *,
    reference_number: str,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> Coroutine[None, None, InvoicePackage]

python

Deprecated: poll export status until KSeF exposes a downloadable package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export().wait() instead.

Parameters

reference_number (str) — Reference number of the export, from ExportHandle.reference_number.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between export status checks.

Returns

Coroutine[None, None, InvoicePackage] — The package metadata once the export is ready.

Raises

KSeFExportTimeoutError — If polling exceeds timeout.

`AsyncInvoicesService.export_and_download`

def export_and_download(
    *,
    filters: InvoicesFilter,
    only_metadata: bool = False,
    compression_type: CompressionType | str | None = None,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> Coroutine[None, None, list[bytes]]

python

Deprecated: schedule an export, wait for it, and download the decrypted package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export().wait() instead; ExportedInvoices.archive is the joined ZIP.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices to export.
only_metadata (bool) — Export only invoice metadata instead of full invoice XML.
compression_type (CompressionType | str | None) — Compression applied to the package; None for the server default.
timeout (float) — Maximum number of seconds to wait for the export to become ready.
poll_interval (float) — Delay in seconds between export status checks.

Returns

Coroutine[None, None, list[bytes]] — The decrypted package parts, in order. Concatenate them to obtain the package archive.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If export key encryption or package decryption fails.
KSeFExportTimeoutError — If polling exceeds timeout.
KSeFExternalTransferError — If external storage rejects a part download or its outcome cannot be determined.

`AsyncInvoiceSubmission`

Import: from ksef2.clients import AsyncInvoiceSubmission

class AsyncInvoiceSubmission(AsyncOperationHandle[SessionInvoiceStatusResponse, SessionInvoiceStatusResponse])

python

AsyncInvoiceSubmission(
    session: AsyncOnlineSessionClient,
    response: invoices.SendInvoiceResponse,
)

python

Handle to one invoice sent into an online session.

Returned by session.send_invoice(). It exposes every field of the submission response, for example reference_number, and polls the invoice's processing status with wait(). It keeps working after the session is closed, so a typical flow sends inside the with block and waits after it.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncInvoiceSubmission.response` (property)

response: invoices.SendInvoiceResponse

python

Get the plain submission response.

Returns

invoices.SendInvoiceResponse — The data model KSeF returned when the invoice was accepted for processing.

`AsyncInvoiceSubmission.get_status`

async def get_status() -> SessionInvoiceStatusResponse

python

Fetch the invoice's current processing status without waiting.

Returns

SessionInvoiceStatusResponse — The processing status of the invoice.

`AsyncInvoiceSubmission.wait`

async def wait(
    *,
    timeout: float = 60.0,
    poll_interval: float = 2.0,
) -> SessionInvoiceStatusResponse

python

Poll until KSeF assigns the invoice a final processing result.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between invoice status checks.

Returns

SessionInvoiceStatusResponse — The final processing status, including the invoice's KSeF number once accepted.

Raises

KSeFInvoiceRejectedError — If invoice processing reaches a failed terminal status. It subclasses KSeFSessionError and keeps the status details and extensions.
KSeFInvoiceProcessingTimeoutError — If polling exceeds timeout.

`AsyncInvoiceSubmission.download_upo`

async def download_upo() -> bytes

python

Download the UPO of this invoice with a single request.

KSeF issues the UPO once it has accepted the invoice. Call wait() first, then this; it never waits or polls.

Returns

bytes — The UPO as XML bytes.

Raises

KSeFNotReadyError — If KSeF has not issued the UPO yet. Call wait() on this submission, then ask again.

`AsyncLimitsClient`

Import: from ksef2.clients import AsyncLimitsClient

class AsyncLimitsClient

python

AsyncLimitsClient(transport: AsyncMiddleware)

python

Async read and override KSeF context, subject, and API rate limits.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncLimitsClient.get_context_limits`

async def get_context_limits() -> ContextLimits

python

Return the effective limits for online and batch sessions.

Returns

ContextLimits — The effective limits for online and batch sessions.

`AsyncLimitsClient.get_subject_limits`

async def get_subject_limits() -> SubjectLimits

python

Return the effective limits for certificate enrollments and issuance.

Returns

SubjectLimits — The effective limits for certificate enrollments and issuance.

`AsyncLimitsClient.get_api_rate_limits`

async def get_api_rate_limits() -> ApiRateLimits

python

Return the effective per-endpoint API rate limits.

Returns

ApiRateLimits — The effective per-endpoint API rate limits.

`AsyncLimitsClient.set_session_limits`

async def set_session_limits(*, limits: ContextLimits) -> None

python

Override session limits for the current subject.

Parameters

limits (ContextLimits) — New session limits.

`AsyncLimitsClient.reset_session_limits`

async def reset_session_limits() -> None

python

Reset session limits to their default values.

`AsyncLimitsClient.set_subject_limits`

async def set_subject_limits(*, limits: SubjectLimits) -> None

python

Override subject-level certificate and enrollment limits.

Parameters

limits (SubjectLimits) — New subject-level limits.

`AsyncLimitsClient.reset_subject_limits`

async def reset_subject_limits() -> None

python

Reset subject-level limits to their default values.

`AsyncLimitsClient.set_api_rate_limits`

async def set_api_rate_limits(*, limits: ApiRateLimits) -> None

python

Override API rate limits for the current subject.

Parameters

limits (ApiRateLimits) — New API rate limits.

`AsyncLimitsClient.reset_api_rate_limits`

async def reset_api_rate_limits() -> None

python

Reset API rate limits to their default values.

`AsyncLimitsClient.set_production_rate_limits`

async def set_production_rate_limits() -> None

python

Copy production rate-limit defaults into the current environment.

`AsyncOnlineSessionClient`

Import: from ksef2.clients import AsyncOnlineSessionClient

class AsyncOnlineSessionClient

python

AsyncOnlineSessionClient(
    transport: AsyncMiddleware,
    state: OnlineSessionResumeState,
    *,
    resumed: bool = False,
)

python

Async client bound to a single online invoice session.

Send invoices with send_invoice(), close the session by leaving the async with block, then call wait() for the terminal session status and download_upo() for the session UPO.

Catch KSeFException for SDK-classified failures raised by this session branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
KSeFClientClosedError — If an operation is attempted after closing the session.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncOnlineSessionClient.reference_number` (property)

reference_number: str

python

Get the online session reference number.

Returns

str — The KSeF reference number of the session.

`AsyncOnlineSessionClient.submission`

def submission(reference_number: str) -> AsyncInvoiceSubmission

python

Get the handle of an invoice sent earlier in this session.

Use it after a restart or on a resumed session to wait() for the invoice or download_upo(). No request is made until you call one.

Parameters

reference_number (str) — Invoice reference number, as exposed by the handle send_invoice() returned.

Returns

AsyncInvoiceSubmission — The invoice submission handle.

`AsyncOnlineSessionClient.send_invoice`

async def send_invoice(invoice_xml: bytes | str) -> AsyncInvoiceSubmission

python

Encrypt and submit one invoice into the open session.

Parameters

invoice_xml (bytes | str) — Invoice XML, valid against the session's schema. A str is encoded as UTF-8.

Returns

AsyncInvoiceSubmission — A handle exposing the submission's reference_number. Call its wait() for the processing result and download_upo() for the invoice UPO.

Raises

KSeFEncryptionError — If invoice encryption fails.

Example

async with auth.online_session(form_code=FormSchema.FA3) as session:
    submission = await session.send_invoice(xml)
    result = await submission.wait(timeout=60)
    print(result.ksef_number)

python

`AsyncOnlineSessionClient.send_invoice_and_wait`

def send_invoice_and_wait(
    *,
    invoice_xml: bytes,
    timeout: float = 60.0,
    poll_interval: float = 2.0,
) -> Coroutine[None, None, SessionInvoiceStatusResponse]

python

Deprecated: submit an invoice and poll until KSeF assigns a final result.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use send_invoice().wait() instead.

Parameters

invoice_xml (bytes) — Invoice XML bytes, valid against the session's schema.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between invoice status checks.

Returns

Coroutine[None, None, SessionInvoiceStatusResponse] — The final processing status, including the invoice's KSeF number once accepted.

Raises

KSeFEncryptionError — If invoice encryption fails.
KSeFInvoiceRejectedError — If invoice processing reaches a failed terminal status. It subclasses KSeFSessionError and keeps the status details and extensions.
KSeFInvoiceProcessingTimeoutError — If polling exceeds timeout.

`AsyncOnlineSessionClient.get_status`

async def get_status() -> SessionStatusResponse

python

Fetch the current state of the online session.

Returns

SessionStatusResponse — The session status, including invoice counters.

`AsyncOnlineSessionClient.list_invoices`

async def list_invoices(
    *,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Fetch one page of invoices submitted in this session.

Parameters

page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices submitted in the session, with a continuation token when more exist.

`AsyncOnlineSessionClient.get_invoice_status`

async def get_invoice_status(
    *,
    invoice_reference_number: str,
) -> SessionInvoiceStatusResponse

python

Fetch processing status for one invoice sent in this session.

Parameters

invoice_reference_number (str) — Reference number of the invoice within the session, as exposed by the handle send_invoice() returned.

Returns

SessionInvoiceStatusResponse — The processing status of the invoice.

`AsyncOnlineSessionClient.wait_for_invoice_ready`

def wait_for_invoice_ready(
    *,
    invoice_reference_number: str,
    timeout: float = 60.0,
    poll_interval: float = 2.0,
) -> Coroutine[None, None, SessionInvoiceStatusResponse]

python

Deprecated: poll invoice status until it succeeds, fails, or times out.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use send_invoice().wait() instead.

Parameters

invoice_reference_number (str) — Reference number of the invoice within the session.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between invoice status checks.

Returns

Coroutine[None, None, SessionInvoiceStatusResponse] — The final processing status, including the invoice's KSeF number once accepted.

Raises

KSeFInvoiceRejectedError — If invoice processing reaches a failed terminal status. It subclasses KSeFSessionError and keeps the status details and extensions.
KSeFInvoiceProcessingTimeoutError — If polling exceeds timeout.

`AsyncOnlineSessionClient.list_failed_invoices`

async def list_failed_invoices(
    *,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Fetch one page of invoices that failed within this session.

Parameters

page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices that failed processing, with a continuation token when more exist.

`AsyncOnlineSessionClient.download_invoice_upo`

async def download_invoice_upo(
    *,
    ksef_number: str | None = None,
    reference_number: str | None = None,
) -> bytes

python

Download the UPO of one invoice accepted in this session.

Parameters

ksef_number (str | None) — KSeF number of the invoice. Pass this or reference_number.
reference_number (str | None) — Reference number of the invoice within the session, as exposed by the handle send_invoice() returned. Pass this or ksef_number.

Returns

bytes — The UPO as XML bytes.

Raises

KSeFValidationError — If both or neither of ksef_number and reference_number are given.

`AsyncOnlineSessionClient.get_invoice_upo_by_ksef_number`

def get_invoice_upo_by_ksef_number(*, ksef_number: str) -> Coroutine[None, None, bytes]

python

Deprecated: download the invoice UPO by KSeF number.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download_invoice_upo(ksef_number=...) instead.

Parameters

ksef_number (str) — KSeF number of an invoice accepted in this session.

Returns

Coroutine[None, None, bytes] — The UPO as XML bytes.

`AsyncOnlineSessionClient.get_invoice_upo_by_reference`

def get_invoice_upo_by_reference(
    *,
    invoice_reference_number: str,
) -> Coroutine[None, None, bytes]

python

Deprecated: download the invoice UPO by session invoice reference number.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download_invoice_upo(reference_number=...) instead.

Parameters

invoice_reference_number (str) — Reference number of the invoice within the session.

Returns

Coroutine[None, None, bytes] — The UPO as XML bytes.

`AsyncOnlineSessionClient.wait`

async def wait(
    *,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> SessionStatusResponse

python

Poll until KSeF reports a terminal status for the closed session.

KSeF processes the session only after it is closed, so call this after leaving the with block or calling aclose().

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between session status checks.

Returns

SessionStatusResponse — The final successful session status, including the session UPO references.

Raises

KSeFSessionError — If the session is still open, or if KSeF reports a failed terminal status.
KSeFOnlineSessionTimeoutError — If polling exceeds timeout.

`AsyncOnlineSessionClient.download_upo`

async def download_upo() -> list[bytes]

python

Download every page of the session UPO in one pass.

Reads the session status once, resolves the UPO page references from it and downloads each page. It never waits or polls: call wait() first.

Returns

list[bytes] — The XML bytes of each UPO page, in order; empty if KSeF issued no UPO because no invoice was accepted.

Raises

KSeFSessionError — If the session is still open.
KSeFNotReadyError — If KSeF has not finished processing the closed session. Call wait() on the session, then ask again.

`AsyncOnlineSessionClient.aclose`

async def aclose() -> None

python

Terminate the online session if it is still open.

Closing an already-closed session is a no-op. A client rebuilt from saved state first asks KSeF whether the session was already closed.

Raises

KSeFApiError — If KSeF rejects the session termination request.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

`AsyncOnlineSessionClient.resume_state`

def resume_state() -> OnlineSessionResumeState

python

Return the sensitive session state needed to resume later.

Returns

OnlineSessionResumeState — The sensitive session state needed to resume later.

`AsyncOnlineSessionClient.get_state`

def get_state() -> OnlineSessionResumeState

python

Deprecated compatibility wrapper for resume_state().

Deprecated

May be removed no earlier than ksef2 2.0.0. Use resume_state() instead.

Returns

OnlineSessionResumeState — The same state as resume_state().

`AsyncOperationHandle`

Import: from ksef2.clients import AsyncOperationHandle

class AsyncOperationHandle(ABC, Generic[_TStatus, _TResult])

python

AsyncOperationHandle(reference_number: str)

python

Client-side handle to an operation that KSeF finishes asynchronously.

A handle keeps a reference to the client that started the operation and the operation's reference number. It is not a data model: it cannot be serialized, and the plain response models stay separate. Subclasses say how to read the operation's status, when it is still pending and how to turn the finished status into the result returned by wait().

Catch KSeFException for SDK-classified failures raised while waiting, and httpx.HTTPError for transport failures.

Methods

`AsyncOperationHandle.reference_number` (property)

reference_number: str

python

Get the KSeF reference number of the operation.

Returns

str — The reference number KSeF assigned to the operation.

`AsyncOperationHandle.get_status`

async def get_status() -> _TStatus

python

Fetch the current status of the operation without waiting.

Returns

_TStatus — The status as KSeF reports it right now.

`AsyncPager`

Import: from ksef2.clients import AsyncPager

class AsyncPager(Generic[_T])

python

AsyncPager(
    pages: Callable[[], AsyncGenerator[list[_T], None]],
    *,
    timeout_error: Callable[[float], BaseException] | None = None,
)

python

Lazy view over a paged collection in KSeF.

Iterating the pager yields every item across all pages, requesting pages as needed. pages() yields one list of items per page, and first_page() fetches only the first. Nothing is requested until the pager is consumed, and every iteration starts from the first page again.

Raises

KSeFApiError — If KSeF returns an API error response while a page is fetched.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncPager.pages`

def pages() -> AsyncIterator[list[_T]]

python

Iterate over the collection one page at a time.

Returns

AsyncIterator[list[_T]] — An iterator yielding the items of each page as a list, in order.

`AsyncPager.first_page`

async def first_page() -> list[_T]

python

Fetch only the first page.

Returns

list[_T] — The items of the first page; empty if the collection is empty.

`AsyncPager.wait`

async def wait(*, timeout: float = 120.0, poll_interval: float = 2.0) -> AsyncPager[_T]

python

Poll until the collection contains at least one item.

Use it when KSeF indexes items with a delay, for example invoices that were accepted a moment ago and are not searchable yet.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between checks.

Returns

AsyncPager[_T] — This pager, now known to be non-empty. Iterating it requests the pages again.

Raises

KSeFInvoiceQueryTimeoutError — For invoice searches, if no invoice appears within timeout.
TimeoutError — For other collections, if no item appears within timeout.

`AsyncPeppolClient`

Import: from ksef2.clients import AsyncPeppolClient

class AsyncPeppolClient

python

AsyncPeppolClient(transport: AsyncMiddleware)

python

Async service for querying Peppol service providers.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncPeppolClient.query`

def query(
    *,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, ListPeppolProvidersResponse]

python

Deprecated: query one page of Peppol service providers.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

params (OffsetPaginationParams | None) — Pagination parameters.

Returns

Coroutine[None, None, ListPeppolProvidersResponse] — One page of providers with pagination info.

`AsyncPeppolClient.all`

async def all(
    *,
    params: OffsetPaginationParams | None = None,
) -> AsyncIterator[PeppolProvider]

python

Deprecated: iterate over all Peppol service providers.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead.

Parameters

params (OffsetPaginationParams | None) — Pagination parameters.

Yields

AsyncIterator[PeppolProvider] — Each Peppol service provider, across all pages.

`AsyncPeppolClient.list`

def list(*, params: OffsetPaginationParams | None = None) -> AsyncPager[PeppolProvider]

python

List the registered Peppol service providers.

Nothing is requested until the result is consumed. Iterate it for every provider, call pages() for page-sized lists or first_page() for one request only.

Parameters

params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[PeppolProvider] — A paging object over the Peppol service providers.

Example

async for provider in auth.peppol.list():
    print(provider.id, provider.name)

python

`AsyncPermissionOperation`

Import: from ksef2.clients import AsyncPermissionOperation

class AsyncPermissionOperation(AsyncOperationHandle[PermissionOperationStatusResponse, PermissionOperationStatusResponse])

python

AsyncPermissionOperation(
    client: AsyncPermissionsClient,
    response: GrantPermissionsResponse,
)

python

Handle to a permission grant or revoke that KSeF applies asynchronously.

Returned by every grant_*() and revoke*() call of auth.permissions. It exposes every field of the response, for example reference_number, and wait() polls the operation until KSeF has finished it. get_operation_status() on the permissions client reads the same status for a reference number you stored earlier.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncPermissionOperation.response` (property)

response: GrantPermissionsResponse

python

Get the plain operation response.

Returns

GrantPermissionsResponse — The data model KSeF returned when it accepted the operation.

`AsyncPermissionOperation.get_status`

async def get_status() -> PermissionOperationStatusResponse

python

Fetch the operation's current status without waiting.

Returns

PermissionOperationStatusResponse — The operation status; status.code is 200 once it has completed successfully.

`AsyncPermissionOperation.wait`

async def wait(
    *,
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> PermissionOperationStatusResponse

python

Poll until KSeF has finished the operation.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between status checks.

Returns

PermissionOperationStatusResponse — The final status, with status.code 200.

Raises

KSeFPermissionOperationFailedError — If KSeF finishes the operation without applying it.
KSeFPermissionOperationTimeoutError — If polling exceeds timeout.

`AsyncPermissionsClient`

Import: from ksef2.clients import AsyncPermissionsClient

class AsyncPermissionsClient

python

AsyncPermissionsClient(transport: AsyncMiddleware)

python

Async high-level API for permission grants, revocations, and queries.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncPermissionsClient.grant_person`

async def grant_person(
    *,
    subject_type: CertificateSubjectIdentifierType,
    subject_value: str,
    permissions: list[PersonPermissionScope],
    description: str,
    first_name: str,
    last_name: str,
) -> AsyncPermissionOperation

python

Grant person-scoped permissions to a subject identifier.

Parameters

subject_type (CertificateSubjectIdentifierType) — Kind of identifier in subject_value (nip, pesel or fingerprint).
subject_value (str) — Identifier of the person receiving the permissions.
permissions (list[PersonPermissionScope]) — Permission scopes to grant; at least one is required.
description (str) — Free-text reason for the grant.
first_name (str) — First name of the person receiving the permissions.
last_name (str) — Last name of the person receiving the permissions.

Returns

AsyncPermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`AsyncPermissionsClient.grant_entity`

async def grant_entity(
    *,
    subject_value: str,
    permissions: list[EntityPermission],
    description: str,
    entity_name: str,
) -> AsyncPermissionOperation

python

Grant entity permissions to a NIP-identified entity.

Parameters

subject_value (str) — NIP of the entity receiving the permissions.
permissions (list[EntityPermission]) — Permissions to grant, each with its delegation flag.
description (str) — Free-text reason for the grant.
entity_name (str) — Full name of the entity receiving the permissions.

Returns

AsyncPermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`AsyncPermissionsClient.grant_authorization`

async def grant_authorization(
    *,
    subject_type: AuthorizationSubjectIdentifierType,
    subject_value: str,
    permission: AuthorizationPermissionType,
    description: str,
    entity_name: str,
) -> AsyncPermissionOperation

python

Grant an authorization permission such as self-invoicing.

Parameters

subject_type (AuthorizationSubjectIdentifierType) — Kind of identifier in subject_value (nip or peppol_id).
subject_value (str) — Identifier of the entity being authorized (NIP or Peppol ID).
permission (AuthorizationPermissionType) — Authorization scope to grant, for example self_invoicing.
description (str) — Free-text reason for the grant.
entity_name (str) — Full name of the entity being authorized.

Returns

AsyncPermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`AsyncPermissionsClient.grant_indirect`

async def grant_indirect(
    *,
    subject_type: CertificateSubjectIdentifierType,
    subject_value: str,
    permissions: list[IndirectPermissionType],
    description: str,
    first_name: str,
    last_name: str,
    target_type: IndirectTargetIdentifierType | None = None,
    target_value: str | None = None,
) -> AsyncPermissionOperation

python

Grant indirect permissions, optionally limited to a target entity.

Parameters

subject_type (CertificateSubjectIdentifierType) — Kind of identifier in subject_value (nip, pesel or fingerprint).
subject_value (str) — Identifier of the person receiving the indirect permissions.
permissions (list[IndirectPermissionType]) — Indirect permission scopes to grant; at least one is required.
description (str) — Free-text reason for the grant.
first_name (str) — First name of the person receiving the permissions.
last_name (str) — Last name of the person receiving the permissions.
target_type (IndirectTargetIdentifierType | None) — Kind of identifier in target_value; None or all_partners scopes the grant to every partner.
target_value (str | None) — Identifier of the partner context the grant is limited to; None when the target is all partners.

Returns

AsyncPermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`AsyncPermissionsClient.grant_subunit`

async def grant_subunit(
    *,
    subject_type: CertificateSubjectIdentifierType,
    subject_value: str,
    context_type: SubunitIdentifierType,
    context_value: str,
    description: str,
    first_name: str,
    last_name: str,
    subunit_name: str | None = None,
) -> AsyncPermissionOperation

python

Grant permissions within a subunit context.

Parameters

subject_type (CertificateSubjectIdentifierType) — Kind of identifier in subject_value (nip, pesel or fingerprint).
subject_value (str) — Identifier of the person receiving the permissions.
context_type (SubunitIdentifierType) — Kind of identifier in context_value (nip or internal_id).
context_value (str) — Identifier of the subunit context the permissions apply to.
description (str) — Free-text reason for the grant.
first_name (str) — First name of the person receiving the permissions.
last_name (str) — Last name of the person receiving the permissions.
subunit_name (str | None) — Display name of the subunit; optional.

Returns

AsyncPermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`AsyncPermissionsClient.grant_eu_entity`

async def grant_eu_entity(
    *,
    subject_value: str,
    permissions: list[EuEntityPermissionType],
    description: str,
) -> AsyncPermissionOperation

python

Grant permissions to an EU entity identified by fingerprint data.

Parameters

subject_value (str) — Fingerprint of the certificate of the person receiving the permissions.
permissions (list[EuEntityPermissionType]) — EU-entity permission scopes to grant.
description (str) — Free-text reason for the grant.

Returns

AsyncPermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`AsyncPermissionsClient.grant_eu_entity_administration`

async def grant_eu_entity_administration(
    *,
    subject_value: str,
    context_type: EuEntityAdminContextIdentifierType,
    context_value: str,
    description: str,
    eu_entity_name: str,
) -> AsyncPermissionOperation

python

Grant administration rights for an EU entity in a VAT UE context.

Parameters

subject_value (str) — Fingerprint of the certificate of the person receiving administration rights.
context_type (EuEntityAdminContextIdentifierType) — Kind of identifier in context_value; always nip_vat_ue.
context_value (str) — NIP-VAT UE identifier of the EU entity.
description (str) — Free-text reason for the grant.
eu_entity_name (str) — Full name of the EU entity.

Returns

AsyncPermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`AsyncPermissionsClient.revoke_authorization`

async def revoke_authorization(*, permission_id: str) -> AsyncPermissionOperation

python

Revoke an authorization permission by permission id.

Parameters

permission_id (str) — Identifier of the authorization permission, from a query result.

Returns

AsyncPermissionOperation — A handle to the asynchronous revoke operation; call wait() to wait until KSeF has applied it.

`AsyncPermissionsClient.revoke`

async def revoke(*, permission_id: str) -> AsyncPermissionOperation

python

Revoke a non-authorization permission by permission id.

Use revoke_authorization() for authorization permissions such as self-invoicing.

Parameters

permission_id (str) — Identifier of the permission, from a query result.

Returns

AsyncPermissionOperation — A handle to the asynchronous revoke operation; call wait() to wait until KSeF has applied it.

Example

operation = await auth.permissions.revoke(permission_id=permission.id)
await operation.wait()

python

`AsyncPermissionsClient.revoke_common`

def revoke_common(
    *,
    permission_id: str,
) -> Coroutine[None, None, GrantPermissionsResponse]

python

Deprecated: revoke a non-authorization permission by permission id.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use revoke() instead.

Parameters

permission_id (str) — Identifier of the permission, from a query result.

Returns

Coroutine[None, None, GrantPermissionsResponse] — The reference of the asynchronous revoke operation; poll get_operation_status() for the result.

`AsyncPermissionsClient.get_attachment_permission_status`

async def get_attachment_permission_status() -> AttachmentPermissionStatus

python

Return whether attachments are currently allowed for the subject.

Returns

AttachmentPermissionStatus — Whether attachments are currently allowed for the subject.

`AsyncPermissionsClient.get_operation_status`

async def get_operation_status(
    *,
    reference_number: str,
) -> PermissionOperationStatusResponse

python

Fetch the status of an asynchronous permission operation.

Parameters

reference_number (str) — Reference number returned by a grant or revoke call.

Returns

PermissionOperationStatusResponse — The operation status; status.code is 200 once it has completed successfully.

`AsyncPermissionsClient.get_entity_roles`

def get_entity_roles(
    *,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, EntityRolesResponse]

python

Deprecated: fetch one page of roles assigned to the authenticated entity.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_entity_roles() instead; first_page() fetches one page.

Parameters

params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

Coroutine[None, None, EntityRolesResponse] — One page of roles assigned to the authenticated entity.

`AsyncPermissionsClient.query_authorizations`

def query_authorizations(
    *,
    query: AuthorizationPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, AuthorizationPermissionsQueryResponse]

python

Deprecated: fetch one page of authorization grants matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_authorizations() instead; first_page() fetches one page.

Parameters

query (AuthorizationPermissionsQuery) — Filters for the authorization grants.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

Coroutine[None, None, AuthorizationPermissionsQueryResponse] — One page of authorization grants.

`AsyncPermissionsClient.query_entities`

def query_entities(
    *,
    query: EntityPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, EntityPermissionsQueryResponse]

python

Deprecated: fetch one page of entity permission grants matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_entities() instead; first_page() fetches one page.

Parameters

query (EntityPermissionsQuery) — Filters for the entity permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

Coroutine[None, None, EntityPermissionsQueryResponse] — One page of entity permissions.

`AsyncPermissionsClient.query_eu_entities`

def query_eu_entities(
    *,
    query: EuEntityPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, EuEntityPermissionsQueryResponse]

python

Deprecated: fetch one page of EU-entity permissions matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_eu_entities() instead; first_page() fetches one page.

Parameters

query (EuEntityPermissionsQuery) — Filters for the EU-entity permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

Coroutine[None, None, EuEntityPermissionsQueryResponse] — One page of EU-entity permissions.

`AsyncPermissionsClient.query_personal`

def query_personal(
    *,
    query: PersonalPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, PersonalPermissionsQueryResponse]

python

Deprecated: fetch one page of permissions held by the authenticated subject.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_personal() instead; first_page() fetches one page.

Parameters

query (PersonalPermissionsQuery) — Filters for the permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

Coroutine[None, None, PersonalPermissionsQueryResponse] — One page of permissions held by the authenticated subject.

`AsyncPermissionsClient.query_persons`

def query_persons(
    *,
    query: PersonPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, PersonPermissionsQueryResponse]

python

Deprecated: fetch one page of person permission grants matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_persons() instead; first_page() fetches one page.

Parameters

query (PersonPermissionsQuery) — Filters for the person permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

Coroutine[None, None, PersonPermissionsQueryResponse] — One page of person permissions.

`AsyncPermissionsClient.query_subordinate_entities`

def query_subordinate_entities(
    *,
    query: SubordinateEntityRolesQuery,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, SubordinateEntityRolesQueryResponse]

python

Deprecated: fetch one page of subordinate entity roles.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_subordinate_entities() instead; first_page() fetches one page.

Parameters

query (SubordinateEntityRolesQuery) — Filters for the subordinate entity roles.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

Coroutine[None, None, SubordinateEntityRolesQueryResponse] — One page of subordinate entity roles.

`AsyncPermissionsClient.query_subunits`

def query_subunits(
    *,
    query: SubunitPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> Coroutine[None, None, SubunitPermissionsQueryResponse]

python

Deprecated: fetch one page of subunit permission grants.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_subunits() instead; first_page() fetches one page.

Parameters

query (SubunitPermissionsQuery) — Filters for the subunit permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

Coroutine[None, None, SubunitPermissionsQueryResponse] — One page of subunit permissions.

`AsyncPermissionsClient.list_entity_roles`

def list_entity_roles(
    *,
    params: OffsetPaginationParams | None = None,
) -> AsyncPager[EntityRole]

python

List the roles assigned to the authenticated entity.

Nothing is requested until the result is consumed. Iterate it for every role, call pages() for page-sized lists or first_page() for one request only.

Parameters

params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[EntityRole] — A paging object over the roles assigned to the authenticated entity.

Example

async for role in auth.permissions.list_entity_roles():
    print(role.role)

python

`AsyncPermissionsClient.list_authorizations`

def list_authorizations(
    query: AuthorizationPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> AsyncPager[AuthorizationGrantDetail]

python

List the authorization grants matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every grant, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (AuthorizationPermissionsQuery) — Filters for the authorization grants.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[AuthorizationGrantDetail] — A paging object over the authorization grants.

Example

async for grant in auth.permissions.list_authorizations(query):
    print(grant.id)

python

`AsyncPermissionsClient.list_entities`

def list_entities(
    query: EntityPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> AsyncPager[EntityPermissionDetail]

python

List the entity permission grants matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every grant, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (EntityPermissionsQuery) — Filters for the entity permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[EntityPermissionDetail] — A paging object over the entity permissions.

Example

async for permission in auth.permissions.list_entities(query):
    print(permission.id)

python

`AsyncPermissionsClient.list_eu_entities`

def list_eu_entities(
    query: EuEntityPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> AsyncPager[EuEntityPermission]

python

List the EU-entity permissions matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every permission, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (EuEntityPermissionsQuery) — Filters for the EU-entity permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[EuEntityPermission] — A paging object over the EU-entity permissions.

Example

async for permission in auth.permissions.list_eu_entities(query):
    print(permission.id)

python

`AsyncPermissionsClient.list_personal`

def list_personal(
    query: PersonalPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> AsyncPager[PersonalPermissionDetail]

python

List the permissions held by the authenticated subject.

Nothing is requested until the result is consumed. Iterate it for every permission, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (PersonalPermissionsQuery) — Filters for the permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[PersonalPermissionDetail] — A paging object over the permissions held by the authenticated subject.

Example

async for permission in auth.permissions.list_personal(query):
    print(permission.id)

python

`AsyncPermissionsClient.list_persons`

def list_persons(
    query: PersonPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> AsyncPager[PersonPermissionDetail]

python

List the person permission grants matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every grant, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (PersonPermissionsQuery) — Filters for the person permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[PersonPermissionDetail] — A paging object over the person permissions.

Example

async for permission in auth.permissions.list_persons(query):
    print(permission.id)

python

`AsyncPermissionsClient.list_subordinate_entities`

def list_subordinate_entities(
    query: SubordinateEntityRolesQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> AsyncPager[SubordinateEntityRoleDetail]

python

List the subordinate entity roles matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every role, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (SubordinateEntityRolesQuery) — Filters for the subordinate entity roles.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[SubordinateEntityRoleDetail] — A paging object over the subordinate entity roles.

Example

async for role in auth.permissions.list_subordinate_entities(query):
    print(role.role)

python

`AsyncPermissionsClient.list_subunits`

def list_subunits(
    query: SubunitPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> AsyncPager[SubunitPermission]

python

List the subunit permission grants matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every grant, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (SubunitPermissionsQuery) — Filters for the subunit permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

AsyncPager[SubunitPermission] — A paging object over the subunit permissions.

Example

async for permission in auth.permissions.list_subunits(query):
    print(permission.id)

python

`AsyncSessionManagementClient`

Import: from ksef2.clients import AsyncSessionManagementClient

class AsyncSessionManagementClient

python

AsyncSessionManagementClient(transport: AsyncMiddleware)

python

Async manage authentication sessions opened through the auth API.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncSessionManagementClient.query`

def query(
    *,
    page_size: int | None = None,
    continuation_token: str | None = None,
) -> Coroutine[None, None, AuthenticationSessionsResponse]

python

Deprecated: fetch one page of authentication sessions.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

page_size (int | None) — Maximum number of sessions to request from KSeF.
continuation_token (str | None) — Cursor returned by a previous page.

Returns

Coroutine[None, None, AuthenticationSessionsResponse] — One page of authentication sessions for the current subject.

`AsyncSessionManagementClient.all`

async def all(
    *,
    page_size: int | None = None,
) -> AsyncIterator[AuthenticationSessionsResponse]

python

Deprecated: iterate through all authentication session pages.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list().pages() instead.

Parameters

page_size (int | None) — Maximum number of sessions to request per page.

Yields

AsyncIterator[AuthenticationSessionsResponse] — Successive pages of authentication sessions until KSeF stops
AsyncIterator[AuthenticationSessionsResponse] — returning a continuation token.

`AsyncSessionManagementClient.terminate_current`

async def terminate_current() -> None

python

Terminate the authentication session backing the current bearer token.

`AsyncSessionManagementClient.terminate`

async def terminate(reference_number: str) -> None

python

Terminate an authentication session by reference number.

Parameters

reference_number (str) — Reference number of the authentication session to terminate.

`AsyncSessionManagementClient.close`

def close(*, reference_number: str) -> Coroutine[None, None, None]

python

Deprecated: terminate an authentication session by reference number.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use terminate() instead.

Parameters

reference_number (str) — Reference number of the authentication session to terminate.

Returns

Coroutine[None, None, None] — None once the session is terminated.

`AsyncSessionManagementClient.list`

def list(*, page_size: int | None = None) -> AsyncPager[AuthenticationSession]

python

List the authentication sessions of the current subject.

Nothing is requested until the result is consumed. Iterate it for every session, call pages() for page-sized lists or first_page() for one request only.

Parameters

page_size (int | None) — Maximum number of sessions to request per page; the KSeF default when None.

Returns

AsyncPager[AuthenticationSession] — A paging object over the authentication sessions.

Example

async for session in auth.sessions.list():
    print(session.reference_number)

python

`AsyncTemporalTestData`

Import: from ksef2.clients import AsyncTemporalTestData

class AsyncTemporalTestData

python

AsyncTemporalTestData(client: AsyncTestDataClient)

python

Async context manager that records testdata mutations and reverts them on exit.

Cleanup failures are logged and suppressed. Direct mutation methods expose the same catch contract as the parent test-data branch.

Raises

KSeFApiError — If KSeF returns an API error response during direct mutations.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncTemporalTestData.create_subject`

async def create_subject(
    *,
    nip: str,
    subject_type: SubjectType,
    description: str,
    subunits: list[SubUnit] | None = None,
    created_date: datetime | None = None,
) -> None

python

Create a TEST subject and register it for context-manager cleanup.

Parameters

nip (str) — NIP of the subject to create.
subject_type (SubjectType) — Kind of subject, such as an enforcement authority or a VAT group.
description (str) — Description of the subject.
subunits (list[SubUnit] | None) — Subunits to create with the subject; None for none.
created_date (datetime | None) — Creation date to record for the subject; None for now.

`AsyncTemporalTestData.delete_subject`

async def delete_subject(*, nip: str) -> None

python

Delete a TEST subject and remove it from cleanup tracking.

Parameters

nip (str) — NIP of the subject to delete.

`AsyncTemporalTestData.create_person`

async def create_person(
    *,
    nip: str,
    pesel: str,
    description: str,
    is_bailiff: bool = False,
    is_deceased: bool = False,
    created_date: datetime | None = None,
) -> None

python

Create a TEST person and register it for context-manager cleanup.

Parameters

nip (str) — NIP of the person.
pesel (str) — PESEL of the person.
description (str) — Description of the person.
is_bailiff (bool) — Whether the person is a bailiff.
is_deceased (bool) — Whether the person is marked as deceased.
created_date (datetime | None) — Creation date to record for the person; None for now.

`AsyncTemporalTestData.delete_person`

async def delete_person(*, nip: str) -> None

python

Delete a TEST person and remove it from cleanup tracking.

Parameters

nip (str) — NIP of the person to delete.

`AsyncTemporalTestData.grant_permissions`

async def grant_permissions(
    *,
    permissions: list[Permission],
    grant_to: Identifier,
    in_context_of: Identifier,
) -> None

python

Grant TEST permissions and register them for cleanup.

Parameters

permissions (list[Permission]) — Permissions to grant.
grant_to (Identifier) — Identifier of the subject receiving the permissions.
in_context_of (Identifier) — Identifier of the context in which the permissions apply.

`AsyncTemporalTestData.revoke_permissions`

async def revoke_permissions(
    *,
    revoke_from: Identifier,
    in_context_of: Identifier,
) -> None

python

Revoke TEST permissions and remove them from cleanup tracking.

Parameters

revoke_from (Identifier) — Identifier of the subject losing the permissions.
in_context_of (Identifier) — Identifier of the context from which the permissions are revoked.

`AsyncTemporalTestData.enable_attachments`

async def enable_attachments(*, nip: str) -> None

python

Enable TEST attachments and register the context for cleanup.

Parameters

nip (str) — NIP of the subject to enable attachments for.

`AsyncTemporalTestData.revoke_attachments`

async def revoke_attachments(*, nip: str, expected_end_date: date | None = None) -> None

python

Revoke TEST attachment access and remove it from cleanup tracking.

Parameters

nip (str) — NIP of the subject to revoke attachments for.
expected_end_date (date | None) — Date on which attachment permission should end; None to revoke immediately.

`AsyncTemporalTestData.block_context`

async def block_context(*, context: AuthContextIdentifier) -> None

python

Block a TEST context and register it for cleanup.

Parameters

context (AuthContextIdentifier) — Authentication context to block.

`AsyncTemporalTestData.unblock_context`

async def unblock_context(*, context: AuthContextIdentifier) -> None

python

Unblock a TEST context and remove it from cleanup tracking.

Parameters

context (AuthContextIdentifier) — Authentication context to unblock.

`AsyncTestDataClient`

Import: from ksef2.clients import AsyncTestDataClient

class AsyncTestDataClient

python

AsyncTestDataClient(transport: AsyncMiddleware)

python

Async helper for TEST-only endpoints that seed and mutate sandbox data.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncTestDataClient.create_subject`

async def create_subject(
    *,
    nip: str,
    subject_type: SubjectType,
    description: str,
    subunits: list[SubUnit] | None = None,
    created_date: datetime | None = None,
) -> None

python

Create a test subject with optional subunits.

Parameters

nip (str) — NIP of the subject to create.
subject_type (SubjectType) — Kind of subject, such as an enforcement authority or a VAT group.
description (str) — Description of the subject.
subunits (list[SubUnit] | None) — Subunits to create with the subject; None for none.
created_date (datetime | None) — Creation date to record for the subject; None for now.

`AsyncTestDataClient.delete_subject`

async def delete_subject(*, nip: str) -> None

python

Delete a test subject by NIP.

Parameters

nip (str) — NIP of the subject to delete.

`AsyncTestDataClient.create_person`

async def create_person(
    *,
    nip: str,
    pesel: str,
    description: str,
    is_bailiff: bool = False,
    is_deceased: bool = False,
    created_date: datetime | None = None,
) -> None

python

Create a test person in the chosen subject.

Parameters

nip (str) — NIP of the person.
pesel (str) — PESEL of the person.
description (str) — Description of the person.
is_bailiff (bool) — Whether the person is a bailiff.
is_deceased (bool) — Whether the person is marked as deceased.
created_date (datetime | None) — Creation date to record for the person; None for now.

`AsyncTestDataClient.delete_person`

async def delete_person(*, nip: str) -> None

python

Delete a test person by subject NIP.

Parameters

nip (str) — NIP of the person to delete.

`AsyncTestDataClient.grant_permissions`

async def grant_permissions(
    *,
    permissions: list[Permission],
    grant_to: Identifier,
    in_context_of: Identifier,
) -> None

python

Grant test permissions in a chosen context.

Parameters

permissions (list[Permission]) — Permissions to grant.
grant_to (Identifier) — Identifier of the subject receiving the permissions.
in_context_of (Identifier) — Identifier of the context in which the permissions apply.

`AsyncTestDataClient.revoke_permissions`

async def revoke_permissions(
    *,
    revoke_from: Identifier,
    in_context_of: Identifier,
) -> None

python

Revoke test permissions in a chosen context.

Parameters

revoke_from (Identifier) — Identifier of the subject losing the permissions.
in_context_of (Identifier) — Identifier of the context from which the permissions are revoked.

`AsyncTestDataClient.enable_attachments`

async def enable_attachments(*, nip: str) -> None

python

Enable invoice attachments for a test subject.

Parameters

nip (str) — NIP of the subject to enable attachments for.

`AsyncTestDataClient.revoke_attachments`

async def revoke_attachments(*, nip: str, expected_end_date: date | None = None) -> None

python

Revoke attachment permissions, optionally effective on a given date.

Parameters

nip (str) — NIP of the subject to revoke attachments for.
expected_end_date (date | None) — Date on which attachment permission should end; None to revoke immediately.

`AsyncTestDataClient.block_context`

async def block_context(*, context: AuthContextIdentifier) -> None

python

Block authentication for a specific test context.

Parameters

context (AuthContextIdentifier) — Authentication context to block.

`AsyncTestDataClient.unblock_context`

async def unblock_context(*, context: AuthContextIdentifier) -> None

python

Unblock authentication for a specific test context.

Parameters

context (AuthContextIdentifier) — Authentication context to unblock.

`AsyncTestDataClient.update_certificate_valid_to`

async def update_certificate_valid_to(*, serial_number: str, valid_to: datetime) -> None

python

Shorten the validity of a certificate in the TEST environment.

Parameters

serial_number (str) — Serial number of the certificate to shorten.
valid_to (datetime) — New, earlier end of the certificate validity period.

`AsyncTestDataClient.temporal`

def temporal() -> AsyncTemporalTestData

python

Return a context manager that automatically cleans up created test data.

Returns

AsyncTemporalTestData — A context manager that automatically cleans up created test data.

`AsyncTokensClient`

Import: from ksef2.clients import AsyncTokensClient

class AsyncTokensClient

python

AsyncTokensClient(transport: AsyncMiddleware)

python

Async high-level API for the KSeF token lifecycle.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AsyncTokensClient.wait_for_activation`

def wait_for_activation(
    *,
    reference_number: str,
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> Coroutine[None, None, TokenStatusResponse]

python

Deprecated: wait until a generated token becomes active or reaches a terminal state.

Deprecated

May be removed no earlier than ksef2 2.0.0. Call wait() on the handle returned by generate() instead.

Parameters

reference_number (str) — Reference returned by generate.
timeout (float) — Maximum number of seconds to wait for activation.
poll_interval (float) — Delay in seconds between status checks.

Returns

Coroutine[None, None, TokenStatusResponse] — The first status response that reports the token as active.

Raises

KSeFApiError — If activation ends in a terminal failure state.
KSeFTokenStatusTimeoutError — If polling exceeds timeout.
httpx.HTTPError — If a status request fails at the transport boundary.

`AsyncTokensClient.generate`

async def generate(
    *,
    permissions: builtins.list[TokenPermission],
    description: str,
) -> AsyncGeneratedToken

python

Create a token and return a handle exposing its one-time credential.

Parameters

permissions (builtins.list[TokenPermission]) — Permissions to include in the generated token.
description (str) — Human-readable label shown in KSeF token listings.

Returns

AsyncGeneratedToken — A handle whose token is the one-time secret KSeF returns only once;
AsyncGeneratedToken — persist it before calling the handle's wait() for activation.

Example

token = await auth.tokens.generate(
    permissions=["invoice_read"], description="reporting"
)
secret = token.token  # store it now, KSeF will not show it again
await token.wait()

python

`AsyncTokensClient.list_page`

def list_page(
    *,
    continuation_token: str | None = None,
    params: TokenListParams | None = None,
) -> Coroutine[None, None, QueryTokensResponse]

python

Deprecated: fetch one page of tokens using optional filters and continuation state.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

continuation_token (str | None) — Token identifying the next page to fetch.
params (TokenListParams | None) — Optional filters and page size for the request.

Returns

Coroutine[None, None, QueryTokensResponse] — A single page of token results.

`AsyncTokensClient.list_all`

async def list_all(
    *,
    params: TokenListParams | None = None,
) -> AsyncIterator[QueryTokensResponse]

python

Deprecated: iterate through all token pages until KSeF stops returning a continuation token.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list().pages() instead.

Parameters

params (TokenListParams | None) — Optional filters and page size applied to every request.

Yields

AsyncIterator[QueryTokensResponse] — Each page returned by the token listing endpoint.

`AsyncTokensClient.get_status`

async def get_status(*, reference_number: str) -> TokenStatusResponse

python

Return the current status of a token.

Parameters

reference_number (str) — Reference number of the token.

Returns

TokenStatusResponse — The token's current lifecycle status.

`AsyncTokensClient.status`

def status(*, reference_number: str) -> Coroutine[None, None, TokenStatusResponse]

python

Deprecated: return the current status of a token.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use get_status() instead.

Parameters

reference_number (str) — Reference number of the token.

Returns

Coroutine[None, None, TokenStatusResponse] — The token's current lifecycle status.

`AsyncTokensClient.revoke`

async def revoke(*, reference_number: str) -> None

python

Revoke a token.

Parameters

reference_number (str) — Reference number of the token to revoke.

`AsyncTokensClient.list`

def list(*, params: TokenListParams | None = None) -> AsyncPager[TokenInfo]

python

List the tokens of the authenticated context.

Nothing is requested until the result is consumed. Iterate it for every token, call pages() for page-sized lists or first_page() for one request only.

Parameters

params (TokenListParams | None) — Optional filters and page size applied to every request.

Returns

AsyncPager[TokenInfo] — A paging object over the tokens matching params.

Example

async for token in auth.tokens.list():
    print(token.reference_number, token.status)

python

`AuthClient`

Import: from ksef2.clients import AuthClient

class AuthClient

python

AuthClient(
    transport: Middleware,
    certificate_store: CertificateStoreProtocol,
    environment: Environment = Environment.PRODUCTION,
    transfer_transport: Middleware | None = None,
    auto_refresh_tokens: bool = True,
)

python

Entry point for creating authenticated SDK clients.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AuthClient.resume`

def resume(state: AuthenticationResumeState | str) -> AuthenticatedClient

python

Rehydrate an authenticated client from saved authentication state.

The client refreshes its access token automatically unless TransportConfig.auto_refresh_tokens is False, so a state whose access token has expired still works while its refresh token is valid.

Parameters

state (AuthenticationResumeState | str) — State previously exported from an authenticated client, or its JSON string.

Returns

AuthenticatedClient — An authenticated client bound to the saved tokens.

`AuthClient.with_token`

def with_token(
    *,
    ksef_token: str,
    nip: str,
    context_type: ContextIdentifierType = 'nip',
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> AuthenticatedClient

python

Authenticate with a KSeF token and return an authenticated client.

Parameters

ksef_token (str) — Token generated in KSeF for the target context.
nip (str) — Context identifier value used for authentication.
context_type (ContextIdentifierType) — Type of identifier represented by nip.
timeout (float) — Maximum number of seconds to wait for authentication.
poll_interval (float) — Delay in seconds between authentication status checks.

Returns

AuthenticatedClient — An authenticated client with redeemed access and refresh tokens.

Raises

NoCertificateAvailableError — If no valid token-encryption certificate is available.
KSeFEncryptionError — If token encryption fails.
KSeFAuthError — If authentication fails.
KSeFAuthPollingTimeoutError — If polling exceeds timeout.

Example

client = Client(Environment.TEST)
auth = client.authentication.with_token(
    ksef_token="your-ksef-token",
    nip="5261040828",
)

python

`AuthClient.with_xades`

def with_xades(
    *,
    nip: str,
    cert: Certificate,
    private_key: XAdESPrivateKey,
    verify_chain: bool = False,
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> AuthenticatedClient

python

Authenticate with an XAdES-signed challenge response.

Parameters

nip (str) — Taxpayer identifier embedded in the authentication request.
cert (Certificate) — Signing certificate accepted by the target environment.
private_key (XAdESPrivateKey) — RSA or EC private key used to sign the XAdES payload.
verify_chain (bool) — Whether KSeF should verify the certificate chain.
timeout (float) — Maximum number of seconds to wait for authentication.
poll_interval (float) — Delay in seconds between authentication status checks.

Returns

AuthenticatedClient — An authenticated client with redeemed access and refresh tokens.

Raises

KSeFAuthError — If authentication fails.
KSeFAuthPollingTimeoutError — If polling exceeds timeout.

Example

from ksef2.xades import load_certificate_from_pem, load_private_key_from_pem

client = Client(Environment.DEMO)
auth = client.authentication.with_xades(
    nip="5261040828",
    cert=load_certificate_from_pem("company.pem"),
    private_key=load_private_key_from_pem("company.key"),
)

python

`AuthClient.with_test_certificate`

def with_test_certificate(
    *,
    nip: str,
    verify_chain: bool = False,
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> AuthenticatedClient

python

Authenticate in the TEST environment using an SDK-generated certificate.

Parameters

nip (str) — NIP to authenticate as; the generated certificate is issued for it.
verify_chain (bool) — Whether KSeF should verify the certificate chain.
timeout (float) — Maximum number of seconds to wait for authentication.
poll_interval (float) — Delay in seconds between authentication status checks.

Returns

AuthenticatedClient — An authenticated client for the NIP's context.

Raises

KSeFUnsupportedEnvironmentError — If the client is not configured for TEST.
KSeFAuthError — If authentication fails.
KSeFAuthPollingTimeoutError — If polling exceeds timeout.

Example

client = Client(Environment.TEST)
auth = client.authentication.with_test_certificate(nip="5261040828")

python

`AuthClient.with_profile`

def with_profile(
    name: str | None = None,
    *,
    config_path: str | Path | None = None,
    timeout: float | None = None,
    poll_interval: float | None = None,
    verify_chain: bool = False,
) -> AuthenticatedClient

python

Authenticate with a local ksef2-cli profile.

Profile selection follows the CLI order: explicit name, KSEF2_PROFILE, then active_profile from the local CLI config. The profile environment must match the root client environment.

Parameters

name (str | None) — Profile name; when None, KSEF2_PROFILE and then the CLI's active profile are used.
config_path (str | Path | None) — Path of the ksef2-cli config file; the default location is used when None.
timeout (float | None) — Maximum number of seconds to wait for authentication; defaults to the profile's polling settings, else 60.
poll_interval (float | None) — Delay in seconds between authentication status checks; defaults to the profile's value, else 1.
verify_chain (bool) — Whether KSeF should verify the certificate chain, for certificate-based profiles.

Returns

AuthenticatedClient — An authenticated client for the profile's context.

Raises

KSeFValidationError — If the profile's environment differs from the client's, or the profile is missing a required secret.
KSeFAuthError — If authentication fails.
KSeFAuthPollingTimeoutError — If polling exceeds timeout.

Example

client = Client(Environment.TEST)
auth = client.authentication.with_profile("test-company")

python

`AuthClient.refresh`

def refresh(*, refresh_token: str) -> RefreshedToken

python

Exchange a refresh token for a new access token.

Parameters

refresh_token (str) — Refresh token obtained during authentication.

Returns

RefreshedToken — A new access token with its expiry time.

`AuthenticatedClient`

Import: from ksef2.clients import AuthenticatedClient

class AuthenticatedClient

python

AuthenticatedClient(
    transport: Middleware,
    auth_tokens: AuthTokens,
    certificate_store: CertificateStoreProtocol,
    environment: Environment = Environment.PRODUCTION,
    transfer_transport: Middleware | None = None,
    refresh_access_token: Refresh | None = None,
)

python

Authenticated entry point for KSeF operations.

Catch KSeFException for SDK-classified failures raised by authenticated branches, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`AuthenticatedClient.auth_tokens` (property)

auth_tokens: AuthTokens

python

Return the authenticated token pair used by this client branch.

Reflects automatic refreshes: after the access token is renewed, this holds the new access token alongside the original refresh token.

Returns

AuthTokens — The current authenticated token pair of this client branch.

`AuthenticatedClient.access_token` (property)

access_token: str

python

Return the bearer access token string used for authenticated calls.

Returns

str — The bearer access token string used for authenticated calls.

`AuthenticatedClient.refresh_token` (property)

refresh_token: str

python

Return the refresh token string paired with the access token.

Returns

str — The refresh token string paired with the access token.

`AuthenticatedClient.resume_state`

def resume_state() -> AuthenticationResumeState

python

Return the authentication state needed to rehydrate this branch later.

The state holds the current tokens, so it includes any access token obtained through automatic refresh.

Returns

AuthenticationResumeState — The authentication state needed to rehydrate this branch later.

`AuthenticatedClient.get_encryption_key`

def get_encryption_key() -> tuple[bytes, bytes, bytes]

python

Generate a session AES key, IV, and encrypted symmetric key payload.

Returns

tuple[bytes, bytes, bytes] — A tuple of the raw AES key, the initialization vector and the AES key encrypted with the KSeF public key.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If symmetric-key encryption fails.

`AuthenticatedClient.online_session`

def online_session(
    *,
    form_code: FormSchema | None = None,
    state: OnlineSessionResumeState | str | None = None,
) -> OnlineSessionClient

python

Open a new online invoice session, or resume one from saved state.

Parameters

form_code (FormSchema | None) — Invoice schema the new session accepts, for example FormSchema.FA3.
state (OnlineSessionResumeState | str | None) — State from session.resume_state(), or its JSON string, to resume an open session.

Returns

OnlineSessionClient — A session client that can be used as a context manager and closes the session on exit.

Raises

KSeFArgumentError — If neither or both of form_code and state are given.
NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If symmetric-key encryption fails.
KSeFValidationError — If state is not valid online session state.

Example

from ksef2.models import FormSchema

with auth.online_session(form_code=FormSchema.FA3) as session:
    submission = session.send_invoice(xml_bytes)
    saved = session.resume_state().to_json()

with auth.online_session(state=saved) as session:
    result = session.submission(reference_number).wait()

python

`AuthenticatedClient.resume_online_session`

def resume_online_session(state: OnlineSessionResumeState) -> OnlineSessionClient

python

Deprecated: rebind an existing serialized online session state to this client.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use online_session(state=...) instead.

Parameters

state (OnlineSessionResumeState) — State previously exported from an online session client.

Returns

OnlineSessionClient — An online session client bound to the saved state.

`AuthenticatedClient.batch_session`

def batch_session(
    *,
    prepared_batch: PreparedBatch | None = None,
    batch_file: BatchFileInfo | None = None,
    form_code: FormSchema | None = None,
    offline_mode: bool | None = None,
    state: BatchSessionResumeState | str | None = None,
) -> BatchSessionClient

python

Open a batch session for upload work, or resume one from saved state.

Parameters

prepared_batch (PreparedBatch | None) — Prepared batch payload created by auth.batch.prepare().
batch_file (BatchFileInfo | None) — Declared ZIP package metadata and encrypted part metadata.
form_code (FormSchema | None) — Invoice schema declared for the batch session when batch_file is provided directly; defaults to FormSchema.FA3.
offline_mode (bool | None) — Whether to declare offline invoicing mode for the batch when batch_file is provided directly; defaults to False.
state (BatchSessionResumeState | str | None) — State from session.resume_state(), or its JSON string, to resume a batch session.

Returns

BatchSessionClient — A bound batch session client exposing presigned upload instructions.

Raises

KSeFArgumentError — If not exactly one of prepared_batch, batch_file and state is given, or form_code or offline_mode is combined with prepared_batch or state.
NoCertificateAvailableError — If certificate-backed encryption material is needed but no valid certificate is available.
KSeFEncryptionError — If symmetric-key encryption fails.
KSeFValidationError — If state is not valid batch session state.

Example

from ksef2.models import BatchInvoice

prepared = auth.batch.prepare(
    [BatchInvoice(file_name="invoice-1.xml", content=xml_bytes)],
)
with auth.batch_session(prepared_batch=prepared) as session:
    session.upload_parts()
    saved = session.resume_state().to_json()

with auth.batch_session(state=saved) as session:
    final = session.wait()

python

`AuthenticatedClient.open_batch_session`

def open_batch_session(
    *,
    batch_file: BatchFileInfo,
    aes_key: bytes,
    iv: bytes,
    encrypted_key: bytes,
    public_key_id: str | None = None,
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    prepared_batch: PreparedBatch | None = None,
) -> BatchSessionClient

python

Deprecated: open a batch session using caller-prepared encryption metadata.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use auth.batch_session() for the standard flow, or raw for caller-supplied encryption material.

Parameters

batch_file (BatchFileInfo) — Declared ZIP package metadata and encrypted part metadata.
aes_key (bytes) — Raw symmetric key used for part encryption.
iv (bytes) — Initialization vector paired with aes_key.
encrypted_key (bytes) — RSA-encrypted symmetric key sent to KSeF.
public_key_id (str | None) — Identifier of the KSeF public key used for encryption.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
prepared_batch (PreparedBatch | None) — Optional prepared batch payload to attach to the returned session so session.upload_parts() can operate without extra args.

Returns

BatchSessionClient — A bound batch session client exposing presigned upload instructions.

Raises

KSeFValidationError — If the batch session request is invalid.

`AuthenticatedClient.resume_batch_session`

def resume_batch_session(state: BatchSessionResumeState) -> BatchSessionClient

python

Deprecated: rebind an existing serialized batch session state to this client.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use batch_session(state=...) instead.

Parameters

state (BatchSessionResumeState) — State previously exported from a batch session client.

Returns

BatchSessionClient — A batch session client bound to the saved state.

`AuthenticatedClient.invoices` (property)

invoices: InvoicesService

python

Return the invoices service with encryption support configured.

Returns

InvoicesService — The invoices service with encryption support configured.

`AuthenticatedClient.batch` (property)

batch: BatchService

python

Return the high-level batch upload workflow service.

The service orchestrates package preparation, session opening, presigned part uploads, session closing, and status polling.

Returns

BatchService — The batch upload service.

`AuthenticatedClient.limits` (property)

limits: LimitsClient

python

Return the authenticated rate-limit branch.

Returns

LimitsClient — The authenticated rate-limit branch.

`AuthenticatedClient.collective_identifiers` (property)

collective_identifiers: CollectiveIdentifiersClient

python

Return the collective invoice identifier branch.

Returns

CollectiveIdentifiersClient — The collective invoice identifier branch.

`AuthenticatedClient.tokens` (property)

tokens: TokensClient

python

Return the authenticated token lifecycle branch.

Returns

TokensClient — The authenticated token lifecycle branch.

`AuthenticatedClient.certificates` (property)

certificates: CertificatesClient

python

Return the authenticated certificate enrollment branch.

Returns

CertificatesClient — The authenticated certificate enrollment branch.

`AuthenticatedClient.sessions` (property)

sessions: SessionManagementClient

python

Return the authenticated session-management branch.

Returns

SessionManagementClient — The authenticated session-management branch.

`AuthenticatedClient.invoice_sessions` (property)

invoice_sessions: InvoiceSessionsClient

python

Return the authenticated invoice-session history branch.

Returns

InvoiceSessionsClient — The authenticated invoice-session history branch.

`AuthenticatedClient.permissions` (property)

permissions: PermissionsClient

python

Return the authenticated permissions branch.

Returns

PermissionsClient — The authenticated permissions branch.

`AuthenticatedClient.testdata` (property)

testdata: TestDataClient

python

Return authenticated TEST-only data mutation helpers.

Returns

TestDataClient — Authenticated TEST-only data mutation helpers.

`AuthenticatedClient.raw` (property)

raw: RawAuthenticatedClient

python

Return raw authenticated endpoints for advanced integrations.

Returns

RawAuthenticatedClient — Raw authenticated endpoints for advanced integrations.

`BatchService`

Import: from ksef2.clients import BatchService

class BatchService

python

BatchService(
    *,
    authed_transport: Middleware,
    upload_transport: Middleware,
    get_encryption_key: Callable[[], SessionEncryptionMaterial],
    open_batch_session: BatchSessionOpener,
)

python

High-level workflow for preparing and sending invoice batches.

Catch KSeFException for SDK-classified failures raised by this service, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`BatchService.prepare`

def prepare(
    invoices: Iterable[bytes | str | Path | BatchInvoice],
    *,
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> PreparedBatch

python

Build a ZIP package, split it, and encrypt each upload part.

Parameters

invoices (Iterable[bytes | str | Path | BatchInvoice]) — Invoices to include. Each item is invoice XML as bytes or str (named invoice-<position>.xml), a Path to an XML file (keeps its file name) or a BatchInvoice with an explicit file name.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
max_part_size (int) — Maximum size of each ZIP part before encryption.

Returns

PreparedBatch — A prepared batch with encrypted part payloads and the metadata required
PreparedBatch — to open a batch session.

Raises

FileNotFoundError — If an invoice XML path does not exist.
NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If the invoice list or part size is invalid.

Example

batch = auth.batch.prepare([xml_1, xml_2])
session = auth.batch.submit(batch)

python

`BatchService.submit`

def submit(
    batch: PreparedBatch | Iterable[bytes | str | Path | BatchInvoice],
    *,
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> BatchSessionClient

python

Open a batch session, upload the parts and close the session.

Parameters

batch (PreparedBatch | Iterable[bytes | str | Path | BatchInvoice]) — A prepared batch from prepare(), or the invoices themselves, which are prepared first (same item types as prepare()).
form_code (FormSchema) — Invoice schema declared for the batch session. Ignored for a prepared batch, which carries its own.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch. Ignored for a prepared batch.
max_part_size (int) — Maximum size of each ZIP part before encryption. Ignored for a prepared batch.

Returns

BatchSessionClient — The closed batch session client. Call its wait() for the terminal status, list_failed_invoices() for rejected invoices and download_upo() for the UPO pages.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If preparation, session opening, or upload validation fails.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

Example

session = auth.batch.submit([xml_1, xml_2])
final = session.wait()
upos = session.download_upo()

python

`BatchService.prepare_batch`

def prepare_batch(
    *,
    invoices: Iterable[BatchInvoice],
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> PreparedBatch

python

Deprecated: build a ZIP package, split it, and encrypt each upload part.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use prepare() instead.

Parameters

invoices (Iterable[BatchInvoice]) — Invoice XML payloads to include in the batch package.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
max_part_size (int) — Maximum size of each ZIP part before encryption.

Returns

PreparedBatch — A prepared batch with encrypted part payloads and the metadata required
PreparedBatch — to open a batch session.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If the invoice list or part size is invalid.

`BatchService.prepare_batch_from_paths`

def prepare_batch_from_paths(
    *,
    invoice_paths: Iterable[Path | str],
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> PreparedBatch

python

Deprecated: load invoice XML files from disk and prepare a batch package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use prepare() with Path items instead.

Parameters

invoice_paths (Iterable[Path | str]) — Paths to XML files that should be added to the batch.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
max_part_size (int) — Maximum size of each ZIP part before encryption.

Returns

PreparedBatch — A prepared batch with encrypted parts ready to be uploaded.

Raises

FileNotFoundError — If an invoice XML path does not exist.
NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If the invoice list or part size is invalid.

`BatchService.open_session`

def open_session(*, prepared_batch: PreparedBatch) -> BatchSessionClient

python

Deprecated: open a batch session for an already prepared package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use auth.batch_session(prepared_batch=...) instead.

Parameters

prepared_batch (PreparedBatch) — Prepared batch payload returned by prepare().

Returns

BatchSessionClient — A session client exposing the upload instructions returned by KSeF.

Raises

KSeFValidationError — If the prepared batch cannot be opened.

`BatchService.upload_parts`

def upload_parts(*, session: BatchSessionClient, prepared_batch: PreparedBatch) -> None

python

Upload all prepared batch parts using the session's presigned URLs.

Parameters

session (BatchSessionClient) — Open batch session that already contains upload instructions.
prepared_batch (PreparedBatch) — Prepared batch whose part ordinals match the session.

Raises

KSeFClientClosedError — If the session client is closed.
KSeFValidationError — If prepared part ordinals do not match session upload instructions.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

`BatchService.submit_prepared_batch`

def submit_prepared_batch(*, prepared_batch: PreparedBatch) -> BatchSessionResumeState

python

Deprecated: open, upload, and close a batch session for a prepared package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use submit() instead.

Parameters

prepared_batch (PreparedBatch) — Prepared batch payload returned by prepare().

Returns

BatchSessionResumeState — Serializable state of the submitted batch session.

Raises

KSeFClientClosedError — If the session client closes before upload.
KSeFValidationError — If the prepared batch cannot be opened or uploaded.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

`BatchService.submit_batch`

def submit_batch(
    *,
    invoices: Iterable[BatchInvoice],
    form_code: FormSchema = FormSchema.FA3,
    offline_mode: bool = False,
    max_part_size: int = MAX_BATCH_PART_SIZE,
) -> BatchSessionResumeState

python

Deprecated: prepare and submit a batch in one call.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use submit() instead.

Parameters

invoices (Iterable[BatchInvoice]) — Invoice XML payloads to include in the batch package.
form_code (FormSchema) — Invoice schema declared for the batch session.
offline_mode (bool) — Whether to declare offline invoicing mode for the batch.
max_part_size (int) — Maximum size of each ZIP part before encryption.

Returns

BatchSessionResumeState — Serializable state of the submitted batch session.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If key or part encryption fails.
KSeFValidationError — If preparation, session opening, or upload validation fails.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

`BatchService.get_status`

def get_status(
    *,
    session: str | BatchSessionResumeState | BatchSessionClient,
) -> SessionStatusResponse

python

Deprecated: fetch the current status of a batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.get_status() instead.

Parameters

session (str | BatchSessionResumeState | BatchSessionClient) — Session reference number, persisted state, or open session client.

Returns

SessionStatusResponse — Current batch session status as reported by KSeF.

`BatchService.list_invoices`

def list_invoices(
    *,
    session: str | BatchSessionResumeState | BatchSessionClient,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Deprecated: fetch one page of accepted invoices from a batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.list_invoices() instead.

Parameters

session (str | BatchSessionResumeState | BatchSessionClient) — Session reference number, persisted state, or open session client.
page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices accepted in the session, with a continuation token when more exist.

`BatchService.list_failed_invoices`

def list_failed_invoices(
    *,
    session: str | BatchSessionResumeState | BatchSessionClient,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Deprecated: fetch one page of failed invoices from a batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.list_failed_invoices() instead.

Parameters

session (str | BatchSessionResumeState | BatchSessionClient) — Session reference number, persisted state, or open session client.
page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices that failed processing, with a continuation token when more exist.

`BatchService.get_upo`

def get_upo(
    *,
    session: str | BatchSessionResumeState | BatchSessionClient,
    upo_reference_number: str,
) -> bytes

python

Deprecated: download the collective UPO for a batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.download_upo() instead.

Parameters

session (str | BatchSessionResumeState | BatchSessionClient) — Session reference number, persisted state, or open session client.
upo_reference_number (str) — UPO page reference returned in the session status.

Returns

bytes — Raw XML bytes of the requested UPO page.

`BatchService.wait_for_completion`

def wait_for_completion(
    *,
    session: str | BatchSessionResumeState | BatchSessionClient,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> SessionStatusResponse

python

Deprecated: poll a batch session until KSeF reports a terminal status.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use BatchSessionClient.wait() instead.

Parameters

session (str | BatchSessionResumeState | BatchSessionClient) — Session reference number, persisted state, or open session client.
timeout (float) — Maximum number of seconds to wait for completion.
poll_interval (float) — Delay between status checks.

Returns

SessionStatusResponse — Final successful batch session status.

Raises

KSeFSessionError — If batch processing reaches a failed terminal status.
KSeFBatchSessionTimeoutError — If polling exceeds timeout.

`BatchSessionClient`

Import: from ksef2.clients import BatchSessionClient

class BatchSessionClient

python

BatchSessionClient(
    transport: Middleware,
    state: BatchSessionResumeState,
    *,
    upload_transport: Middleware | None = None,
    prepared_batch: PreparedBatch | None = None,
    access_token: str | None = None,
    resumed: bool = False,
)

python

Client for managing an active batch session.

Catch KSeFException for SDK-classified failures raised by this session branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
KSeFClientClosedError — If an operation is attempted after closing the session.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`BatchSessionClient.reference_number` (property)

reference_number: str

python

Get the batch session reference number.

Returns

str — The KSeF reference number of the session.

`BatchSessionClient.access_token` (property)

access_token: str

python

Deprecated compatibility accessor for the current bearer token.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use AuthenticatedClient.access_token instead.

Returns

str — The bearer access token the session was opened with.

Raises

KSeFValidationError — If the session state carries no bearer token.

`BatchSessionClient.aes_key` (property)

aes_key: bytes

python

Get the AES key for encrypting batch files.

Returns

bytes — The raw AES key.

`BatchSessionClient.iv` (property)

iv: bytes

python

Get the initialization vector for AES encryption.

Returns

bytes — The raw initialization vector.

`BatchSessionClient.part_upload_requests` (property)

part_upload_requests: list[PartUploadRequest]

python

Get the upload instructions for each file part.

Returns

list[PartUploadRequest] — One upload request, with presigned URL and headers, per part.

`BatchSessionClient.resume_state`

def resume_state() -> BatchSessionResumeState

python

Return the sensitive session state needed to resume later.

Returns

BatchSessionResumeState — The sensitive session state needed to resume later.

`BatchSessionClient.get_state`

def get_state() -> BatchSessionResumeState

python

Deprecated compatibility wrapper for resume_state().

Deprecated

May be removed no earlier than ksef2 2.0.0. Use resume_state() instead.

Returns

BatchSessionResumeState — The same state as resume_state().

`BatchSessionClient.get_status`

def get_status() -> SessionStatusResponse

python

Fetch the current processing state of the batch session.

Returns

SessionStatusResponse — The session status, including invoice counters.

`BatchSessionClient.list_invoices`

def list_invoices(
    *,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Fetch one page of invoices submitted in this batch session.

Parameters

page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices submitted in the session, with a continuation token when more exist.

`BatchSessionClient.list_failed_invoices`

def list_failed_invoices(
    *,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Fetch one page of failed invoices from this batch session.

Parameters

page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices that failed processing, with a continuation token when more exist.

`BatchSessionClient.get_upo`

def get_upo(*, upo_reference_number: str) -> bytes

python

Deprecated: download one page of the collective UPO for the batch session.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download_upo() instead.

Parameters

upo_reference_number (str) — Reference number of the UPO, taken from the session status.

Returns

bytes — The UPO as XML bytes.

`BatchSessionClient.wait`

def wait(*, timeout: float = 120.0, poll_interval: float = 2.0) -> SessionStatusResponse

python

Poll until KSeF reports a terminal status for the closed batch session.

KSeF processes the batch after the session is closed, so call this after auth.batch.submit() or after leaving the with block of a batch session.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between session status checks.

Returns

SessionStatusResponse — The final successful session status, including the session UPO references.

Raises

KSeFSessionError — If the session is still open, or if batch processing reaches a failed terminal status.
KSeFBatchSessionTimeoutError — If polling exceeds timeout.

`BatchSessionClient.download_upo`

def download_upo() -> list[bytes]

python

Download every page of the collective UPO in one pass.

Reads the session status once, resolves the UPO page references from it and downloads each page. It never waits or polls: call wait() first.

Returns

list[bytes] — The XML bytes of each UPO page, in order; empty if KSeF issued no UPO because no invoice was accepted.

Raises

KSeFSessionError — If the session is still open.
KSeFNotReadyError — If KSeF has not finished processing the closed batch. Call wait() on the session, then ask again.

`BatchSessionClient.upload_parts`

def upload_parts() -> None

python

Upload the prepared batch parts using the session's presigned URLs.

Raises

KSeFClientClosedError — If the upload window has already been closed.
KSeFValidationError — If the session has no prepared batch payload attached.
KSeFBatchUploadError — If external storage rejects an upload or its outcome cannot be determined. Call recovery_state() on the error to deliberately recover the sensitive batch state.

`BatchSessionClient.close`

def close() -> None

python

Close the batch session and start processing.

`CertificateEnrollment`

Import: from ksef2.clients import CertificateEnrollment

class CertificateEnrollment(OperationHandle[CertificateEnrollmentStatusResponse, CertificateEnrollmentStatusResponse])

python

CertificateEnrollment(
    client: CertificatesClient,
    response: CertificateEnrollmentResponse,
)

python

Handle to a certificate enrollment that KSeF issues asynchronously.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`CertificateEnrollment.timestamp` (property)

timestamp: datetime

python

Get when KSeF accepted the enrollment request.

Returns

datetime — The acceptance time reported by KSeF.

`CertificateEnrollment.response` (property)

response: CertificateEnrollmentResponse

python

Get the plain enrollment response.

Returns

CertificateEnrollmentResponse — The data model KSeF returned when it accepted the enrollment.

`CertificateEnrollment.get_status`

def get_status() -> CertificateEnrollmentStatusResponse

python

Fetch the enrollment's current status without waiting.

Returns

CertificateEnrollmentStatusResponse — The enrollment status, including the certificate serial number once issued.

`CertificateEnrollment.wait`

def wait(
    *,
    timeout: float = 60.0,
    poll_interval: float = 2.0,
) -> CertificateEnrollmentStatusResponse

python

Poll until KSeF has issued the certificate.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between status checks.

Returns

CertificateEnrollmentStatusResponse — The final status, with the issued certificate's serial number.

Raises

KSeFCertificateEnrollmentFailedError — If KSeF rejects, cancels or fails the enrollment.
KSeFCertificateEnrollmentTimeoutError — If polling exceeds timeout.

`CertificatesClient`

Import: from ksef2.clients import CertificatesClient

class CertificatesClient

python

CertificatesClient(transport: Middleware)

python

High-level API for certificate enrollment, retrieval, and search.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`CertificatesClient.get_limits`

def get_limits() -> CertificateLimitsResponse

python

Return the effective certificate enrollment and issuance quotas.

Returns

CertificateLimitsResponse — The effective certificate enrollment and issuance quotas.

`CertificatesClient.get_enrollment_data`

def get_enrollment_data() -> CertificateEnrollmentData

python

Return subject data needed to prepare a CSR for enrollment.

Returns

CertificateEnrollmentData — Subject data needed to prepare a CSR for enrollment.

`CertificatesClient.enroll`

def enroll(
    *,
    certificate_name: str,
    certificate_type: CertificateTypeValue,
    csr: str,
    valid_from: datetime | str | None = None,
) -> CertificateEnrollment

python

Request issuance of a certificate from a CSR.

Parameters

certificate_name (str) — Name to give the certificate.
certificate_type (CertificateTypeValue) — Certificate type to request, authentication or offline.
csr (str) — PKCS#10 certificate signing request, DER-encoded and Base64-encoded.
valid_from (datetime | str | None) — Requested start of validity as a datetime or ISO 8601 string; None for immediately.

Returns

CertificateEnrollment — A handle to the enrollment; call wait() to wait until KSeF has issued the certificate.

Example

enrollment = auth.certificates.enroll(
    certificate_name="signing", certificate_type="offline", csr=csr
)
status = enrollment.wait()
print(status.certificate_serial_number)

python

`CertificatesClient.get_enrollment_status`

def get_enrollment_status(
    *,
    reference_number: str,
) -> CertificateEnrollmentStatusResponse

python

Fetch the current status of a certificate enrollment request.

Parameters

reference_number (str) — Reference number returned by enroll().

Returns

CertificateEnrollmentStatusResponse — The enrollment status, including the certificate serial number once issued.

`CertificatesClient.retrieve`

def retrieve(
    *,
    certificate_serial_numbers: builtins.list[CertificateSerialNumber],
) -> RetrievedCertificatesList

python

Download issued certificates by serial number.

Parameters

certificate_serial_numbers (builtins.list[CertificateSerialNumber]) — Serial numbers of the certificates to download.

Returns

RetrievedCertificatesList — The requested certificates with their DER data.

`CertificatesClient.revoke`

def revoke(
    *,
    certificate_serial_number: CertificateSerialNumber,
    reason: RevocationReason | None = None,
) -> None

python

Revoke a certificate, optionally providing a revocation reason.

Parameters

certificate_serial_number (CertificateSerialNumber) — Serial number of the certificate to revoke.
reason (RevocationReason | None) — Reason for the revocation; None to omit it.

`CertificatesClient.query`

def query(
    *,
    name: str | None = None,
    certificate_serial_number: CertificateSerialNumber | None = None,
    certificate_type: CertificateTypeValue | None = None,
    status: CertificateStatusValue | None = None,
    expires_after: datetime | str | None = None,
    params: OffsetPaginationParams | None = None,
) -> CertificatesInfoList

python

Deprecated: fetch one page of certificate search results.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

name (str | None) — Match this certificate name.
certificate_serial_number (CertificateSerialNumber | None) — Match this certificate serial number.
certificate_type (CertificateTypeValue | None) — Match this certificate type.
status (CertificateStatusValue | None) — Match this lifecycle status.
expires_after (datetime | str | None) — Match certificates that expire after this datetime or ISO 8601 string.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

CertificatesInfoList — One page of certificate metadata.

`CertificatesClient.all`

def all(
    *,
    certificate_serial_number: CertificateSerialNumber | None = None,
    name: str | None = None,
    certificate_type: CertificateTypeValue | None = None,
    status: CertificateStatusValue | None = None,
    expires_after: datetime | str | None = None,
    params: OffsetPaginationParams | None = None,
) -> Iterator[CertificateInfo]

python

Deprecated: iterate over all certificates matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead.

Parameters

certificate_serial_number (CertificateSerialNumber | None) — Match this certificate serial number.
name (str | None) — Match this certificate name.
certificate_type (CertificateTypeValue | None) — Match this certificate type.
status (CertificateStatusValue | None) — Match this lifecycle status.
expires_after (datetime | str | None) — Match certificates that expire after this datetime or ISO 8601 string.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Yields

CertificateInfo — Each matching certificate, across all pages.

`CertificatesClient.list`

def list(
    *,
    name: str | None = None,
    certificate_serial_number: CertificateSerialNumber | None = None,
    certificate_type: CertificateTypeValue | None = None,
    status: CertificateStatusValue | None = None,
    expires_after: datetime | str | None = None,
    params: OffsetPaginationParams | None = None,
) -> Pager[CertificateInfo]

python

List the certificates of the authenticated context.

Nothing is requested until the result is consumed. Iterate it for every matching certificate, call pages() for page-sized lists or first_page() for one request only.

Parameters

name (str | None) — Match this certificate name.
certificate_serial_number (CertificateSerialNumber | None) — Match this certificate serial number.
certificate_type (CertificateTypeValue | None) — Match this certificate type.
status (CertificateStatusValue | None) — Match this lifecycle status.
expires_after (datetime | str | None) — Match certificates that expire after this datetime or ISO 8601 string.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[CertificateInfo] — A paging object over the matching certificates.

Example

for certificate in auth.certificates.list(status="active"):
    print(certificate.certificate_serial_number)

python

`Client`

Import: from ksef2.clients import Client

class Client

python

Client(
    environment: Environment = Environment.PRODUCTION,
    *,
    transport_config: TransportConfig | None = None,
    http_client: httpx.Client | None = None,
    certificate_store: stores.CertificateStoreProtocol | None = None,
)

python

Root KSeF SDK client responsible for transport and lifecycle management.

Branch properties only create child clients. Branch operations document their own API, validation, and transport failures.

Raises

KSeFClientClosedError — If a branch is accessed after the client is closed.
KSeFUnsupportedEnvironmentError — If a TEST-only branch is accessed outside Environment.TEST.

Example

from ksef2 import Client, Environment

with Client(Environment.TEST) as client:
    auth = client.authentication.with_test_certificate(nip="5261040828")
    for invoice in auth.invoices.all_metadata(filters=filters):
        print(invoice.ksef_number)

python

Methods

`Client.authentication` (property)

authentication: AuthClient

python

Return the authentication entry point.

Returns

AuthClient — The authentication branch, used to log in and obtain an authenticated client.

Raises

KSeFClientClosedError — If the root client has been closed.

`Client.encryption` (property)

encryption: EncryptionClient

python

Return the public encryption-certificate client.

Returns

EncryptionClient — The client for downloading KSeF public-key certificates.

Raises

KSeFClientClosedError — If the root client has been closed.

`Client.testdata` (property)

testdata: TestDataClient

python

Return the TEST-only data seeding client.

Returns

TestDataClient — The client for creating and removing TEST-environment data.

Raises

KSeFClientClosedError — If the root client has been closed.
KSeFUnsupportedEnvironmentError — If the client environment is not TEST.

`Client.peppol` (property)

peppol: PeppolClient

python

Return the public Peppol provider client.

Returns

PeppolClient — The client for listing registered Peppol providers.

Raises

KSeFClientClosedError — If the root client has been closed.

`Client.raw` (property)

raw: RawClient

python

Return raw unauthenticated endpoints for advanced integrations.

Returns

RawClient — Raw unauthenticated endpoints for advanced integrations.

`Client.authenticated`

def authenticated(auth_tokens: AuthTokens) -> AuthenticatedClient

python

Deprecated compatibility wrapper for authentication.resume().

Deprecated

May be removed no earlier than ksef2 2.0.0. Use authentication.resume() with AuthenticationResumeState.from_tokens() instead.

Parameters

auth_tokens (AuthTokens) — Access and refresh tokens from a previous authentication.

Returns

AuthenticatedClient — An authenticated client bound to the tokens.

`Client.close`

def close() -> None

python

Close owned resources and invalidate cached child clients.

`CollectiveIdentifiersClient`

Import: from ksef2.clients import CollectiveIdentifiersClient

class CollectiveIdentifiersClient

python

CollectiveIdentifiersClient(transport: Middleware)

python

API for generating and querying collective invoice identifiers.

Methods

`CollectiveIdentifiersClient.generate`

def generate(
    *,
    invoices: builtins.list[CollectiveIdentifierInvoice],
) -> GenerateCollectiveIdentifierResponse

python

Generate a collective identifier for the supplied invoices.

Parameters

invoices (builtins.list[CollectiveIdentifierInvoice]) — Invoices to group, with optional payment details.

Returns

GenerateCollectiveIdentifierResponse — The generated collective identifier number.

`CollectiveIdentifiersClient.query`

def query(
    *,
    filters: CollectiveIdentifiersQuery,
    continuation_token: str | None = None,
    params: CollectiveIdentifierParams | None = None,
) -> CollectiveIdentifiersPage

python

Deprecated: fetch one page of collective identifiers visible in the context.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

filters (CollectiveIdentifiersQuery) — Criteria selecting collective identifiers.
continuation_token (str | None) — Token from the previous page's response; None requests the first page.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

CollectiveIdentifiersPage — One page of matching collective identifiers.

`CollectiveIdentifiersClient.query_all`

def query_all(
    *,
    filters: CollectiveIdentifiersQuery,
    params: CollectiveIdentifierParams | None = None,
) -> Iterator[CollectiveIdentifiersPage]

python

Deprecated: iterate through every page matching a collective identifier query.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list().pages() instead.

Parameters

filters (CollectiveIdentifiersQuery) — Criteria selecting collective identifiers.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Yields

CollectiveIdentifiersPage — Each page of matching collective identifiers, following continuation tokens.

`CollectiveIdentifiersClient.query_by_ksef_number`

def query_by_ksef_number(
    *,
    ksef_number: str,
    continuation_token: str | None = None,
    params: CollectiveIdentifierParams | None = None,
) -> CollectiveIdentifierReferencesPage

python

Deprecated: fetch one page of identifiers associated with a KSeF invoice.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_for_invoice() instead; first_page() fetches one page.

Parameters

ksef_number (str) — KSeF number of the invoice.
continuation_token (str | None) — Token from the previous page's response; None requests the first page.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

CollectiveIdentifierReferencesPage — One page of collective identifiers that reference the invoice.

`CollectiveIdentifiersClient.query_all_by_ksef_number`

def query_all_by_ksef_number(
    *,
    ksef_number: str,
    params: CollectiveIdentifierParams | None = None,
) -> Iterator[CollectiveIdentifierReferencesPage]

python

Deprecated: iterate through identifiers associated with one KSeF invoice.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_for_invoice().pages() instead.

Parameters

ksef_number (str) — KSeF number of the invoice.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Yields

CollectiveIdentifierReferencesPage — Each page of identifiers that reference the invoice, following continuation tokens.

`CollectiveIdentifiersClient.list_all_invoices`

def list_all_invoices(
    *,
    collective_identifier_numbers: builtins.list[str],
    params: CollectiveIdentifierParams | None = None,
) -> Iterator[CollectiveIdentifierInvoicesPage]

python

Deprecated: iterate through every invoice page for the supplied collective identifiers.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_invoices().pages() instead.

Parameters

collective_identifier_numbers (builtins.list[str]) — Collective identifier numbers to expand (1–10).
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Yields

CollectiveIdentifierInvoicesPage — Each page of invoices, following continuation tokens.

`CollectiveIdentifiersClient.list_for_invoice`

def list_for_invoice(
    ksef_number: str,
    *,
    params: CollectiveIdentifierParams | None = None,
) -> Pager[CollectiveIdentifierReference]

python

List the collective identifiers that reference one KSeF invoice.

Nothing is requested until the result is consumed.

Parameters

ksef_number (str) — KSeF number of the invoice.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

Pager[CollectiveIdentifierReference] — A paging object over the collective identifiers referencing the invoice.

Example

for reference in auth.collective_identifiers.list_for_invoice(ksef_number):
    print(reference.collective_identifier_number)

python

`CollectiveIdentifiersClient.list_invoices`

def list_invoices(
    collective_identifier_numbers: builtins.list[str],
    *,
    params: CollectiveIdentifierParams | None = None,
) -> Pager[CollectiveIdentifierInvoiceDetails]

python

List the invoices inside the supplied collective identifiers.

Nothing is requested until the result is consumed. Unlike the former list_invoices(), which fetched one page, this returns a paging object.

Parameters

collective_identifier_numbers (builtins.list[str]) — Collective identifier numbers to expand (1–10).
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

Pager[CollectiveIdentifierInvoiceDetails] — A paging object over the invoices belonging to the identifiers.

Example

for invoice in auth.collective_identifiers.list_invoices([number]):
    print(invoice.ksef_number)

python

`CollectiveIdentifiersClient.list`

def list(
    filters: CollectiveIdentifiersQuery,
    *,
    params: CollectiveIdentifierParams | None = None,
) -> Pager[CollectiveIdentifierSummary]

python

List the collective identifiers visible in the context.

Nothing is requested until the result is consumed. Iterate it for every identifier, call pages() for page-sized lists or first_page() for one request only.

Parameters

filters (CollectiveIdentifiersQuery) — Criteria selecting collective identifiers.
params (CollectiveIdentifierParams | None) — Page size; defaults are used when None.

Returns

Pager[CollectiveIdentifierSummary] — A paging object over the matching collective identifiers.

Example

for identifier in auth.collective_identifiers.list(filters):
    print(identifier.collective_identifier_number)

python

`EncryptionClient`

Import: from ksef2.clients import EncryptionClient

class EncryptionClient

python

EncryptionClient(transport: Middleware)

python

Access to public encryption certificates published by KSeF.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`EncryptionClient.get_certificates`

def get_certificates(
    *,
    usage: list[CertUsage] | None = None,
) -> list[PublicKeyCertificate]

python

Return public certificates, optionally filtered by supported usage.

Parameters

usage (list[CertUsage] | None) — Return only certificates valid for these usages; None for all.

Returns

list[PublicKeyCertificate] — The public-key certificates published by KSeF.

`ExportedInvoices`

Import: from ksef2.clients import ExportedInvoices

class ExportedInvoices

python

ExportedInvoices(parts: list[bytes], *, package: InvoicePackage | None = None)

python

Decrypted result of an invoice export.

The decrypted package parts are joined into one ZIP archive, which holds one <ksef_number>.xml file per invoice and a _metadata.json file. The ZIP is opened lazily; construct this object through ExportJob.wait(). An export that matched no invoices has an empty archive.

Methods

`ExportedInvoices.archive` (property)

archive: bytes

python

Get the raw ZIP archive.

Returns

bytes — The decrypted and joined package parts, exactly as KSeF built the archive.

`ExportedInvoices.package` (property)

package: InvoicePackage | None

python

Get the package metadata reported by the export status.

Returns

InvoicePackage | None — Invoice count, size, truncation flag and the dates for continuing a truncated export; None if unknown.

`ExportedInvoices.metadata` (property)

metadata: list[InvoiceMetadata] | None

python

Get the invoice metadata stored in the package.

Returns

list[InvoiceMetadata] | None — The parsed invoices list of _metadata.json, or None if the package has no such file.

Raises

KSeFValidationError — If _metadata.json is not valid metadata.
zipfile.BadZipFile — If the archive is not a valid ZIP file.

`ExportedInvoices.invoices`

def invoices() -> Iterator[tuple[str, bytes]]

python

Iterate over the invoices in the package.

Yields

tuple[str, bytes] — A (ksef_number, xml) pair for each invoice file, in archive order.

Raises

zipfile.BadZipFile — If the archive is not a valid ZIP file.

`ExportedInvoices.save`

def save(directory: Path | str) -> list[Path]

python

Extract the package into a directory.

Every entry path is validated first, and nothing is written if one is unsafe: absolute paths, .. segments, backslashes, drive letters and NUL bytes are refused.

Parameters

directory (Path | str) — Directory to extract into; created if missing.

Returns

list[Path] — The paths of the written files.

Raises

ValueError — If an archive entry path would escape directory.
zipfile.BadZipFile — If the archive is not a valid ZIP file.
OSError — If a file or directory cannot be written.

`ExportJob`

Import: from ksef2.clients import ExportJob

class ExportJob(OperationHandle[InvoiceExportStatusResponse, ExportedInvoices])

python

ExportJob(
    state: ExportResumeState,
    *,
    get_status: Callable[[], InvoiceExportStatusResponse],
    download_parts: Callable[[InvoicePackage], list[bytes]],
)

python

Handle to a scheduled invoice export.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`ExportJob.resume_state`

def resume_state() -> ExportResumeState

python

Return the sensitive state needed to resume the export later.

Returns

ExportResumeState — The export's reference number and AES key material.

`ExportJob.get_status`

def get_status() -> InvoiceExportStatusResponse

python

Fetch the export's current status without waiting.

Returns

InvoiceExportStatusResponse — The export status, with package metadata once the export is ready.

`ExportJob.wait`

def wait(*, timeout: float = 120.0, poll_interval: float = 2.0) -> ExportedInvoices

python

Poll until the export is ready, then download and decrypt the package.

Parameters

timeout (float) — Maximum number of seconds to wait for the export to become ready.
poll_interval (float) — Delay in seconds between export status checks.

Returns

ExportedInvoices — The decrypted package: iterate invoices(), read metadata or save() it to disk.

Raises

KSeFExportFailedError — If KSeF reports that the export failed, was cancelled or expired.
KSeFExportTimeoutError — If polling exceeds timeout.
KSeFEncryptionError — If a downloaded package part cannot be decrypted.
KSeFExternalTransferError — If external storage rejects a part download or its outcome cannot be determined.

`GeneratedToken`

Import: from ksef2.clients import GeneratedToken

class GeneratedToken(OperationHandle[TokenStatusResponse, TokenStatusResponse])

python

GeneratedToken(client: TokensClient, response: GenerateTokenResponse)

python

Handle to a token KSeF is still activating.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`GeneratedToken.token` (property)

token: str

python

Get the one-time token value.

Returns

str — The secret KSeF returned once at generation; store it securely.

`GeneratedToken.response` (property)

response: GenerateTokenResponse

python

Get the plain generation response.

Returns

GenerateTokenResponse — The data model KSeF returned when the token was generated.

`GeneratedToken.to_sensitive_dict`

def to_sensitive_dict() -> dict[str, str]

python

Export the one-time token for deliberately protected persistence.

Returns

dict[str, str] — A dictionary with the reference number and the plain token.

`GeneratedToken.get_status`

def get_status() -> TokenStatusResponse

python

Fetch the token's current status without waiting.

Returns

TokenStatusResponse — The token's current lifecycle status.

`GeneratedToken.wait`

def wait(*, timeout: float = 60.0, poll_interval: float = 1.0) -> TokenStatusResponse

python

Poll until the token becomes active or reaches a terminal state.

Parameters

timeout (float) — Maximum number of seconds to wait for activation.
poll_interval (float) — Delay in seconds between status checks.

Returns

TokenStatusResponse — The first status response that reports the token as active.

Raises

KSeFApiError — If activation ends in a terminal failure state.
KSeFTokenStatusTimeoutError — If polling exceeds timeout.

`InvoiceSessionsClient`

Import: from ksef2.clients import InvoiceSessionsClient

class InvoiceSessionsClient

python

InvoiceSessionsClient(transport: Middleware)

python

Browse historical invoice sessions and download their UPO documents.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`InvoiceSessionsClient.download_upo`

def download_upo(
    session_reference_number: str,
    *,
    invoice_reference_number: str | None = None,
    ksef_number: str | None = None,
    upo_reference_number: str | None = None,
) -> bytes

python

Download one UPO from stored references without opening a session.

Parameters

session_reference_number (str) — Reference number of the invoice session.
invoice_reference_number (str | None) — Invoice reference within that session. Pass exactly one of this, ksef_number or upo_reference_number.
ksef_number (str | None) — KSeF number of an accepted invoice in that session.
upo_reference_number (str | None) — Reference number of one session UPO page.

Returns

bytes — The invoice UPO or session UPO page as XML bytes.

Raises

KSeFValidationError — If no selector or multiple selectors are given.
KSeFNotReadyError — If KSeF has not issued the requested UPO yet. Check processing status before retrying. Rejected invoices may never receive a UPO.

`InvoiceSessionsClient.query`

def query(
    *,
    session_type: str,
    continuation_token: str | None = None,
    params: ListSessionsQuery | None = None,
    statuses: builtins.list[SessionStatus | SessionStatusEnum] | None = None,
) -> ListSessionsResponse

python

Deprecated: fetch one page of invoice session history for the chosen session type.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

session_type (str) — Invoice session family to browse, such as "online" or "batch".
continuation_token (str | None) — Cursor returned by a previous page.
params (ListSessionsQuery | None) — Optional query object with pagination and filter settings.
statuses (builtins.list[SessionStatus | SessionStatusEnum] | None) — Optional list of status filters applied on top of params.

Returns

ListSessionsResponse — One page of historical invoice sessions.

`InvoiceSessionsClient.all`

def all(
    *,
    session_type: str,
    params: ListSessionsQuery | None = None,
) -> Iterator[ListSessionsResponse]

python

Deprecated: iterate through all pages of invoice session history.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list().pages() instead.

Parameters

session_type (str) — Invoice session family to browse, such as "online" or "batch".
params (ListSessionsQuery | None) — Optional query object with pagination and filter settings.

Yields

ListSessionsResponse — Successive pages of historical invoice sessions until KSeF stops
ListSessionsResponse — returning a continuation token.

`InvoiceSessionsClient.list`

def list(
    session_type: str,
    *,
    params: ListSessionsQuery | None = None,
    statuses: builtins.list[SessionStatus | SessionStatusEnum] | None = None,
) -> Pager[SessionSummary]

python

List the invoice sessions of the chosen type, newest history first as KSeF returns it.

Nothing is requested until the result is consumed. Iterate it for every session, call pages() for page-sized lists or first_page() for one request only.

Parameters

session_type (str) — Invoice session family to browse, such as "online" or "batch".
params (ListSessionsQuery | None) — Optional query object with page size and filter settings.
statuses (builtins.list[SessionStatus | SessionStatusEnum] | None) — Optional list of status filters applied on top of params.

Returns

Pager[SessionSummary] — A paging object over the matching invoice sessions.

Example

for session in auth.invoice_sessions.list("online"):
    print(session.reference_number, session.status)

python

`InvoicesService`

Import: from ksef2.clients import InvoicesService

class InvoicesService

python

InvoicesService(
    transport: Middleware,
    download_transport: Middleware,
    certificate_store: CertificateStoreProtocol,
    *,
    client: InvoicesClient | None = None,
    ensure_encryption_certificates_loaded: Callable[[], None] | None = None,
)

python

Invoice workflows with encryption, polling, and export helpers.

Catch KSeFException for SDK-classified failures raised by this service, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`InvoicesService.search`

def search(
    filters: InvoicesFilter,
    params: InvoiceMetadataParams | None = None,
) -> Pager[InvoiceMetadata]

python

Search invoice metadata, following KSeF page and truncation mechanics.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
params (InvoiceMetadataParams | None) — Page size and sort order; defaults are used when None.

Returns

Pager[InvoiceMetadata] — A paging object over the metadata of matching invoices.

Raises

KSeFMetadataPaginationError — If KSeF returns inconsistent pagination boundaries while paging.

Example

from ksef2.models import InvoicesFilter

filters = InvoicesFilter.for_seller(date_from="2026-01-01T00:00:00+01:00")
for invoice in auth.invoices.search(filters):
    print(invoice.ksef_number, invoice.gross_amount)

python

`InvoicesService.download`

def download(
    ksef_number: str,
    *,
    timeout: float | None = None,
    poll_interval: float = 2.0,
) -> bytes

python

Download one processed invoice by KSeF number.

Parameters

ksef_number (str) — KSeF number of the invoice.
timeout (float | None) — Seconds to keep polling while KSeF has not made the invoice available yet; None tries once.
poll_interval (float) — Delay in seconds between download attempts when timeout is set.

Returns

bytes — The invoice XML as bytes.

Raises

KSeFApiError — If KSeF rejects the download, for example because the invoice is not processed yet and timeout is None.
KSeFInvoiceDownloadTimeoutError — If timeout is set and polling exceeds it.

Example

xml = auth.invoices.download(ksef_number, timeout=60)

python

`InvoicesService.export`

def export(
    filters: InvoicesFilter | None = None,
    *,
    state: ExportResumeState | str | None = None,
    only_metadata: bool = False,
    compression_type: CompressionType | str | None = None,
) -> ExportJob

python

Schedule an encrypted invoice export, or resume one scheduled earlier.

Pass exactly one of filters (schedule a new export) or state (resume the export described by a saved ExportResumeState).

Parameters

filters (InvoicesFilter | None) — Criteria selecting the invoices to export.
state (ExportResumeState | str | None) — State from ExportJob.resume_state(), or its JSON string, to resume an export after a restart. Cannot be combined with the other arguments.
only_metadata (bool) — Export only invoice metadata instead of full invoice XML.
compression_type (CompressionType | str | None) — Compression applied to the package; None for the server default.

Returns

ExportJob — A handle to the export. Call its wait() to download the decrypted package.

Raises

KSeFArgumentError — If neither or both of filters and state are given, or state is combined with scheduling options.
NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If export key encryption fails.
KSeFValidationError — If state is not valid export resume state.

Example

job = auth.invoices.export(filters)
saved = job.resume_state().to_json()  # persist before waiting

job = auth.invoices.export(state=saved)  # after a restart
package = job.wait()
package.save("out/")

python

`InvoicesService.query_metadata`

def query_metadata(
    *,
    filters: InvoicesFilter,
    params: InvoiceMetadataParams | None = None,
) -> QueryInvoicesMetadataResponse

python

Deprecated: fetch one invoice metadata page matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use search() instead; first_page() fetches one page.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
params (InvoiceMetadataParams | None) — Page size, page offset and sort order; defaults are used when None.

Returns

QueryInvoicesMetadataResponse — One page of invoice metadata.

`InvoicesService.query_metadata_pages`

def query_metadata_pages(
    *,
    filters: InvoicesFilter,
    params: InvoiceMetadataParams | None = None,
) -> Iterator[QueryInvoicesMetadataResponse]

python

Deprecated: fetch metadata pages, following KSeF page and truncation mechanics.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use search().pages() instead.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
params (InvoiceMetadataParams | None) — Page size, page offset and sort order; defaults are used when None.

Yields

QueryInvoicesMetadataResponse — Each page of invoice metadata in order.

Raises

KSeFMetadataPaginationError — If KSeF returns inconsistent pagination boundaries.

`InvoicesService.all_metadata`

def all_metadata(
    *,
    filters: InvoicesFilter,
    params: InvoiceMetadataParams | None = None,
) -> Iterator[InvoiceMetadata]

python

Deprecated: iterate over all invoice metadata items matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use search() instead.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
params (InvoiceMetadataParams | None) — Page size and sort order; defaults are used when None.

Yields

InvoiceMetadata — Metadata of each matching invoice, across all pages.

Raises

KSeFMetadataPaginationError — If KSeF returns inconsistent pagination boundaries.

`InvoicesService.download_invoice`

def download_invoice(*, ksef_number: str) -> bytes

python

Deprecated: download one processed invoice by KSeF number.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download() instead.

Parameters

ksef_number (str) — KSeF number of the invoice.

Returns

bytes — The invoice XML as bytes.

`InvoicesService.wait_for_invoice_download`

def wait_for_invoice_download(
    *,
    ksef_number: str,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> bytes

python

Deprecated: poll until KSeF makes a processed invoice available for download.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download(ksef_number, timeout=...) instead.

Parameters

ksef_number (str) — KSeF number of the invoice.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between download attempts.

Returns

bytes — The invoice XML as bytes.

Raises

KSeFInvoiceDownloadTimeoutError — If polling exceeds timeout.

`InvoicesService.schedule_export`

def schedule_export(
    *,
    filters: InvoicesFilter,
    only_metadata: bool = False,
    compression_type: CompressionType | str | None = None,
) -> ExportHandle

python

Deprecated: schedule an encrypted invoice export.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export() instead.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices to export.
only_metadata (bool) — Export only invoice metadata instead of full invoice XML.
compression_type (CompressionType | str | None) — Compression applied to the package; None for the server default.

Returns

ExportHandle — A handle holding the export reference number and the keys needed to decrypt the package.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If export key encryption fails.

`InvoicesService.get_export_status`

def get_export_status(*, reference_number: str) -> InvoiceExportStatusResponse

python

Deprecated: fetch the current status for a scheduled invoice export.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export() and ExportJob.get_status() instead.

Parameters

reference_number (str) — Reference number of the export, from ExportHandle.reference_number.

Returns

InvoiceExportStatusResponse — The export status, with package metadata once the export is ready.

`InvoicesService.fetch_package`

def fetch_package(
    *,
    package: InvoicePackage,
    export: ExportHandle,
    target_directory: Path | str = '.',
) -> list[Path]

python

Deprecated: download and decrypt all parts of an export package to disk.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export().wait() and ExportedInvoices.save() instead.

Parameters

package (InvoicePackage) — Package metadata from the export status.
export (ExportHandle) — Handle returned when the export was scheduled; supplies the decryption keys.
target_directory (Path | str) — Directory to write the decrypted parts to. Defaults to the current directory.

Returns

list[Path] — The paths of the written part files.

Raises

KSeFEncryptionError — If a downloaded package part cannot be decrypted.
ValueError — If a package part name is unsafe for local filesystem output.
OSError — If the target directory or output file cannot be written.
KSeFExternalTransferError — If external storage rejects a part download or its outcome cannot be determined.

`InvoicesService.fetch_package_bytes`

def fetch_package_bytes(*, package: InvoicePackage, export: ExportHandle) -> list[bytes]

python

Deprecated: download and decrypt all parts of an export package in memory.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export().wait() and ExportedInvoices.archive instead.

Parameters

package (InvoicePackage) — Package metadata from the export status.
export (ExportHandle) — Handle returned when the export was scheduled; supplies the decryption keys.

Returns

list[bytes] — The decrypted package parts, in order.

Raises

KSeFEncryptionError — If a downloaded package part cannot be decrypted.
KSeFExternalTransferError — If external storage rejects a part download or its outcome cannot be determined.

`InvoicesService.wait_for_invoices`

def wait_for_invoices(
    *,
    filters: InvoicesFilter,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> QueryInvoicesMetadataResponse

python

Deprecated: poll invoice metadata until at least one invoice matches the filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use search(filters).wait() instead.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between queries.

Returns

QueryInvoicesMetadataResponse — The first metadata page that contains at least one invoice.

Raises

KSeFInvoiceQueryTimeoutError — If polling exceeds timeout.

`InvoicesService.wait_for_export_package`

def wait_for_export_package(
    *,
    reference_number: str,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> InvoicePackage

python

Deprecated: poll export status until KSeF exposes a downloadable package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export().wait() instead.

Parameters

reference_number (str) — Reference number of the export, from ExportHandle.reference_number.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between export status checks.

Returns

InvoicePackage — The package metadata once the export is ready.

Raises

KSeFExportTimeoutError — If polling exceeds timeout.

`InvoicesService.export_and_download`

def export_and_download(
    *,
    filters: InvoicesFilter,
    only_metadata: bool = False,
    compression_type: CompressionType | str | None = None,
    timeout: float = 120.0,
    poll_interval: float = 2.0,
) -> list[bytes]

python

Deprecated: schedule an export, wait for it, and download the decrypted package.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use export().wait() instead; ExportedInvoices.archive is the joined ZIP.

Parameters

filters (InvoicesFilter) — Criteria selecting the invoices to export.
only_metadata (bool) — Export only invoice metadata instead of full invoice XML.
compression_type (CompressionType | str | None) — Compression applied to the package; None for the server default.
timeout (float) — Maximum number of seconds to wait for the export to become ready.
poll_interval (float) — Delay in seconds between export status checks.

Returns

list[bytes] — The decrypted package parts, in order. Concatenate them to obtain the package archive.

Raises

NoCertificateAvailableError — If no valid symmetric-key certificate is available.
KSeFEncryptionError — If export key encryption or package decryption fails.
KSeFExportTimeoutError — If polling exceeds timeout.
KSeFExternalTransferError — If external storage rejects a part download or its outcome cannot be determined.

`InvoiceSubmission`

Import: from ksef2.clients import InvoiceSubmission

class InvoiceSubmission(OperationHandle[SessionInvoiceStatusResponse, SessionInvoiceStatusResponse])

python

InvoiceSubmission(session: OnlineSessionClient, response: invoices.SendInvoiceResponse)

python

Handle to one invoice sent into an online session.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`InvoiceSubmission.response` (property)

response: invoices.SendInvoiceResponse

python

Get the plain submission response.

Returns

invoices.SendInvoiceResponse — The data model KSeF returned when the invoice was accepted for processing.

`InvoiceSubmission.get_status`

def get_status() -> SessionInvoiceStatusResponse

python

Fetch the invoice's current processing status without waiting.

Returns

SessionInvoiceStatusResponse — The processing status of the invoice.

`InvoiceSubmission.wait`

def wait(
    *,
    timeout: float = 60.0,
    poll_interval: float = 2.0,
) -> SessionInvoiceStatusResponse

python

Poll until KSeF assigns the invoice a final processing result.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between invoice status checks.

Returns

SessionInvoiceStatusResponse — The final processing status, including the invoice's KSeF number once accepted.

Raises

KSeFInvoiceRejectedError — If invoice processing reaches a failed terminal status. It subclasses KSeFSessionError and keeps the status details and extensions.
KSeFInvoiceProcessingTimeoutError — If polling exceeds timeout.

`InvoiceSubmission.download_upo`

def download_upo() -> bytes

python

Download the UPO of this invoice with a single request.

KSeF issues the UPO once it has accepted the invoice. Call wait() first, then this; it never waits or polls.

Returns

bytes — The UPO as XML bytes.

Raises

KSeFNotReadyError — If KSeF has not issued the UPO yet. Call wait() on this submission, then ask again.

`LimitsClient`

Import: from ksef2.clients import LimitsClient

class LimitsClient

python

LimitsClient(transport: Middleware)

python

Read and override KSeF context, subject, and API rate limits.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`LimitsClient.get_context_limits`

def get_context_limits() -> ContextLimits

python

Return the effective limits for online and batch sessions.

Returns

ContextLimits — The effective limits for online and batch sessions.

`LimitsClient.get_subject_limits`

def get_subject_limits() -> SubjectLimits

python

Return the effective limits for certificate enrollments and issuance.

Returns

SubjectLimits — The effective limits for certificate enrollments and issuance.

`LimitsClient.get_api_rate_limits`

def get_api_rate_limits() -> ApiRateLimits

python

Return the effective per-endpoint API rate limits.

Returns

ApiRateLimits — The effective per-endpoint API rate limits.

`LimitsClient.set_session_limits`

def set_session_limits(*, limits: ContextLimits) -> None

python

Override session limits for the current subject.

Parameters

limits (ContextLimits) — New session limits.

`LimitsClient.reset_session_limits`

def reset_session_limits() -> None

python

Reset session limits to their default values.

`LimitsClient.set_subject_limits`

def set_subject_limits(*, limits: SubjectLimits) -> None

python

Override subject-level certificate and enrollment limits.

Parameters

limits (SubjectLimits) — New subject-level limits.

`LimitsClient.reset_subject_limits`

def reset_subject_limits() -> None

python

Reset subject-level limits to their default values.

`LimitsClient.set_api_rate_limits`

def set_api_rate_limits(*, limits: ApiRateLimits) -> None

python

Override API rate limits for the current subject.

Parameters

limits (ApiRateLimits) — New API rate limits.

`LimitsClient.reset_api_rate_limits`

def reset_api_rate_limits() -> None

python

Reset API rate limits to their default values.

`LimitsClient.set_production_rate_limits`

def set_production_rate_limits() -> None

python

Copy production rate-limit defaults into the current environment.

`OnlineSessionClient`

Import: from ksef2.clients import OnlineSessionClient

class OnlineSessionClient

python

OnlineSessionClient(
    transport: Middleware,
    state: OnlineSessionResumeState,
    *,
    resumed: bool = False,
)

python

Client bound to a single online invoice session.

Send invoices with send_invoice(), close the session by leaving the with block, then call wait() for the terminal session status and download_upo() for the session UPO.

Catch KSeFException for SDK-classified failures raised by this session branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
KSeFClientClosedError — If an operation is attempted after closing the session.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`OnlineSessionClient.reference_number` (property)

reference_number: str

python

Get the online session reference number.

Returns

str — The KSeF reference number of the session.

`OnlineSessionClient.submission`

def submission(reference_number: str) -> InvoiceSubmission

python

Get the handle of an invoice sent earlier in this session.

Use it after a restart or on a resumed session to wait() for the invoice or download_upo(). No request is made until you call one.

Parameters

reference_number (str) — Invoice reference number, as exposed by the handle send_invoice() returned.

Returns

InvoiceSubmission — The invoice submission handle.

`OnlineSessionClient.send_invoice`

def send_invoice(invoice_xml: bytes | str) -> InvoiceSubmission

python

Encrypt and submit one invoice into the open session.

Parameters

invoice_xml (bytes | str) — Invoice XML, valid against the session's schema. A str is encoded as UTF-8.

Returns

InvoiceSubmission — A handle exposing the submission's reference_number. Call its wait() for the processing result and download_upo() for the invoice UPO.

Raises

KSeFEncryptionError — If invoice encryption fails.

Example

with auth.online_session(form_code=FormSchema.FA3) as session:
    submission = session.send_invoice(xml)
    result = submission.wait(timeout=60)
    print(result.ksef_number)

python

`OnlineSessionClient.send_invoice_and_wait`

def send_invoice_and_wait(
    *,
    invoice_xml: bytes,
    timeout: float = 60.0,
    poll_interval: float = 2.0,
) -> SessionInvoiceStatusResponse

python

Deprecated: submit an invoice and poll until KSeF assigns a final result.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use send_invoice().wait() instead.

Parameters

invoice_xml (bytes) — Invoice XML bytes, valid against the session's schema.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between invoice status checks.

Returns

SessionInvoiceStatusResponse — The final processing status, including the invoice's KSeF number once accepted.

Raises

KSeFEncryptionError — If invoice encryption fails.
KSeFInvoiceRejectedError — If invoice processing reaches a failed terminal status. It subclasses KSeFSessionError and keeps the status details and extensions.
KSeFInvoiceProcessingTimeoutError — If polling exceeds timeout.

`OnlineSessionClient.get_status`

def get_status() -> SessionStatusResponse

python

Fetch the current state of the online session.

Returns

SessionStatusResponse — The session status, including invoice counters.

`OnlineSessionClient.list_invoices`

def list_invoices(
    *,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Fetch one page of invoices submitted in this session.

Parameters

page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices submitted in the session, with a continuation token when more exist.

`OnlineSessionClient.get_invoice_status`

def get_invoice_status(*, invoice_reference_number: str) -> SessionInvoiceStatusResponse

python

Fetch processing status for one invoice sent in this session.

Parameters

invoice_reference_number (str) — Reference number of the invoice within the session, as exposed by the handle send_invoice() returned.

Returns

SessionInvoiceStatusResponse — The processing status of the invoice.

`OnlineSessionClient.wait_for_invoice_ready`

def wait_for_invoice_ready(
    *,
    invoice_reference_number: str,
    timeout: float = 60.0,
    poll_interval: float = 2.0,
) -> SessionInvoiceStatusResponse

python

Deprecated: poll invoice status until it succeeds, fails, or times out.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use send_invoice().wait() instead.

Parameters

invoice_reference_number (str) — Reference number of the invoice within the session.
timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between invoice status checks.

Returns

SessionInvoiceStatusResponse — The final processing status, including the invoice's KSeF number once accepted.

Raises

KSeFInvoiceRejectedError — If invoice processing reaches a failed terminal status. It subclasses KSeFSessionError and keeps the status details and extensions.
KSeFInvoiceProcessingTimeoutError — If polling exceeds timeout.

`OnlineSessionClient.list_failed_invoices`

def list_failed_invoices(
    *,
    page_size: int = 10,
    continuation_token: str | None = None,
) -> SessionInvoicesResponse

python

Fetch one page of invoices that failed within this session.

Parameters

page_size (int) — Number of invoices per page (10–1000).
continuation_token (str | None) — Token from the previous page's response; None requests the first page.

Returns

SessionInvoicesResponse — One page of invoices that failed processing, with a continuation token when more exist.

`OnlineSessionClient.download_invoice_upo`

def download_invoice_upo(
    *,
    ksef_number: str | None = None,
    reference_number: str | None = None,
) -> bytes

python

Download the UPO of one invoice accepted in this session.

Parameters

ksef_number (str | None) — KSeF number of the invoice. Pass this or reference_number.
reference_number (str | None) — Reference number of the invoice within the session, as exposed by the handle send_invoice() returned. Pass this or ksef_number.

Returns

bytes — The UPO as XML bytes.

Raises

KSeFValidationError — If both or neither of ksef_number and reference_number are given.

`OnlineSessionClient.get_invoice_upo_by_ksef_number`

def get_invoice_upo_by_ksef_number(*, ksef_number: str) -> bytes

python

Deprecated: download the invoice UPO by KSeF number.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download_invoice_upo(ksef_number=...) instead.

Parameters

ksef_number (str) — KSeF number of an invoice accepted in this session.

Returns

bytes — The UPO as XML bytes.

`OnlineSessionClient.get_invoice_upo_by_reference`

def get_invoice_upo_by_reference(*, invoice_reference_number: str) -> bytes

python

Deprecated: download the invoice UPO by session invoice reference number.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use download_invoice_upo(reference_number=...) instead.

Parameters

invoice_reference_number (str) — Reference number of the invoice within the session.

Returns

bytes — The UPO as XML bytes.

`OnlineSessionClient.wait`

def wait(*, timeout: float = 120.0, poll_interval: float = 2.0) -> SessionStatusResponse

python

Poll until KSeF reports a terminal status for the closed session.

KSeF processes the session only after it is closed, so call this after leaving the with block or calling close().

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between session status checks.

Returns

SessionStatusResponse — The final successful session status, including the session UPO references.

Raises

KSeFSessionError — If the session is still open, or if KSeF reports a failed terminal status.
KSeFOnlineSessionTimeoutError — If polling exceeds timeout.

`OnlineSessionClient.download_upo`

def download_upo() -> list[bytes]

python

Download every page of the session UPO in one pass.

Reads the session status once, resolves the UPO page references from it and downloads each page. It never waits or polls: call wait() first.

Returns

list[bytes] — The XML bytes of each UPO page, in order; empty if KSeF issued no UPO because no invoice was accepted.

Raises

KSeFSessionError — If the session is still open.
KSeFNotReadyError — If KSeF has not finished processing the closed session. Call wait() on the session, then ask again.

`OnlineSessionClient.close`

def close() -> None

python

Terminate the online session if it is still open.

Closing an already-closed session is a no-op. A client rebuilt from saved state first asks KSeF whether the session was already closed.

Raises

KSeFApiError — If KSeF rejects the session termination request.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

`OnlineSessionClient.resume_state`

def resume_state() -> OnlineSessionResumeState

python

Return the sensitive session state needed to resume later.

Returns

OnlineSessionResumeState — The sensitive session state needed to resume later.

`OnlineSessionClient.get_state`

def get_state() -> OnlineSessionResumeState

python

Deprecated compatibility wrapper for resume_state().

Deprecated

May be removed no earlier than ksef2 2.0.0. Use resume_state() instead.

Returns

OnlineSessionResumeState — The same state as resume_state().

`OperationHandle`

Import: from ksef2.clients import OperationHandle

class OperationHandle(ABC, Generic[_TStatus, _TResult])

python

OperationHandle(reference_number: str)

python

Client-side handle to an operation that KSeF finishes asynchronously.

Catch KSeFException for SDK-classified failures raised while waiting, and httpx.HTTPError for transport failures.

Methods

`OperationHandle.reference_number` (property)

reference_number: str

python

Get the KSeF reference number of the operation.

Returns

str — The reference number KSeF assigned to the operation.

`OperationHandle.get_status`

def get_status() -> _TStatus

python

Fetch the current status of the operation without waiting.

Returns

_TStatus — The status as KSeF reports it right now.

`Pager`

Import: from ksef2.clients import Pager

class Pager(Generic[_T])

python

Pager(
    pages: Callable[[], Generator[list[_T], None]],
    *,
    timeout_error: Callable[[float], BaseException] | None = None,
)

python

Lazy view over a paged collection in KSeF.

Raises

KSeFApiError — If KSeF returns an API error response while a page is fetched.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`Pager.pages`

def pages() -> Iterator[list[_T]]

python

Iterate over the collection one page at a time.

Returns

Iterator[list[_T]] — An iterator yielding the items of each page as a list, in order.

`Pager.first_page`

def first_page() -> list[_T]

python

Fetch only the first page.

Returns

list[_T] — The items of the first page; empty if the collection is empty.

`Pager.wait`

def wait(*, timeout: float = 120.0, poll_interval: float = 2.0) -> Pager[_T]

python

Poll until the collection contains at least one item.

Use it when KSeF indexes items with a delay, for example invoices that were accepted a moment ago and are not searchable yet.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between checks.

Returns

Pager[_T] — This pager, now known to be non-empty. Iterating it requests the pages again.

Raises

KSeFInvoiceQueryTimeoutError — For invoice searches, if no invoice appears within timeout.
TimeoutError — For other collections, if no item appears within timeout.

`PeppolClient`

Import: from ksef2.clients import PeppolClient

class PeppolClient

python

PeppolClient(transport: Middleware)

python

Service for querying Peppol service providers.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`PeppolClient.query`

def query(
    *,
    params: OffsetPaginationParams | None = None,
) -> ListPeppolProvidersResponse

python

Deprecated: query one page of Peppol service providers.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

params (OffsetPaginationParams | None) — Pagination parameters.

Returns

ListPeppolProvidersResponse — One page of providers with pagination info.

`PeppolClient.all`

def all(*, params: OffsetPaginationParams | None = None) -> Iterator[PeppolProvider]

python

Deprecated: iterate over all Peppol service providers.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead.

Parameters

params (OffsetPaginationParams | None) — Pagination parameters.

Yields

PeppolProvider — Each Peppol service provider, across all pages.

`PeppolClient.list`

def list(*, params: OffsetPaginationParams | None = None) -> Pager[PeppolProvider]

python

List the registered Peppol service providers.

Nothing is requested until the result is consumed. Iterate it for every provider, call pages() for page-sized lists or first_page() for one request only.

Parameters

params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[PeppolProvider] — A paging object over the Peppol service providers.

Example

for provider in auth.peppol.list():
    print(provider.id, provider.name)

python

`PermissionOperation`

Import: from ksef2.clients import PermissionOperation

class PermissionOperation(OperationHandle[PermissionOperationStatusResponse, PermissionOperationStatusResponse])

python

PermissionOperation(client: PermissionsClient, response: GrantPermissionsResponse)

python

Handle to a permission grant or revoke that KSeF applies asynchronously.

Raises

KSeFApiError — If KSeF returns an API error response.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`PermissionOperation.response` (property)

response: GrantPermissionsResponse

python

Get the plain operation response.

Returns

GrantPermissionsResponse — The data model KSeF returned when it accepted the operation.

`PermissionOperation.get_status`

def get_status() -> PermissionOperationStatusResponse

python

Fetch the operation's current status without waiting.

Returns

PermissionOperationStatusResponse — The operation status; status.code is 200 once it has completed successfully.

`PermissionOperation.wait`

def wait(
    *,
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> PermissionOperationStatusResponse

python

Poll until KSeF has finished the operation.

Parameters

timeout (float) — Maximum number of seconds to wait before giving up.
poll_interval (float) — Delay in seconds between status checks.

Returns

PermissionOperationStatusResponse — The final status, with status.code 200.

Raises

KSeFPermissionOperationFailedError — If KSeF finishes the operation without applying it.
KSeFPermissionOperationTimeoutError — If polling exceeds timeout.

`PermissionsClient`

Import: from ksef2.clients import PermissionsClient

class PermissionsClient

python

PermissionsClient(transport: Middleware)

python

High-level API for permission grants, revocations, and queries.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`PermissionsClient.grant_person`

def grant_person(
    *,
    subject_type: CertificateSubjectIdentifierType,
    subject_value: str,
    permissions: list[PersonPermissionScope],
    description: str,
    first_name: str,
    last_name: str,
) -> PermissionOperation

python

Grant person-scoped permissions to a subject identifier.

Parameters

subject_type (CertificateSubjectIdentifierType) — Kind of identifier in subject_value (nip, pesel or fingerprint).
subject_value (str) — Identifier of the person receiving the permissions.
permissions (list[PersonPermissionScope]) — Permission scopes to grant; at least one is required.
description (str) — Free-text reason for the grant.
first_name (str) — First name of the person receiving the permissions.
last_name (str) — Last name of the person receiving the permissions.

Returns

PermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`PermissionsClient.grant_entity`

def grant_entity(
    *,
    subject_value: str,
    permissions: list[EntityPermission],
    description: str,
    entity_name: str,
) -> PermissionOperation

python

Grant entity permissions to a NIP-identified entity.

Parameters

subject_value (str) — NIP of the entity receiving the permissions.
permissions (list[EntityPermission]) — Permissions to grant, each with its delegation flag.
description (str) — Free-text reason for the grant.
entity_name (str) — Full name of the entity receiving the permissions.

Returns

PermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`PermissionsClient.grant_authorization`

def grant_authorization(
    *,
    subject_type: AuthorizationSubjectIdentifierType,
    subject_value: str,
    permission: AuthorizationPermissionType,
    description: str,
    entity_name: str,
) -> PermissionOperation

python

Grant an authorization permission such as self-invoicing.

Parameters

subject_type (AuthorizationSubjectIdentifierType) — Kind of identifier in subject_value (nip or peppol_id).
subject_value (str) — Identifier of the entity being authorized (NIP or Peppol ID).
permission (AuthorizationPermissionType) — Authorization scope to grant, for example self_invoicing.
description (str) — Free-text reason for the grant.
entity_name (str) — Full name of the entity being authorized.

Returns

PermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`PermissionsClient.grant_indirect`

def grant_indirect(
    *,
    subject_type: CertificateSubjectIdentifierType,
    subject_value: str,
    permissions: list[IndirectPermissionType],
    description: str,
    first_name: str,
    last_name: str,
    target_type: IndirectTargetIdentifierType | None = None,
    target_value: str | None = None,
) -> PermissionOperation

python

Grant indirect permissions, optionally limited to a target entity.

Parameters

subject_type (CertificateSubjectIdentifierType) — Kind of identifier in subject_value (nip, pesel or fingerprint).
subject_value (str) — Identifier of the person receiving the indirect permissions.
permissions (list[IndirectPermissionType]) — Indirect permission scopes to grant; at least one is required.
description (str) — Free-text reason for the grant.
first_name (str) — First name of the person receiving the permissions.
last_name (str) — Last name of the person receiving the permissions.
target_type (IndirectTargetIdentifierType | None) — Kind of identifier in target_value; None or all_partners scopes the grant to every partner.
target_value (str | None) — Identifier of the partner context the grant is limited to; None when the target is all partners.

Returns

PermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`PermissionsClient.grant_subunit`

def grant_subunit(
    *,
    subject_type: CertificateSubjectIdentifierType,
    subject_value: str,
    context_type: SubunitIdentifierType,
    context_value: str,
    description: str,
    first_name: str,
    last_name: str,
    subunit_name: str | None = None,
) -> PermissionOperation

python

Grant permissions within a subunit context.

Parameters

subject_type (CertificateSubjectIdentifierType) — Kind of identifier in subject_value (nip, pesel or fingerprint).
subject_value (str) — Identifier of the person receiving the permissions.
context_type (SubunitIdentifierType) — Kind of identifier in context_value (nip or internal_id).
context_value (str) — Identifier of the subunit context the permissions apply to.
description (str) — Free-text reason for the grant.
first_name (str) — First name of the person receiving the permissions.
last_name (str) — Last name of the person receiving the permissions.
subunit_name (str | None) — Display name of the subunit; optional.

Returns

PermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`PermissionsClient.grant_eu_entity`

def grant_eu_entity(
    *,
    subject_value: str,
    permissions: list[EuEntityPermissionType],
    description: str,
) -> PermissionOperation

python

Grant permissions to an EU entity identified by fingerprint data.

Parameters

subject_value (str) — Fingerprint of the certificate of the person receiving the permissions.
permissions (list[EuEntityPermissionType]) — EU-entity permission scopes to grant.
description (str) — Free-text reason for the grant.

Returns

PermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`PermissionsClient.grant_eu_entity_administration`

def grant_eu_entity_administration(
    *,
    subject_value: str,
    context_type: EuEntityAdminContextIdentifierType,
    context_value: str,
    description: str,
    eu_entity_name: str,
) -> PermissionOperation

python

Grant administration rights for an EU entity in a VAT UE context.

Parameters

subject_value (str) — Fingerprint of the certificate of the person receiving administration rights.
context_type (EuEntityAdminContextIdentifierType) — Kind of identifier in context_value; always nip_vat_ue.
context_value (str) — NIP-VAT UE identifier of the EU entity.
description (str) — Free-text reason for the grant.
eu_entity_name (str) — Full name of the EU entity.

Returns

PermissionOperation — A handle to the asynchronous grant operation; call wait() to wait until KSeF has applied it.

`PermissionsClient.revoke_authorization`

def revoke_authorization(*, permission_id: str) -> PermissionOperation

python

Revoke an authorization permission by permission id.

Parameters

permission_id (str) — Identifier of the authorization permission, from a query result.

Returns

PermissionOperation — A handle to the asynchronous revoke operation; call wait() to wait until KSeF has applied it.

`PermissionsClient.revoke`

def revoke(*, permission_id: str) -> PermissionOperation

python

Revoke a non-authorization permission by permission id.

Use revoke_authorization() for authorization permissions such as self-invoicing.

Parameters

permission_id (str) — Identifier of the permission, from a query result.

Returns

PermissionOperation — A handle to the asynchronous revoke operation; call wait() to wait until KSeF has applied it.

Example

operation = auth.permissions.revoke(permission_id=permission.id)
operation.wait()

python

`PermissionsClient.revoke_common`

def revoke_common(*, permission_id: str) -> GrantPermissionsResponse

python

Deprecated: revoke a non-authorization permission by permission id.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use revoke() instead.

Parameters

permission_id (str) — Identifier of the permission, from a query result.

Returns

GrantPermissionsResponse — The reference of the asynchronous revoke operation; poll get_operation_status() for the result.

`PermissionsClient.get_attachment_permission_status`

def get_attachment_permission_status() -> AttachmentPermissionStatus

python

Return whether attachments are currently allowed for the subject.

Returns

AttachmentPermissionStatus — Whether attachments are currently allowed for the subject.

`PermissionsClient.get_operation_status`

def get_operation_status(*, reference_number: str) -> PermissionOperationStatusResponse

python

Fetch the status of an asynchronous permission operation.

Parameters

reference_number (str) — Reference number returned by a grant or revoke call.

Returns

PermissionOperationStatusResponse — The operation status; status.code is 200 once it has completed successfully.

`PermissionsClient.get_entity_roles`

def get_entity_roles(
    *,
    params: OffsetPaginationParams | None = None,
) -> EntityRolesResponse

python

Deprecated: fetch one page of roles assigned to the authenticated entity.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_entity_roles() instead; first_page() fetches one page.

Parameters

params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

EntityRolesResponse — One page of roles assigned to the authenticated entity.

`PermissionsClient.query_authorizations`

def query_authorizations(
    *,
    query: AuthorizationPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> AuthorizationPermissionsQueryResponse

python

Deprecated: fetch one page of authorization grants matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_authorizations() instead; first_page() fetches one page.

Parameters

query (AuthorizationPermissionsQuery) — Filters for the authorization grants.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

AuthorizationPermissionsQueryResponse — One page of authorization grants.

`PermissionsClient.query_entities`

def query_entities(
    *,
    query: EntityPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> EntityPermissionsQueryResponse

python

Deprecated: fetch one page of entity permission grants matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_entities() instead; first_page() fetches one page.

Parameters

query (EntityPermissionsQuery) — Filters for the entity permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

EntityPermissionsQueryResponse — One page of entity permissions.

`PermissionsClient.query_eu_entities`

def query_eu_entities(
    *,
    query: EuEntityPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> EuEntityPermissionsQueryResponse

python

Deprecated: fetch one page of EU-entity permissions matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_eu_entities() instead; first_page() fetches one page.

Parameters

query (EuEntityPermissionsQuery) — Filters for the EU-entity permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

EuEntityPermissionsQueryResponse — One page of EU-entity permissions.

`PermissionsClient.query_personal`

def query_personal(
    *,
    query: PersonalPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> PersonalPermissionsQueryResponse

python

Deprecated: fetch one page of permissions held by the authenticated subject.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_personal() instead; first_page() fetches one page.

Parameters

query (PersonalPermissionsQuery) — Filters for the permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

PersonalPermissionsQueryResponse — One page of permissions held by the authenticated subject.

`PermissionsClient.query_persons`

def query_persons(
    *,
    query: PersonPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> PersonPermissionsQueryResponse

python

Deprecated: fetch one page of person permission grants matching the provided filters.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_persons() instead; first_page() fetches one page.

Parameters

query (PersonPermissionsQuery) — Filters for the person permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

PersonPermissionsQueryResponse — One page of person permissions.

`PermissionsClient.query_subordinate_entities`

def query_subordinate_entities(
    *,
    query: SubordinateEntityRolesQuery,
    params: OffsetPaginationParams | None = None,
) -> SubordinateEntityRolesQueryResponse

python

Deprecated: fetch one page of subordinate entity roles.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_subordinate_entities() instead; first_page() fetches one page.

Parameters

query (SubordinateEntityRolesQuery) — Filters for the subordinate entity roles.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

SubordinateEntityRolesQueryResponse — One page of subordinate entity roles.

`PermissionsClient.query_subunits`

def query_subunits(
    *,
    query: SubunitPermissionsQuery,
    params: OffsetPaginationParams | None = None,
) -> SubunitPermissionsQueryResponse

python

Deprecated: fetch one page of subunit permission grants.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list_subunits() instead; first_page() fetches one page.

Parameters

query (SubunitPermissionsQuery) — Filters for the subunit permissions.
params (OffsetPaginationParams | None) — Page size and offset; defaults are used when None.

Returns

SubunitPermissionsQueryResponse — One page of subunit permissions.

`PermissionsClient.list_entity_roles`

def list_entity_roles(
    *,
    params: OffsetPaginationParams | None = None,
) -> Pager[EntityRole]

python

List the roles assigned to the authenticated entity.

Nothing is requested until the result is consumed. Iterate it for every role, call pages() for page-sized lists or first_page() for one request only.

Parameters

params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[EntityRole] — A paging object over the roles assigned to the authenticated entity.

Example

for role in auth.permissions.list_entity_roles():
    print(role.role)

python

`PermissionsClient.list_authorizations`

def list_authorizations(
    query: AuthorizationPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> Pager[AuthorizationGrantDetail]

python

List the authorization grants matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every grant, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (AuthorizationPermissionsQuery) — Filters for the authorization grants.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[AuthorizationGrantDetail] — A paging object over the authorization grants.

Example

for grant in auth.permissions.list_authorizations(query):
    print(grant.id)

python

`PermissionsClient.list_entities`

def list_entities(
    query: EntityPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> Pager[EntityPermissionDetail]

python

List the entity permission grants matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every grant, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (EntityPermissionsQuery) — Filters for the entity permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[EntityPermissionDetail] — A paging object over the entity permissions.

Example

for permission in auth.permissions.list_entities(query):
    print(permission.id)

python

`PermissionsClient.list_eu_entities`

def list_eu_entities(
    query: EuEntityPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> Pager[EuEntityPermission]

python

List the EU-entity permissions matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every permission, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (EuEntityPermissionsQuery) — Filters for the EU-entity permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[EuEntityPermission] — A paging object over the EU-entity permissions.

Example

for permission in auth.permissions.list_eu_entities(query):
    print(permission.id)

python

`PermissionsClient.list_personal`

def list_personal(
    query: PersonalPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> Pager[PersonalPermissionDetail]

python

List the permissions held by the authenticated subject.

Nothing is requested until the result is consumed. Iterate it for every permission, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (PersonalPermissionsQuery) — Filters for the permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[PersonalPermissionDetail] — A paging object over the permissions held by the authenticated subject.

Example

for permission in auth.permissions.list_personal(query):
    print(permission.id)

python

`PermissionsClient.list_persons`

def list_persons(
    query: PersonPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> Pager[PersonPermissionDetail]

python

List the person permission grants matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every grant, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (PersonPermissionsQuery) — Filters for the person permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[PersonPermissionDetail] — A paging object over the person permissions.

Example

for permission in auth.permissions.list_persons(query):
    print(permission.id)

python

`PermissionsClient.list_subordinate_entities`

def list_subordinate_entities(
    query: SubordinateEntityRolesQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> Pager[SubordinateEntityRoleDetail]

python

List the subordinate entity roles matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every role, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (SubordinateEntityRolesQuery) — Filters for the subordinate entity roles.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[SubordinateEntityRoleDetail] — A paging object over the subordinate entity roles.

Example

for role in auth.permissions.list_subordinate_entities(query):
    print(role.role)

python

`PermissionsClient.list_subunits`

def list_subunits(
    query: SubunitPermissionsQuery,
    *,
    params: OffsetPaginationParams | None = None,
) -> Pager[SubunitPermission]

python

List the subunit permission grants matching the provided filters.

Nothing is requested until the result is consumed. Iterate it for every grant, call pages() for page-sized lists or first_page() for one request only.

Parameters

query (SubunitPermissionsQuery) — Filters for the subunit permissions.
params (OffsetPaginationParams | None) — Page size and offset of the first page; defaults are used when None.

Returns

Pager[SubunitPermission] — A paging object over the subunit permissions.

Example

for permission in auth.permissions.list_subunits(query):
    print(permission.id)

python

`SessionManagementClient`

Import: from ksef2.clients import SessionManagementClient

class SessionManagementClient

python

SessionManagementClient(transport: Middleware)

python

Manage authentication sessions opened through the auth API.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`SessionManagementClient.query`

def query(
    *,
    page_size: int | None = None,
    continuation_token: str | None = None,
) -> AuthenticationSessionsResponse

python

Deprecated: fetch one page of authentication sessions.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

page_size (int | None) — Maximum number of sessions to request from KSeF.
continuation_token (str | None) — Cursor returned by a previous page.

Returns

AuthenticationSessionsResponse — One page of authentication sessions for the current subject.

`SessionManagementClient.all`

def all(*, page_size: int | None = None) -> Iterator[AuthenticationSessionsResponse]

python

Deprecated: iterate through all authentication session pages.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list().pages() instead.

Parameters

page_size (int | None) — Maximum number of sessions to request per page.

Yields

AuthenticationSessionsResponse — Successive pages of authentication sessions until KSeF stops
AuthenticationSessionsResponse — returning a continuation token.

`SessionManagementClient.terminate_current`

def terminate_current() -> None

python

Terminate the authentication session backing the current bearer token.

`SessionManagementClient.terminate`

def terminate(reference_number: str) -> None

python

Terminate an authentication session by reference number.

Parameters

reference_number (str) — Reference number of the authentication session to terminate.

`SessionManagementClient.close`

def close(*, reference_number: str) -> None

python

Deprecated: terminate an authentication session by reference number.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use terminate() instead.

Parameters

reference_number (str) — Reference number of the authentication session to terminate.

Returns

None — None once the session is terminated.

`SessionManagementClient.list`

def list(*, page_size: int | None = None) -> Pager[AuthenticationSession]

python

List the authentication sessions of the current subject.

Nothing is requested until the result is consumed. Iterate it for every session, call pages() for page-sized lists or first_page() for one request only.

Parameters

page_size (int | None) — Maximum number of sessions to request per page; the KSeF default when None.

Returns

Pager[AuthenticationSession] — A paging object over the authentication sessions.

Example

for session in auth.sessions.list():
    print(session.reference_number)

python

`TemporalTestData`

Import: from ksef2.clients import TemporalTestData

class TemporalTestData

python

TemporalTestData(client: TestDataClient)

python

Context manager that records testdata mutations and reverts them on exit.

Cleanup failures are logged and suppressed. Direct mutation methods expose the same catch contract as the parent test-data branch.

Raises

KSeFApiError — If KSeF returns an API error response during direct mutations.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`TemporalTestData.create_subject`

def create_subject(
    *,
    nip: str,
    subject_type: SubjectType,
    description: str,
    subunits: list[SubUnit] | None = None,
    created_date: datetime | None = None,
) -> None

python

Create a TEST subject and register it for context-manager cleanup.

Parameters

nip (str) — NIP of the subject to create.
subject_type (SubjectType) — Kind of subject, such as an enforcement authority or a VAT group.
description (str) — Description of the subject.
subunits (list[SubUnit] | None) — Subunits to create with the subject; None for none.
created_date (datetime | None) — Creation date to record for the subject; None for now.

`TemporalTestData.delete_subject`

def delete_subject(*, nip: str) -> None

python

Delete a TEST subject and remove it from cleanup tracking.

Parameters

nip (str) — NIP of the subject to delete.

`TemporalTestData.create_person`

def create_person(
    *,
    nip: str,
    pesel: str,
    description: str,
    is_bailiff: bool = False,
    is_deceased: bool = False,
    created_date: datetime | None = None,
) -> None

python

Create a TEST person and register it for context-manager cleanup.

Parameters

nip (str) — NIP of the person.
pesel (str) — PESEL of the person.
description (str) — Description of the person.
is_bailiff (bool) — Whether the person is a bailiff.
is_deceased (bool) — Whether the person is marked as deceased.
created_date (datetime | None) — Creation date to record for the person; None for now.

`TemporalTestData.delete_person`

def delete_person(*, nip: str) -> None

python

Delete a TEST person and remove it from cleanup tracking.

Parameters

nip (str) — NIP of the person to delete.

`TemporalTestData.grant_permissions`

def grant_permissions(
    *,
    permissions: list[Permission],
    grant_to: Identifier,
    in_context_of: Identifier,
) -> None

python

Grant TEST permissions and register them for cleanup.

Parameters

permissions (list[Permission]) — Permissions to grant.
grant_to (Identifier) — Identifier of the subject receiving the permissions.
in_context_of (Identifier) — Identifier of the context in which the permissions apply.

`TemporalTestData.revoke_permissions`

def revoke_permissions(*, revoke_from: Identifier, in_context_of: Identifier) -> None

python

Revoke TEST permissions and remove them from cleanup tracking.

Parameters

revoke_from (Identifier) — Identifier of the subject losing the permissions.
in_context_of (Identifier) — Identifier of the context from which the permissions are revoked.

`TemporalTestData.enable_attachments`

def enable_attachments(*, nip: str) -> None

python

Enable TEST attachments and register the context for cleanup.

Parameters

nip (str) — NIP of the subject to enable attachments for.

`TemporalTestData.revoke_attachments`

def revoke_attachments(*, nip: str, expected_end_date: date | None = None) -> None

python

Revoke TEST attachment access and remove it from cleanup tracking.

Parameters

nip (str) — NIP of the subject to revoke attachments for.
expected_end_date (date | None) — Date on which attachment permission should end; None to revoke immediately.

`TemporalTestData.block_context`

def block_context(*, context: AuthContextIdentifier) -> None

python

Block a TEST context and register it for cleanup.

Parameters

context (AuthContextIdentifier) — Authentication context to block.

`TemporalTestData.unblock_context`

def unblock_context(*, context: AuthContextIdentifier) -> None

python

Unblock a TEST context and remove it from cleanup tracking.

Parameters

context (AuthContextIdentifier) — Authentication context to unblock.

`TestDataClient`

Import: from ksef2.clients import TestDataClient

class TestDataClient

python

TestDataClient(transport: Middleware)

python

Helper for TEST-only endpoints that seed and mutate sandbox data.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`TestDataClient.create_subject`

def create_subject(
    *,
    nip: str,
    subject_type: SubjectType,
    description: str,
    subunits: list[SubUnit] | None = None,
    created_date: datetime | None = None,
) -> None

python

Create a test subject with optional subunits.

Parameters

nip (str) — NIP of the subject to create.
subject_type (SubjectType) — Kind of subject, such as an enforcement authority or a VAT group.
description (str) — Description of the subject.
subunits (list[SubUnit] | None) — Subunits to create with the subject; None for none.
created_date (datetime | None) — Creation date to record for the subject; None for now.

`TestDataClient.delete_subject`

def delete_subject(*, nip: str) -> None

python

Delete a test subject by NIP.

Parameters

nip (str) — NIP of the subject to delete.

`TestDataClient.create_person`

def create_person(
    *,
    nip: str,
    pesel: str,
    description: str,
    is_bailiff: bool = False,
    is_deceased: bool = False,
    created_date: datetime | None = None,
) -> None

python

Create a test person in the chosen subject.

Parameters

nip (str) — NIP of the person.
pesel (str) — PESEL of the person.
description (str) — Description of the person.
is_bailiff (bool) — Whether the person is a bailiff.
is_deceased (bool) — Whether the person is marked as deceased.
created_date (datetime | None) — Creation date to record for the person; None for now.

`TestDataClient.delete_person`

def delete_person(*, nip: str) -> None

python

Delete a test person by subject NIP.

Parameters

nip (str) — NIP of the person to delete.

`TestDataClient.grant_permissions`

def grant_permissions(
    *,
    permissions: list[Permission],
    grant_to: Identifier,
    in_context_of: Identifier,
) -> None

python

Grant test permissions in a chosen context.

Parameters

permissions (list[Permission]) — Permissions to grant.
grant_to (Identifier) — Identifier of the subject receiving the permissions.
in_context_of (Identifier) — Identifier of the context in which the permissions apply.

`TestDataClient.revoke_permissions`

def revoke_permissions(*, revoke_from: Identifier, in_context_of: Identifier) -> None

python

Revoke test permissions in a chosen context.

Parameters

revoke_from (Identifier) — Identifier of the subject losing the permissions.
in_context_of (Identifier) — Identifier of the context from which the permissions are revoked.

`TestDataClient.enable_attachments`

def enable_attachments(*, nip: str) -> None

python

Enable invoice attachments for a test subject.

Parameters

nip (str) — NIP of the subject to enable attachments for.

`TestDataClient.revoke_attachments`

def revoke_attachments(*, nip: str, expected_end_date: date | None = None) -> None

python

Revoke attachment permissions, optionally effective on a given date.

Parameters

nip (str) — NIP of the subject to revoke attachments for.
expected_end_date (date | None) — Date on which attachment permission should end; None to revoke immediately.

`TestDataClient.block_context`

def block_context(*, context: AuthContextIdentifier) -> None

python

Block authentication for a specific test context.

Parameters

context (AuthContextIdentifier) — Authentication context to block.

`TestDataClient.unblock_context`

def unblock_context(*, context: AuthContextIdentifier) -> None

python

Unblock authentication for a specific test context.

Parameters

context (AuthContextIdentifier) — Authentication context to unblock.

`TestDataClient.update_certificate_valid_to`

def update_certificate_valid_to(*, serial_number: str, valid_to: datetime) -> None

python

Shorten the validity of a certificate in the TEST environment.

Parameters

serial_number (str) — Serial number of the certificate to shorten.
valid_to (datetime) — New, earlier end of the certificate validity period.

`TestDataClient.temporal`

def temporal() -> TemporalTestData

python

Return a context manager that automatically cleans up created test data.

Returns

TemporalTestData — A context manager that automatically cleans up created test data.

`TokensClient`

Import: from ksef2.clients import TokensClient

class TokensClient

python

TokensClient(transport: Middleware)

python

High-level API for the KSeF token lifecycle.

Catch KSeFException for SDK-classified failures raised by this branch, and httpx.HTTPError for transport failures.

Raises

KSeFApiError — If KSeF returns an API error response. Catch KSeFAuthError for authentication or authorization failures and KSeFRateLimitError for throttling.
KSeFValidationError — If a KSeF response cannot be parsed into SDK models.
httpx.HTTPError — If the HTTP transport fails before KSeF returns a response.

Methods

`TokensClient.wait_for_activation`

def wait_for_activation(
    *,
    reference_number: str,
    timeout: float = 60.0,
    poll_interval: float = 1.0,
) -> TokenStatusResponse

python

Deprecated: wait until a generated token becomes active or reaches a terminal state.

Deprecated

May be removed no earlier than ksef2 2.0.0. Call wait() on the handle returned by generate() instead.

Parameters

reference_number (str) — Reference returned by generate.
timeout (float) — Maximum number of seconds to wait for activation.
poll_interval (float) — Delay in seconds between status checks.

Returns

TokenStatusResponse — The first status response that reports the token as active.

Raises

KSeFApiError — If activation ends in a terminal failure state.
KSeFTokenStatusTimeoutError — If polling exceeds timeout.
httpx.HTTPError — If a status request fails at the transport boundary.

`TokensClient.generate`

def generate(
    *,
    permissions: builtins.list[TokenPermission],
    description: str,
) -> GeneratedToken

python

Create a token and return a handle exposing its one-time credential.

Parameters

permissions (builtins.list[TokenPermission]) — Permissions to include in the generated token.
description (str) — Human-readable label shown in KSeF token listings.

Returns

GeneratedToken — A handle whose token is the one-time secret KSeF returns only once;
GeneratedToken — persist it before calling the handle's wait() for activation.

Example

token = auth.tokens.generate(
    permissions=["invoice_read"], description="reporting"
)
secret = token.token  # store it now, KSeF will not show it again
token.wait()

python

`TokensClient.list_page`

def list_page(
    *,
    continuation_token: str | None = None,
    params: TokenListParams | None = None,
) -> QueryTokensResponse

python

Deprecated: fetch one page of tokens using optional filters and continuation state.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list() instead; first_page() fetches one page.

Parameters

continuation_token (str | None) — Token identifying the next page to fetch.
params (TokenListParams | None) — Optional filters and page size for the request.

Returns

QueryTokensResponse — A single page of token results.

`TokensClient.list_all`

def list_all(*, params: TokenListParams | None = None) -> Iterator[QueryTokensResponse]

python

Deprecated: iterate through all token pages until KSeF stops returning a continuation token.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use list().pages() instead.

Parameters

params (TokenListParams | None) — Optional filters and page size applied to every request.

Yields

QueryTokensResponse — Each page returned by the token listing endpoint.

`TokensClient.get_status`

def get_status(*, reference_number: str) -> TokenStatusResponse

python

Return the current status of a token.

Parameters

reference_number (str) — Reference number of the token.

Returns

TokenStatusResponse — The token's current lifecycle status.

`TokensClient.status`

def status(*, reference_number: str) -> TokenStatusResponse

python

Deprecated: return the current status of a token.

Deprecated

May be removed no earlier than ksef2 2.0.0. Use get_status() instead.

Parameters

reference_number (str) — Reference number of the token.

Returns

TokenStatusResponse — The token's current lifecycle status.

`TokensClient.revoke`

def revoke(*, reference_number: str) -> None

python

Revoke a token.

Parameters

reference_number (str) — Reference number of the token to revoke.

`TokensClient.list`

def list(*, params: TokenListParams | None = None) -> Pager[TokenInfo]

python

List the tokens of the authenticated context.

Nothing is requested until the result is consumed. Iterate it for every token, call pages() for page-sized lists or first_page() for one request only.

Parameters

params (TokenListParams | None) — Optional filters and page size applied to every request.

Returns

Pager[TokenInfo] — A paging object over the tokens matching params.

Example

for token in auth.tokens.list():
    print(token.reference_number, token.status)

python

Classes

AsyncAuthClient

AsyncAuthClient.resume

AsyncAuthClient.with_token

AsyncAuthClient.with_xades

AsyncAuthClient.with_test_certificate

AsyncAuthClient.with_profile

AsyncAuthClient.refresh

AsyncAuthenticatedClient

AsyncAuthenticatedClient.auth_tokens (property)

AsyncAuthenticatedClient.access_token (property)

AsyncAuthenticatedClient.refresh_token (property)

AsyncAuthenticatedClient.resume_state

AsyncAuthenticatedClient.get_encryption_key

AsyncAuthenticatedClient.online_session

AsyncAuthenticatedClient.resume_online_session

AsyncAuthenticatedClient.batch_session

AsyncAuthenticatedClient.open_batch_session

AsyncAuthenticatedClient.resume_batch_session

AsyncAuthenticatedClient.invoices (property)

AsyncAuthenticatedClient.batch (property)

AsyncAuthenticatedClient.limits (property)

AsyncAuthenticatedClient.collective_identifiers (property)

AsyncAuthenticatedClient.tokens (property)

AsyncAuthenticatedClient.certificates (property)

AsyncAuthenticatedClient.sessions (property)

AsyncAuthenticatedClient.invoice_sessions (property)

AsyncAuthenticatedClient.permissions (property)

AsyncAuthenticatedClient.testdata (property)

AsyncAuthenticatedClient.raw (property)

AsyncBatchService

AsyncBatchService.prepare

AsyncBatchService.submit

AsyncBatchService.prepare_batch

AsyncBatchService.prepare_batch_from_paths

AsyncBatchService.open_session

AsyncBatchService.upload_parts

AsyncBatchService.submit_prepared_batch

AsyncBatchService.submit_batch

AsyncBatchService.get_status

AsyncBatchService.list_invoices

AsyncBatchService.list_failed_invoices

AsyncBatchService.get_upo

AsyncBatchService.wait_for_completion

AsyncBatchSessionClient

AsyncBatchSessionClient.reference_number (property)

AsyncBatchSessionClient.access_token (property)

AsyncBatchSessionClient.aes_key (property)

AsyncBatchSessionClient.iv (property)

AsyncBatchSessionClient.part_upload_requests (property)

AsyncBatchSessionClient.resume_state

AsyncBatchSessionClient.get_state

AsyncBatchSessionClient.get_status

AsyncBatchSessionClient.list_invoices

AsyncBatchSessionClient.list_failed_invoices

AsyncBatchSessionClient.get_upo

AsyncBatchSessionClient.wait

AsyncBatchSessionClient.download_upo

AsyncBatchSessionClient.upload_parts

AsyncBatchSessionClient.aclose

AsyncCertificateEnrollment

AsyncCertificateEnrollment.timestamp (property)

AsyncCertificateEnrollment.response (property)

AsyncCertificateEnrollment.get_status

AsyncCertificateEnrollment.wait

AsyncCertificatesClient

AsyncCertificatesClient.get_limits

AsyncCertificatesClient.get_enrollment_data

AsyncCertificatesClient.enroll

AsyncCertificatesClient.get_enrollment_status

AsyncCertificatesClient.retrieve

AsyncCertificatesClient.revoke

AsyncCertificatesClient.query

AsyncCertificatesClient.all

AsyncCertificatesClient.list

AsyncClient

AsyncClient.authentication (property)

AsyncClient.encryption (property)

AsyncClient.peppol (property)

AsyncClient.testdata (property)

`AsyncAuthClient`

`AsyncAuthClient.resume`

`AsyncAuthClient.with_token`

`AsyncAuthClient.with_xades`

`AsyncAuthClient.with_test_certificate`

`AsyncAuthClient.with_profile`

`AsyncAuthClient.refresh`

`AsyncAuthenticatedClient`

`AsyncAuthenticatedClient.auth_tokens` (property)

`AsyncAuthenticatedClient.access_token` (property)

`AsyncAuthenticatedClient.refresh_token` (property)

`AsyncAuthenticatedClient.resume_state`

`AsyncAuthenticatedClient.get_encryption_key`

`AsyncAuthenticatedClient.online_session`

`AsyncAuthenticatedClient.resume_online_session`

`AsyncAuthenticatedClient.batch_session`

`AsyncAuthenticatedClient.open_batch_session`

`AsyncAuthenticatedClient.resume_batch_session`

`AsyncAuthenticatedClient.invoices` (property)

`AsyncAuthenticatedClient.batch` (property)

`AsyncAuthenticatedClient.limits` (property)

`AsyncAuthenticatedClient.collective_identifiers` (property)

`AsyncAuthenticatedClient.tokens` (property)

`AsyncAuthenticatedClient.certificates` (property)

`AsyncAuthenticatedClient.sessions` (property)

`AsyncAuthenticatedClient.invoice_sessions` (property)

`AsyncAuthenticatedClient.permissions` (property)

`AsyncAuthenticatedClient.testdata` (property)

`AsyncAuthenticatedClient.raw` (property)

`AsyncBatchService`

`AsyncBatchService.prepare`

`AsyncBatchService.submit`

`AsyncBatchService.prepare_batch`

`AsyncBatchService.prepare_batch_from_paths`

`AsyncBatchService.open_session`

`AsyncBatchService.upload_parts`

`AsyncBatchService.submit_prepared_batch`

`AsyncBatchService.submit_batch`

`AsyncBatchService.get_status`

`AsyncBatchService.list_invoices`

`AsyncBatchService.list_failed_invoices`

`AsyncBatchService.get_upo`

`AsyncBatchService.wait_for_completion`

`AsyncBatchSessionClient`

`AsyncBatchSessionClient.reference_number` (property)

`AsyncBatchSessionClient.access_token` (property)

`AsyncBatchSessionClient.aes_key` (property)

`AsyncBatchSessionClient.iv` (property)

`AsyncBatchSessionClient.part_upload_requests` (property)

`AsyncBatchSessionClient.resume_state`

`AsyncBatchSessionClient.get_state`

`AsyncBatchSessionClient.get_status`

`AsyncBatchSessionClient.list_invoices`

`AsyncBatchSessionClient.list_failed_invoices`

`AsyncBatchSessionClient.get_upo`

`AsyncBatchSessionClient.wait`

`AsyncBatchSessionClient.download_upo`

`AsyncBatchSessionClient.upload_parts`

`AsyncBatchSessionClient.aclose`

`AsyncCertificateEnrollment`

`AsyncCertificateEnrollment.timestamp` (property)

`AsyncCertificateEnrollment.response` (property)

`AsyncCertificateEnrollment.get_status`

`AsyncCertificateEnrollment.wait`

`AsyncCertificatesClient`

`AsyncCertificatesClient.get_limits`

`AsyncCertificatesClient.get_enrollment_data`

`AsyncCertificatesClient.enroll`

`AsyncCertificatesClient.get_enrollment_status`

`AsyncCertificatesClient.retrieve`

`AsyncCertificatesClient.revoke`

`AsyncCertificatesClient.query`

`AsyncCertificatesClient.all`

`AsyncCertificatesClient.list`

`AsyncClient`

`AsyncClient.authentication` (property)

`AsyncClient.encryption` (property)

`AsyncClient.peppol` (property)

`AsyncClient.testdata` (property)

`AsyncClient.raw` (property)