from __future__ import annotations
from datetime import date
from sqlalchemy import and_, case, func, or_
from sqlalchemy import Column, Enum, ForeignKey, UUID as UUIDType
from sqlalchemy import Table
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import mapped_column, relationship, Mapped
from uuid import uuid4, UUID
from onegov.core.collection import GenericCollection, Pagination
from onegov.core.orm import Base
from onegov.core.orm.mixins import ContentMixin
from onegov.core.utils import toggle
from onegov.file import MultiAssociatedFiles
from onegov.org import _
from onegov.org.models.extensions import AccessExtension
from onegov.org.models.extensions import GeneralFileLinkExtension
from onegov.search import ORMSearchable, SearchIndex
from onegov.search.utils import language_from_locale
from typing import Literal, Self, TypeAlias, TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Collection
from sqlalchemy.orm import Query, Session
from sqlalchemy.sql import ColumnElement
from onegov.org.models import MeetingItem
from onegov.org.models import RISParliamentarian
from onegov.org.models import RISParliamentaryGroup
from onegov.org.request import OrgRequest
[docs]
PoliticalBusinessType: TypeAlias = Literal[
'inquiry', # Anfrage
'report and proposal', # Bericht und Antrag
'urgent interpellation', # Dringliche Interpellation
'invitation', # Einladung
'interpellation', # Interpellation
'commission report', # Kommissionsbericht
'motion', # Motion
'postulate', # Postulat
'resolution', # Resolution
'election', # Wahl
'parliamentary statement', # Parlamentarische Erklärung
'miscellaneous', # Verschiedenes
]
[docs]
PoliticalBusinessStatus: TypeAlias = Literal[
'abgeschrieben',
'beantwortet',
'erheblich_erklaert',
'erledigt',
'nicht_erheblich_erklaert',
'nicht_zustandegekommen',
'pendent_exekutive',
'pendent_legislative',
'rueckzug',
'umgewandelt',
'zurueckgewiesen',
'ueberwiesen',
]
[docs]
POLITICAL_BUSINESS_TYPE: dict[PoliticalBusinessType, str] = {
'inquiry': _('Inquiry'),
'report and proposal': _('Report and Proposal'),
'urgent interpellation': _('Urgent Interpellation'),
'invitation': _('Invitation'),
'interpellation': _('Interpellation'),
'commission report': _('Commission Report'),
'motion': _('Motion'),
'postulate': _('Postulate'),
'resolution': _('Resolution'),
'election': _('Election'),
'parliamentary statement': _('Parliamentary Statement'),
'miscellaneous': _('Miscellaneous'),
}
# FIXME: i18n
[docs]
POLITICAL_BUSINESS_STATUS: dict[PoliticalBusinessStatus, str] = {
'abgeschrieben': 'Abgeschrieben',
'beantwortet': 'Beantwortet',
'erheblich_erklaert': 'Erheblich erklärt',
'erledigt': 'Erledigt',
'nicht_erheblich_erklaert': 'Nicht erheblich erklärt',
'nicht_zustandegekommen': 'Nicht zustandegekommen',
'pendent_exekutive': 'Pendent Exekutive',
'pendent_legislative': 'Pendent Legislative',
'rueckzug': 'Rückzug',
'umgewandelt': 'Umgewandelt',
'zurueckgewiesen': 'Zurückgewiesen',
'ueberwiesen': 'Überwiesen',
}
# join table between political businesses and parliamentary groups
[docs]
par_political_business_parliamentary_groups = Table(
'par_political_business_parliamentary_groups',
Base.metadata,
Column(
'political_business_id',
UUIDType(as_uuid=True),
ForeignKey('par_political_businesses.id', ondelete='CASCADE'),
primary_key=True,
),
Column(
'parliamentary_group_id',
UUIDType(as_uuid=True),
ForeignKey('par_parliamentary_groups.id', ondelete='CASCADE'),
primary_key=True,
),
)
[docs]
class PoliticalBusiness(
AccessExtension,
MultiAssociatedFiles,
Base,
ContentMixin,
GeneralFileLinkExtension,
ORMSearchable
):
[docs]
GERMAN_STATUS_NAME_TO_VALUE_MAP: dict[str, str] = {
'Abgeschrieben': 'written_off',
'Beantwortet': 'answered',
'Erheblich erklärt': 'declared_significant',
'Erledigt': 'completed',
'Nicht erheblich erklärt': 'declared_insignificant',
'Nicht zustandegekommen': 'not_realized',
'Pendent Exekutive': 'pending_executive',
'Pendent Legislative': 'pending_legislative',
'Rückzug': 'withdrawn',
'Umgewandelt': 'converted',
'Zurückgewiesen': 'rejected',
'Überwiesen': 'referred',
}
# Politisches Geschäft
[docs]
__tablename__ = 'par_political_businesses'
[docs]
fts_type_title = _('Political Businesses')
[docs]
fts_title_property = 'title'
[docs]
fts_properties = {
'title': {'type': 'text', 'weight': 'A'},
'number': {'type': 'text', 'weight': 'A'}
}
@property
[docs]
def fts_suggestion(self) -> list[str]:
if self.number is None:
return [self.title]
return [
f'{self.title} {self.number}',
f'{self.number} {self.title}'
]
#: Internal ID
[docs]
id: Mapped[UUID] = mapped_column(
primary_key=True,
default=uuid4,
)
#: The title of the agenda item
#: number of the agenda item
[docs]
number: Mapped[str | None]
#: business type of the agenda item
[docs]
political_business_type: Mapped[PoliticalBusinessType] = mapped_column(
Enum(
*POLITICAL_BUSINESS_TYPE.keys(),
name='par_political_business_type',
),
)
#: status of the political business
[docs]
status: Mapped[PoliticalBusinessStatus | None] = mapped_column(
Enum(
*POLITICAL_BUSINESS_STATUS.keys(),
name='par_political_business_status',
),
)
#: entry date of political business
[docs]
entry_date: Mapped[date | None]
#: may have participants (Verfasser/Beteiligte) depending on the type
[docs]
participants: Mapped[list[PoliticalBusinessParticipation]] = (
relationship(
back_populates='political_business',
order_by='desc(PoliticalBusinessParticipation.participant_type)',
)
)
#: parliamentary groups (Fraktionen)
[docs]
parliamentary_groups: Mapped[list[RISParliamentaryGroup]] = (
relationship(
secondary=par_political_business_parliamentary_groups,
back_populates='political_businesses',
passive_deletes=True
)
)
[docs]
meeting_items: Mapped[list[MeetingItem]] = relationship(
back_populates='political_business',
)
@hybrid_property
[docs]
def display_name(self) -> str:
return f'{self.number} {self.title}' if self.number else self.title
@display_name.inplace.expression
@classmethod
[docs]
def _display_name_expression(cls) -> ColumnElement[str]:
return func.concat(
func.coalesce(cls.number, ''),
case(
(and_(cls.number.isnot(None), cls.number != ''), ' '),
else_=''
),
cls.title
)
[docs]
def __repr__(self) -> str:
return (f'<Political Business {self.number}, '
f'{self.title}, {self.political_business_type}>')
[docs]
class PoliticalBusinessParticipation(Base, ContentMixin):
""" A participant of a political business, e.g. a parliamentarian. """
[docs]
__tablename__ = 'par_political_business_participants'
#: Internal ID
[docs]
id: Mapped[UUID] = mapped_column(
primary_key=True,
default=uuid4,
)
#: The id of the political business
[docs]
political_business_id: Mapped[UUID] = mapped_column(
ForeignKey('par_political_businesses.id'),
)
#: The id of the parliamentarian
[docs]
parliamentarian_id: Mapped[UUID] = mapped_column(
ForeignKey('par_parliamentarians.id'),
)
#: the role of the parliamentarian in the political business
[docs]
participant_type: Mapped[str | None]
#: the related political business
[docs]
political_business: Mapped[PoliticalBusiness] = relationship(
back_populates='participants',
)
#: the related parliamentarian
[docs]
parliamentarian: Mapped[RISParliamentarian] = relationship(
back_populates='political_businesses',
)
[docs]
def __repr__(self) -> str:
return (f'<Political Business Participation '
f'{self.parliamentarian.title}, '
f'{self.political_business.title}, '
f'{self.participant_type}>')
[docs]
class PoliticalBusinessCollection(
GenericCollection[PoliticalBusiness],
Pagination[PoliticalBusiness]
):
def __init__(
self,
request: OrgRequest,
page: int = 0,
term: str | None = None,
status: Collection[PoliticalBusinessStatus] | None = None,
types: Collection[PoliticalBusinessType] | None = None,
years: Collection[int] | None = None,
) -> None:
super().__init__(request.session)
[docs]
self.status = set(status) if status else set()
[docs]
self.types = set(types) if types else set()
[docs]
self.years = set(years) if years else set()
@property
[docs]
def q(self) -> str | None:
return self.term
@property
[docs]
def model_class(self) -> type[PoliticalBusiness]:
return PoliticalBusiness
[docs]
def __eq__(self, other: object) -> bool:
return (
isinstance(other, self.__class__)
and self.page == other.page
)
[docs]
def query(self) -> Query[PoliticalBusiness]:
query = super().query()
if self.term:
language = self.request.locale
if language_from_locale(language) == 'simple':
language = 'simple'
query = query.join(
SearchIndex,
SearchIndex.owner_id_uuid == PoliticalBusiness.id
)
query = query.filter(SearchIndex.data_vector.op('@@')(
func.websearch_to_tsquery(language, self.term)
))
if self.types:
query = query.filter(
PoliticalBusiness.political_business_type.in_(self.types)
)
if self.status:
query = query.filter(
PoliticalBusiness.status.in_(self.status)
)
if self.years:
query = query.filter(
or_(*[
and_(
PoliticalBusiness.entry_date.isnot(None),
PoliticalBusiness.entry_date >= date(year, 1, 1),
PoliticalBusiness.entry_date < date(year + 1, 1, 1),
)
for year in self.years
])
)
return query.order_by(self.model_class.entry_date.desc(),
self.model_class.title)
[docs]
def subset(self) -> Query[PoliticalBusiness]:
return self.query()
[docs]
def page_by_index(self, index: int) -> Self:
return self.__class__(
self.request,
term=self.term,
page=index,
status=self.status,
types=self.types,
years=self.years,
)
@property
[docs]
def page_index(self) -> int:
return self.page
[docs]
def for_filter(
self,
status: PoliticalBusinessStatus | None = None,
type: PoliticalBusinessType | None = None,
year: int | None = None,
) -> Self:
status_ = toggle(self.status, status)
types = toggle(self.types, type)
years = toggle(self.years, year)
return self.__class__(
self.request,
page=0,
term=self.term,
status=status_,
types=types,
years=years,
)
[docs]
def years_for_entries(self) -> list[int]:
""" Returns a list of years for which there are entries in the db """
year = func.extract('year', PoliticalBusiness.entry_date).label('year')
years = self.session.query(year).filter(
PoliticalBusiness.entry_date.isnot(None)
).distinct().order_by(year.desc())
# convert to a list of integers, remove duplicates, and sort
return sorted({int(year[0]) for year in years}, reverse=True)
[docs]
def by_display_name(self, display_name: str) -> PoliticalBusiness | None:
""" Returns the given political business by display name or None. """
return (
self.query()
.filter(PoliticalBusiness.display_name == display_name)
.first()
)
[docs]
def by_parliamentarian(
self,
parliamentarian_id: UUID
) -> Query[PoliticalBusiness]:
""" Returns political businesses by given parliamentarian id """
return (
self.session.query(PoliticalBusiness)
.filter(PoliticalBusiness.participants.any(
PoliticalBusinessParticipation.parliamentarian_id ==
parliamentarian_id
))
.order_by(
PoliticalBusiness.entry_date.desc(),
PoliticalBusiness.title
)
)
[docs]
class PoliticalBusinessParticipationCollection(
GenericCollection[PoliticalBusinessParticipation]
):
def __init__(self, session: Session, active: bool | None = None):
super().__init__(session)
@property
[docs]
def model_class(self) -> type[PoliticalBusinessParticipation]:
return PoliticalBusinessParticipation
[docs]
def by_parliamentarian_id(
self,
parliamentarian_id: UUID
) -> Query[PoliticalBusinessParticipation]:
return self.query().filter_by(parliamentarian_id=parliamentarian_id)