pas.importer.zug_kub_importer

Functions

_serialize_model_objects(→ list[dict[str, Any]])	Convert model objects to JSON-serializable dictionaries.
import_zug_kub_data(→ dict[str, ...)	Imports data from KUB JSON files within a single transaction,

Module Contents

pas.importer.zug_kub_importer._serialize_model_objects(obj_list: list[Any]) → list[dict[str, Any]]

Convert model objects to JSON-serializable dictionaries.

pas.importer.zug_kub_importer.import_zug_kub_data(session: sqlalchemy.orm.Session, people_data: collections.abc.Sequence[onegov.pas.importer.json_import.PersonData], organization_data: collections.abc.Sequence[onegov.pas.importer.json_import.OrganizationData], membership_data: collections.abc.Sequence[onegov.pas.importer.json_import.MembershipData], user_id: uuid.UUID | None = None, import_type: str = 'cli', logger: logging.Logger | None = None, create_import_log: bool = True) → dict[str, onegov.pas.importer.json_import.ImportCategoryResult]

Imports data from KUB JSON files within a single transaction, logs the outcome, and returns details of changes including processed counts.

Args:

session: Database session people_data: People data to import organization_data: Organization data to import membership_data: Membership data to import user_id: ID of user performing the import (optional) import_type: Type of import logger: Optional logger to use instead of module logger create_import_log: Whether to create ImportLog (True for web form,

False for CLI/orchestrator)

Returns a dictionary where keys are categories (e.g., ‘parliamentarians’) and values are dictionaries containing ‘created’ (list), ‘updated’ (list), and ‘processed’ (int).

Rolls back changes within this import if an internal error occurs. Logs the attempt regardless of success or failure.