pas.importer.json_import ======================== .. py:module:: pas.importer.json_import Classes ------- .. autoapisummary:: pas.importer.json_import.OrganizationData pas.importer.json_import.EmailData pas.importer.json_import.AddressData pas.importer.json_import.PhoneNumberData pas.importer.json_import.UrlData pas.importer.json_import.OrganizationDataWithinMembership pas.importer.json_import.PersonData pas.importer.json_import.MembershipData pas.importer.json_import.ImportCategoryResult pas.importer.json_import.DataImporter pas.importer.json_import.PeopleImporter pas.importer.json_import.OrganizationImporter pas.importer.json_import.MembershipImporter Functions --------- .. autoapisummary:: pas.importer.json_import.count_unique_fullnames pas.importer.json_import.import_zug_kub_data Module Contents --------------- .. py:class:: OrganizationData Bases: :py:obj:`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) .. py:attribute:: created :type: str .. py:attribute:: description :type: str .. py:attribute:: htmlUrl :type: str .. py:attribute:: id :type: str .. py:attribute:: isActive :type: bool .. py:attribute:: memberCount :type: int .. py:attribute:: modified :type: str .. py:attribute:: name :type: str .. py:attribute:: organizationTypeTitle :type: Literal['Kommission', 'Fraktion', 'Kantonsrat', 'Sonstige'] | None .. py:attribute:: primaryEmail :type: None .. py:attribute:: status :type: int .. py:attribute:: thirdPartyId :type: str | None .. py:attribute:: url :type: str .. py:class:: EmailData Bases: :py:obj:`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) .. py:attribute:: id :type: str .. py:attribute:: label :type: str .. py:attribute:: email :type: str .. py:attribute:: isDefault :type: bool .. py:attribute:: thirdPartyId :type: str | None .. py:attribute:: modified :type: str .. py:attribute:: created :type: str .. py:class:: AddressData Bases: :py:obj:`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) .. py:attribute:: formattedAddress :type: str .. py:attribute:: id :type: str .. py:attribute:: label :type: str .. py:attribute:: isDefault :type: bool .. py:attribute:: organisationName :type: str .. py:attribute:: organisationNameAddOn1 :type: str .. py:attribute:: organisationNameAddOn2 :type: str .. py:attribute:: addressLine1 :type: str .. py:attribute:: addressLine2 :type: str .. py:attribute:: street :type: str .. py:attribute:: houseNumber :type: str .. py:attribute:: dwellingNumber :type: str .. py:attribute:: postOfficeBox :type: str .. py:attribute:: swissZipCode :type: str .. py:attribute:: swissZipCodeAddOn :type: str .. py:attribute:: swissZipCodeId :type: str .. py:attribute:: foreignZipCode :type: str .. py:attribute:: locality :type: str .. py:attribute:: town :type: str .. py:attribute:: countryIdISO2 :type: str .. py:attribute:: countryName :type: str .. py:attribute:: thirdPartyId :type: str | None .. py:attribute:: modified :type: str .. py:attribute:: created :type: str .. py:class:: PhoneNumberData Bases: :py:obj:`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) .. py:attribute:: id :type: str .. py:attribute:: label :type: str .. py:attribute:: phoneNumber :type: str .. py:attribute:: phoneCategory :type: int .. py:attribute:: otherPhoneCategory :type: str | None .. py:attribute:: phoneCategoryText :type: str .. py:attribute:: isDefault :type: bool .. py:attribute:: thirdPartyId :type: str | None .. py:attribute:: modified :type: str .. py:attribute:: created :type: str .. py:class:: UrlData Bases: :py:obj:`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) .. py:attribute:: id :type: str .. py:attribute:: label :type: str .. py:attribute:: url :type: str | None .. py:attribute:: isDefault :type: bool .. py:attribute:: thirdPartyId :type: str | None .. py:attribute:: modified :type: str .. py:attribute:: created :type: str .. py:class:: OrganizationDataWithinMembership Bases: :py:obj:`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) .. py:attribute:: created :type: str .. py:attribute:: description :type: str .. py:attribute:: htmlUrl :type: str .. py:attribute:: id :type: str .. py:attribute:: isActive :type: bool .. py:attribute:: memberCount :type: int .. py:attribute:: modified :type: str .. py:attribute:: name :type: str .. py:attribute:: organizationTypeTitle :type: Literal['Kommission', 'Fraktion', 'Kantonsrat', 'Sonstige'] | None .. py:attribute:: primaryEmail :type: EmailData | None .. py:attribute:: status :type: int .. py:attribute:: thirdPartyId :type: str | None .. py:attribute:: url :type: str .. py:attribute:: organizationType :type: int .. py:attribute:: primaryAddress :type: AddressData | None .. py:attribute:: primaryPhoneNumber :type: PhoneNumberData | None .. py:attribute:: primaryUrl :type: UrlData | None .. py:attribute:: statusDisplay :type: str .. py:attribute:: tags :type: list[str] .. py:attribute:: type :type: str .. py:class:: PersonData Bases: :py:obj:`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) .. py:attribute:: created :type: str .. py:attribute:: firstName :type: str .. py:attribute:: fullName :type: str .. py:attribute:: htmlUrl :type: str .. py:attribute:: id :type: str .. py:attribute:: isActive :type: bool .. py:attribute:: modified :type: str .. py:attribute:: officialName :type: str .. py:attribute:: personTypeTitle :type: str | None .. py:attribute:: primaryEmail :type: EmailData | None .. py:attribute:: salutation :type: str .. py:attribute:: tags :type: list[str] .. py:attribute:: thirdPartyId :type: str .. py:attribute:: title :type: str .. py:attribute:: url :type: str .. py:attribute:: username :type: str | None .. py:class:: MembershipData Bases: :py:obj:`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) .. py:attribute:: department :type: str .. py:attribute:: description :type: str .. py:attribute:: emailReceptionType :type: str .. py:attribute:: end :type: str | bool | None .. py:attribute:: id :type: str .. py:attribute:: isDefault :type: bool .. py:attribute:: organization :type: OrganizationDataWithinMembership .. py:attribute:: person :type: PersonData .. py:attribute:: primaryAddress :type: AddressData | None .. py:attribute:: primaryEmail :type: EmailData | None .. py:attribute:: primaryPhoneNumber :type: PhoneNumberData | None .. py:attribute:: primaryUrl :type: UrlData | None .. py:attribute:: email :type: EmailData | None .. py:attribute:: phoneNumber :type: PhoneNumberData | None .. py:attribute:: address :type: AddressData | None .. py:attribute:: urlField :type: UrlData | None .. py:attribute:: role :type: str .. py:attribute:: start :type: str | bool | None .. py:attribute:: text :type: str .. py:attribute:: thirdPartyId :type: str | None .. py:attribute:: type :type: str .. py:attribute:: typedId :type: str .. py:attribute:: url :type: str .. py:class:: ImportCategoryResult Bases: :py:obj:`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) .. py:attribute:: created :type: list[Any] .. py:attribute:: updated :type: list[Any] .. py:attribute:: processed :type: int .. py:class:: DataImporter(session: sqlalchemy.orm.Session) Base class for all importers with common functionality. .. py:attribute:: session .. py:method:: parse_date(date_string: str | None) -> datetime.date | None .. py:method:: _bulk_save(objects: list[Any], object_type: str) -> int Save a batch of objects to the database and return the count. .. py:class:: PeopleImporter(session: sqlalchemy.orm.Session) Bases: :py:obj:`DataImporter` Importer for Parliamentarian data from api/v2/people endpoint (People.json) API Inconsistency Note: The API design is inconsistent regarding address information. While 'people.json', which is expected to be the primary source for person details, *lacks* address information, 'memberships.json' *does* embed address data within the 'person' object and sometimes at the membership level. This requires us to extract address data from 'memberships.json' instead of the more logical 'people.json' endpoint. 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 .. py:attribute:: person_attribute_map :type: dict[str, str] .. py:method:: bulk_import(people_data: collections.abc.Sequence[PersonData]) -> tuple[dict[str, onegov.pas.models.Parliamentarian], dict[str, list[onegov.pas.models.Parliamentarian]], int] 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. .. py:method:: _update_parliamentarian_attributes(parliamentarian: onegov.pas.models.Parliamentarian, person_data: PersonData) -> bool Updates attributes of an existing Parliamentarian object. Returns True if any attributes were changed, False otherwise. .. py:method:: _create_parliamentarian(person_data: PersonData) -> onegov.pas.models.Parliamentarian | None Creates a single new Parliamentarian object from person data. .. py:class:: OrganizationImporter(session: sqlalchemy.orm.Session) Bases: :py:obj:`DataImporter` Importer for organization data from organizations.json. .. py:method:: bulk_import(organizations_data: collections.abc.Sequence[OrganizationData]) -> tuple[dict[str, onegov.pas.models.Commission], dict[str, onegov.pas.models.ParliamentaryGroup], dict[str, onegov.pas.models.Party], dict[str, Any], dict[str, dict[str, list[onegov.pas.models.Commission | onegov.pas.models.Party]]], dict[str, int]] 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. .. py:class:: MembershipImporter(session: sqlalchemy.orm.Session) Bases: :py:obj:`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 .. py:attribute:: session .. py:attribute:: parliamentarian_map :type: dict[str, onegov.pas.models.Parliamentarian] .. py:attribute:: commission_map :type: dict[str, onegov.pas.models.Commission] .. py:attribute:: parliamentary_group_map :type: dict[str, onegov.pas.models.ParliamentaryGroup] .. py:attribute:: party_map :type: dict[str, onegov.pas.models.Party] .. py:attribute:: other_organization_map :type: dict[str, Any] .. py:method:: init(session: sqlalchemy.orm.Session, parliamentarian_map: dict[str, onegov.pas.models.Parliamentarian], commission_map: dict[str, onegov.pas.models.Commission], parliamentary_group_map: dict[str, onegov.pas.models.ParliamentaryGroup], party_map: dict[str, onegov.pas.models.Party], other_organization_map: dict[str, Any]) -> None Initialize the importer with maps of objects by their external KUB ID. .. py:method:: _extract_and_update_or_create_missing_parliamentarians(memberships_data: collections.abc.Sequence[MembershipData]) -> dict[str, list[onegov.pas.models.Parliamentarian]] 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. .. py:method:: _update_parliamentarian_from_membership(parliamentarian: onegov.pas.models.Parliamentarian, person_data: PersonData, membership_data: MembershipData) -> bool 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. .. py:method:: _create_parliamentarian_from_membership(person_data: PersonData, membership_data: MembershipData) -> onegov.pas.models.Parliamentarian | None Creates a new Parliamentarian object using data primarily from a membership record. .. py:method:: bulk_import(memberships_data: collections.abc.Sequence[MembershipData]) -> tuple[dict[str, ImportCategoryResult | dict[str, list[Any]]], dict[str, int]] 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. .. py:method:: _create_commission_membership(parliamentarian: onegov.pas.models.Parliamentarian, commission: onegov.pas.models.Commission, membership_data: MembershipData) -> onegov.pas.models.CommissionMembership | None Create a CommissionMembership object. .. py:method:: _update_commission_membership(membership: onegov.pas.models.CommissionMembership, membership_data: MembershipData) -> bool Updates an existing CommissionMembership object. Returns True if changed. .. py:method:: _create_parliamentarian_role(parliamentarian: onegov.pas.models.Parliamentarian, role: onegov.pas.models.parliamentarian_role.Role, parliamentary_group: onegov.pas.models.ParliamentaryGroup | None = None, parliamentary_group_role: onegov.pas.models.parliamentarian_role.ParliamentaryGroupRole | None = None, party: onegov.pas.models.Party | None = None, party_role: onegov.pas.models.parliamentarian_role.PartyRole | None = None, additional_information: str | None = None, start_date: str | None = None, end_date: str | None = None) -> onegov.pas.models.ParliamentarianRole | None .. py:method:: _update_parliamentarian_role(role_obj: onegov.pas.models.ParliamentarianRole, role: onegov.pas.models.parliamentarian_role.Role, parliamentary_group: onegov.pas.models.ParliamentaryGroup | None = None, parliamentary_group_role: onegov.pas.models.parliamentarian_role.ParliamentaryGroupRole | None = None, party: onegov.pas.models.Party | None = None, party_role: onegov.pas.models.parliamentarian_role.PartyRole | None = None, additional_information: str | None = None, start_date: str | None = None, end_date: str | None = None) -> bool Updates an existing ParliamentarianRole object. Returns True if changed. .. py:method:: _map_to_commission_role(role_text: str) -> Literal['president', 'extended_member', 'guest', 'member'] Map a role text to a CommissionMembership role enum value. .. py:method:: _map_to_parliamentarian_role(role_text: str) -> onegov.pas.models.parliamentarian_role.Role Map a role text to a parliamentarian role enum value. .. py:method:: _map_to_parliamentary_group_role(role_text: str) -> onegov.pas.models.parliamentarian_role.ParliamentaryGroupRole | None Map a role text to a ParliamentaryGroupRole enum value. .. py:method:: extract_role_org_type_pairs(memberships_data: collections.abc.Sequence[MembershipData]) -> list[str] :classmethod: Extract unique combinations of role and organizationTypeTitle from JSON data. Useful way to get an overview of the data. .. py:function:: count_unique_fullnames(people_data: list[Any], organizations_data: list[Any], memberships_data: list[Any]) -> set[str] Counts unique 'fullName' entries across different data structures. .. py:function:: import_zug_kub_data(session: sqlalchemy.orm.Session, people_data: collections.abc.Sequence[PersonData], organization_data: collections.abc.Sequence[OrganizationData], membership_data: collections.abc.Sequence[MembershipData], user_id: uuid.UUID | None = None) -> dict[str, ImportCategoryResult] Imports data from KUB JSON files within a single transaction, logs the outcome, and returns details of changes including processed counts. 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.