from onegov.core.orm import Base
from onegov.core.orm.mixins import TimestampMixin
from onegov.core.orm.types import UUID
from onegov.election_day import _
from onegov.election_day.models.election import Election
from onegov.election_day.models.vote import Vote
from sqlalchemy import Column
from sqlalchemy import desc
from sqlalchemy import Enum
from sqlalchemy import ForeignKey
from sqlalchemy import Text
from sqlalchemy.orm import object_session
from sqlalchemy.orm import relationship
from uuid import uuid4
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import uuid
from onegov.core.types import AppenderQuery
from sqlalchemy.orm import Query
from typing import Literal
from typing import TypeAlias
[docs]
UploadType: TypeAlias = Literal['vote', 'proporz', 'majorz']
[docs]
UPLOAD_TYPE_LABELS = (
('vote', _('Vote')),
('proporz', _('Election based on proportional representation')),
('majorz', _('Election based on the simple majority system')),
)
[docs]
class DataSource(Base, TimestampMixin):
""" Stores the data source of an upload. """
[docs]
__tablename__ = 'upload_data_source'
#: Identifies the data source
[docs]
id: 'Column[uuid.UUID]' = Column(
UUID, # type:ignore[arg-type]
primary_key=True,
default=uuid4
)
#: The name of the upload configuration
[docs]
name: 'Column[str]' = Column(Text, nullable=False)
#: The token used to authenticate
[docs]
token: 'Column[uuid.UUID]' = Column(
UUID, # type:ignore[arg-type]
nullable=False,
default=uuid4
)
#: The type of upload
[docs]
type: 'Column[UploadType]' = Column(
Enum( # type:ignore[arg-type]
'vote',
'majorz',
'proporz',
name='type_of_data_source'
),
nullable=False
)
#: A configuration may contain n items
[docs]
items: 'relationship[AppenderQuery[DataSourceItem]]' = relationship(
'DataSourceItem',
cascade='all, delete-orphan',
lazy='dynamic',
back_populates='source',
)
@property
[docs]
def label(self) -> str:
return dict(UPLOAD_TYPE_LABELS)[self.type]
[docs]
def query_candidates(self) -> 'Query[Election | Vote]':
""" Returns a list of available votes or elections matching the
type of the source. """
session = object_session(self)
if self.type == 'vote':
query = session.query(Vote)
query = query.order_by(
desc(Vote.date),
Vote.domain,
Vote.shortcode,
Vote.title
)
return query
query = session.query(Election).filter(Election.type == self.type)
query = query.order_by(
desc(Election.date),
Election.domain,
Election.shortcode,
Election.title
)
return query
[docs]
class DataSourceItem(Base, TimestampMixin):
""" Stores the configuration of an auto upload. """
[docs]
__tablename__ = 'upload_data_source_item'
[docs]
id: 'Column[uuid.UUID]' = Column(
UUID, # type:ignore[arg-type]
primary_key=True,
default=uuid4
)
#: the upload configuration result belongs to
[docs]
source_id: 'Column[uuid.UUID]' = Column(
UUID, # type:ignore[arg-type]
ForeignKey(DataSource.id),
nullable=False
)
#: the district
[docs]
district: 'Column[str | None]' = Column(Text, nullable=True)
#: the vote / election number
[docs]
number: 'Column[str | None]' = Column(Text, nullable=True)
#: the election
[docs]
election_id: 'Column[str | None]' = Column(
Text,
ForeignKey(Election.id, onupdate='CASCADE'),
nullable=True
)
[docs]
election: 'relationship[Election | None]' = relationship(
'Election',
back_populates='data_sources'
)
#: the vote
[docs]
vote_id: 'Column[str | None]' = Column(
Text,
ForeignKey(Vote.id, onupdate='CASCADE'),
nullable=True
)
[docs]
vote: 'relationship[Vote | None]' = relationship(
'Vote',
back_populates='data_sources'
)
[docs]
source: 'relationship[DataSource]' = relationship(
DataSource,
back_populates='items'
)
@property
[docs]
def item(self) -> Election | Vote | None:
""" Returns the vote or election. """
if self.source.type == 'vote':
return self.vote
else:
return self.election
@property
[docs]
def name(self) -> str:
item = self.item
if item:
return item.title or ''
return ''