diff options
| -rw-r--r-- | app.py | 68 | ||||
| -rw-r--r-- | db.py | 34 | ||||
| -rw-r--r-- | migrations/0005_indices.sql | 7 |
3 files changed, 65 insertions, 44 deletions
@@ -79,27 +79,33 @@ def index(): def create(): form = request.form - if "title" not in form or len(form["title"]) == 0: + title = form.get("title") + title = title.strip() if title is not None else None + if title is None or len(title) == 0: return error_page("Title is required") - if len(form["title"]) > TITLE_MAX_LENGTH: + if len(title) > TITLE_MAX_LENGTH: return error_page(f"Title must be {TITLE_MAX_LENGTH} characters or fewer") - if "description" in form and len(form["description"]) > DESCRIPTION_MAX_LENGTH: + description = form.get("description") + description = description.strip() if description is not None else None + if description is not None and len(description) > DESCRIPTION_MAX_LENGTH: return error_page(f"Description must be {DESCRIPTION_MAX_LENGTH} characters or fewer") - if "author_name" not in form or len(form["author_name"]) == 0: + author_name = form.get("author_name") + author_name = author_name.strip() if author_name is not None else None + if author_name is None or len(author_name) == 0: return error_page("Author name is required") - if len(form["author_name"]) > AUTHOR_NAME_MAX_LENGTH: + if len(author_name) > AUTHOR_NAME_MAX_LENGTH: return error_page(f"Author name must be {AUTHOR_NAME_MAX_LENGTH} characters or fewer") - if "author_email" in form and len(form["author_email"]) > AUTHOR_EMAIL_MAX_LENGTH: + author_email = form.get("author_email") + author_email = author_email.strip() if author_email is not None else None + if author_email is not None and len(author_email) > AUTHOR_EMAIL_MAX_LENGTH: return error_page(f"Author email must be {AUTHOR_EMAIL_MAX_LENGTH} characters or fewer") - choices = [] poll = db.create_poll( - form["title"], - form["description"], - form["author_name"], - form["author_email"], + title, + description, + author_name, + author_email, "is_whole_day" in form, - choices, ) if email_client.email_enabled: @@ -179,17 +185,17 @@ def vote_poll(id): return error_page("Invalid poll ID", 400) form = request.form - if "voter_name" not in form or len(form["voter_name"]) == 0: + voter_name = form.get("voter_name") + voter_name = voter_name.strip() if voter_name is not None else None + if voter_name is None or len(voter_name) == 0: return error_page("Voter name is required") - if len(form["voter_name"]) > VOTER_NAME_MAX_LENGTH: + if len(voter_name) > VOTER_NAME_MAX_LENGTH: return error_page(f"Voter name must be {VOTER_NAME_MAX_LENGTH} characters or fewer") poll = db.get_poll(id) if poll is None: return error_page("Poll not found") - voter_name: str = form["voter_name"] - selections: dict[str, int] = {} for choice in poll.choices: selections[choice.id] = 0 @@ -200,6 +206,8 @@ def vote_poll(id): selections[choice_id] = 1 manage_code = db.vote_poll(id, voter_name, selections) + if manage_code is None: + return error_page("That name is already in use") if email_client.email_enabled: def task(): @@ -240,25 +248,33 @@ def update_poll_info(code): form = request.form - if "title" not in form or len(form["title"]) == 0: + title = form.get("title") + title = title.strip() if title is not None else None + if title is None or len(title) == 0: return error_page("Title is required") - if len(form["title"]) > TITLE_MAX_LENGTH: + if len(title) > TITLE_MAX_LENGTH: return error_page(f"Title must be {TITLE_MAX_LENGTH} characters or fewer") - if "description" in form and len(form["description"]) > DESCRIPTION_MAX_LENGTH: + description = form.get("description") + description = description.strip() if description is not None else None + if description is not None and len(description) > DESCRIPTION_MAX_LENGTH: return error_page(f"Description must be {DESCRIPTION_MAX_LENGTH} characters or fewer") - if "author_name" not in form or len(form["author_name"]) == 0: + author_name = form.get("author_name") + author_name = author_name.strip() if author_name is not None else None + if author_name is None or len(author_name) == 0: return error_page("Author name is required") - if len(form["author_name"]) > AUTHOR_NAME_MAX_LENGTH: + if len(author_name) > AUTHOR_NAME_MAX_LENGTH: return error_page(f"Author name must be {AUTHOR_NAME_MAX_LENGTH} characters or fewer") - if "author_email" in form and len(form["author_email"]) > AUTHOR_EMAIL_MAX_LENGTH: + author_email = form.get("author_email") + author_email = author_email.strip() if author_email is not None else None + if author_email is not None and len(author_email) > AUTHOR_EMAIL_MAX_LENGTH: return error_page(f"Author email must be {AUTHOR_EMAIL_MAX_LENGTH} characters or fewer") changed = db.update_poll_info( code, - form["title"], - form["description"], - form["author_name"], - form["author_email"], + title, + description, + author_name, + author_email, "is_whole_day" in form, ) @@ -179,11 +179,10 @@ def get_poll(id: str): raise e def create_poll(title: str, - description: str, + description: str | None, author_name: str, - author_email: str, - is_whole_day: bool, - choices: list[Choice]): + author_email: str | None, + is_whole_day: bool): with db.cursor() as (conn, cur): try: @@ -197,19 +196,14 @@ def create_poll(title: str, poll = tuple_to_poll(poll_t) - for choice in choices: - cur.execute("INSERT INTO choices (poll_id, start_datetime, end_datetime)" - "VALUES (%s, %s, %s)", - (poll.id, choice.start_datetime, choice.end_datetime)) - conn.commit() return poll except Exception as e: conn.rollback() raise e -def vote_poll(poll_id: str, voter_name: str, selections: dict[str, int]) -> str: - """Returns the manage code of the vote.""" +def vote_poll(poll_id: str, voter_name: str, selections: dict[str, int]) -> str | None: + """Returns the manage code of the vote or None if the vote failed on unique constraint.""" with db.cursor() as (conn, cur): try: manage_code = str(uuid.uuid4()) @@ -220,6 +214,9 @@ def vote_poll(poll_id: str, voter_name: str, selections: dict[str, int]) -> str: (poll_id, voter_name, choice_id, value, manage_code)) conn.commit() return manage_code + except psycopg2.errors.UniqueViolation: + conn.rollback() + return None except Exception as e: conn.rollback() raise e @@ -240,12 +237,12 @@ def get_poll_by_code(code: str) -> Poll | None: raise e def update_poll_info( - code, - title, - description, - author_name, - author_email, - is_whole_day, + code: str, + title: str, + description: str | None, + author_name: str, + author_email: str | None, + is_whole_day: bool, ) -> str | None: """Returns the id of the updated poll or None if not found.""" with db.cursor() as (conn, cur): @@ -295,7 +292,8 @@ def get_polls_by_codes(codes: list[str]) -> list[Poll]: with db.cursor() as (conn, cur): try: codes_t = tuple(codes) - cur.execute("SELECT * FROM polls WHERE manage_code IN %s" + cur.execute("SELECT * FROM polls " + "WHERE manage_code IN %s " "ORDER BY pub_date DESC", (codes_t,)) poll_ts = cur.fetchall() polls = [tuple_to_poll(poll_t) for poll_t in poll_ts] diff --git a/migrations/0005_indices.sql b/migrations/0005_indices.sql new file mode 100644 index 0000000..83d0d09 --- /dev/null +++ b/migrations/0005_indices.sql @@ -0,0 +1,7 @@ +ALTER TABLE votes ADD CONSTRAINT idx_votes_unique_voter_name UNIQUE (poll_id, choice_id, voter_name); + +CREATE INDEX idx_votes_poll_id_voter_name ON votes (poll_id, voter_name); +CREATE INDEX idx_choices_poll_id_start_datetime ON choices (poll_id, start_datetime); +CREATE INDEX idx_votes_choice_id ON votes (choice_id); +CREATE INDEX idx_votes_manage_code ON votes (manage_code); +CREATE INDEX idx_polls_manage_code_pub_date ON polls (manage_code, pub_date); |