-
Notifications
You must be signed in to change notification settings - Fork 0
/
crud.py
118 lines (106 loc) · 5.35 KB
/
crud.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from sqlalchemy.orm import Session, load_only
from sqlalchemy import func
import uuid
import models, schemas
from typing import Optional, List
from fastapi import Request, HTTPException
import email_scheduler
import datetime
def get_event(db: Session, event_id: uuid.UUID):
event = db.query(models.Event).get(event_id)
is_custom_answers_added = event.is_custom_answers_added
answers = []
for answer in event.answers:
if answer.is_custom:
if is_custom_answers_added:
answers.append(answer.answer)
else:
answers.append(answer.answer)
event = event.__dict__
event['answers'] = answers
event['admin_email'] = ''
event['admin_token'] = ''
return event
def create_event(db: Session, event: schemas.EventCreate, answers: List[str]):
db_event = models.Event(**event.dict())
db.add(db_event)
db.flush()
db.commit()
for answer in answers:
db_answer = models.Answer(answer=answer, event_id=db_event.id)
db.add(db_answer)
db.flush()
db.commit()
if event.will_email_admin and event.admin_email != '':
email_scheduler.add_job(str(db_event.id), event.voting_deadline, event.admin_email, event.question)
return db_event
def patch_event_admin_email(db: Session, admin_email: str, will_email_admin: bool, event_id: uuid.UUID): # TODO set cron or other to send email
db_event = db.query(models.Event).get(event_id)
db_event.will_email_admin = will_email_admin
db_event.admin_email = admin_email
db.commit()
return db_event
def create_vote(db: Session, votes: List[dict], event_id: uuid.UUID, voter_username: str, voter_email: str):
event = db.query(models.Event).get(event_id)
if datetime.datetime.now() > event.voting_deadline:
raise HTTPException(status_code=400, detail="Voting has reached its deadline and is now closed.")
if not voter_username:
voter_username = 'anonymous' + uuid.uuid4().hex
if voter_email != '':
db_voter = db.query(models.Voter).filter(models.Voter.email == voter_email).first()
else:
db_voter = db.query(models.Voter).filter(models.Voter.username == voter_username).first()
if not db_voter:
db_voter = models.Voter(username=voter_username, email=voter_email)
db.add(db_voter)
db.flush()
db.commit()
can_write_custom = event.can_write_custom
is_vote_changeable = event.is_vote_changeable # TODO if same username or email, say u can either change vote or not! CAN'T multiple vote which means see if there is a voter id and answer id or event id already in place
answers = {}
for answer in event.answers:
if not answer.is_custom:
answers[answer.answer] = answer.id
rank = 0
for vote in votes:
try:
answer_id = answers[vote['choice']]
except KeyError:
if can_write_custom:
db_answer = models.Answer(answer=vote['choice'], event_id=event_id, is_custom=True)
db.add(db_answer)
db.flush()
db.commit()
answer_id = db_answer.id
else:
raise HTTPException(status_code=400, detail= "Cannot add custom answer")
if vote['dates']:
for date in vote['dates']:
date_start = datetime.datetime.strptime(date['start'].split(' (')[0], '%a %b %d %Y %H:%M:%S GMT%z') # todo this works for locale en_us, best if client converts locale to timestamp without timezone and sends that value to api
date_end = datetime.datetime.strptime(date['end'].split(' (')[0], '%a %b %d %Y %H:%M:%S GMT%z')
db_vote = models.Vote(answer_id=answer_id, voter_id=db_voter.id, rank=rank, date_start=date_start, date_end=date_end)
db.add(db_vote)
else:
db_vote = models.Vote(answer_id=answer_id, voter_id=db_voter.id, rank=rank)
db.add(db_vote)
rank += 1
db.flush()
db.commit()
return {"success": 1}
def get_result(db: Session, event_id: uuid.UUID): # TODO pagination?
# rows = db.query(models.Answer.answer, func.count(func.distinct(models.Answer.answer, models.Voter.username))).join(models.Event).join(models.Vote).filter(models.Event.id==event_id).group_by(models.Answer.answer).group_by(models.Voter.username).all()
rows = db.query(models.Answer.answer, func.count(models.Answer.answer)).join(models.Event).filter(models.Event.id==event_id).group_by(models.Answer.answer).all()
event = db.query(models.Event).filter(models.Event.id==event_id).first()
return {'rows': dict(rows), 'question': event.question, 'voting_deadline': event.voting_deadline}
# TODO this voting deadline does not return timezone, is it saved in db to begin with?
def get_voters(db: Session, event_id: uuid.UUID, answer: dict): # TODO pagination?
voters = db.query(models.Voter.username, models.Vote.date_start, models.Vote.date_end).join(models.Vote).join(models.Answer).filter(models.Answer.event_id==event_id).filter(models.Answer.answer==answer['answer']).all()
return voters
def get_user(db: Session, email: str):
user = db.query(models.Voter).filter(models.Voter.email==email).first()
return user
def create_user(db: Session, email: str, password: str):
db_user = models.Voter(email=email, password=password)
db.add(db_user)
db.commit()
return {'id': db_user.id, 'email': db_user.email}