pas.importer.json_import

Classes

`OrganizationData`	dict() -> new empty dictionary
`EmailData`	dict() -> new empty dictionary
`AddressData`	dict() -> new empty dictionary
`PhoneNumberData`	dict() -> new empty dictionary
`UrlData`	dict() -> new empty dictionary
`OrganizationDataWithinMembership`	dict() -> new empty dictionary
`PersonData`	dict() -> new empty dictionary
`MembershipData`	dict() -> new empty dictionary
`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.
`import_zug_kub_data`(→ dict[str, ImportCategoryResult])	Imports data from KUB JSON files within a single transaction,

Module Contents

class pas.importer.json_import.OrganizationData[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: str[source]

description: str[source]

htmlUrl: str[source]

id: str[source]

isActive: bool[source]

memberCount: int[source]

modified: str[source]

name: str[source]

organizationTypeTitle: Literal['Kommission', 'Fraktion', 'Kantonsrat', 'Sonstige'] | None[source]

primaryEmail: None[source]

status: int[source]

thirdPartyId: str | None[source]

url: str[source]

class pas.importer.json_import.EmailData[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)

id: str[source]

label: str[source]

email: str[source]

isDefault: bool[source]

thirdPartyId: str | None[source]

modified: str[source]

created: str[source]

class pas.importer.json_import.AddressData[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)

formattedAddress: str[source]

id: str[source]

label: str[source]

isDefault: bool[source]

organisationName: str[source]

organisationNameAddOn1: str[source]

organisationNameAddOn2: str[source]

addressLine1: str[source]

addressLine2: str[source]

street: str[source]

houseNumber: str[source]

dwellingNumber: str[source]

postOfficeBox: str[source]

swissZipCode: str[source]

swissZipCodeAddOn: str[source]

swissZipCodeId: str[source]

foreignZipCode: str[source]

locality: str[source]

town: str[source]

countryIdISO2: str[source]

countryName: str[source]

thirdPartyId: str | None[source]

modified: str[source]

created: str[source]

class pas.importer.json_import.PhoneNumberData[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)

id: str[source]

label: str[source]

phoneNumber: str[source]

phoneCategory: int[source]

otherPhoneCategory: str | None[source]

phoneCategoryText: str[source]

isDefault: bool[source]

thirdPartyId: str | None[source]

modified: str[source]

created: str[source]

class pas.importer.json_import.UrlData[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)

id: str[source]

label: str[source]

url: str | None[source]

isDefault: bool[source]

thirdPartyId: str | None[source]

modified: str[source]

created: str[source]

class pas.importer.json_import.OrganizationDataWithinMembership[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: str[source]

description: str[source]

htmlUrl: str[source]

id: str[source]

isActive: bool[source]

memberCount: int[source]

modified: str[source]

name: str[source]

organizationTypeTitle: Literal['Kommission', 'Fraktion', 'Kantonsrat', 'Sonstige'] | None[source]

primaryEmail: EmailData | None[source]

status: int[source]

thirdPartyId: str | None[source]

url: str[source]

organizationType: int[source]

primaryAddress: AddressData | None[source]

primaryPhoneNumber: PhoneNumberData | None[source]

primaryUrl: UrlData | None[source]

statusDisplay: str[source]

tags: list[str][source]

type: str[source]

class pas.importer.json_import.PersonData[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: str[source]

firstName: str[source]

fullName: str[source]

htmlUrl: str[source]

id: str[source]

isActive: bool[source]

modified: str[source]

officialName: str[source]

personTypeTitle: str | None[source]

primaryEmail: EmailData | None[source]

salutation: str[source]

tags: list[str][source]

thirdPartyId: str[source]

title: str[source]

url: str[source]

username: str | None[source]

class pas.importer.json_import.MembershipData[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)

department: str[source]

description: str[source]

emailReceptionType: str[source]

end: str | bool | None[source]

id: str[source]

isDefault: bool[source]

organization: OrganizationDataWithinMembership[source]

person: PersonData[source]

primaryAddress: AddressData | None[source]

primaryEmail: EmailData | None[source]

primaryPhoneNumber: PhoneNumberData | None[source]

primaryUrl: UrlData | None[source]

email: EmailData | None[source]

phoneNumber: PhoneNumberData | None[source]

address: AddressData | None[source]

urlField: UrlData | None[source]

role: str[source]

start: str | bool | None[source]

text: str[source]

thirdPartyId: str | None[source]

type: str[source]

typedId: str[source]

url: str[source]

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)[source]

Base class for all importers with common functionality.

session[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)[source]

Bases: 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

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

bulk_import(people_data: collections.abc.Sequence[PersonData]) → tuple[dict[str, onegov.pas.models.Parliamentarian], dict[str, list[onegov.pas.models.Parliamentarian]], 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.Parliamentarian, person_data: PersonData) → bool[source]: Updates attributes of an existing Parliamentarian object. Returns True if any attributes were changed, False otherwise.

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

class pas.importer.json_import.OrganizationImporter(session: sqlalchemy.orm.Session)[source]

Bases: DataImporter

Importer for organization data from organizations.json.

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]][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)[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

session[source]

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

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

parliamentary_group_map: dict[str, onegov.pas.models.ParliamentaryGroup][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.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[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[MembershipData]) → dict[str, list[onegov.pas.models.Parliamentarian]][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.Parliamentarian, person_data: PersonData, membership_data: 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: PersonData, membership_data: MembershipData) → onegov.pas.models.Parliamentarian | None[source]: Creates a new Parliamentarian object using data primarily from a membership record.

bulk_import(memberships_data: collections.abc.Sequence[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.Parliamentarian, commission: onegov.pas.models.Commission, membership_data: MembershipData) → onegov.pas.models.CommissionMembership | None[source]: Create a CommissionMembership object.

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

_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[source]

_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[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.pas.models.parliamentarian_role.Role[source]: Map a role text to a parliamentarian role enum value.

_map_to_parliamentary_group_role(role_text: str) → onegov.pas.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[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.

pas.importer.json_import.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][source]

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.