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

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

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 = None[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

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

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.

_create_commission_membership(parliamentarian: onegov.pas.models.PASParliamentarian, commission: onegov.pas.models.PASCommission, membership_data: onegov.pas.importer.types.MembershipData) → onegov.pas.models.PASCommissionMembership | None[source]: Create a CommissionMembership object.

_update_commission_membership(membership: onegov.pas.models.PASCommissionMembership, membership_data: onegov.pas.importer.types.MembershipData) → bool[source]: Updates an existing CommissionMembership object. Returns True if changed.

_create_parliamentarian_role(parliamentarian: onegov.pas.models.PASParliamentarian, 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_date: str | None = None, end_date: str | None = None) → onegov.pas.models.PASParliamentarianRole | None[source]

_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_date: str | None = None, end_date: str | None = None) → bool[source]: Updates an existing ParliamentarianRole object. Returns True if changed.

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