add some named tuples for db models

src/atproto2/atproto_oauth.py src/atproto/atproto_oauth.py

···
 from authlib.oauth2.rfc7636 import create_s256_code_challenge
 from requests import Response
 
+from ..types import OAuthAuthRequest, OAuthSession
+
 from .atproto_security import is_safe_url, hardened_http
 
 
···
 # Completes the auth flow by sending an initial auth token request.
 # Returns token response (dict) and DPoP nonce (str)
 def initial_token_request(
-    auth_request: dict[str, str],
+    auth_request: OAuthAuthRequest,
     code: str,
     app_url: str,
     client_secret_jwk: Key,
 ) -> tuple[dict[str, str], str]:
-    authserver_url = auth_request["authserver_iss"]
+    authserver_url = auth_request.authserver_iss
 
     # Re-fetch server metadata
     authserver_meta = fetch_authserver_meta(authserver_url)
···
         "redirect_uri": redirect_uri,
         "grant_type": "authorization_code",
         "code": code,
-        "code_verifier": auth_request["pkce_verifier"],
+        "code_verifier": auth_request.pkce_verifier,
         "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
         "client_assertion": client_assertion,
     }
 
     # Create DPoP header JWT, using the existing DPoP signing key for this account/session
     token_url = authserver_meta["token_endpoint"]
-    dpop_private_jwk = JsonWebKey.import_key(
-        json.loads(auth_request["dpop_private_jwk"])
-    )
-    dpop_authserver_nonce = auth_request["dpop_authserver_nonce"]
+    dpop_private_jwk = JsonWebKey.import_key(json.loads(auth_request.dpop_private_jwk))
+    dpop_authserver_nonce = auth_request.dpop_authserver_nonce
     dpop_proof = authserver_dpop_jwt(
         "POST", token_url, dpop_authserver_nonce, dpop_private_jwk
     )
···
 
 # Returns token response (dict) and DPoP nonce (str)
 def refresh_token_request(
-    user: dict,
+    user: OAuthSession,
     app_url: str,
     client_secret_jwk: Key,
 ) -> tuple[dict[str, str], str]:
-    authserver_url = user["authserver_iss"]
+    authserver_url = user.authserver_iss
 
     # Re-fetch server metadata
     authserver_meta = fetch_authserver_meta(authserver_url)
···
     params = {
         "client_id": client_id,
         "grant_type": "refresh_token",
-        "refresh_token": user["refresh_token"],
+        "refresh_token": user.refresh_token,
         "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
         "client_assertion": client_assertion,
     }
 
     # Create DPoP header JWT, using the existing DPoP signing key for this account/session
     token_url = authserver_meta["token_endpoint"]
-    dpop_private_jwk = JsonWebKey.import_key(json.loads(user["dpop_private_jwk"]))
-    dpop_authserver_nonce = user["dpop_authserver_nonce"]
+    dpop_private_jwk = JsonWebKey.import_key(json.loads(user.dpop_private_jwk))
+    dpop_authserver_nonce = user.dpop_authserver_nonce
     dpop_proof = authserver_dpop_jwt(
         "POST", token_url, dpop_authserver_nonce, dpop_private_jwk
     )
···
 def pds_dpop_jwt(
     method: str,
     url: str,
-    access_token: str,
-    nonce: str,
+    access_token: str | None,
+    nonce: str | None,
     dpop_private_jwk: Key,
 ) -> str:
     dpop_pub_jwk = json.loads(dpop_private_jwk.as_json(is_private=False))
···
 def pds_authed_req(
     method: str,
     url: str,
-    user: dict[str, str],
+    user: OAuthSession,
     db: sqlite3.Connection,
     body: dict[str, Any] | None = None,
 ) -> Response | None:
-    dpop_private_jwk = JsonWebKey.import_key(json.loads(user["dpop_private_jwk"]))
-    dpop_pds_nonce = user["dpop_pds_nonce"]
-    access_token = user["access_token"]
+    dpop_private_jwk = JsonWebKey.import_key(json.loads(user.dpop_private_jwk))
+    dpop_pds_nonce = user.dpop_pds_nonce
+    access_token = user.access_token
 
     response: Response | None = None
 
···
             cur = db.cursor()
             _ = cur.execute(
                 "UPDATE oauth_session SET dpop_pds_nonce = ? WHERE did = ?;",
-                [dpop_pds_nonce, user["did"]],
+                [dpop_pds_nonce, user.did],
             )
             db.commit()
             cur.close()

src/db.py

···
-import sqlite3
-
 from flask import Flask, g
+
+import sqlite3
 
 
 def get_db(app: Flask) -> sqlite3.Connection:

src/main.py

···
 from typing import Any
 import json
 
-from .atproto2 import PdsUrl, get_record, resolve_did_from_handle, resolve_pds_from_did
-from .atproto2.atproto_oauth import pds_authed_req
+from .atproto import PdsUrl, get_record, resolve_did_from_handle, resolve_pds_from_did
+from .atproto.atproto_oauth import pds_authed_req
 from .db import close_db_connection, get_db, init_db
 from .oauth import oauth
+from .types import OAuthSession
 
 app = Flask(__name__)
 _ = app.config.from_prefixed_env()
···
 
 @app.before_request
 def load_user_to_context():
+    user: OAuthSession | None = None
     did: str | None = session.get("user_did")
-    if did is None:
-        g.user = None
-    else:
+    if did is not None:
         db = get_db(app)
-        g.user = db.execute(
+        row = db.execute(
             "select * from oauth_session where did = ?",
             (did,),
         ).fetchone()
+        user = OAuthSession(**row)
+    g.user = user
 
 
-def get_user() -> dict[str, str] | None:
+def get_user() -> OAuthSession | None:
     return g.user
 
 
···
     if user is None:
         return redirect("/login")
 
-    did: str = user["did"]
-    pds: str = user["pds_url"]
-    handle: str | None = user["handle"]
+    did: str = user.did
+    pds: str = user.pds_url
+    handle: str | None = user.handle
 
     profile, from_bluesky = load_profile(pds, did, reload=True)
     links = load_links(pds, did, reload=True) or [{"background": "#fa0"}]
···
 
     put_record(
         user=user,
-        pds=user["pds_url"],
-        repo=user["did"],
+        pds=user.pds_url,
+        repo=user.did,
         collection=f"{SCHEMA}.actor.profile",
         rkey="self",
         record={
···
 
     put_record(
         user=user,
-        pds=user["pds_url"],
-        repo=user["did"],
+        pds=user.pds_url,
+        repo=user.did,
         collection=f"{SCHEMA}.actor.links",
         rkey="self",
         record={
···
 
 
 def put_record(
-    user: dict[str, str],
+    user: OAuthSession,
     pds: PdsUrl,
     repo: str,
     collection: str,
···
     if user is not None:
         db = get_db(app)
         cursor = db.cursor()
-        _ = cursor.execute("delete from oauth_session where did = ?", (user["did"],))
+        _ = cursor.execute("delete from oauth_session where did = ?", (user.did,))
         db.commit()
         cursor.close()
     session.clear()

src/oauth.py

···
 
 import json
 
-from .atproto2.atproto_identity import is_valid_did, is_valid_handle
-from .atproto2.atproto_oauth import initial_token_request, send_par_auth_request
-from .atproto2.atproto_security import is_safe_url
-from .atproto2 import (
+from .atproto.atproto_identity import is_valid_did, is_valid_handle
+from .atproto.atproto_oauth import initial_token_request, send_par_auth_request
+from .atproto.atproto_security import is_safe_url
+from .atproto import (
     pds_endpoint_from_doc,
     resolve_authserver_from_pds,
     resolve_authserver_meta,
     resolve_identity,
 )
+from .types import OAuthAuthRequest
 from .db import get_db
 
 oauth = Blueprint("oauth", __name__, url_prefix="/oauth")
-
-
-oauth_auth_requests: dict[str, dict[str, str]] = {}
 
 
 @oauth.get("/start")
···
 
     par_request_uri: str = resp.json()["request_uri"]
     current_app.logger.debug(f"saving oauth_auth_request to DB  state={state}")
-    oauth_auth_requests[state] = {
-        "authserver_iss": authserver_meta["issuer"],
-        "did": did or "",  # TODO: use actual typing
-        "handle": handle or "",
-        "pds_url": pds_url or "",
-        "pkce_verifier": pkce_verifier,
-        "scope": scope,
-        "dpop_authserver_nonce": dpop_authserver_nonce,
-        "dpop_private_jwk": dpop_private_jwk.as_json(is_private=True),
-    }
+
+    db = get_db(current_app)
+    cursor = db.cursor()
+    _ = cursor.execute(
+        "insert or replace into oauth_auth_requests values (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (
+            state,
+            authserver_meta["issuer"],
+            did,
+            handle,
+            pds_url,
+            pkce_verifier,
+            scope,
+            dpop_authserver_nonce,
+            dpop_private_jwk.as_json(is_private=True),
+        ),
+    )
+    db.commit()
+    cursor.close()
 
     auth_endpoint = authserver_meta["authorization_endpoint"]
     assert is_safe_url(auth_endpoint)
···
     authserver_iss = request.args["iss"]
     authorization_code = request.args["code"]
 
-    auth_request = oauth_auth_requests.get(state)
-    if auth_request is None:
+    db = get_db(current_app)
+    cursor = db.cursor()
+
+    row = cursor.execute(
+        "select * from oauth_auth_requests where state = ?", (state,)
+    ).fetchone()
+    try:
+        auth_request = OAuthAuthRequest(**row)
+    except TypeError:
         return redirect(url_for("page_login"), 303)
 
     current_app.logger.debug(f"Deleting auth request for state={state}")
-    _ = oauth_auth_requests.pop(state)
+    _ = cursor.execute("delete from oauth_auth_requests where state = ?", (state,))
+    db.commit()
 
-    assert auth_request["authserver_iss"] == authserver_iss
-    # assert state ????
+    assert auth_request.authserver_iss == authserver_iss
+    assert auth_request.state == state
 
     app_url = request.url_root.replace("http://", "https://")
     CLIENT_SECRET_JWK = JsonWebKey.import_key(current_app.config["CLIENT_SECRET_JWK"])
···
 
     row = auth_request
 
-    did = auth_request["did"]
-    if row["did"]:
+    did = auth_request.did
+    if row.did:
         # If we started with an account identifier, this is simple
-        did, handle, pds_url = row["did"], row["handle"], row["pds_url"]
+        did, handle, pds_url = row.did, row.handle, row.pds_url
         assert tokens["sub"] == did
     else:
         did = tokens["sub"]
···
         authserver_url = resolve_authserver_from_pds(pds_url)
         assert authserver_url == authserver_iss
 
-    assert row["scope"] == tokens["scope"]
+    assert row.scope == tokens["scope"]
 
     current_app.logger.debug("storing user did and handle")
     db = get_db(current_app)
···
             tokens["refresh_token"],
             dpop_authserver_nonce,
             None,
-            auth_request["dpop_private_jwk"],
+            auth_request.dpop_private_jwk,
         ),
     )
     db.commit()
     cursor.close()
 
     session["user_did"] = did
-    session["user_handle"] = auth_request["handle"]
+    session["user_handle"] = auth_request.handle
 
     return redirect(url_for("page_login"))
 

src/schema.sql

···
-create table if not exists oauth_session (
+create table if not exists oauth_auth_requests (
+    state text not null primary key,
+    authserver_iss text not null,
+    did text,
+    handle text,
+    pds_url text,
+    pkce_verifier text not null,
+    scope text not null,
+    dpop_authserver_nonce text not null,
+    dpop_private_jwk text not null
+) strict, without rowid;
+
+create table if not exists oauth_sessions (
     did text not null primary key,
     handle text,
     pds_url text not null,

src/types.py

···
+from typing import NamedTuple
+
+
+class OAuthAuthRequest(NamedTuple):
+    state: str
+    authserver_iss: str
+    did: str | None
+    handle: str | None
+    pds_url: str | None
+    pkce_verifier: str
+    scope: str
+    dpop_authserver_nonce: str
+    dpop_private_jwk: str
+
+
+class OAuthSession(NamedTuple):
+    did: str
+    handle: str | None
+    pds_url: str
+    authserver_iss: str
+    access_token: str | None
+    refresh_token: str | None
+    dpop_authserver_nonce: str
+    dpop_pds_nonce: str | None
+    dpop_private_jwk: str