Source code for election_day.collections.election_compounds

from onegov.core.collection import Pagination
from onegov.election_day.models import ElectionCompound
from sqlalchemy import cast
from sqlalchemy import desc
from sqlalchemy import distinct
from sqlalchemy import extract
from sqlalchemy import Integer


from typing import TYPE_CHECKING
if TYPE_CHECKING:
    from datetime import date
    from sqlalchemy.orm import Query, Session
    from typing import Self


[docs] class ElectionCompoundCollection(Pagination[ElectionCompound]):
[docs] page: int
def __init__( self, session: 'Session', page: int = 0, year: int | None = None ): super().__init__(page)
[docs] self.session = session
[docs] self.year = year
[docs] def __eq__(self, other: object) -> bool: if not isinstance(other, self.__class__): return False return self.year == other.year and self.page == other.page
[docs] def subset(self) -> 'Query[ElectionCompound]': query = self.query() query = query.order_by( desc(ElectionCompound.date), ElectionCompound.shortcode, ElectionCompound.title ) if self.year: query = query.filter( extract('year', ElectionCompound.date) == self.year ) return query
@property
[docs] def page_index(self) -> int: return self.page
[docs] def page_by_index(self, index: int) -> 'Self': return self.__class__(self.session, index, self.year)
[docs] def for_year(self, year: int | None) -> 'Self': return self.__class__(self.session, 0, year)
[docs] def query(self) -> 'Query[ElectionCompound]': return self.session.query(ElectionCompound)
[docs] def get_latest(self) -> list[ElectionCompound] | None: """ Returns the election compounds with the latest date. """ latest_date = self.query().with_entities(ElectionCompound.date) latest_date = latest_date.order_by(desc(ElectionCompound.date)) latest_date = latest_date.limit(1).scalar() return self.by_date(latest_date) if latest_date else None
[docs] def get_years(self) -> list[int]: """ Returns a list of years for which there are election compounds. """ year = cast(extract('year', ElectionCompound.date), Integer) query = self.session.query(distinct(year)) query = query.order_by(desc(year)) return [year for year, in query]
[docs] def by_date(self, date: 'date') -> list[ElectionCompound]: """ Returns the election compounds on the given date. """ query = self.query() query = query.filter(ElectionCompound.date == date) query = query.order_by( ElectionCompound.shortcode, ElectionCompound.title ) return query.all()
[docs] def by_year(self, year: int | None = None) -> list[ElectionCompound]: """ Returns the election compounds for the current/given year. """ year = year or self.year query = self.query() query = query.filter(extract('year', ElectionCompound.date) == year) query = query.order_by( ElectionCompound.date, ElectionCompound.shortcode, ElectionCompound.title ) return query.all()
[docs] def by_id(self, id: str) -> ElectionCompound | None: """ Returns the election compound by id. """ query = self.query() query = query.filter(ElectionCompound.id == id) return query.first()