Registration

registration

Service class for handling registration workflows in Kaleidoscope.

This module provides methods for driving the registration lifecycle of operation files, including creating registrations, listing registration files for an operation, pushing intermediate status updates (e.g. after external validation), and submitting final results.

The typical webhook-driven workflow is:

Receive a registration_submitted webhook event containing a file_url and record_ids.
Download the file using :meth:FilesService.download_file.
Process the file externally.
(Optional) Push intermediate status via :meth:RegistrationService.push_status (e.g. validated, validation_failed).
Submit final results via :meth:RegistrationService.submit_results.

Classes:

Name	Description
`RegistrationFileStatusEnum`	Lifecycle states for a registration file.
`RegistrationResultStatusEnum`	Terminal statuses for a results submission.
`RegistrationResultError`	Per-record error from a failed registration.
`RegistrationFile`	A registration attempt for a file within an operation.
`RegistrationFileWithErrors`	A registration file with any per-record errors attached.
`RegistrationService`	Service for the registration endpoints.

RegistrationFileStatusEnum

Bases: str, Enum

Lifecycle states for a registration file.

Attributes:

Name	Type	Description
`PENDING`		Registration created, awaiting processing.
`VALIDATED`		External validation succeeded.
`VALIDATION_FAILED`		External validation failed.
`SUBMITTED`		Results have been submitted for processing.
`COMPLETED`		Registration completed successfully.
`FAILED`		Registration failed.

RegistrationResultStatusEnum

Bases: str, Enum

Terminal statuses reported when submitting registration results.

Attributes:

Name	Type	Description
`SUCCESS`		Processing succeeded.
`ERROR`		Processing failed.

RegistrationResultError

Bases: _KaleidoscopeBaseModel

A per-record error surfaced from a failed registration.

Attributes:

Name	Type	Description
`source_record_id`	`str`	The ID of the source record that failed.
`status`	`str`	The per-record status reported by the processor.
`error_message`	`str`	Human-readable error description.

RegistrationFile

Bases: _KaleidoscopeBaseModel

A registration attempt for a file within an operation.

Attributes:

Name	Type	Description
`workspace_id`	`str`	ID of the workspace.
`operation_id`	`str`	ID of the operation this registration belongs to.
`file_id`	`str`	ID of the file being registered.
`status`	`RegistrationFileStatusEnum`	Current lifecycle state.
`message`	`str \| None`	Status message set by the processor.
`created_by`	`str`	ID of the user who created the registration.
`last_updated_by`	`str`	ID of the user who last updated the registration.
`created_at`	`str`	ISO timestamp when the registration was created.
`updated_at`	`str`	ISO timestamp when the registration was last updated.

RegistrationFileWithErrors

Bases: RegistrationFile

A registration file with any per-record errors attached.

Attributes:

Name	Type	Description
`record_errors`	`list[RegistrationResultError]`	Errors for records that failed to register. Empty unless `status` is `failed`.

RegistrationService

RegistrationService(client: KaleidoscopeClient)

Service class for registration workflows.

Provides methods to create registrations, list registration files for an operation, push intermediate status updates, and submit final results.

Methods:

Name	Description
`register_file`	Register a file with an operation.
`get_registration_files`	List all registration files for an operation.
`push_status`	Push the current status of a registration file.
`submit_results`	Submit registration results for a processed file.

Source code in kalbio/registration.py

def __init__(self, client: "KaleidoscopeClient"):
    self._client = client

register_file

register_file(operation_id: str, file_id: str) -> RegistrationFile

Register a file with an operation.

Creates a registration for a previously uploaded file, submitting it for external registration processing.

Parameters:

Name	Type	Description	Default
`operation_id`	`str`	The operation to register the file with.	required
`file_id`	`str`	The uploaded file to register.	required

Returns:

Name	Type	Description
`RegistrationFile`	`RegistrationFile`	The newly created registration.

Raises:

Type	Description
`KalbioAPIError`	If the API request fails.

Source code in kalbio/registration.py

def register_file(self, operation_id: str, file_id: str) -> RegistrationFile:
    """Register a file with an operation.

    Creates a registration for a previously uploaded file, submitting
    it for external registration processing.

    Args:
        operation_id (str): The operation to register the file with.
        file_id (str): The uploaded file to register.

    Returns:
        RegistrationFile: The newly created registration.

    Raises:
        KalbioAPIError: If the API request fails.
    """
    url = f"/operations/{operation_id}/register/{file_id}"
    resp = self._client._post(url, {})
    return RegistrationFile.model_validate(resp)

get_registration_files

get_registration_files(operation_id: str) -> list[RegistrationFileWithErrors]

List all registration files for an operation.

Parameters:

Name	Type	Description	Default
`operation_id`	`str`	The operation to list registrations for.	required

Returns:

Type	Description
`list[RegistrationFileWithErrors]`	List[RegistrationFileWithErrors]: Registration files for the
`list[RegistrationFileWithErrors]`	operation. `record_errors` will be populated for entries
`list[RegistrationFileWithErrors]`	whose `status` is `failed`.

Raises:

Type	Description
`KalbioAPIError`	If the API request fails.

Source code in kalbio/registration.py

def get_registration_files(
    self, operation_id: str
) -> List[RegistrationFileWithErrors]:
    """List all registration files for an operation.

    Args:
        operation_id (str): The operation to list registrations for.

    Returns:
        List[RegistrationFileWithErrors]: Registration files for the
        operation. ``record_errors`` will be populated for entries
        whose ``status`` is ``failed``.

    Raises:
        KalbioAPIError: If the API request fails.
    """
    url = f"/operations/{operation_id}/registrations"
    resp = self._client._get(url)
    return TypeAdapter(List[RegistrationFileWithErrors]).validate_python(resp)

push_status

push_status(operation_id: str, file_id: str, status: RegistrationFileStatusEnum, message: str | None = None) -> bool

Push the current status of a registration file.

Used to report intermediate lifecycle updates from an external processor (e.g. validated, validation_failed) without submitting full results.

Parameters:

Name	Type	Description	Default
`operation_id`	`str`	The operation the registration belongs to.	required
`file_id`	`str`	The registration file to update.	required
`status`	`RegistrationFileStatusEnum`	The status to push.	required
`message`	`str`	Optional message to attach to the status update.	`None`

Returns:

Name	Type	Description
`bool`	`bool`	True if the request succeeded (204).

Raises:

Type	Description
`KalbioAPIError`	If the API request fails.

Source code in kalbio/registration.py

def push_status(
    self,
    operation_id: str,
    file_id: str,
    status: RegistrationFileStatusEnum,
    message: Optional[str] = None,
) -> bool:
    """Push the current status of a registration file.

    Used to report intermediate lifecycle updates from an external
    processor (e.g. ``validated``, ``validation_failed``) without
    submitting full results.

    Args:
        operation_id (str): The operation the registration belongs to.
        file_id (str): The registration file to update.
        status (RegistrationFileStatusEnum): The status to push.
        message (str, optional): Optional message to attach to the
            status update.

    Returns:
        bool: True if the request succeeded (204).

    Raises:
        KalbioAPIError: If the API request fails.
    """
    payload: dict[str, Any] = {"status": status}
    if message is not None:
        payload["message"] = message

    url = f"/operations/{operation_id}/registration_files/{file_id}/status"
    return self._client._post_no_content(url, payload)

submit_results

submit_results(operation_id: str, file_id: str, status: RegistrationResultStatusEnum, message: str | None = None, key_field_names: list[str] | None = None, records: list[dict[str, Any]] | None = None) -> bool

Submit registration results for a processed file.

Called after downloading and processing a file from a registration_submitted webhook event.

Parameters:

Name	Type	Description	Default
`operation_id`	`str`	The operation ID from the webhook event.	required
`file_id`	`str`	The file ID from the webhook event.	required
`status`	`RegistrationResultStatusEnum`	Overall result status.	required
`message`	`str`	Message to attach to the results.	`None`
`key_field_names`	`list[str]`	Names of key fields in the result data (e.g. `["id", "name"]`).	`None`
`records`	`list[dict[str, Any]]`	Per-record results. Each dict should contain `record_id` (str, required), `status` (str, required), and optionally `data` (dict[str, Any]) and `error_message` (str).	`None`

Returns:

Name	Type	Description
`bool`	`bool`	True if the results were submitted successfully (204).

Raises:

Type	Description
`KalbioAPIError`	If the API request fails.

Source code in kalbio/registration.py

def submit_results(
    self,
    operation_id: str,
    file_id: str,
    status: RegistrationResultStatusEnum,
    message: Optional[str] = None,
    key_field_names: Optional[list[str]] = None,
    records: Optional[list[dict[str, Any]]] = None,
) -> bool:
    """Submit registration results for a processed file.

    Called after downloading and processing a file from a
    ``registration_submitted`` webhook event.

    Args:
        operation_id (str): The operation ID from the webhook event.
        file_id (str): The file ID from the webhook event.
        status (RegistrationResultStatusEnum): Overall result status.
        message (str, optional): Message to attach to the results.
        key_field_names (list[str], optional): Names of key fields in the
            result data (e.g. ``["id", "name"]``).
        records (list[dict[str, Any]], optional): Per-record results. Each
            dict should contain ``record_id`` (str, required),
            ``status`` (str, required), and optionally ``data``
            (dict[str, Any]) and ``error_message`` (str).

    Returns:
        bool: True if the results were submitted successfully (204).

    Raises:
        KalbioAPIError: If the API request fails.
    """
    payload: dict[str, Any] = {"status": status}
    if message is not None:
        payload["message"] = message
    if key_field_names is not None:
        payload["key_field_names"] = key_field_names
    if records is not None:
        payload["records"] = records

    url = f"/operations/{operation_id}/register/{file_id}/results"
    return self._client._post_no_content(url, payload)