From a0d81cf601bb9d7af252c63f06f932217a161468 Mon Sep 17 00:00:00 2001 From: Ava Silver Date: Sun, 2 Aug 2026 16:49:43 -0400 Subject: [PATCH 1/2] add Session typing --- car/app.py | 100 ++++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 72 insertions(+), 28 deletions(-) diff --git a/car/app.py b/car/app.py index a6a3fd6..fb9396d 100644 --- a/car/app.py +++ b/car/app.py @@ -7,7 +7,7 @@ import secrets import time from collections.abc import Callable -from typing import Any, cast +from typing import Any, TypedDict, cast # 3p from flask import ( @@ -20,7 +20,7 @@ redirect, render_template, request, - session, + session, # type: ignore # we ignore this so we can better type it later url_for, ) from typing_extensions import TypeIs @@ -44,7 +44,35 @@ ) PHONEBANK_MIN_DELAY = 60 * 15 -SETTINGS_KEYS = { + + +class Session(TypedDict, total=False): + # settings + use_map: int + autolink: int + zoom_phone: int + dark_mode: int + + # state + admin: bool + phonebank: bool + return_to: str + canvasser: str + phonebank_turf: ID + last_turf: ID + last_door: ID | None + last_commit: float + turfs: list[ID] + chosen_voter: ID + previous_voter: ID | None + + phone_code: str + phone_paired: bool + + +session: Session + +SETTINGS_KEYS: Session = { "use_map": 1, "autolink": 1, "zoom_phone": 0, @@ -54,7 +82,16 @@ app = Flask(__name__) cache = utils.MemoryCache() -phones = {} + + +class Phone(TypedDict, total=False): + zoom_phone: int + data: str + voter: ID | None + seen: float + + +phones: dict[str, Phone] = {} os.chdir(DATA_ROOT) @@ -128,39 +165,42 @@ def after_request(resp): return resp -def ensure_turf_accessible(turf_id) -> TypeIs[int]: +def ensure_turf_accessible(turf_id) -> TypeIs[ID]: + if not isinstance(turf_id, int): + abort(400) + if "canvasser" not in session: abort(403) - if session["admin"]: + if session.get("admin"): return True - if turf_id in session["turfs"]: + if turf_id in session.get("turfs", []): return True abort(403) def restrict_admin(): - if session["admin"]: + if session.get("admin"): return abort(403) -def ensure_voter_accessible(voter): - if session["admin"]: +def ensure_voter_accessible(voter: Voter): + if session.get("admin"): return if g.phonebank: - if voter.id == session["chosen_voter"]: + if voter.id == session.get("chosen_voter"): return # also check householding info for phonebanking # (if the last voter we worked on is in the household, then it's ok) for voter_set in householding.household_info_by_phones(voter).values(): - if session["chosen_voter"] in [v.id for v in voter_set]: + if session.get("chosen_voter") in [v.id for v in voter_set]: return # check for householding by door @@ -174,8 +214,8 @@ def ensure_voter_accessible(voter): abort(403) -def ensure_door_accessible(door): - if session["admin"]: +def ensure_door_accessible(door: Door): + if session.get("admin"): return turf_id = session.get("last_turf") @@ -186,7 +226,7 @@ def ensure_door_accessible(door): def to_last_turf(): - return redirect(url_for("show_turf", id=session["last_turf"])) + return redirect(url_for("show_turf", id=session.get("last_turf"))) @app.route("/login/", methods=["GET", "POST"]) @@ -253,14 +293,14 @@ def login(login_code=None): flash("Turf already on list.") if "canvasser" not in session: - session["canvasser"] = request.form.get("canvasser") + session["canvasser"] = request.form["canvasser"] return redirect(session.pop("return_to", "/")) @app.route("/logout/") def logout(): - session.clear() + cast(dict, session).clear() flash("Logged out!") return redirect(url_for("login")) @@ -312,17 +352,17 @@ def is_phone(k: str): @app.route("/") def index(): - if not session["admin"] and len(session["turfs"]) == 1: - return redirect(url_for("show_turf", id=session["turfs"][0])) + if not session.get("admin") and len(session.get("turfs", [])) == 1: + return redirect(url_for("show_turf", id=session.get("turfs", [])[0])) - if session["admin"]: + if session.get("admin"): my_geoturfs = geoturfs else: my_geoturfs = geoturfs.copy() my_geoturfs["features"] = [ x for x in my_geoturfs["features"] - if x["properties"]["car_id"] in session["turfs"] + if x["properties"]["car_id"] in session.get("turfs", []) ] return render_template( @@ -470,8 +510,9 @@ def finish_turf(id): def show_door(id: ID): door = db.get_door_by_id(id) ensure_door_accessible(door) - - turf = db.get_turf_by_id(session.get("last_turf")) + last_turf = session.get("last_turf") + assert ensure_turf_accessible(last_turf) + turf = db.get_turf_by_id(last_turf) voters = [db.get_voter_by_id(voter_id) for voter_id in door.voters] # filter out "New Voter" @@ -575,7 +616,8 @@ def show_voter(id: ID): voter = db.get_voter_by_id(id) if g.phonebank and session.get("phone_paired"): if not request.headers.get("HX-Preloaded"): - phones[session["phone_code"]]["voter"] = id + # TODO do we wanna add more validation around this? + phones[session.get("phone_code", "")]["voter"] = id ensure_voter_accessible(voter) @@ -792,7 +834,7 @@ def phonebank_next_voter(turf_id): return redirect(url_for("show_voter", id=voter_id, keep_previous=1)) flash("No voters to contact in this phonebank.") - if not session["admin"] and len(session["turfs"]) == 1: + if not session.get("admin") and len(session.get("turfs", [])) == 1: session["turfs"] = [] return redirect(url_for("login")) @@ -812,7 +854,7 @@ def phonebank_previous_voter(): def pair_phone_generate_code(): session["return_to"] = request.args.get("return", "/") phone_code = session["phone_code"] = secrets.token_hex() - phones[phone_code] = {"zoom_phone": session["zoom_phone"]} + phones[phone_code] = {"zoom_phone": session.get("zoom_phone", 0)} qr_code = utils.qr_code(f"{BASE_URL}/pair_phone/{phone_code}/") return render_template("phone_pair.html", code=qr_code) @@ -836,14 +878,16 @@ def pair_phone(code): ) return redirect(request.args.get("return", "/")) - session["zoom_phone"] = phones[code]["zoom_phone"] + session["zoom_phone"] = phones[code].get("zoom_phone", 0) phones[code].update({"seen": time.time(), "data": "Phone paired!", "voter": None}) return render_template("phone_paired.html", code=code) @app.route("/pair_phone//voter/") def phone_voter_html(code): - voter = db.get_voter_by_id(phones[code]["voter"]) + if (voter_id := phones[code].get("voter")) is None: + abort(404) + voter = db.get_voter_by_id(voter_id) return render_template("phone_voter.html", voter=voter) From 3985ba04167d302e2e5f9bbf4f0cd484112b9598 Mon Sep 17 00:00:00 2001 From: Ava Silver Date: Sun, 2 Aug 2026 17:07:30 -0400 Subject: [PATCH 2/2] fix typing --- car/script/export_geocoded_voters.py | 2 +- car/script/update_voter_turfs.py | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/car/script/export_geocoded_voters.py b/car/script/export_geocoded_voters.py index 5c59d49..14f25c4 100644 --- a/car/script/export_geocoded_voters.py +++ b/car/script/export_geocoded_voters.py @@ -9,7 +9,7 @@ include_voters = set() group_id = os.getenv("TURF_GROUP") -for group in database.group: +for group in database.groups: if group.external_id == group_id: include_voters.update(group.voters) diff --git a/car/script/update_voter_turfs.py b/car/script/update_voter_turfs.py index ca6ba28..07f32c5 100644 --- a/car/script/update_voter_turfs.py +++ b/car/script/update_voter_turfs.py @@ -8,7 +8,7 @@ from ..model import ID, Database, Turf, has_geocode -TURF_DATA_PATH = os.getenv("TURF_DATA_PATH") +TURF_DATA_PATH = os.getenv("TURF_DATA_PATH", "") TURF_GROUP_ID = os.getenv("TURF_GROUP") database = Database.get() @@ -32,6 +32,7 @@ def sync_turf_props(): cur.close() turf_group = get_turf_group() + assert turf_group active_turf_ids = { turf.id for turf in database.turfs if turf.group_id == turf_group.id }