1
- import dataclasses
2
1
import logging
2
+ from dataclasses import dataclass
3
3
from datetime import datetime
4
4
5
5
import sqlalchemy
9
9
from officers .constants import OfficerPosition
10
10
from officers .tables import OfficerInfo , OfficerTerm
11
11
12
+
13
+ @dataclass
14
+ class ElectionParameters :
15
+ """
16
+ Dataclass encompassing the data that can go into an Election.
17
+ """
18
+ slug : str
19
+ name : str
20
+ officer_id : str
21
+ type : str
22
+ date : datetime
23
+ end_date : datetime
24
+ survey_link : str
25
+
26
+
12
27
_logger = logging .getLogger (__name__ )
13
28
14
29
async def get_election (db_session : database .DBSession , election_slug : str ) -> Election | None :
@@ -20,18 +35,18 @@ async def get_election(db_session: database.DBSession, election_slug: str) -> El
20
35
result = await db_session .scalar (query )
21
36
return result
22
37
23
- async def create_election (params : dict [ str , datetime ] , db_session : database .DBSession ) -> None :
38
+ async def create_election (params : ElectionParameters , db_session : database .DBSession ) -> None :
24
39
"""
25
40
Creates a new election with given parameters.
26
41
Does not validate if an election _already_ exists
27
42
"""
28
- election = Election (slug = params [ " slug" ] ,
29
- name = params [ " name" ] ,
30
- officer_id = params [ " officer_id" ] ,
31
- type = params [ " type" ] ,
32
- date = params [ " date" ] ,
33
- end_date = params [ " end_date" ] ,
34
- websurvey = params [ "websurvey" ] )
43
+ election = Election (slug = params . slug ,
44
+ name = params . name ,
45
+ officer_id = params . officer_id ,
46
+ type = params . type ,
47
+ date = params . date ,
48
+ end_date = params . end_date ,
49
+ survey_link = params . survey_link )
35
50
db_session .add (election )
36
51
37
52
async def delete_election (slug : str , db_session : database .DBSession ) -> None :
@@ -42,25 +57,25 @@ async def delete_election(slug: str, db_session: database.DBSession) -> None:
42
57
query = sqlalchemy .delete (Election ).where (Election .slug == slug )
43
58
await db_session .execute (query )
44
59
45
- async def update_election (params : dict [ str , datetime ] , db_session : database .DBSession ) -> None :
60
+ async def update_election (params : ElectionParameters , db_session : database .DBSession ) -> None :
46
61
"""
47
62
Updates an election with the provided parameters.
48
63
Take care as this will replace values with None if not populated.
49
64
You _cannot_ change the name or slug, you should instead delete and create a new election.
50
65
Does not validate if an election _already_ exists
51
66
"""
52
67
53
- election = (await db_session .execute (sqlalchemy .select (Election ).filter_by (slug = params [ " slug" ] ))).scalar_one ()
68
+ election = (await db_session .execute (sqlalchemy .select (Election ).filter_by (slug = params . slug ))).scalar_one ()
54
69
55
- if params [ " date" ] is not None :
56
- election .date = params [ " date" ]
57
- if params [ " type" ] is not None :
58
- election .type = params [ " type" ]
59
- if params [ " end_date" ] is not None :
60
- election .end_date = params [ " end_date" ]
61
- if params [ "websurvey" ] is not None :
62
- election .websurvey = params [ "websurvey" ]
70
+ if params . date is not None :
71
+ election .date = params . date
72
+ if params . type is not None :
73
+ election .type = params . type
74
+ if params . end_date is not None :
75
+ election .end_date = params . end_date
76
+ if params . survey_link is not None :
77
+ election .survey_link = params . survey_link
63
78
64
- query = sqlalchemy .update (Election ).where (Election .slug == params [ " slug" ] ).values (election )
79
+ query = sqlalchemy .update (Election ).where (Election .slug == params . slug ).values (election )
65
80
await db_session .execute (query )
66
81
0 commit comments