"""Vote models for raw roll-call data and member positions.""" from __future__ import annotations from datetime import date, datetime from typing import TYPE_CHECKING from sqlalchemy import DateTime, ForeignKey, Index, UniqueConstraint from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship from pipelines.orm.data_science_dev.base import ( DataScienceDevBase, DataScienceDevTableBase, ) if TYPE_CHECKING: from pipelines.orm.data_science_dev.congress.context import ( VoteActionMatch, VoteClassification, VoteContextAudit, VoteMeasureLink, VotePositionMeaning, VoteTextTarget, ) from pipelines.orm.data_science_dev.congress.legislator import Legislator class VoteRecord(DataScienceDevBase): """Links a vote to a legislator with their position (Yea, Nay, etc.).""" __tablename__ = "vote_record" vote_id: Mapped[int] = mapped_column( ForeignKey("main.vote.id", ondelete="CASCADE"), primary_key=True, ) legislator_id: Mapped[int] = mapped_column( ForeignKey("main.legislator.id", ondelete="CASCADE"), primary_key=True, ) position: Mapped[str] vote: Mapped[Vote] = relationship("Vote", back_populates="vote_records") legislator: Mapped[Legislator] = relationship( "Legislator", back_populates="vote_records" ) class Vote(DataScienceDevTableBase): """Raw roll call vote facts from House or Senate vote sources.""" __tablename__ = "vote" __table_args__ = ( UniqueConstraint( "congress", "chamber", "session_number", "roll_number", name="uq_vote_congress_chamber_session_number_roll_number", ), Index("ix_vote_date", "vote_date"), Index("ix_vote_congress_chamber", "congress", "chamber"), ) congress: Mapped[int] chamber: Mapped[str] session_year: Mapped[int] session_number: Mapped[int] roll_number: Mapped[int] vote_type: Mapped[str | None] question: Mapped[str | None] result: Mapped[str | None] result_text: Mapped[str | None] vote_date: Mapped[date] vote_datetime: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) raw_vote_source_url: Mapped[str | None] yea_count: Mapped[int | None] nay_count: Mapped[int | None] not_voting_count: Mapped[int | None] present_count: Mapped[int | None] raw_bill_ref: Mapped[dict | None] = mapped_column(JSONB) raw_amendment_ref: Mapped[dict | None] = mapped_column(JSONB) raw_nomination_ref: Mapped[dict | None] = mapped_column(JSONB) raw_treaty_ref: Mapped[dict | None] = mapped_column(JSONB) raw_vote_source_artifact_id: Mapped[int | None] = mapped_column( ForeignKey("main.source_artifact.id", ondelete="SET NULL") ) vote_records: Mapped[list[VoteRecord]] = relationship( "VoteRecord", back_populates="vote", cascade="all, delete-orphan", ) action_matches: Mapped[list[VoteActionMatch]] = relationship( "VoteActionMatch", back_populates="vote", cascade="all, delete-orphan", ) classification: Mapped[VoteClassification | None] = relationship( "VoteClassification", back_populates="vote", cascade="all, delete-orphan", uselist=False, ) vote_measure_links: Mapped[list[VoteMeasureLink]] = relationship( "VoteMeasureLink", back_populates="vote", cascade="all, delete-orphan", ) text_target: Mapped[VoteTextTarget | None] = relationship( "VoteTextTarget", back_populates="vote", cascade="all, delete-orphan", uselist=False, ) position_meaning: Mapped[VotePositionMeaning | None] = relationship( "VotePositionMeaning", back_populates="vote", cascade="all, delete-orphan", uselist=False, ) context_audit_rows: Mapped[list[VoteContextAudit]] = relationship( "VoteContextAudit", back_populates="vote", cascade="all, delete-orphan", )