ksef2.clients

Public sync and async client entry points.

Klasy

AsyncAuthClient

Import: from ksef2.clients import AsyncAuthClient

class AsyncAuthClient

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

Async entry point for creating authenticated SDK clients.

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

Zgłasza

• 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.

Metody

AsyncAuthClient.resume

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

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.

Parametry

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

Zwraca

• 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

Authenticate with a KSeF token and return an authenticated client.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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",
)

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

Authenticate with an XAdES-signed challenge response.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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"),
)

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

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

Parametry

• 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.

Zwraca

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

Zgłasza

• 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")

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

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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")

AsyncAuthClient.refresh

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

Exchange a refresh token for a new access token.

Parametry

• refresh_token (str) — Refresh token obtained during authentication.

Zwraca

• RefreshedToken — A new access token with its expiry time.

AsyncAuthenticatedClient

Import: from ksef2.clients import AsyncAuthenticatedClient

class AsyncAuthenticatedClient

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

Authenticated async entry point for KSeF operations.

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

Zgłasza

• 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.

Metody

AsyncAuthenticatedClient.auth_tokens (właściwość)

auth_tokens: AuthTokens

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.

Zwraca

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

AsyncAuthenticatedClient.access_token (właściwość)

access_token: str

Return the bearer access token string used for authenticated calls.

Zwraca

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

AsyncAuthenticatedClient.refresh_token (właściwość)

refresh_token: str

Return the refresh token string paired with the access token.

Zwraca

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

AsyncAuthenticatedClient.resume_state

def resume_state() -> AuthenticationResumeState

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.

Zwraca

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

AsyncAuthenticatedClient.get_encryption_key

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

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

Zwraca

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

Zgłasza

• 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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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()

AsyncAuthenticatedClient.resume_online_session

def resume_online_session(state: OnlineSessionResumeState) -> AsyncOnlineSessionClient

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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()

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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• KSeFValidationError — If the batch session request is invalid.

AsyncAuthenticatedClient.resume_batch_session

def resume_batch_session(state: BatchSessionResumeState) -> AsyncBatchSessionClient

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.

Parametry

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

Zwraca

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

AsyncAuthenticatedClient.invoices (właściwość)

invoices: AsyncInvoicesService

Return the invoices service with encryption support configured.

Zwraca

• AsyncInvoicesService — The invoices service with encryption support configured.

AsyncAuthenticatedClient.batch (właściwość)

batch: AsyncBatchService

Return the high-level batch upload workflow service.

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

Zwraca

• AsyncBatchService — The batch upload service.

AsyncAuthenticatedClient.limits (właściwość)

limits: AsyncLimitsClient

Return the authenticated rate-limit branch.

Zwraca

• AsyncLimitsClient — The authenticated rate-limit branch.

AsyncAuthenticatedClient.collective_identifiers (właściwość)

collective_identifiers: AsyncCollectiveIdentifiersClient

Return the collective invoice identifier branch.

Zwraca

• AsyncCollectiveIdentifiersClient — The collective invoice identifier branch.

AsyncAuthenticatedClient.tokens (właściwość)

tokens: AsyncTokensClient

Return the authenticated token lifecycle branch.

Zwraca

• AsyncTokensClient — The authenticated token lifecycle branch.

AsyncAuthenticatedClient.certificates (właściwość)

certificates: AsyncCertificatesClient

Return the authenticated certificate enrollment branch.

Zwraca

• AsyncCertificatesClient — The authenticated certificate enrollment branch.

AsyncAuthenticatedClient.sessions (właściwość)

sessions: AsyncSessionManagementClient

Return the authenticated session-management branch.

Zwraca

• AsyncSessionManagementClient — The authenticated session-management branch.

AsyncAuthenticatedClient.invoice_sessions (właściwość)

invoice_sessions: AsyncInvoiceSessionsClient

Return the authenticated invoice-session history branch.

Zwraca

• AsyncInvoiceSessionsClient — The authenticated invoice-session history branch.

AsyncAuthenticatedClient.permissions (właściwość)

permissions: AsyncPermissionsClient

Return the authenticated permissions branch.

Zwraca

• AsyncPermissionsClient — The authenticated permissions branch.

AsyncAuthenticatedClient.testdata (właściwość)

testdata: AsyncTestDataClient

Return authenticated TEST-only data mutation helpers.

Zwraca

• AsyncTestDataClient — Authenticated TEST-only data mutation helpers.

AsyncAuthenticatedClient.raw (właściwość)

raw: AsyncRawAuthenticatedClient

Return raw authenticated endpoints for advanced async integrations.

Zwraca

• AsyncRawAuthenticatedClient — Raw authenticated endpoints for advanced async integrations.

AsyncBatchService

Import: from ksef2.clients import AsyncBatchService

class AsyncBatchService

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

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.

Zgłasza

• 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.

Metody

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

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

Parametry

• 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.

Zwraca

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

Zgłasza

• 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)

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

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

Parametry

• 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.

Zwraca

• 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.

Zgłasza

• 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()

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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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]

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.

Parametry

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

Zwraca

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

Zgłasza

• KSeFValidationError — If the prepared batch cannot be opened.

AsyncBatchService.upload_parts

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

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

Parametry

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

Zgłasza

• 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]

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.

Parametry

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

Zwraca

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

Zgłasza

• 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]

Deprecated: prepare and submit a batch in one call.

Deprecated

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

Parametry

• 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.

Zwraca

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

Zgłasza

• 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]

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

• 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.

Zwraca

• 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]

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.

Parametry

• 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.

Zwraca

• 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]

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.

Parametry

• 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.

Zwraca

• 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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

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

AsyncBatchSessionClient

Import: from ksef2.clients import AsyncBatchSessionClient

class AsyncBatchSessionClient

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

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.

Zgłasza

• 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.

Metody

AsyncBatchSessionClient.reference_number (właściwość)

reference_number: str

Get the batch session reference number.

Zwraca

• str — The KSeF reference number of the session.

AsyncBatchSessionClient.access_token (właściwość)

access_token: str

Deprecated compatibility accessor for the current bearer token.

Deprecated

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

Zwraca

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

Zgłasza

• KSeFValidationError — If the session state carries no bearer token.

AsyncBatchSessionClient.aes_key (właściwość)

aes_key: bytes

Get the AES key for encrypting batch files.

Zwraca

• bytes — The raw AES key.

AsyncBatchSessionClient.iv (właściwość)

iv: bytes

Get the initialization vector for AES encryption.

Zwraca

• bytes — The raw initialization vector.

AsyncBatchSessionClient.part_upload_requests (właściwość)

part_upload_requests: list[PartUploadRequest]

Get the upload instructions for each file part.

Zwraca

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

AsyncBatchSessionClient.resume_state

def resume_state() -> BatchSessionResumeState

Return the sensitive session state needed to resume later.

Zwraca

• BatchSessionResumeState — The sensitive session state needed to resume later.

AsyncBatchSessionClient.get_state

def get_state() -> BatchSessionResumeState

Deprecated compatibility wrapper for resume_state().

Deprecated

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

Zwraca

• BatchSessionResumeState — The same state as resume_state().

AsyncBatchSessionClient.get_status

async def get_status() -> SessionStatusResponse

Fetch the current processing state of the batch session.

Zwraca

• SessionStatusResponse — The session status, including invoice counters.

AsyncBatchSessionClient.list_invoices

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

Fetch one page of invoices submitted in this batch session.

Parametry

• 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.

Zwraca

• 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

Fetch one page of failed invoices from this batch session.

Parametry

• 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.

Zwraca

• 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]

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.

Parametry

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

Zwraca

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

AsyncBatchSessionClient.wait

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

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.

Parametry

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

Zwraca

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

Zgłasza

• 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]

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.

Zwraca

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

Zgłasza

• 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

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

Zgłasza

• 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

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])

AsyncCertificateEnrollment(
    client: AsyncCertificatesClient,
    response: CertificateEnrollmentResponse,
)

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.

Zgłasza

• 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.

Metody

AsyncCertificateEnrollment.timestamp (właściwość)

timestamp: datetime

Get when KSeF accepted the enrollment request.

Zwraca

• datetime — The acceptance time reported by KSeF.

AsyncCertificateEnrollment.response (właściwość)

response: CertificateEnrollmentResponse

Get the plain enrollment response.

Zwraca

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

AsyncCertificateEnrollment.get_status

async def get_status() -> CertificateEnrollmentStatusResponse

Fetch the enrollment's current status without waiting.

Zwraca

• 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

Poll until KSeF has issued the certificate.

Parametry

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

Zwraca

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

Zgłasza

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

AsyncCertificatesClient

Import: from ksef2.clients import AsyncCertificatesClient

class AsyncCertificatesClient

AsyncCertificatesClient(transport: AsyncMiddleware)

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.

Zgłasza

• 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.

Metody

AsyncCertificatesClient.get_limits

async def get_limits() -> CertificateLimitsResponse

Return the effective certificate enrollment and issuance quotas.

Zwraca

• CertificateLimitsResponse — The effective certificate enrollment and issuance quotas.

AsyncCertificatesClient.get_enrollment_data

async def get_enrollment_data() -> CertificateEnrollmentData

Return subject data needed to prepare a CSR for enrollment.

Zwraca

• 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

Request issuance of a certificate from a CSR.

Parametry

• 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.

Zwraca

• 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)

AsyncCertificatesClient.get_enrollment_status

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

Fetch the current status of a certificate enrollment request.

Parametry

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

Zwraca

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

AsyncCertificatesClient.retrieve

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

Download issued certificates by serial number.

Parametry

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

Zwraca

• RetrievedCertificatesList — The requested certificates with their DER data.

AsyncCertificatesClient.revoke

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

Revoke a certificate, optionally providing a revocation reason.

Parametry

• 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]

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.

Parametry

• 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.

Zwraca

• 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]

Deprecated: iterate over all certificates matching the provided filters.

Deprecated

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

Parametry

• 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.

Zwraca (generator)

• 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]

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.

Parametry

• 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.

Zwraca

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

Example

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

AsyncClient

Import: from ksef2.clients import AsyncClient

class AsyncClient

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

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.

Zgłasza

• 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)

Metody

AsyncClient.authentication (właściwość)

authentication: AsyncAuthClient

Return the authentication entry point.

Zwraca

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

Zgłasza

• KSeFClientClosedError — If the root client has been closed.

AsyncClient.encryption (właściwość)

encryption: AsyncEncryptionClient

Return the public encryption-certificate client.

Zwraca

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

Zgłasza

• KSeFClientClosedError — If the root client has been closed.

AsyncClient.peppol (właściwość)

peppol: AsyncPeppolClient

Return the public Peppol provider client.

Zwraca

• AsyncPeppolClient — The client for listing registered Peppol providers.

Zgłasza

• KSeFClientClosedError — If the root client has been closed.

AsyncClient.testdata (właściwość)

testdata: AsyncTestDataClient

Return the TEST-only data seeding client.

Zwraca

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

Zgłasza

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

AsyncClient.raw (właściwość)

raw: AsyncRawClient

Return raw unauthenticated endpoints for advanced async integrations.

Zwraca

• AsyncRawClient — Raw unauthenticated endpoints for advanced async integrations.

AsyncClient.authenticated

def authenticated(auth_tokens: AuthTokens) -> AsyncAuthenticatedClient

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.

Parametry

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

Zwraca

• AsyncAuthenticatedClient — An authenticated client bound to the tokens.

AsyncClient.aclose

async def aclose() -> None

Close owned resources and invalidate cached child clients.

AsyncCollectiveIdentifiersClient

Import: from ksef2.clients import AsyncCollectiveIdentifiersClient

class AsyncCollectiveIdentifiersClient

AsyncCollectiveIdentifiersClient(transport: AsyncMiddleware)

API for generating and querying collective invoice identifiers.

Metody

AsyncCollectiveIdentifiersClient.generate

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

Generate a collective identifier for the supplied invoices.

Parametry

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

Zwraca

• 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]

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.

Parametry

• 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.

Zwraca

• 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]

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.

Parametry

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

Zwraca (generator)

• 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]

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.

Parametry

• 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.

Zwraca

• 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]

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.

Parametry

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

Zwraca (generator)

• 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]

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.

Parametry

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

Zwraca (generator)

• 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]

List the collective identifiers that reference one KSeF invoice.

Nothing is requested until the result is consumed.

Parametry

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

Zwraca

• 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)

AsyncCollectiveIdentifiersClient.list_invoices

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

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.

Parametry

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

Zwraca

• 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)

AsyncCollectiveIdentifiersClient.list

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

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.

Parametry

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

Zwraca

• 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)

AsyncEncryptionClient

Import: from ksef2.clients import AsyncEncryptionClient

class AsyncEncryptionClient

AsyncEncryptionClient(transport: AsyncMiddleware)

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.

Zgłasza

• 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.

Metody

AsyncEncryptionClient.get_certificates

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

Return public certificates, optionally filtered by supported usage.

Parametry

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

Zwraca

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

AsyncExportJob

Import: from ksef2.clients import AsyncExportJob

class AsyncExportJob(AsyncOperationHandle[InvoiceExportStatusResponse, ExportedInvoices])

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

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.

Zgłasza

• 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.

Metody

AsyncExportJob.resume_state

def resume_state() -> ExportResumeState

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.

Zwraca

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

AsyncExportJob.get_status

async def get_status() -> InvoiceExportStatusResponse

Fetch the export's current status without waiting.

Zwraca

• 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

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

Parametry

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

Zwraca

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

Zgłasza

• 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])

AsyncGeneratedToken(client: AsyncTokensClient, response: GenerateTokenResponse)

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.

Zgłasza

• 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.

Metody

AsyncGeneratedToken.token (właściwość)

token: str

Get the one-time token value.

Zwraca

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

AsyncGeneratedToken.response (właściwość)

response: GenerateTokenResponse

Get the plain generation response.

Zwraca

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

AsyncGeneratedToken.to_sensitive_dict

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

Export the one-time token for deliberately protected persistence.

Zwraca

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

AsyncGeneratedToken.get_status

async def get_status() -> TokenStatusResponse

Fetch the token's current status without waiting.

Zwraca

• TokenStatusResponse — The token's current lifecycle status.

AsyncGeneratedToken.wait

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

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

Parametry

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

Zwraca

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

Zgłasza

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

AsyncInvoiceSessionsClient

Import: from ksef2.clients import AsyncInvoiceSessionsClient

class AsyncInvoiceSessionsClient

AsyncInvoiceSessionsClient(transport: AsyncMiddleware)

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.

Zgłasza

• 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.

Metody

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

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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]

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.

Parametry

• 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.

Zwraca

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

AsyncInvoiceSessionsClient.all

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

Deprecated: iterate through all pages of invoice session history.

Deprecated

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

Parametry

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

Zwraca (generator)

• 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]

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.

Parametry

• 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.

Zwraca

• 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)

AsyncInvoicesService

Import: from ksef2.clients import AsyncInvoicesService

class AsyncInvoicesService

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

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.

Zgłasza

• 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.

Metody

AsyncInvoicesService.search

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

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.

Parametry

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

Zwraca

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

Zgłasza

• 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)

AsyncInvoicesService.download

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

Download one processed invoice by KSeF number.

Parametry

• 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.

Zwraca

• bytes — The invoice XML as bytes.

Zgłasza

• 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)

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

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).

Parametry

• 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.

Zwraca

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

Zgłasza

• 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/")

AsyncInvoicesService.query_metadata

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

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca (generator)

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

Zgłasza

• KSeFMetadataPaginationError — If KSeF returns inconsistent pagination boundaries.

AsyncInvoicesService.all_metadata

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

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.

Parametry

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

Zwraca (generator)

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

Zgłasza

• KSeFMetadataPaginationError — If KSeF returns inconsistent pagination boundaries.

AsyncInvoicesService.download_invoice

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

Deprecated: download one processed invoice by KSeF number.

Deprecated

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

Parametry

• ksef_number (str) — KSeF number of the invoice.

Zwraca

• 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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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]

Deprecated: schedule an encrypted invoice export.

Deprecated

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

Parametry

• 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.

Zwraca

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

Zgłasza

• 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]

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.

Parametry

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

Zwraca

• 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]]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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]]

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.

Parametry

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

Zwraca

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

Zgłasza

• 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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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]]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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])

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

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.

Zgłasza

• 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.

Metody

AsyncInvoiceSubmission.response (właściwość)

response: invoices.SendInvoiceResponse

Get the plain submission response.

Zwraca

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

AsyncInvoiceSubmission.get_status

async def get_status() -> SessionInvoiceStatusResponse

Fetch the invoice's current processing status without waiting.

Zwraca

• SessionInvoiceStatusResponse — The processing status of the invoice.

AsyncInvoiceSubmission.wait

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

Poll until KSeF assigns the invoice a final processing result.

Parametry

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

Zwraca

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

Zgłasza

• 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

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.

Zwraca

• bytes — The UPO as XML bytes.

Zgłasza

• 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

AsyncLimitsClient(transport: AsyncMiddleware)

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.

Zgłasza

• 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.

Metody

AsyncLimitsClient.get_context_limits

async def get_context_limits() -> ContextLimits

Return the effective limits for online and batch sessions.

Zwraca

• ContextLimits — The effective limits for online and batch sessions.

AsyncLimitsClient.get_subject_limits

async def get_subject_limits() -> SubjectLimits

Return the effective limits for certificate enrollments and issuance.

Zwraca

• SubjectLimits — The effective limits for certificate enrollments and issuance.

AsyncLimitsClient.get_api_rate_limits

async def get_api_rate_limits() -> ApiRateLimits

Return the effective per-endpoint API rate limits.

Zwraca

• ApiRateLimits — The effective per-endpoint API rate limits.

AsyncLimitsClient.set_session_limits

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

Override session limits for the current subject.

Parametry

• limits (ContextLimits) — New session limits.

AsyncLimitsClient.reset_session_limits

async def reset_session_limits() -> None

Reset session limits to their default values.

AsyncLimitsClient.set_subject_limits

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

Override subject-level certificate and enrollment limits.

Parametry

• limits (SubjectLimits) — New subject-level limits.

AsyncLimitsClient.reset_subject_limits

async def reset_subject_limits() -> None

Reset subject-level limits to their default values.

AsyncLimitsClient.set_api_rate_limits

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

Override API rate limits for the current subject.

Parametry

• limits (ApiRateLimits) — New API rate limits.

AsyncLimitsClient.reset_api_rate_limits

async def reset_api_rate_limits() -> None

Reset API rate limits to their default values.

AsyncLimitsClient.set_production_rate_limits

async def set_production_rate_limits() -> None

Copy production rate-limit defaults into the current environment.

AsyncOnlineSessionClient

Import: from ksef2.clients import AsyncOnlineSessionClient

class AsyncOnlineSessionClient

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

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.

Zgłasza

• 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.

Metody

AsyncOnlineSessionClient.reference_number (właściwość)

reference_number: str

Get the online session reference number.

Zwraca

• str — The KSeF reference number of the session.

AsyncOnlineSessionClient.submission

def submission(reference_number: str) -> AsyncInvoiceSubmission

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.

Parametry

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

Zwraca

• AsyncInvoiceSubmission — The invoice submission handle.

AsyncOnlineSessionClient.send_invoice

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

Encrypt and submit one invoice into the open session.

Parametry

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

Zwraca

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

Zgłasza

• 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)

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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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

Fetch the current state of the online session.

Zwraca

• SessionStatusResponse — The session status, including invoice counters.

AsyncOnlineSessionClient.list_invoices

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

Fetch one page of invoices submitted in this session.

Parametry

• 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.

Zwraca

• 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

Fetch processing status for one invoice sent in this session.

Parametry

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

Zwraca

• 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]

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.

Parametry

• 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.

Zwraca

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

Zgłasza

• 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

Fetch one page of invoices that failed within this session.

Parametry

• 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.

Zwraca

• 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

Download the UPO of one invoice accepted in this session.

Parametry

• 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.

Zwraca

• bytes — The UPO as XML bytes.

Zgłasza

• 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]

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca

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

AsyncOnlineSessionClient.wait

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

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().

Parametry

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

Zwraca

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

Zgłasza

• 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]

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.

Zwraca

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

Zgłasza

• 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

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.

Zgłasza

• 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

Return the sensitive session state needed to resume later.

Zwraca

• OnlineSessionResumeState — The sensitive session state needed to resume later.

AsyncOnlineSessionClient.get_state

def get_state() -> OnlineSessionResumeState

Deprecated compatibility wrapper for resume_state().

Deprecated

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

Zwraca

• OnlineSessionResumeState — The same state as resume_state().

AsyncOperationHandle

Import: from ksef2.clients import AsyncOperationHandle

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

AsyncOperationHandle(reference_number: str)

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.

Metody

AsyncOperationHandle.reference_number (właściwość)

reference_number: str

Get the KSeF reference number of the operation.

Zwraca

• str — The reference number KSeF assigned to the operation.

AsyncOperationHandle.get_status

async def get_status() -> _TStatus

Fetch the current status of the operation without waiting.

Zwraca

• _TStatus — The status as KSeF reports it right now.

AsyncPager

Import: from ksef2.clients import AsyncPager

class AsyncPager(Generic[_T])

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

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.

Zgłasza

• 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.

Metody

AsyncPager.pages

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

Iterate over the collection one page at a time.

Zwraca

• 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]

Fetch only the first page.

Zwraca

• 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]

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.

Parametry

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

Zwraca

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

Zgłasza

• 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

AsyncPeppolClient(transport: AsyncMiddleware)

Async service for querying Peppol service providers.

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

Zgłasza

• 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.

Metody

AsyncPeppolClient.query

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

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.

Parametry

• params (OffsetPaginationParams | None) — Pagination parameters.

Zwraca

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

AsyncPeppolClient.all

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

Deprecated: iterate over all Peppol service providers.

Deprecated

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

Parametry

• params (OffsetPaginationParams | None) — Pagination parameters.

Zwraca (generator)

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

AsyncPeppolClient.list

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

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.

Parametry

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

Zwraca

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

Example

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

AsyncPermissionOperation

Import: from ksef2.clients import AsyncPermissionOperation

class AsyncPermissionOperation(AsyncOperationHandle[PermissionOperationStatusResponse, PermissionOperationStatusResponse])

AsyncPermissionOperation(
    client: AsyncPermissionsClient,
    response: GrantPermissionsResponse,
)

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.

Zgłasza

• 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.

Metody

AsyncPermissionOperation.response (właściwość)

response: GrantPermissionsResponse

Get the plain operation response.

Zwraca

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

AsyncPermissionOperation.get_status

async def get_status() -> PermissionOperationStatusResponse

Fetch the operation's current status without waiting.

Zwraca

• 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

Poll until KSeF has finished the operation.

Parametry

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

Zwraca

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

Zgłasza

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

AsyncPermissionsClient

Import: from ksef2.clients import AsyncPermissionsClient

class AsyncPermissionsClient

AsyncPermissionsClient(transport: AsyncMiddleware)

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.

Zgłasza

• 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.

Metody

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

Grant person-scoped permissions to a subject identifier.

Parametry

• 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.

Zwraca

• 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

Grant entity permissions to a NIP-identified entity.

Parametry

• 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.

Zwraca

• 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

Grant an authorization permission such as self-invoicing.

Parametry

• 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.

Zwraca

• 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

Grant indirect permissions, optionally limited to a target entity.

Parametry

• 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.

Zwraca

• 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

Grant permissions within a subunit context.

Parametry

• 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.

Zwraca

• 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

Grant permissions to an EU entity identified by fingerprint data.

Parametry

• 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.

Zwraca

• 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

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

Parametry

• 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.

Zwraca

• 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

Revoke an authorization permission by permission id.

Parametry

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

Zwraca

• 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

Revoke a non-authorization permission by permission id.

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

Parametry

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

Zwraca

• 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()

AsyncPermissionsClient.revoke_common

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

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

Deprecated

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

Parametry

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

Zwraca

• 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

Return whether attachments are currently allowed for the subject.

Zwraca

• AttachmentPermissionStatus — Whether attachments are currently allowed for the subject.

AsyncPermissionsClient.get_operation_status

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

Fetch the status of an asynchronous permission operation.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca

• 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]

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.

Parametry

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

Zwraca

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

AsyncPermissionsClient.list_entity_roles

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

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.

Parametry

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

Zwraca

• 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)

AsyncPermissionsClient.list_authorizations

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

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.

Parametry

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

Zwraca

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

Example

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

AsyncPermissionsClient.list_entities

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

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.

Parametry

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

Zwraca

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

Example

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

AsyncPermissionsClient.list_eu_entities

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

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.

Parametry

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

Zwraca

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

Example

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

AsyncPermissionsClient.list_personal

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

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.

Parametry

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

Zwraca

• 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)

AsyncPermissionsClient.list_persons

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

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.

Parametry

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

Zwraca

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

Example

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

AsyncPermissionsClient.list_subordinate_entities

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

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.

Parametry

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

Zwraca

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

Example

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

AsyncPermissionsClient.list_subunits

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

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.

Parametry

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

Zwraca

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

Example

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

AsyncSessionManagementClient

Import: from ksef2.clients import AsyncSessionManagementClient

class AsyncSessionManagementClient

AsyncSessionManagementClient(transport: AsyncMiddleware)

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.

Zgłasza

• 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.

Metody

AsyncSessionManagementClient.query

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

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.

Parametry

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

Zwraca

• 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]

Deprecated: iterate through all authentication session pages.

Deprecated

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

Parametry

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

Zwraca (generator)

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

AsyncSessionManagementClient.terminate_current

async def terminate_current() -> None

Terminate the authentication session backing the current bearer token.

AsyncSessionManagementClient.terminate

async def terminate(reference_number: str) -> None

Terminate an authentication session by reference number.

Parametry

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

AsyncSessionManagementClient.close

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

Deprecated: terminate an authentication session by reference number.

Deprecated

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

Parametry

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

Zwraca

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

AsyncSessionManagementClient.list

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

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.

Parametry

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

Zwraca

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

Example

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

AsyncTemporalTestData

Import: from ksef2.clients import AsyncTemporalTestData

class AsyncTemporalTestData

AsyncTemporalTestData(client: AsyncTestDataClient)

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.

Zgłasza

• 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.

Metody

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

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

Parametry

• 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

Delete a TEST subject and remove it from cleanup tracking.

Parametry

• 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

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

Parametry

• 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

Delete a TEST person and remove it from cleanup tracking.

Parametry

• 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

Grant TEST permissions and register them for cleanup.

Parametry

• 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

Revoke TEST permissions and remove them from cleanup tracking.

Parametry

• 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

Enable TEST attachments and register the context for cleanup.

Parametry

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