Skip to content

Second Elections PR #105

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 7 commits into
base: main
Choose a base branch
from
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 11 additions & 10 deletions src/alembic/versions/243190df5588_create_election_tables.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,34 +28,35 @@ def upgrade() -> None:
sa.Column("datetime_start_nominations", sa.DateTime(), nullable=False),
sa.Column("datetime_start_voting", sa.DateTime(), nullable=False),
sa.Column("datetime_end_voting", sa.DateTime(), nullable=False),
sa.Column("avaliable_positions", sa.Text(), nullable=False),
sa.Column("survey_link", sa.String(length=300), nullable=True),
sa.PrimaryKeyConstraint("slug")
)
op.create_table(
"election_nominee",
"election_nominee_info",
sa.Column("computing_id", sa.String(length=32), nullable=False),
sa.Column("full_name", sa.String(length=64), nullable=False),
sa.Column("facebook", sa.String(length=128), nullable=True),
sa.Column("linked_in", sa.String(length=128), nullable=True),
sa.Column("instagram", sa.String(length=128), nullable=True),
sa.Column("email", sa.String(length=64), nullable=True),
sa.Column("discord", sa.String(length=32), nullable=True),
sa.Column("discord_id", sa.String(length=32), nullable=True),
#sa.Column("discord", sa.String(length=32), nullable=True),
#sa.Column("discord_id", sa.String(length=32), nullable=True),
sa.Column("discord_username", sa.String(length=32), nullable=True),
sa.PrimaryKeyConstraint("computing_id")
)
op.create_table(
"nominee_application",
"election_nominee_application",
sa.Column("computing_id", sa.String(length=32), nullable=False),
sa.Column("nominee_election", sa.String(length=32), nullable=False),
sa.Column("speech", sa.Text(), nullable=True),
sa.Column("nominee_election", sa.String(length=64), nullable=False),
sa.Column("position", sa.String(length=64), nullable=False),
sa.Column("speech", sa.Text(), nullable=True),
sa.ForeignKeyConstraint(["computing_id"], ["election_nominee.computing_id"]),
sa.ForeignKeyConstraint(["nominee_election"], ["election.slug"]),
sa.PrimaryKeyConstraint("computing_id", "nominee_election")
sa.PrimaryKeyConstraint("computing_id", "nominee_election", "position")
)


def downgrade() -> None:
op.drop_table("nominee_application")
op.drop_table("election_nominee")
op.drop_table("election_nominee_application")
op.drop_table("election_nominee_info")
op.drop_table("election")
137 changes: 117 additions & 20 deletions src/elections/crud.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,49 +3,146 @@
import sqlalchemy
from sqlalchemy.ext.asyncio import AsyncSession

from elections.tables import Election
from elections.tables import Election, NomineeApplication, NomineeInfo

_logger = logging.getLogger(__name__)

async def get_all_elections(db_session: AsyncSession) -> list[Election] | None:
# TODO: can this return None?
election_list = (await db_session.scalars(
sqlalchemy
.select(Election)
)).all()
return election_list

async def get_election(db_session: AsyncSession, election_slug: str) -> Election | None:
return await db_session.scalar(
sqlalchemy
.select(Election)
.where(Election.slug == election_slug)
)

async def create_election(db_session: AsyncSession, election: Election) -> None:
async def create_election(db_session: AsyncSession, election: Election):
"""
Creates a new election with given parameters.
Does not validate if an election _already_ exists
"""
db_session.add(election)

async def update_election(db_session: AsyncSession, new_election: Election):
"""
Attempting to change slug will fail. Instead, you must create a new election.
"""
await db_session.execute(
sqlalchemy
.update(Election)
.where(Election.slug == new_election.slug)
.values(new_election.to_update_dict())
)

async def delete_election(db_session: AsyncSession, slug: str) -> None:
"""
Deletes a given election by its slug.
Does not validate if an election exists
Deletes a given election by its slug. Does not validate if an election exists
"""
await db_session.execute(
sqlalchemy
.delete(Election)
.where(Election.slug == slug)
)

async def update_election(db_session: AsyncSession, new_election: Election) -> bool:
"""
You attempting to change the name or slug will fail. Instead, you must create a new election.
"""
target_slug = new_election.slug
target_election = await get_election(db_session, target_slug)

if target_election is None:
return False
else:
await db_session.execute(
sqlalchemy
.update(Election)
.where(Election.slug == target_slug)
.values(new_election.to_update_dict())
# ------------------------------------------------------- #

# TODO: switch to only using one of application or registration
async def get_all_registrations(
db_session: AsyncSession,
computing_id: str,
election_slug: str
) -> list[NomineeApplication] | None:
registrations = (await db_session.scalars(
sqlalchemy
.select(NomineeApplication)
.where(
NomineeApplication.computing_id == computing_id
and NomineeApplication.election_slug == election_slug
)
return True
)).all()
return registrations

async def get_all_registrations_in_election(
db_session: AsyncSession,
election_slug: str,
) -> list[NomineeApplication] | None:
registrations = (await db_session.scalars(
sqlalchemy
.select(NomineeApplication)
.where(
NomineeApplication.election_slug == election_slug
)
)).all()
return registrations

async def add_registration(
db_session: AsyncSession,
initial_application: NomineeApplication
):
db_session.add(initial_application)

async def update_registration(
db_session: AsyncSession,
initial_application: NomineeApplication
):
await db_session.execute(
sqlalchemy
.update(NomineeApplication)
.where(
NomineeApplication.computing_id == initial_application.computing_id
and NomineeApplication.nominee_election == initial_application.nominee_election
and NomineeApplication.position == initial_application.position
)
.values(initial_application.to_update_dict())
)

async def delete_registration(
db_session: AsyncSession,
computing_id: str,
election_slug: str,
position: str
):
await db_session.execute(
sqlalchemy
.delete(NomineeApplication)
.where(
NomineeApplication.computing_id == computing_id
and NomineeApplication.nominee_election == election_slug
and NomineeApplication.position == position
)
)

# ------------------------------------------------------- #

async def get_nominee_info(
db_session: AsyncSession,
computing_id: str,
) -> NomineeInfo | None:
return await db_session.scalar(
sqlalchemy
.select(NomineeInfo)
.where(NomineeInfo.computing_id == computing_id)
)

async def create_nominee_info(
db_session: AsyncSession,
info: NomineeInfo,
):
db_session.add(info)

async def update_nominee_info(
db_session: AsyncSession,
info: NomineeInfo,
):
await db_session.execute(
sqlalchemy
.update(NomineeInfo)
.where(NomineeInfo.computing_id == info.computing_id)
.values(info.to_update_dict())
)
117 changes: 105 additions & 12 deletions src/elections/tables.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from datetime import datetime

from sqlalchemy import (
Column,
DateTime,
Expand All @@ -14,9 +16,20 @@
DISCORD_NICKNAME_LEN,
)
from database import Base
from officers.constants import COUNCIL_REP_ELECTION_POSITIONS, GENERAL_ELECTION_POSITIONS

# If you wish to add more elections & defaults, please see `create_election`
election_types = ["general_election", "by_election", "council_rep_election"]

DEFAULT_POSITIONS_GENERAL_ELECTION = ",".join(GENERAL_ELECTION_POSITIONS)
DEFAULT_POSITIONS_BY_ELECTION = ",".join(GENERAL_ELECTION_POSITIONS)
DEFAULT_POSITIONS_COUNCIL_REP_ELECTION = ",".join(COUNCIL_REP_ELECTION_POSITIONS)

STATUS_BEFORE_NOMINATIONS = "before_nominations"
STATUS_NOMINATIONS = "nominations"
STATUS_VOTING = "voting"
STATUS_AFTER_VOTING = "after_voting"

MAX_ELECTION_NAME = 64
MAX_ELECTION_SLUG = 64

Expand All @@ -30,9 +43,13 @@ class Election(Base):
datetime_start_nominations = Column(DateTime, nullable=False)
datetime_start_voting = Column(DateTime, nullable=False)
datetime_end_voting = Column(DateTime, nullable=False)

# a csv list of positions which must be elements of OfficerPosition
avaliable_positions = Column(Text, nullable=False)
survey_link = Column(String(300))

def serializable_dict(self) -> dict:
def private_details(self, at_time: datetime) -> dict:
# is serializable
return {
"slug": self.slug,
"name": self.name,
Expand All @@ -42,10 +59,28 @@ def serializable_dict(self) -> dict:
"datetime_start_voting": self.datetime_start_voting.isoformat(),
"datetime_end_voting": self.datetime_end_voting.isoformat(),

"status": self.status(at_time),
"avaliable_positions": self.avaliable_positions,
"survey_link": self.survey_link,
}

def public_details(self) -> dict:
def public_details(self, at_time: datetime) -> dict:
# is serializable
return {
"slug": self.slug,
"name": self.name,
"type": self.type,

"datetime_start_nominations": self.datetime_start_nominations.isoformat(),
"datetime_start_voting": self.datetime_start_voting.isoformat(),
"datetime_end_voting": self.datetime_end_voting.isoformat(),

"status": self.status(at_time),
"avaliable_positions": self.avaliable_positions,
}

def public_metadata(self, at_time: datetime) -> dict:
# is serializable
return {
"slug": self.slug,
"name": self.name,
Expand All @@ -54,6 +89,8 @@ def public_details(self) -> dict:
"datetime_start_nominations": self.datetime_start_nominations.isoformat(),
"datetime_start_voting": self.datetime_start_voting.isoformat(),
"datetime_end_voting": self.datetime_end_voting.isoformat(),

"status": self.status(at_time),
}

def to_update_dict(self) -> dict:
Expand All @@ -66,31 +103,87 @@ def to_update_dict(self) -> dict:
"datetime_start_voting": self.datetime_start_voting,
"datetime_end_voting": self.datetime_end_voting,

"avaliable_positions": self.avaliable_positions,
"survey_link": self.survey_link,
}

# Each row represents a nominee of a given election
class Nominee(Base):
__tablename__ = "election_nominee"
def status(self, at_time: datetime) -> str:
if at_time <= self.datetime_start_nominations:
return STATUS_BEFORE_NOMINATIONS
elif at_time <= self.datetime_start_voting:
return STATUS_NOMINATIONS
elif at_time <= self.datetime_end_voting:
return STATUS_VOTING
else:
return STATUS_AFTER_VOTING

class NomineeInfo(Base):
__tablename__ = "election_nominee_info"

# Previously named sfuid
computing_id = Column(String(COMPUTING_ID_LEN), primary_key=True)
full_name = Column(String(64), nullable=False)
facebook = Column(String(128))
linked_in = Column(String(128))
instagram = Column(String(128))
email = Column(String(64))
discord = Column(String(DISCORD_NAME_LEN))
discord_id = Column(String(DISCORD_ID_LEN))
#discord = Column(String(DISCORD_NAME_LEN))
#discord_id = Column(String(DISCORD_ID_LEN))
discord_username = Column(String(DISCORD_NICKNAME_LEN))

def to_update_dict(self) -> dict:
return {
"computing_id": self.computing_id,
"full_name": self.full_name,

"linked_in": self.linked_in,
"instagram": self.instagram,
"email": self.email,
"discord_username": self.discord_username,
}

def as_serializable(self) -> dict:
# NOTE: this function is currently the same as to_update_dict since the contents
# have a different invariant they're upholding, which may cause them to change if a
# new property is introduced. For example, dates must be converted into strings
# to be serialized, but must not for update dictionaries.
return {
"computing_id": self.computing_id,
"full_name": self.full_name,

"linked_in": self.linked_in,
"instagram": self.instagram,
"email": self.email,
"discord_username": self.discord_username,
}

class NomineeApplication(Base):
__tablename__ = "nominee_application"
__tablename__ = "election_nominee_application"

# TODO: add index for nominee_election?
computing_id = Column(ForeignKey("election_nominee.computing_id"), primary_key=True)
nominee_election = Column(ForeignKey("election.slug"), primary_key=True)
position = Column(String(64), primary_key=True)

speech = Column(Text)
position = Column(String(64), nullable=False)

__table_args__ = (
PrimaryKeyConstraint(computing_id, nominee_election),
PrimaryKeyConstraint(computing_id, nominee_election, position),
)

def serializable_dict(self) -> dict:
return {
"computing_id": self.computing_id,
"nominee_election": self.nominee_election,
"position": self.position,

"speech": self.speech,
}

def to_update_dict(self) -> dict:
return {
"computing_id": self.computing_id,
"nominee_election": self.nominee_election,
"position": self.position,

"speech": self.speech,
}

Loading
Loading