backend/accounts.py (3960 bytes)
1 """Who an account is, and keeping one email to exactly one account. 2 3 Every visitor starts anonymous, identified by a cookie. An email gets attached 4 in one of two ways - following a sign-in link, or paying (Stripe collects one at 5 checkout) - and from then on that email is the account. When the email already 6 belongs to another row, the anonymous row is folded into it, so whatever someone 7 converted before signing in is still there afterwards. 8 """ 9 10 from __future__ import annotations 11 12 import re 13 import secrets 14 15 from sqlalchemy.orm import Session 16 17 from .db import Account, Job, Purchase 18 19 # Deliberately loose: the real test of an address is whether the link arrives. 20 _EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") 21 22 23 class InvalidEmail(ValueError): 24 pass 25 26 27 def normalize_email(raw: str | None) -> str: 28 email = (raw or "").strip().lower() 29 if not email or len(email) > 320 or not _EMAIL.match(email): 30 raise InvalidEmail("That does not look like an email address.") 31 return email 32 33 34 def new_account(session: Session) -> Account: 35 account = Account(device_token=secrets.token_urlsafe(24)) 36 session.add(account) 37 session.flush() 38 return account 39 40 41 def by_email(session: Session, email: str) -> Account | None: 42 return session.query(Account).filter(Account.email == email).first() 43 44 45 def resolve(session: Session, account: Account) -> Account: 46 """Follow merged_into to the account that survived.""" 47 seen = {account.id} 48 while account.merged_into: 49 target = session.get(Account, account.merged_into) 50 if target is None or target.id in seen: 51 break 52 seen.add(target.id) 53 account = target 54 return account 55 56 57 def merge(session: Session, src: Account, dst: Account) -> None: 58 """Fold `src` into `dst`: its books, its purchases and anything it paid for.""" 59 if src.id == dst.id: 60 return 61 session.query(Job).filter(Job.account_id == src.id).update( 62 {Job.account_id: dst.id}, synchronize_session=False 63 ) 64 session.query(Purchase).filter(Purchase.account_id == src.id).update( 65 {Purchase.account_id: dst.id}, synchronize_session=False 66 ) 67 from . import captions, subrep # they import api, which imports this module 68 69 captions.merge(session, src, dst) 70 subrep.merge(session, src, dst) 71 dst.purchased_credits += src.purchased_credits 72 src.purchased_credits = 0 73 # Add the free credits used, not the free credits left: a second device 74 # must not bring a second set of free credits into one account. 75 dst.free_credits_used += src.free_credits_used 76 src.free_credits_used = 0 77 if src.stripe_customer_id and not dst.stripe_customer_id: 78 dst.stripe_customer_id = src.stripe_customer_id 79 if src.subscription_id and not dst.subscription_id: 80 dst.subscription_id = src.subscription_id 81 dst.subscription_status = src.subscription_status 82 dst.subscription_period_end = src.subscription_period_end 83 src.stripe_customer_id = None 84 src.subscription_id = src.subscription_status = None 85 src.subscription_period_end = None 86 src.merged_into = dst.id 87 88 89 def adopt_email(session: Session, account: Account, email: str) -> Account: 90 """Attach `email` to `account`, returning whichever account ends up owning it. 91 92 The caller must point the browser at the returned account - it is not 93 always the one passed in. 94 """ 95 account = resolve(session, account) 96 if account.email == email: 97 return account 98 99 owner = by_email(session, email) 100 101 if account.email is None: 102 if owner is None: 103 account.email = email 104 return account 105 # Known email, anonymous device: bring the device's work along. 106 merge(session, account, owner) 107 return owner 108 109 # Signed in as someone else already. Switch users; never merge two people. 110 if owner is None: 111 owner = new_account(session) 112 owner.email = email 113 return owner