pas.importer.orchestrator

Attributes

log

Exceptions

APIAccessibilityError

Raised when the KUB API is not accessible.

Classes

UpdateResult
KubImporter	KUB API importer that manages shared session and state for both

Functions

_fetch_custom_data_worker(→ None)	Worker thread that fetches API data for parliamentarians.
create_mock_file_data(→ onegov.core.types.LaxFileDict)

Module Contents

pas.importer.orchestrator.log

exception pas.importer.orchestrator.APIAccessibilityError

Bases: Exception

Raised when the KUB API is not accessible.

class pas.importer.orchestrator.UpdateResult

parliamentarian_id: str

title: str

custom_values: dict[str, str]

addresses: list[dict[str, Any]]

sex: int | None

error: str | None = None

pas.importer.orchestrator._fetch_custom_data_worker(parliamentarian_queue: queue.Queue[Any], result_queue: queue.Queue[UpdateResult], session: requests.Session, base_url: str) → None

Worker thread that fetches API data for parliamentarians.

pas.importer.orchestrator.create_mock_file_data(data: list[dict[str, Any]], filename: str) → onegov.core.types.LaxFileDict

class pas.importer.orchestrator.KubImporter(token: str, base_url: str, output: onegov.pas.log.OutputHandler | None = None)

KUB API importer that manages shared session and state for both data import and custom data updates.

token

base_url

output = None

session

__enter__() → Self

__exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: types.TracebackType | None) → None

_check_api_accessibility() → None

Check if the API is accessible.

_update_address_fields(parliamentarian: Any, addresses: list[dict[str, Any]]) → list[str]

Update parliamentarian address fields based on addresses data.

Returns list of updated field names.

_update_gender_field(parliamentarian: Any, sex: int) → bool

_fetch_api_data_with_pagination(endpoint: str) → list[dict[str, Any]]

Fetches all data from a paginated API endpoint using the shared session.

Args:

endpoint: The API endpoint (e.g., ‘people’, ‘organizations’,

‘memberships’)

Returns:

List of all records from all pages

update_custom_data(request: onegov.pas.request.PasRequest, app: onegov.pas.app.PasApp, max_workers: int = 3, import_log_id: uuid.UUID | None = None) → tuple[int, int, list[dict[str, Any]]]

Multi-threaded function that updates parliamentarians with custom field data.

Uses a queue-based approach where worker threads fetch API data concurrently while the main thread handles all database updates to maintain thread safety.

Returns:

Tuple of (updated_count, error_count, output_messages)

import_data(request: onegov.pas.request.PasRequest, app: onegov.pas.app.PasApp, import_type: str = 'automatic') → tuple[dict[str, Any], list[Any], list[Any], list[Any], uuid.UUID]

Performs KUB data import using the shared session.

Returns:

Tuple of (import_results, people_data, organization_data,

membership_data, import_log_id)

run_full_sync(request: onegov.pas.request.PasRequest, app: onegov.pas.app.PasApp, import_type: str, update_custom: bool = True, max_workers: int = 3) → tuple[dict[str, Any], uuid.UUID]

This is the main entry point for cronjobs and automated imports. Complete KUB synchronization including import and custom data update.

Returns:

Tuple of (combined_results, import_log_id)