pas.importer.json_import

Complex JSON import system with interdependent importers.

WARNING: PeopleImporter, OrganizationImporter, and MembershipImporter must be used together in the correct order, otherwise database foreign key violations will occur. The importers cannot work independently due to tight coupling between entities.

Classes

`ImportCategoryResult`	dict() -> new empty dictionary
`DataImporter`	Base class for all importers with common functionality.
`PeopleImporter`	Importer for Parliamentarian data from api/v2/people endpoint
`OrganizationImporter`	Importer for organization data from organizations.json.
`MembershipImporter`	Importer for membership data from memberships.json.

Functions

`periods_overlap`(→ bool)	Whether two possibly open ended date ranges have any day in common.
`count_unique_fullnames`(→ set[str])	Counts unique 'fullName' entries across different data structures.

Module Contents

class pas.importer.json_import.ImportCategoryResult[source]

Bases: TypedDict

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object’s

(key, value) pairs

dict(iterable) -> new dictionary initialized as if via:: d = {} for k, v in iterable:

d[k] = v
dict(**kwargs) -> new dictionary initialized with the name=value pairs: in the keyword argument list. For example: dict(one=1, two=2)

created: list[Any][source]

updated: list[Any][source]

processed: int[source]

pas.importer.json_import.periods_overlap(start_a: datetime.date | None, end_a: datetime.date | None, start_b: datetime.date | None, end_b: datetime.date | None) → bool[source]: Whether two possibly open ended date ranges have any day in common.

class pas.importer.json_import.DataImporter(session: sqlalchemy.orm.Session, logger: logging.Logger | None = None)[source]

Base class for all importers with common functionality.

session[source]

logger[source]

parse_date(date_string: str | None) → datetime.date | None[source]

_bulk_save(objects: list[Any], object_type: str) → int[source]: Save a batch of objects to the database and return the count.

class pas.importer.json_import.PeopleImporter(session: sqlalchemy.orm.Session, logger: logging.Logger | None = None)[source]

Bases: DataImporter

Importer for Parliamentarian data from api/v2/people endpoint: (People.json)

API Inconsistency Note: Adresses are not imported here. The source of the address in the API is convoluted: While ‘people.json’, is expected to be the primary source for person details, it lacks address information. The memberships.json embeds inconsistent address data - sometimes in the person object, sometimes at membership level. This differs from the canonical address returned by api/v2/people/<uuid>. We must use the direct API endpoint’s address, requiring KubImporter.update_custom_data for a second pass to fetch correct addresses for exports.

Whatever the reasons for this, we need to be careful to avoid potential inconsistencies if addresses are not carefully managed across both endpoints in the source system.

person_attribute_map: dict[str, str][source]

bulk_import(people_data: collections.abc.Sequence[onegov.pas.importer.types.PersonData]) → tuple[dict[str, onegov.pas.models.PASParliamentarian], dict[str, list[onegov.pas.models.PASParliamentarian]], int][source]

Imports people from JSON data.

Returns:: A tuple containing: - A map of external KUB ID to Parliamentarian objects. - A dictionary with lists of created and updated parliamentarians. - The total number of people records processed from the input.

_update_parliamentarian_attributes(parliamentarian: onegov.pas.models.PASParliamentarian, person_data: onegov.pas.importer.types.PersonData) → bool[source]: Updates attributes of an existing Parliamentarian object. Returns True if any attributes were changed, False otherwise.

_create_parliamentarian(person_data: onegov.pas.importer.types.PersonData) → onegov.pas.models.PASParliamentarian | None[source]: Creates a single new Parliamentarian object from person data.

class pas.importer.json_import.OrganizationImporter(session: sqlalchemy.orm.Session, logger: logging.Logger | None = None)[source]

Bases: DataImporter

Importer for organization data from organizations.json.

bulk_import(organizations_data: collections.abc.Sequence[onegov.pas.importer.types.OrganizationData]) → tuple[dict[str, onegov.pas.models.PASCommission], dict[str, onegov.pas.models.PASParliamentaryGroup], dict[str, onegov.pas.models.Party], dict[str, Any], dict[str, dict[str, list[onegov.pas.models.PASCommission | onegov.pas.models.Party]]], dict[str, int]][source]

Imports organizations from JSON data. Returns maps, details, and counts.

Returns:: Tuple containing maps of external IDs to: - Commissions - Parliamentary Groups - Parties (created from parliamentary groups) - Other organizations - Dictionary containing lists of created/updated objects per type.

class pas.importer.json_import.MembershipImporter(session: sqlalchemy.orm.Session, logger: logging.Logger | None = None)[source]

Bases: DataImporter

Importer for membership data from memberships.json.

Get an overview of the possible pairs: see extract_role_org_type_pairs:

Role - Organization Type Combinations:

Assistentin Leitung Staatskanzlei - Sonstige Frau Landammann - Sonstige Landammann - Sonstige Landschreiber - Sonstige Mitglied - Kommission Mitglied des Kantonsrates - Kantonsrat Nationalrat - Sonstige Parlamentarier - Sonstige Parlamentarierin - Sonstige Protokollführerin - Sonstige Präsident des Kantonsrates - Kantonsrat Präsident/-in - Kommission Regierungsrat - Sonstige Regierungsrätin - Sonstige Standesweibel - Sonstige Standesweibelin - Sonstige Statthalter - Sonstige Stv. Landschreiberin - Sonstige Stv. Protokollführer - Sonstige Stv. Protokollführerin - Sonstige Stv. Standesweibelin - Sonstige Ständerat - Sonstige

Total unique combinations: 22

Roles and commission memberships are identified by the id of the KUB membership they originate from (external_kub_id). A person may hold the same role in the same organization more than once (leaving and returning), so anything derived from person, organization and role is not unique and would collapse those periods into a single row.

Rows which still lack an external_kub_id are matched by person, organization and start date, and adopt the id of the membership they were imported from. Rows that keep an empty id were entered by hand and are left alone.

parliamentarian_map: dict[str, onegov.pas.models.PASParliamentarian][source]

commission_map: dict[str, onegov.pas.models.PASCommission][source]

parliamentary_group_map: dict[str, onegov.pas.models.PASParliamentaryGroup][source]

party_map: dict[str, onegov.pas.models.Party][source]

other_organization_map: dict[str, Any][source]

payload_ids: set[uuid.UUID][source]

seen_membership_ids: set[uuid.UUID][source]

seen_role_ids: set[uuid.UUID][source]

memberships_by_kub_id: dict[uuid.UUID, onegov.pas.models.PASCommissionMembership][source]

memberships_by_bucket: dict[MembershipBucket, list[onegov.pas.models.PASCommissionMembership]][source]

roles_by_kub_id: dict[uuid.UUID, onegov.pas.models.PASParliamentarianRole][source]

roles_by_bucket: dict[RoleBucket, list[onegov.pas.models.PASParliamentarianRole]][source]

init(session: sqlalchemy.orm.Session, parliamentarian_map: dict[str, onegov.pas.models.PASParliamentarian], commission_map: dict[str, onegov.pas.models.PASCommission], parliamentary_group_map: dict[str, onegov.pas.models.PASParliamentaryGroup], party_map: dict[str, onegov.pas.models.Party], other_organization_map: dict[str, Any]) → None[source]: Initialize the importer with maps of objects by their external KUB ID.

_extract_and_update_or_create_missing_parliamentarians(memberships_data: collections.abc.Sequence[onegov.pas.importer.types.MembershipData]) → dict[str, list[onegov.pas.models.PASParliamentarian]][source]

Extracts/updates/creates parliamentarians found only in membership data.

Returns a dictionary with lists of created and updated parliamentarians found during this process.

If a parliamentarian is not already known (from people.json import), check if they exist in the DB. If yes, update them with potentially missing info (address, email) from membership data. If no, create a new parliamentarian. Updates self.parliamentarian_map accordingly.

_update_parliamentarian_from_membership(parliamentarian: onegov.pas.models.PASParliamentarian, person_data: onegov.pas.importer.types.PersonData, membership_data: onegov.pas.importer.types.MembershipData) → bool[source]: Updates an existing parliamentarian with info from membership data, focusing on potentially missing fields like email and address. Returns True if any changes were made, False otherwise.

_create_parliamentarian_from_membership(person_data: onegov.pas.importer.types.PersonData, membership_data: onegov.pas.importer.types.MembershipData) → onegov.pas.models.PASParliamentarian | None[source]: Creates a new Parliamentarian object using data primarily from a membership record.

bulk_import(memberships_data: collections.abc.Sequence[onegov.pas.importer.types.MembershipData]) → tuple[dict[str, ImportCategoryResult | dict[str, list[Any]]], dict[str, int]][source]

Imports memberships from JSON data based on organization type.

Returns:

A dictionary containing lists of created/updated objects and processed counts per category.
A dictionary containing the count of processed items per type.

_parse_kub_id(membership_data: onegov.pas.importer.types.MembershipData) → uuid.UUID | None[source]: The id of the membership in KUB, if it is a valid UUID.

_membership_dates(membership_data: onegov.pas.importer.types.MembershipData) → tuple[datetime.date | None, datetime.date | None][source]

_prefetch(parliamentarian_ids: list[uuid.UUID], memberships_data: collections.abc.Sequence[onegov.pas.importer.types.MembershipData]) → None[source]: Loads the roles and memberships of the parliamentarians we are about to import, indexed by KUB id and by organization.

_register_membership(membership: onegov.pas.models.PASCommissionMembership) → None[source]

_register_role(role: onegov.pas.models.PASParliamentarianRole, bucket: RoleBucket) → None[source]

_match[T: (onegov.pas.models.PASCommissionMembership, onegov.pas.models.PASParliamentarianRole)](candidates: collections.abc.Sequence[T], role: str, start: datetime.date | None, end: datetime.date | None) → tuple[Match, T | None][source]

Decides what to do with a membership from KUB, given the rows we already have for the same person and organization.

duplicate: KUB holds two memberships for the same period, we only keep one row
adopt: the row predates external_kub_id (or its membership is gone from KUB), it takes over the id of this membership
create: a period we haven’t seen before

_import_commission_membership(parliamentarian: onegov.pas.models.PASParliamentarian, commission: onegov.pas.models.PASCommission, kub_id: uuid.UUID | None, role: onegov.parliament.models.commission_membership.MembershipRole, start: datetime.date | None, end: datetime.date | None) → tuple[onegov.pas.models.PASCommissionMembership | None, bool][source]

Creates, updates or merges a single commission membership.

Returns the membership that was created or changed (if any) and whether it is a new one.

_import_parliamentarian_role(parliamentarian: onegov.pas.models.PASParliamentarian, kub_id: uuid.UUID | None, role: onegov.parliament.models.parliamentarian_role.Role, parliamentary_group: onegov.pas.models.PASParliamentaryGroup | None = None, parliamentary_group_role: onegov.parliament.models.parliamentarian_role.ParliamentaryGroupRole | None = None, party: onegov.pas.models.Party | None = None, party_role: onegov.parliament.models.parliamentarian_role.PartyRole | None = None, additional_information: str | None = None, start: datetime.date | None = None, end: datetime.date | None = None, org_type: str | None = None) → tuple[onegov.pas.models.PASParliamentarianRole | None, bool][source]

Creates, updates or merges a single parliamentarian role.

Returns the role that was created or changed (if any) and whether it is a new one.

_update_parliamentarian_role(role_obj: onegov.pas.models.PASParliamentarianRole, role: onegov.parliament.models.parliamentarian_role.Role, parliamentary_group: onegov.pas.models.PASParliamentaryGroup | None = None, parliamentary_group_role: onegov.parliament.models.parliamentarian_role.ParliamentaryGroupRole | None = None, party: onegov.pas.models.Party | None = None, party_role: onegov.parliament.models.parliamentarian_role.PartyRole | None = None, additional_information: str | None = None, start: datetime.date | None = None, end: datetime.date | None = None, org_type: str | None = None) → bool[source]: Updates an existing ParliamentarianRole object. Returns True if changed.

_delete_vanished(parliamentarian_ids: list[uuid.UUID]) → int[source]

Deletes the roles and memberships whose KUB membership is gone.

Rows without an external_kub_id were entered by hand and are never touched.

_map_to_commission_role(role_text: str) → Literal['president', 'extended_member', 'guest', 'member'][source]: Map a role text to a CommissionMembership role enum value.

_map_to_parliamentarian_role(role_text: str) → onegov.parliament.models.parliamentarian_role.Role[source]: Map a role text to a parliamentarian role enum value.

_map_to_parliamentary_group_role(role_text: str) → onegov.parliament.models.parliamentarian_role.ParliamentaryGroupRole | None[source]: Map a role text to a ParliamentaryGroupRole enum value.

classmethod extract_role_org_type_pairs(memberships_data: collections.abc.Sequence[onegov.pas.importer.types.MembershipData]) → list[str][source]: Extract unique combinations of role and organizationTypeTitle from JSON data. Useful way to get an overview of the data.

pas.importer.json_import.count_unique_fullnames(people_data: list[Any], organizations_data: list[Any], memberships_data: list[Any]) → set[str][source]: Counts unique ‘fullName’ entries across different data structures.