app/main.py
Rama main · réplica de lectura del repositorio original.
"""Portfolio tracker entrypoint.
Run locally:
APP_CONFIG_FILE=./config.yaml APP_DATABASE__PATH=./stocktake.db \
uvicorn app.main:app --reload
Access control has two independent layers, on purpose:
* `require_login` (middleware) — nothing but the probes, the static files and
the login/setup pages is reachable logged out.
* `tenancy` — every query for personal data is filtered to the session's user,
so even a route that forgets to filter cannot serve another portfolio.
"""
from __future__ import annotations
import asyncio
import datetime as dt
import json
import logging
import math
import re
import threading
import urllib.parse
from contextlib import asynccontextmanager, contextmanager
from decimal import Decimal
from http import HTTPStatus
from pathlib import Path
from typing import Annotated
from urllib.parse import quote_plus
from zoneinfo import ZoneInfo
from fastapi import Form, HTTPException, Request
from fastapi.encoders import jsonable_encoder
from fastapi.exception_handlers import http_exception_handler
from fastapi.exceptions import RequestValidationError
from fastapi.responses import (
HTMLResponse,
JSONResponse,
PlainTextResponse,
RedirectResponse,
Response,
)
from pydantic import BaseModel, Field
from sqlalchemy import func, select, text, update
from sqlalchemy.orm import Session as DbSession
from sqlalchemy.orm import object_session
from starlette.exceptions import HTTPException as StarletteHTTPException
from appcore import (
create_app,
ensure_utc,
load_config,
oidc,
)
from appcore.config import DatabaseSettings
from . import (
auth,
branding,
brokerage,
brokerdesign,
calendarview,
chart_templates,
charts_build,
clock,
columns,
configfile,
cpi,
exports,
features,
federation,
fields,
fyreport,
invites,
lifecycle,
logbuffer,
maintenance,
markettime,
memory,
metrics,
money,
moves,
navigation,
passkeys,
plans,
pricefeed,
providers,
queries,
setupwizard,
simulate,
sorting,
statements,
submissions,
tenancy,
textfield,
theming,
twofactor,
wording,
)
from . import (
db as database,
)
from .api import router as api_router
from .imports_web import router as imports_router
from .models import (
ACTION_KINDS,
ASSET_CLASSES,
MARKET_OPEN,
MEMBER_ROLES,
NAV_STYLES,
ROLE_BLURBS,
ROLE_LABELS,
ROW_ID_MAX,
THEMES,
TIME_ZONES_SHOWN,
WRITE_ROLES,
ApiKey,
CorporateAction,
Dividend,
ExternalIdentity,
HoldingPref,
Instrument,
InvestmentPlan,
InvestmentPlanEntry,
PlannedPurchase,
Portfolio,
PortfolioInvite,
PortfolioMember,
Price,
RowId,
SavedChart,
Trade,
User,
UserSession,
currency_problem,
exchange_problem,
row_id,
ticker_problem,
)
from .settings import PortfolioSettings
log = logging.getLogger(__name__)
APP_DIR = Path(__file__).parent
# Where the published documentation lives. ONE constant rather than a URL typed
# into each template that links out, because a search-and-replace across
# templates is how one gets missed.
DOCS_URL = "https://jarrodruggiero.github.io/stocktake"
settings = load_config(PortfolioSettings)
# Before anything can ask what day it is: a container runs in UTC, and every
# date decision in this app belongs to the portfolio's own zone.
clock.configure(settings.timezone)
# Root-logger setup so app INFO lines (price feed runs, warnings) reach the pod
# logs — uvicorn only configures its own loggers, leaving ours invisible.
logging.basicConfig(
level=str(settings.log_level).upper(),
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
# Keep the last few hundred WARNING-and-above lines in memory for the Settings
# page's console. Installed here rather than in the lifespan so it is
# collecting before the first request — startup is exactly when the
# interesting failures happen.
logbuffer.install()
# Where the database comes from, and whether there is one at all. Connecting is
# deliberately conditional so a fresh install can serve the wizard —
# decisions.md #41.
database.set_migrations(str(APP_DIR / "migrations"))
database.start(settings)
def SessionLocal():
"""A database session.
A function rather than the sessionmaker itself, because the factory behind
it is built at connect time and can be replaced by the wizard. Every call
site keeps working unchanged; the tests still repoint this name directly.
"""
return database.session()
# Reachable without a session: probes, the login/setup flow, and the assets
# those pages need. `/login/code`, `/session/status` and `/metrics` are public
# to the MIDDLEWARE only, and each guards itself — decisions.md #29.
PUBLIC_PATHS = {"/healthz", "/readyz", "/login", "/login/code", "/login/recover",
# Signing in with a passkey happens before there is a session,
# so both halves of the ceremony are public. Each verifies the
# pre-auth CSRF token, and the second only trusts a signature.
"/login/passkey", "/login/passkey/options",
# The provider redirects the browser back here before there is
# a session; `state` is what makes the callback trustworthy.
"/login/oidc", "/login/oidc/callback",
# The provider POSTs here with no cookie and no CSRF token.
# The signature on the `logout_token` is the authentication —
# see `oidc_backchannel_logout`, and decisions.md #127.
"/oidc/backchannel-logout",
"/setup", "/session/status", "/metrics"}
# Two prefixes the login middleware waves through, each of which guards
# itself. **Both failure modes are silent publication of an endpoint:**
# * a route under /api/ that does not open with `auth.api_session(...)`
# * a route under /setup/ that does not open with `_wizard_step()`
# `/invite/` is public because the person opening one has no account yet. The
# token in the URL is the only authorisation, and every route under it checks
# it — decisions.md #116.
PUBLIC_PREFIXES = ("/static/", "/api/", "/setup/", "/invite/")
# The subset that must answer with NO database at all. Everything else — /login
# included — needs one to say anything useful, and gets sent to the wizard
# instead of a 500 from its first query.
NO_DATABASE_PATHS = {"/healthz", "/readyz", "/setup", "/metrics"}
NO_DATABASE_PREFIXES = ("/static/", "/setup/")
def _db_ready() -> bool:
"""Readiness, which is not the same question as "is there a database".
An unconfigured install is *ready*: it can serve the wizard, which is the
only thing there is to serve and the only way to make it into anything
more. Reporting not-ready would be a deadlock — Kubernetes would keep
traffic away from the page that fixes it.
"""
if not database.is_ready():
return True
try:
with SessionLocal() as s:
s.execute(text("SELECT 1"))
return True
except Exception:
return False
# --------------------------------------------------------------------------- #
# Price feed
# --------------------------------------------------------------------------- #
# Last feed outcome, shown on the dashboard so a stalled feed is visible
# instead of silently serving stale prices. In-memory is fine: one process.
feed_status: dict = {"running": False, "finished": None, "ok": None, "error": None,
"counts": None, "degraded": {}, "quoted": None, "quotes": 0}
_feed_lock = threading.Lock()
def _run_feed() -> bool:
if not _feed_lock.acquire(blocking=False):
return True # already in flight (manual refresh raced the schedule)
feed_status["running"] = True
try:
# Market data is shared and the FX repair pass spans every user's
# trades, so this is the one place tenancy is deliberately bypassed.
with tenancy.unscoped_session(SessionLocal) as s:
counts = pricefeed.run_feed(s, settings)
# Only the symbols a fallback served, or that nobody could serve. The
# normal case is silence; a list here means Yahoo let us down for
# something, which is exactly what goes unnoticed otherwise.
feed_status.update(
ok=True,
error=None,
counts=counts,
degraded={
key: summary
for key, summary in pricefeed.last_run_sources.items()
if "failed" in summary or "no rows" in summary
},
)
return True
except Exception as exc:
log.exception("price feed run failed")
feed_status.update(ok=False, error=f"{type(exc).__name__}: {exc}", counts=None)
return False
finally:
# In `finally` so a run is counted whichever branch returned. Reads the
# outcome back off `feed_status` rather than tracking a second flag,
# because both branches have just written it and two sources of the
# same truth is how they drift.
metrics.record_feed_run(ok=bool(feed_status["ok"]))
feed_status.update(
running=False,
finished=dt.datetime.now(ZoneInfo(settings.price_feed.timezone)),
)
_feed_lock.release()
_quote_lock = threading.Lock()
def _run_quotes() -> int:
"""One batched live-price refresh. Never raises: a quote is a nicety."""
if not _quote_lock.acquire(blocking=False):
return 0
try:
with tenancy.unscoped_session(SessionLocal) as s:
written = pricefeed.refresh_quotes(s, settings)
if written:
# The panel says "Last refreshed <time>", and before this it only
# knew about the daily close run — so a dashboard whose prices were
# five minutes old reported them as an hour old, and the one place
# somebody checks that question answered it wrongly. A mechanism
# nobody can see is indistinguishable from one that is not running.
feed_status.update(
quoted=dt.datetime.now(ZoneInfo(settings.price_feed.timezone)),
quotes=written,
)
return written
except Exception:
# A failed quote does NOT set `ok=False`: that flag drives the alert on
# the thing which keeps history correct, and a missing live price makes
# nothing wrong — the page falls back to the last close. Raising the
# alarm here would train exactly the habit this design removed.
log.warning("quote refresh failed", exc_info=True)
return 0
finally:
_quote_lock.release()
def _live_exchanges(open_positions: list) -> list[str]:
"""Which exchanges are trading right now, among things actually HELD.
Built from the positions the page is showing, not the instrument table —
decisions.md #16. Held crypto counts; unheld crypto cannot speak for it.
"""
return sorted({
h.instrument.exchange
for h in open_positions
if h.instrument.yahoo_symbol and pricefeed.is_trading(h.instrument.exchange)
})
def _anyone_signed_in(db: DbSession) -> bool:
"""Whether ANY account currently has a live session.
"Anyone", not "this person": one user signing out while another is still
signed in must not stop the polling. Logging out deletes the row.
Timeouts are recomputed the way `auth` does rather than trusting
`expires_at`, which carries only the sliding cap. Compared in Python via
`ensure_utc` because SQLite returns naive datetimes.
"""
now = dt.datetime.now(dt.timezone.utc)
idle = dt.timedelta(minutes=settings.auth.session_idle_minutes)
absolute = dt.timedelta(days=settings.auth.session_absolute_days)
for last_seen, created, awaiting in db.execute(
select(UserSession.last_seen_at, UserSession.created_at,
UserSession.awaiting_totp)
).all():
if awaiting: # half-way through sign-in is not signed in
continue
if ensure_utc(last_seen) + idle > now and ensure_utc(created) + absolute > now:
return True
return False
def _should_poll_quotes() -> bool:
"""Both conditions, in one session: somebody here AND a market moving."""
try:
with tenancy.unscoped_session(SessionLocal) as s:
return _anyone_signed_in(s) and pricefeed.any_market_trading(s)
except Exception:
log.warning("could not decide whether to poll quotes", exc_info=True)
return False
async def _quote_loop() -> None:
"""Live prices while somebody is signed in and something is trading.
Nobody there means nobody can see the answer; every market shut means it
cannot have changed. Either way the app goes quiet until the evening run.
"""
loop = asyncio.get_running_loop()
await _await_database("live quotes")
while True:
try:
if await loop.run_in_executor(None, _should_poll_quotes):
await loop.run_in_executor(None, _run_quotes)
except Exception:
log.exception("quote loop iteration failed")
# Read each time round: the settings page applies it without a restart.
await asyncio.sleep(max(1, settings.price_feed.quote_interval_minutes) * 60)
def refresh_quotes_soon() -> None:
"""Kick off a live refresh without waiting for it.
Called at sign-in: the first thing somebody does is look at what their
holdings are worth, and the poll below would otherwise leave them on the
last close for up to a whole interval. Only with the feed on as well as
its quotes, because `price_feed.enabled` governs every outbound request
(decisions.md #9), and only while a market is open, as the poll asks.
"""
if not (settings.price_feed.enabled and settings.price_feed.quotes_enabled):
return
try:
asyncio.get_running_loop().run_in_executor(None, _quotes_if_trading)
except RuntimeError: # no loop (CLI/tests) — the poll will catch it
pass
def _quotes_if_trading() -> int:
"""A live refresh, unless every market is shut: then the price is the
close the daily run already has, and nothing can have moved."""
try:
with tenancy.unscoped_session(SessionLocal) as s:
trading = pricefeed.any_market_trading(s)
except Exception:
log.warning("could not tell whether a market is open", exc_info=True)
return 0
return _run_quotes() if trading else 0
async def _await_database(what: str) -> None:
"""Block until the wizard has chosen a database.
Both background loops start at process startup, and on a fresh install
there is nothing for them to talk to yet — the wizard has not run. Without
this they fire immediately and fill the very first page of the log with
tracebacks, which is a poor first impression and buries the one line that
matters ("Uvicorn running on…").
Polling rather than an event because the connection can also be made from
outside this loop's world, and a poll cannot miss an edge.
"""
if database.is_ready():
return
log.info("%s waiting: no database configured yet (finish the setup wizard)", what)
while not database.is_ready():
await asyncio.sleep(2)
log.info("%s starting: database is now configured", what)
async def _feed_loop() -> None:
loop = asyncio.get_running_loop()
await _await_database("price feed")
# Catch-up on boot (first deploy backfills to 2020), then daily at hh:mm.
ok = await loop.run_in_executor(None, _run_feed)
# The schedule is read on every pass: the settings page applies it
# without a restart.
while True:
now = dt.datetime.now(ZoneInfo(settings.price_feed.timezone))
target = now.replace(
hour=settings.price_feed.hour, minute=settings.price_feed.minute,
second=0, microsecond=0,
)
if target <= now:
target += dt.timedelta(days=1)
if not ok:
# A failed run retries every 30 min rather than leaving prices
# stale until tomorrow; the daily fire time still wins if sooner.
target = min(target, now + dt.timedelta(minutes=30))
await asyncio.sleep((target - now).total_seconds())
ok = await loop.run_in_executor(None, _run_feed)
def _next_run(now: dt.datetime, hour: int, minute: int) -> dt.datetime:
"""The next occurrence of hh:mm, today or tomorrow."""
target = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
return target if target > now else target + dt.timedelta(days=1)
async def _maintenance_loop() -> None:
"""Housekeeping, once at startup and daily thereafter.
Its own loop rather than a step inside the feed's: the two answer to
different things, and turning the feed off should not silently stop
expiring sessions.
"""
loop = asyncio.get_running_loop()
await _await_database("maintenance")
# The zone is read on every pass, like the feed's.
while True:
try:
await loop.run_in_executor(
None, maintenance.run, SessionLocal, settings)
except Exception:
log.exception("maintenance run failed")
now = dt.datetime.now(ZoneInfo(settings.price_feed.timezone or settings.timezone))
target = _next_run(now, settings.maintenance.hour, settings.maintenance.minute)
await asyncio.sleep((target - now).total_seconds())
@asynccontextmanager
async def lifespan(_app):
# One line at startup saying what this pod is actually holding. Without it,
# "why does an app serving six pages want 190 MiB" needs a profiler; with
# it, the answer is in the first line of the log.
memory.log_snapshot("at startup")
tasks = []
if settings.price_feed.enabled:
tasks.append(asyncio.create_task(_feed_loop()))
if settings.price_feed.enabled and settings.price_feed.quotes_enabled:
tasks.append(asyncio.create_task(_quote_loop()))
if settings.maintenance.enabled:
tasks.append(asyncio.create_task(_maintenance_loop()))
yield
for task in tasks:
task.cancel()
app = create_app(
settings,
templates_dir=str(APP_DIR / "templates"),
static_dir=str(APP_DIR / "static"),
readiness_check=_db_ready,
lifespan=lifespan,
)
app.state.session_factory = SessionLocal
templates = app.state.templates
# --------------------------------------------------------------------------- #
# Template filters/globals
# --------------------------------------------------------------------------- #
def _money_plain(v) -> str:
return money.plain(v)
def _money(v, currency: str | None = None):
"""The digits, no symbol — for table cells, whose column header names the
currency. See `app/money.py` for why the symbol is not repeated per cell.
Wrapped so dollar values can be blurred by the privacy toggle ("glass
brick"). Use money_plain inside HTML attributes (placeholders etc.).
"""
from markupsafe import Markup, escape
return Markup(f'<span class="m">{escape(money.plain(v))}</span>')
def _cash(v, currency: str | None = None):
"""The amount WITH its symbol, for headline numbers that have no column
header to carry it. The symbol goes inside the span so the privacy overlay
blurs it along with the digits rather than leaving it sharp beside one."""
from markupsafe import Markup, escape
return Markup(f'<span class="m">{escape(money.text(v, currency))}</span>')
# The admin cog renders in the top bar on every page, so it is a global rather
# than something every route has to remember to pass.
templates.env.globals["admin_nav"] = navigation.ADMIN_ITEM
# What a person reads on a page, from the one module that knows it. A global
# because the login card needs it while signed OUT, where there is no route
# context to hang it on — and because the alternative is the name typed into a
# template, which is what `branding.NAME` exists to stop.
templates.env.globals["app_name"] = branding.NAME
# The currency totals are computed in. A global so no template writes "AUD",
# which is what makes T28b's per-portfolio reporting currency one change.
templates.env.globals["reporting_ccy"] = money.REPORTING
# Exposed to every template, so a link to the docs is `{{ docs_url }}` rather
# than an address repeated in a dozen files.
templates.env.globals["docs_url"] = DOCS_URL
# A fresh id each time a creating form renders — see app/submissions.py.
templates.env.globals["submission_id"] = submissions.mint
templates.env.globals["role_labels"] = ROLE_LABELS
templates.env.globals["action_label"] = queries.action_label
# Spreadsheet column letters: A..Z, AA, AB. Nobody counts to the 29th column.
templates.env.globals["column_letter"] = brokerdesign.column_letter
# A trade's moment on the exchange's clock, offset attached, for a <time
# datetime="...">. localtime.js rewrites those to the viewer's clock when the
# account asked for it; without JS the exchange's own time still renders.
templates.env.globals["market_stamp"] = lambda d, t, ex: (
markettime.aware(d, t, ex).isoformat() if markettime.aware(d, t, ex) else ""
)
def _trend(value) -> str:
"""`up`, `down`, or nothing: no change is neither a gain nor a loss."""
if value is None or value == 0:
return ""
return "up" if value > 0 else "down"
def _arrow(value):
"""Direction as a glyph beside the colour, which ~8% of men cannot tell
apart. Nothing for no change."""
from markupsafe import Markup
trend = _trend(value)
if not trend:
return ""
return Markup(f'<span class="dir" aria-hidden="true">{"▲" if trend == "up" else "▼"}</span>')
templates.env.filters["trend"] = _trend
templates.env.filters["plural"] = wording.plural
templates.env.globals["fy_label"] = fyreport.fy_label
templates.env.globals["class_label"] = lambda value: ASSET_CLASSES.get(value, value)
templates.env.filters["arrow"] = _arrow
templates.env.filters["money"] = _money
templates.env.filters["cash"] = _cash
templates.env.filters["money_plain"] = _money_plain
templates.env.filters["cpi_quarter"] = cpi.name
templates.env.filters["qty"] = lambda v: f"{v.normalize():f}" if v is not None else "—"
# Signed, as a change is, but no change has no sign.
templates.env.filters["pct"] = (lambda v: "—" if v is None else "0.00%" if v == 0
else f"{v * 100:+.2f}%")
# Cache-buster for static assets: browsers kept serving stale charts.js across
# releases (StaticFiles sends no Cache-Control). Any asset change → new URL.
templates.env.globals["asset_v"] = str(
int(max(p.stat().st_mtime for p in (APP_DIR / "static").rglob("*") if p.is_file()))
)
# --------------------------------------------------------------------------- #
# Session plumbing
# --------------------------------------------------------------------------- #
# Both live in auth.py so the imports router shares them verbatim.
scoped = auth.scoped_session
@contextmanager
def anon():
"""A session for the pre-login flows (users, sessions, login attempts)."""
db = SessionLocal()
try:
yield db
db.commit()
except Exception:
db.rollback()
raise
finally:
db.close()
def _users_exist(db: DbSession) -> bool:
return (db.scalar(select(func.count()).select_from(User)) or 0) > 0
def _set_session_cookie(resp: Response, raw_token: str) -> None:
# A session now exists, so somebody is about to look at what their holdings
# are worth. Kick a live refresh off here rather than in each of the five
# sign-in paths (password, second factor, recovery code, passkey, setup) —
# this is the one line all of them share, and it is exactly the moment the
# answer stops being hypothetical.
refresh_quotes_soon()
resp.set_cookie(
settings.auth.cookie_name,
raw_token,
max_age=settings.auth.session_ttl_days * 86400,
httponly=True,
samesite="lax",
secure=settings.auth.cookie_secure,
path="/",
)
# MUST comfortably outlive the session idle window, or the login page you were
# just sent to fails its own CSRF check — decisions.md #94.
PRE_AUTH_CSRF_SECONDS = 12 * 3600
def _set_pre_auth_csrf(resp: Response, token: str) -> None:
resp.set_cookie(
auth.PRE_AUTH_CSRF_COOKIE,
token,
max_age=PRE_AUTH_CSRF_SECONDS,
# A double-submit token, paired with a hidden form field. Scripts read
# the field, never the cookie, so a script on the page gets nothing here.
httponly=True,
samesite="lax",
secure=settings.auth.cookie_secure,
path="/",
)
OIDC_STATE_COOKIE = "pf_oidc"
def _set_oidc_cookie(resp: Response, token: str) -> None:
"""The token naming this sign-in attempt, for the callback to find.
HttpOnly and SameSite=Lax: the provider redirects the browser BACK here, a
top-level navigation, which Lax allows — Strict would drop the cookie on
exactly the request that needs it, and the sign-in would fail with nothing
on screen to explain why.
"""
resp.set_cookie(
OIDC_STATE_COOKIE, token,
max_age=int(federation.STATE_TTL.total_seconds()),
httponly=True, samesite="lax",
secure=settings.auth.cookie_secure, path="/",
)
def _redirect(url: str) -> RedirectResponse:
return RedirectResponse(url, status_code=303)
def _listings(ctx) -> dict:
"""`holding_href` and `listing_ref` for templates. Each names a holding's
exchange only where this portfolio holds its ticker on more than one,
which is read from the request's session the first time one is used."""
session = object_session(ctx.user) if getattr(ctx, "user", None) is not None else None
found: list[set[str]] = []
def ambiguous() -> set[str]:
if not found:
found.append(queries.ambiguous_tickers(session) if session is not None else set())
return found[0]
return {
"holding_href": lambda ticker, exchange=None, **params: queries.holding_path(
ticker, exchange, ambiguous(), **params),
"listing_ref": lambda ticker, exchange: queries.listing_ref(ticker, exchange, ambiguous()),
}
def _holding_page(db: DbSession, inst: Instrument, **params) -> str:
"""Where a change to this holding lands: its page, with the exchange in
the address when the ticker is held here on another one too."""
return queries.holding_path(inst.ticker, inst.exchange, queries.ambiguous_tickers(db),
**params)
def _render(request: Request, ctx, name: str, extra: dict,
status_code: int = 200) -> HTMLResponse:
"""Render with what the chrome always needs: auth, the nav list, and which
entry to highlight (routes pass `active_nav`)."""
return templates.TemplateResponse(
request,
name,
{
"auth": ctx,
**_listings(ctx),
"nav": navigation.nav_for(ctx, settings),
# The idle overlay reads this off <body> so the warning lead time is
# configurable without templating JavaScript.
"idle_warning_seconds": settings.auth.idle_warning_seconds,
# Set here rather than per route: it belongs to every page.
"theme_css": theming.stylesheet(ctx.user.theme_colors),
**extra,
},
status_code=status_code,
)
@app.middleware("http")
async def optional_features(request: Request, call_next):
"""Refuse the routes owned by a switched-off optional feature.
Middleware rather than a check in each route, because a feature owns a
*family* of paths — `/schedule`, `/schedule/save`, `/schedule/skip`, `/schedule/complete` —
and a per-route guard is a list that falls behind the routes. `features.py`
names one prefix and this covers everything under it, POSTs included: with
the nav entry hidden, a stale tab is exactly how a write to a switched-off
feature arrives.
**410 Gone, not 404.** The page exists and is turned off; saying "not
found" would send somebody hunting for a typo. The message names the
feature and says where to turn it back on.
"""
feature = features.owning(request.url.path)
if feature is not None and not features.enabled(settings, feature.key):
return PlainTextResponse(
f"{feature.name} is switched off for this install. "
"An administrator can turn it back on under Admin → Settings.",
status_code=410,
)
return await call_next(request)
@app.middleware("http")
async def refuse_nul(request: Request, call_next):
"""A NUL character anywhere a request carries text is refused, for every route.
Nobody types one, and Postgres refuses text containing one even to look
something up, so wherever one reached a query it was a 500, and on SQLite it
was stored. One check here rather than one per field, which is how #60's
gaps happened: decisions.md #133.
An upload is let through, because a file is bytes and a PDF is full of NULs.
Only a form or JSON body is read, and Starlette hands the route the same
body afterwards.
"""
if "\x00" in request.url.path or b"%00" in request.scope.get("query_string", b""):
return _nul_refused()
kind = request.headers.get("content-type", "")
if kind.startswith(("application/x-www-form-urlencoded", "application/json")):
body = await request.body()
# Encoded as %00 in a form, \u0000 in JSON, or sent raw.
if b"%00" in body or b"\\u0000" in body or b"\x00" in body:
return _nul_refused()
return await call_next(request)
def _nul_refused() -> PlainTextResponse:
return PlainTextResponse("Text can't contain a NUL character.", status_code=400)
def _wants_page(request: Request) -> bool:
"""A person in a browser, who gets a page rather than JSON. The API's
clients, and a page's own fetch() calls, ask for anything else."""
return (not request.url.path.startswith("/api/")
and "text/html" in request.headers.get("accept", ""))
def _error_page(request: Request, status: int, message: str,
headers: dict | None = None) -> HTMLResponse:
"""Every refusal and fault a browser meets. No session is read for it:
a fault may be the database itself, so the page has no nav."""
heading = ("Something went wrong" if status >= 500
else {403: "Not allowed", 404: "Not found"}.get(status, "That didn't work"))
return templates.TemplateResponse(
request, "error.html",
{"auth": None, "heading": heading, "message": message[:1].upper() + message[1:]},
status_code=status, headers=headers)
# A refusal that says no more than its status line gets a sentence instead.
_SAID_PLAINLY = {404: "There's nothing at this address.",
405: "This address doesn't take that kind of request."}
@app.exception_handler(StarletteHTTPException)
async def _refused(request: Request, exc: StarletteHTTPException):
"""A refusal, as a page in a browser. There was no HTML handler, and every
one of them answered a person with `{"detail": ...}`."""
if exc.status_code < 400 or not _wants_page(request):
return await http_exception_handler(request, exc)
phrase = HTTPStatus(exc.status_code).phrase
message = exc.detail if isinstance(exc.detail, str) and exc.detail != phrase else ""
return _error_page(request, exc.status_code,
message or _SAID_PLAINLY.get(exc.status_code, f"{phrase}."),
exc.headers)
@app.exception_handler(Exception)
async def _fault(request: Request, exc: Exception):
"""A fault, as a page in a browser, saying nothing of the code. The
server still logs the traceback: Starlette raises it again after this."""
if _wants_page(request):
return _error_page(request, 500, "This could not be finished. The application's "
"log has the details.")
return PlainTextResponse("Internal Server Error", status_code=500)
@app.exception_handler(RequestValidationError)
async def _validation_failed(request: Request, exc: RequestValidationError):
"""FastAPI's own 422, able to say what it refused.
The default echoes each bad value back as `input`. A bare NaN in a JSON
body (Python's parser accepts one) cannot be written back out as JSON, so a
request that was rightly refused ended in a 500 instead. A number that is
not finite goes back as text. A browser gets a page naming the fields.
"""
if _wants_page(request):
fields = sorted({str(error["loc"][-1]).replace("_", " ") for error in exc.errors()})
return _error_page(request, 422, f"Missing or unreadable: {', '.join(fields)}.")
return JSONResponse(status_code=422,
content={"detail": jsonable_encoder(_finite(exc.errors()))})
def _finite(value):
if isinstance(value, float) and not math.isfinite(value):
return str(value)
if isinstance(value, dict):
return {k: _finite(v) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [_finite(v) for v in value]
return value
@app.middleware("http")
async def security_headers(request: Request, call_next):
"""Headers every response carries.
Nothing loads from a CDN, so every CSP source is `'self'`.
**`'unsafe-inline'` is present and is the weak part** — the templates still
carry inline blocks and handlers, so the real XSS defence is the escaping,
not this: decisions.md #97.
"""
response = await call_next(request)
response.headers.setdefault("X-Content-Type-Options", "nosniff")
response.headers.setdefault("X-Frame-Options", "DENY")
response.headers.setdefault("Referrer-Policy", "same-origin")
response.headers.setdefault(
"Content-Security-Policy",
"default-src 'self'; "
# See the note above: inline handlers remain in the templates.
"script-src 'self' 'unsafe-inline'; "
"style-src 'self' 'unsafe-inline'; "
"img-src 'self' data:; "
"font-src 'self'; "
"connect-src 'self'; "
"form-action 'self'; "
"base-uri 'self'; "
"frame-ancestors 'none'",
)
return response
@app.middleware("http")
async def require_login(request: Request, call_next):
"""Single gate for the whole app — a new route is protected by default.
Unauthenticated HTML requests are sent to /login; anything else gets a 401
rather than a redirect, so a stale fetch doesn't silently render a page.
"""
path = request.url.path
# Serve-without-a-database: static files, the health probes, the wizard
# itself and the metrics endpoint, which guards its own access.
if path.startswith(NO_DATABASE_PREFIXES) or path in NO_DATABASE_PATHS:
return await call_next(request)
# No database means no sessions, no users, and nothing any other route can
# do. Everything goes to the wizard rather than to a 500 from the first
# query — this is the state a fresh container starts in.
#
# This is checked BEFORE the public paths, not after. /login is public and
# opens a session to look for the account, so an unconfigured app answered
# it with `NotConfigured` — a 500 on the one page somebody lands on after
# any redirect, which reads as the app being broken beyond recovery rather
# than as "finish setting me up".
if not database.is_ready():
if "text/html" not in request.headers.get("accept", ""):
return PlainTextResponse("not set up yet — visit /setup", status_code=503)
return _redirect("/setup")
if path in PUBLIC_PATHS or path.startswith(PUBLIC_PREFIXES):
return await call_next(request)
with anon() as db:
ctx = auth.load_auth(request, db)
logged_in = ctx is not None
first_run = not logged_in and not _users_exist(db)
must_change = logged_in and ctx.user.must_change_password
if not logged_in:
if "text/html" not in request.headers.get("accept", ""):
return PlainTextResponse("not authenticated", status_code=401)
return _redirect("/setup" if first_run else "/login")
# A temporary password must be replaced before anything else is reachable.
if must_change and path not in ("/profile", "/profile/password", "/logout"):
return _redirect("/profile")
# The wizard creates the database and the account BEFORE it has finished —
# step 3 has to write the account somewhere — so from there on a session
# exists and every page answers. Leaving mid-wizard then means a
# half-configured app with no timezone, no portfolio and no config file
# written, reached by deleting `/setup` from the address bar. Setup is
# finished when the draft is discarded, so this closes the gap without the
# wizard needing to know about it. `/logout` stays open: abandoning it is a
# legitimate thing to want.
if setupwizard.in_progress() and path != "/logout":
if "text/html" not in request.headers.get("accept", ""):
return PlainTextResponse("setup is not finished", status_code=503)
return _redirect(_wizard_url())
return await call_next(request)
app.include_router(imports_router)
app.include_router(api_router)
# --------------------------------------------------------------------------- #
# First run: the setup wizard — the HTTP half of app/setupwizard.py.
#
# EVERY route under /setup must open with `_wizard_step()`: /setup/ is public,
# so one without that call is a route anybody can post to (decisions.md #92).
# --------------------------------------------------------------------------- #
def _accounts_exist() -> bool:
"""Whether anybody has signed up, answered safely with no database."""
if not database.is_ready():
return False
try:
with anon() as db:
return _users_exist(db)
except Exception:
# A configured database that will not answer. Not "no accounts" — the
# wizard must not offer to create one over the top of a database it
# simply cannot reach.
log.exception("could not check for existing accounts")
return True
class _WizardRefused(Exception):
"""A step that no longer applies, carrying where to go instead."""
def __init__(self, response):
self.response = response
# Which step URL the wizard is actually on. NOT `/setup`: welcome closes the
# moment an account exists, so redirecting there sends somebody to /login, which
# sends them to /, which arrives back here — a loop.
_WIZARD_URLS = {
"welcome": "/setup", "database": "/setup/database", "account": "/setup/profile",
"recovery": "/setup/recovery", "2fa": "/setup/2fa", "portfolio": "/setup/portfolio",
"features": "/setup/features", "environment": "/setup/environment",
"finish": "/setup/finish",
}
def _wizard_url() -> str:
step = setupwizard.next_step(
database_configured=database.source(settings).configured,
account_exists=_accounts_exist(),
)
return _WIZARD_URLS.get(step, "/setup/finish")
def _wizard_step(request: Request, step: str):
"""Gate one wizard step, returning the auth context for the later ones.
The rule is per-step rather than one blanket check, because the steps are
open at different times and for different reasons:
* `welcome` and `database` are open only while there is no account. They
run without a session — there is nowhere to keep one.
* `account` is the same, plus it needs a database to write to.
* everything after needs a live session AND a wizard in progress. That
second half is what closes these pages afterwards: `finish` discards the
draft, and a restart has none, so the settings steps stop answering.
"""
if step in ("welcome", "database", "account"):
if _accounts_exist():
raise _WizardRefused(_redirect("/login"))
if step == "account" and not database.is_ready():
raise _WizardRefused(_redirect("/setup"))
return None
if not setupwizard.in_progress():
raise _WizardRefused(_redirect("/" if _accounts_exist() else "/setup"))
# A step this run does not offer is not enterable by typing its URL either.
# The rail would otherwise be describing a flow the router disagrees with,
# which is the shape of the loop in decisions.md #109.
if step in setupwizard.skipped():
raise _WizardRefused(_redirect(_wizard_url()))
db = SessionLocal()
try:
ctx = auth.load_auth(request, db)
finally:
db.close()
if ctx is None:
raise _WizardRefused(_redirect("/login"))
return ctx
# The steps that close the moment an account exists: they run before there is
# a session, and `_wizard_step` sends anyone arriving later to /login. Offering
# one as a destination is offering a loop — /login bounces a signed-in user to
# /, and / bounces a wizard in progress back here.
_CLOSED_ONCE_SIGNED_IN = ("welcome", "database", "account")
def _previous_step_url(rail: list[dict], *, closed: tuple[str, ...]) -> str | None:
"""The nearest step before the active one that can still be entered.
A wizard whose only way back is the browser button is one where noticing a
mistake on the summary means starting again. Steps this deployment skips
have nothing to go back TO, and the early steps shut once an account
exists — so the answer is the nearest one that is neither, or nothing.
`closed` is passed rather than read from the constant because those steps
are shut ONCE SIGNED IN and not before: treating them as shut throughout
took the back link off the database and account steps, where going back is
exactly what somebody correcting a typo needs.
"""
seen: list[str] = []
for entry in rail:
if entry.get("state") == "active":
break
if entry.get("state") != "skipped":
seen.append(entry.get("key") or "")
for key in reversed(seen):
if key not in closed and key in _WIZARD_URLS:
return _WIZARD_URLS[key]
return None
def _wizard_page(request: Request, step: str, extra: dict, *,
ctx=None, skip: set[str] | None = None,
rail: str | None = None) -> HTMLResponse:
"""Render a step with the rail, the CSRF token and the chrome it needs.
`rail` is for the pages that are part of a step without being one — the
recovery codes belong to two-factor, and giving them a rail entry of their
own would make the flow look longer than it is.
"""
# Union rather than default, so a page that skips a step of its own does
# not also un-skip the ones this run never offers.
skip = (skip or set()) | setupwizard.skipped()
token = ctx.session.csrf_token if ctx else auth.ensure_pre_auth_csrf(request)
resp = templates.TemplateResponse(
request,
f"setup_{step}.html",
{
"auth": None, # the wizard never draws the app's navigation
"csrf": token,
"step": step,
"steps": setupwizard.progress(rail or step, skip=skip),
# The previous step somebody can actually return to. Derived from
# the rail rather than a fixed order, so a step this deployment
# skips is skipped going backwards too.
"wizard_back": _previous_step_url(
setupwizard.progress(rail or step, skip=skip),
closed=_CLOSED_ONCE_SIGNED_IN if _accounts_exist() else ()),
"error": None,
**extra,
},
)
if ctx is None:
_set_pre_auth_csrf(resp, token)
return resp
def _skipped_steps() -> set[str]:
"""Steps this deployment will not be shown, so the rail can grey them out
rather than silently renumbering itself."""
skip = set()
if database.source(settings).configured:
skip.add("database")
if not configfile.creatable().writable:
# Neither of these can be saved, so neither is offered — a form that
# discards what you type is worse than one that is not there. Optional
# features live in config.yaml exactly like the environment settings,
# so they skip for the same reason and keep every feature on.
skip.add("environment")
skip.add("features")
return skip
@app.exception_handler(_WizardRefused)
async def _wizard_refused(request: Request, exc: _WizardRefused):
return exc.response
# --- step 1: welcome ------------------------------------------------------- #
@app.get("/setup", response_class=HTMLResponse)
def setup_welcome(request: Request):
_wizard_step(request, "welcome")
source = database.source(settings)
writable = configfile.creatable()
# A configured database is TESTED here, not merely reported as present.
# "Configured" only says a connection string exists; the useful answer is
# whether it works, and finding out on this page beats finding out three
# steps later with an account half-created.
probe = database.probe(settings.database) if source.configured else None
return _wizard_page(request, "welcome", {
"db_source": source,
"db_description": database.describe(settings.database) if source.configured else None,
"db_probe": probe,
"config_path": str(configfile.config_path()),
"config_writable": writable,
}, skip=_skipped_steps())
@app.post("/setup")
async def setup_welcome_next(request: Request):
_wizard_step(request, "welcome")
await auth.verify_pre_auth_csrf(request)
setupwizard.draft().completed("welcome")
if not database.source(settings).configured:
return _redirect("/setup/database")
if not database.is_ready():
# Configured but not connected: the connection failed at boot. Saying
# so beats sending them into an account form that cannot save anything.
return _redirect("/setup/database")
return _redirect("/setup/profile")
# --- step 2: database ------------------------------------------------------ #
@app.get("/setup/database", response_class=HTMLResponse)
def setup_database(request: Request, tested: str = "", ok: str = ""):
_wizard_step(request, "database")
source = database.source(settings)
if source.configured and database.is_ready():
return _redirect("/setup/profile")
return _wizard_page(request, "database", {
"db_source": source,
"db_description": database.describe(settings.database) if source.configured else None,
"default_path": settings.database.path or "/data/stocktake.db",
"tested": tested,
"tested_ok": ok == "1",
"form": {},
}, skip=_skipped_steps())
def _database_from_form(form) -> DatabaseSettings:
"""Build settings from the submitted form, raising ValueError to show."""
kind = (form.get("kind") or "sqlite").strip()
if kind == "sqlite":
return setupwizard.sqlite_settings(str(form.get("path") or ""))
return setupwizard.postgres_settings(
host=str(form.get("host") or ""),
port=str(form.get("port") or "5432"),
name=str(form.get("name") or ""),
user=str(form.get("user") or ""),
password=str(form.get("password") or ""),
)
def _database_form_page(request: Request, form, message: str, ok: bool) -> HTMLResponse:
"""Redraw the database step keeping what was typed — losing a host and port
someone just worked out, because the password was wrong, is its own small
cruelty, and retyping is where the next typo comes from.
The password is dropped here rather than carried. That is the second of two
layers: the template's password input has no `value` either, so it cannot
echo one even if this dict grew one. Both, because either alone is one edit
away from putting a database password into the page source.
"""
page = _wizard_page(request, "database", {
"db_source": database.source(settings),
"db_description": None,
"default_path": settings.database.path or "/data/stocktake.db",
"tested": message,
"tested_ok": ok,
"form": {k: v for k, v in form.items() if k not in ("_csrf", "password")},
}, skip=_skipped_steps())
return page
@app.post("/setup/database/test")
async def setup_database_test(request: Request):
"""Connect and report, writing nothing. The whole point of the button."""
_wizard_step(request, "database")
await auth.verify_pre_auth_csrf(request)
form = await request.form()
try:
chosen = _database_from_form(form)
except ValueError as exc:
return _database_form_page(request, form, str(exc), False)
# Off the event loop: anybody can press this before the first account
# exists, and a slow server would otherwise stop every other request,
# /healthz included, until the liveness probe restarted the pod.
result = await asyncio.to_thread(database.probe, chosen)
return _database_form_page(request, form, result.detail, result.ok)
@app.post("/setup/database")
async def setup_database_save(request: Request):
"""Connect for real: migrate, build the session factory, and remember the
choice so the last step can write it to the config file.
Probed first even though `connect()` would fail anyway — a failure here
would surface as an Alembic traceback, and the probe's explanation is the
one worth reading.
"""
_wizard_step(request, "database")
await auth.verify_pre_auth_csrf(request)
form = await request.form()
try:
chosen = _database_from_form(form)
except ValueError as exc:
return _database_form_page(request, form, str(exc), False)
# Both off the event loop, as the test button is: connecting migrates.
result = await asyncio.to_thread(database.probe, chosen)
if not result.ok:
return _database_form_page(request, form, result.detail, False)
try:
await asyncio.to_thread(database.connect, chosen)
except Exception as exc:
log.exception("wizard could not connect the chosen database")
return _database_form_page(
request, form,
f"Connected, but preparing the database failed: {exc}", False)
current = setupwizard.draft()
current.database = chosen
current.completed("database")
return _redirect("/setup/profile")
# --- step 3: the account --------------------------------------------------- #
@app.get("/setup/profile", response_class=HTMLResponse)
def setup_account(request: Request, error: str = ""):
_wizard_step(request, "account")
return _wizard_page(request, "account", {
"db_description": database.describe(database.current() or settings.database),
"error": error or None,
"want_2fa": setupwizard.draft().want_2fa,
}, skip=_skipped_steps())
@app.post("/setup/profile")
async def setup_account_create(
request: Request,
name: str = Form(...),
email: str = Form(...),
password: str = Form(...),
want_2fa: str = Form(""),
):
_wizard_step(request, "account")
await auth.verify_pre_auth_csrf(request)
problem = (auth.password_problem(password) or await _confirm_problem(request, password)
or auth.email_problem(email) or textfield.problem(name, User.name, "Name"))
if problem:
setupwizard.draft().want_2fa = bool(want_2fa)
return _wizard_page(request, "account", {
"db_description": database.describe(database.current() or settings.database),
"error": problem,
"want_2fa": bool(want_2fa),
"typed": {"name": name, "email": email},
}, skip=_skipped_steps())
with anon() as db:
if _users_exist(db):
raise HTTPException(403, "Setup already completed")
user = User(
email=email.strip().lower(),
name=_name_or_email(name, email),
password_hash=await asyncio.to_thread(auth.hash_password, password),
is_admin=True, # the bootstrap account manages the others
)
db.add(user)
db.flush()
# Named now and renamed on the portfolio step. Created here rather than
# there because a session needs a portfolio to be scoped to — and
# because an abandoned wizard should still leave a working install.
portfolio = Portfolio(name=f"{user.name}'s portfolio"[:80])
db.add(portfolio)
db.flush()
db.add(PortfolioMember(portfolio_id=portfolio.id, user_id=user.id, role="owner"))
db.flush()
_claim_legacy_rows(db, user.id, portfolio.id)
twofactor.ensure_recovery_codes(db, user)
raw = auth.create_session(db, user, settings, active_portfolio_id=portfolio.id)
current = setupwizard.draft()
current.completed("account")
current.want_2fa = bool(want_2fa)
current.portfolio_name = portfolio.name
resp = _redirect("/setup/recovery")
_set_session_cookie(resp, raw)
return resp
# --- step 4: recovery codes ------------------------------------------------ #
@app.get("/setup/recovery", response_class=HTMLResponse)
def setup_recovery(request: Request):
"""The codes, shown once, before the second factor is even offered.
This order is the point. They are what makes skipping 2FA survivable, and
they are the whole of account recovery for an app with no reset email — so
they are not a footnote to enrolling an authenticator, they come first.
"""
_wizard_step(request, "recovery")
current = setupwizard.draft()
with scoped(request) as (ctx, db):
# Issued once for the whole wizard. A reload or a step back re-renders
# the same list; a fresh set is a deliberate act from the account page.
if not current.recovery_codes:
current.recovery_codes = twofactor.generate_recovery_codes(db, ctx.user)
ctx.user.recovery_codes_seen_at = dt.datetime.now(dt.timezone.utc)
db.flush()
codes = current.recovery_codes
current.completed("recovery")
return _wizard_page(request, "recovery", {
"codes": codes,
"totp_on": False,
# Derived, not "/setup/portfolio": whether two-factor comes next is the
# tickbox's business, and a hardcoded target here is how a step gets
# skipped by the link that was supposed to lead to it.
"continue_url": _WIZARD_URLS.get(setupwizard.next_step(
database_configured=True, account_exists=True), "/setup/portfolio"),
}, ctx=ctx, skip=_skipped_steps())
# --- step 5: two-factor ---------------------------------------------------- #
@app.get("/setup/2fa", response_class=HTMLResponse)
def setup_twofactor(request: Request, error: str = ""):
"""Offered only when the account step asked for it.
Same ceremony as `/profile/2fa`, in wizard chrome — including resuming an
enrolment already in progress, which matters more here: a wrong code
re-renders this page, and regenerating would invalidate the QR on the way
back from the mistake.
"""
ctx = _wizard_step(request, "2fa")
with scoped(request) as (inner, db):
enrol = _enrolment(db, inner.user)
return _wizard_page(request, "2fa", {
"error": error or None,
"secret": enrol["secret"],
"qr": enrol["qr"],
}, ctx=ctx, skip=_skipped_steps())
@app.post("/setup/2fa")
async def setup_twofactor_enable(request: Request, code: str = Form(...)):
_wizard_step(request, "2fa")
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
# The pending secret on the row, never one posted with the form.
secret = ctx.user.totp_secret
if not twofactor.accept_code(db, ctx.user, code):
return _redirect("/setup/2fa?error=" + quote_plus(
"That code isn't right — check the app and try again."))
# Does NOT reissue recovery codes: they were handed out on the previous
# step and belong to the account, not to this factor.
twofactor.enable(db, ctx.user, secret)
log.info("2FA enabled for user %s during setup", ctx.user.id)
setupwizard.draft().completed("2fa")
return _redirect("/setup/portfolio")
@app.post("/setup/2fa/skip")
async def setup_twofactor_skip(request: Request):
"""Not a dead end: somebody who cannot reach their phone right now would
otherwise be stuck on a step they asked for. Clearing the flag also takes
it out of the rail, so the numbering matches what is left to do.
"""
_wizard_step(request, "2fa")
# The session token, not the pre-auth one: the account exists by now, so
# this page was rendered with `ctx` and carries the signed-in token.
with scoped(request) as (_ctx, db):
await auth.verify_csrf(request, db)
setupwizard.draft().want_2fa = False
return _redirect("/setup/portfolio")
# --- step 5: the first portfolio ------------------------------------------- #
@app.get("/setup/portfolio", response_class=HTMLResponse)
def setup_portfolio(request: Request, error: str = ""):
ctx = _wizard_step(request, "portfolio")
current = setupwizard.draft()
return _wizard_page(request, "portfolio", {
"error": error,
"name": current.portfolio_name or "My portfolio",
"price_feed": current.price_feed,
"timezone": current.timezone or settings.timezone,
"zones": setupwizard.timezones(),
}, ctx=ctx, skip=_skipped_steps())
@app.post("/setup/portfolio")
async def setup_portfolio_save(
request: Request,
name: str = Form(...),
timezone: str = Form(...),
price_feed: str = Form(""),
):
_wizard_step(request, "portfolio")
problem = setupwizard.timezone_problem(timezone)
if not name.strip():
problem = "Give the portfolio a name."
if problem:
return _redirect("/setup/portfolio?error=" + quote_plus(problem))
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
portfolio = db.get(Portfolio, ctx.active_portfolio_id)
portfolio.name = name.strip()[:80]
db.flush()
current = setupwizard.draft()
current.portfolio_name = name.strip()[:80]
current.timezone = timezone.strip()
current.price_feed = price_feed.lower() in ("1", "true", "on", "yes")
current.completed("portfolio")
# The timezone decides what "today" means everywhere, so it applies now
# rather than at the restart — a wizard finished on the wrong date would be
# a strange first impression.
settings.timezone = current.timezone
clock.configure(current.timezone)
if "features" in _skipped_steps():
return _redirect("/setup/finish" if "environment" in _skipped_steps()
else "/setup/environment")
return _redirect("/setup/features")
# --- step 6: optional features --------------------------------------------- #
@app.get("/setup/features", response_class=HTMLResponse)
def setup_features(request: Request):
"""Which optional parts of the app this install wants.
Everything defaults ON, and the page says these are changeable later —
a first-run wizard is the worst moment to ask somebody to decide what they
will want, so the honest framing is "you can change your mind".
"""
ctx = _wizard_step(request, "features")
current = setupwizard.draft()
return _wizard_page(request, "features", {
"features": features.FEATURES,
"chosen": current.features,
}, ctx=ctx, skip=_skipped_steps())
@app.post("/setup/features")
async def setup_features_save(request: Request):
"""A checkbox that is off is simply absent from the submission, so the
chosen set is built from what IS there — every declared feature not named
was unticked."""
_wizard_step(request, "features")
with scoped(request) as (_, db):
await auth.verify_csrf(request, db)
form = await request.form()
ticked = set(form.getlist("feature"))
current = setupwizard.draft()
current.features = {f.key: (f.key in ticked) for f in features.FEATURES}
current.completed("features")
if "environment" in _skipped_steps():
return _redirect("/setup/finish")
return _redirect("/setup/environment")
# --- step 7: environment --------------------------------------------------- #
@app.get("/setup/environment", response_class=HTMLResponse)
def setup_environment(request: Request, error: str = ""):
ctx = _wizard_step(request, "environment")
current = setupwizard.draft()
return _wizard_page(request, "environment", {
"error": error,
"proxies": "\n".join(current.trusted_proxies),
"external_url": current.external_url or "",
"supervision": lifecycle.supervision(),
}, ctx=ctx, skip=_skipped_steps())
@app.post("/setup/environment")
async def setup_environment_save(
request: Request,
trusted_proxies: str = Form(""),
external_url: str = Form(""),
):
_wizard_step(request, "environment")
with scoped(request) as (_, db):
await auth.verify_csrf(request, db)
current = setupwizard.draft()
try:
current.trusted_proxies = setupwizard.parse_proxies(trusted_proxies)
setupwizard.parse_external_url(external_url) # validated, kept raw
except ValueError as exc:
return _redirect("/setup/environment?error=" + quote_plus(str(exc)))
current.external_url = external_url.strip() or None
current.completed("environment")
return _redirect("/setup/finish")
# --- step 7: finish -------------------------------------------------------- #
@app.get("/setup/finish", response_class=HTMLResponse)
def setup_finish(request: Request, error: str = ""):
ctx = _wizard_step(request, "finish")
current = setupwizard.draft()
values = setupwizard.config_values(current)
writable = configfile.creatable()
# Only what the file does not already provide. Somebody who wrote their own
# config.yaml was being told to set values it already had — issue #18.
outstanding = configfile.not_yet_provided(values)
return _wizard_page(request, "finish", {
"error": error,
"values": values,
"yaml": configfile.render(values),
"outstanding_yaml": configfile.render(outstanding) if outstanding else "",
"config_path": str(configfile.config_path()),
"config_writable": writable,
"restart_for": setupwizard.needs_restart(current),
"supervision": lifecycle.supervision(),
"external_url": current.external_url,
}, ctx=ctx, skip=_skipped_steps())
@app.post("/setup/finish")
async def setup_finish_save(request: Request):
"""Write the config, then either land on the dashboard or on the page that
carries you across a restart."""
_wizard_step(request, "finish")
with scoped(request) as (_, db):
await auth.verify_csrf(request, db)
current = setupwizard.draft()
values = setupwizard.config_values(current)
if configfile.creatable().writable:
try:
configfile.write_tree(values)
except OSError as exc:
log.exception("wizard could not write the config file")
return _redirect("/setup/finish?error=" + quote_plus(
f"Could not write {configfile.config_path()}: {exc}"))
# Always, when there is something a restart is needed to pick up. It used
# to be a tickbox, which made "finish setup" able to mean "finish setup and
# ignore what I just configured".
wants_restart = bool(setupwizard.needs_restart(current))
target = _post_setup_url(current)
setupwizard.discard()
if wants_restart and lifecycle.supervision().restarts:
return _restarting_page(request, target)
return _redirect(target)
def _post_setup_url(current) -> str:
"""Where to send someone once setup is done.
Their own external URL if they gave one — the app may well be about to
become unreachable at the address they are currently using, which is
exactly the case the waiting page exists for.
"""
url = setupwizard.parse_external_url(current.external_url or "")
return url.url + "/" if url is not None else "/"
def _restarting_page(request: Request, target: str):
"""The page that survives the restart, as the Finish step's own answer.
Rendered, THEN the stop is requested — the other order sends a SIGTERM
before the browser has the page telling it what is happening, and the
person is left looking at a connection error with no idea whether the
wizard worked.
It was a GET of its own, taking where to go next from the address. Only
Finish, gated and token-checked, draws it now, and the target is the one
it worked out.
"""
resp = templates.TemplateResponse(
request,
"setup_restarting.html",
{"auth": None, "target": target, "supervision": lifecycle.supervision()},
)
lifecycle.request_stop("setup wizard finished")
return resp
def _claim_legacy_rows(db: DbSession, user_id: int, portfolio_id: int) -> None:
"""Hand every pre-auth row to the first account.
Rows created before accounts existed have no owner. They join the first portfolio. Instrument notes were
personal too (sale prices, strategy) — they move into that portfolio's
holding_pref and are cleared from the shared catalogue in the same
transaction.
"""
# These rows have no owner yet, so a user filter could never find them.
tenancy.allow_unscoped(db)
claimed = 0
for model in (Trade, Dividend, PlannedPurchase, InvestmentPlan):
rows = db.scalars(select(model).where(model.portfolio_id.is_(None))).all()
for row in rows:
row.portfolio_id = portfolio_id
if getattr(row, "user_id", "absent") is None:
row.user_id = user_id
claimed += len(rows)
for inst in db.scalars(select(Instrument)).all():
if inst.note or inst.drp:
db.add(
HoldingPref(
portfolio_id=portfolio_id,
instrument_id=inst.id,
drp=inst.drp,
note=inst.note,
)
)
inst.note = None
db.flush()
log.info("claimed %d pre-auth rows into portfolio %d", claimed, portfolio_id)
def _passkeys_usable(request: Request) -> bool:
"""Whether this connection can run the ceremony at all.
Both halves matter: configured by the deployment, and reached over HTTPS —
a browser refuses WebAuthn outside a secure context, so offering the button
on plain HTTP is offering a failure with no visible cause.
"""
return passkeys.configured(settings) and auth.is_secure_request(request, settings)
def _oidc_context() -> dict:
"""Whether to offer provider sign-in, and what to call it.
No secure-context requirement, unlike passkeys: OIDC is a redirect, and the
provider enforces its own transport. The redirect URL is the thing that has
to be right, and it is declared rather than guessed.
"""
return {"oidc_usable": federation.configured(settings),
"oidc_label": settings.auth.oidc.button_label}
@app.get("/login", response_class=HTMLResponse)
def login_form(request: Request, timeout: str = "", error: str = ""):
with anon() as db:
if auth.load_auth(request, db) is not None:
return _redirect("/")
if not _users_exist(db):
return _redirect("/setup")
token = auth.ensure_pre_auth_csrf(request)
resp = templates.TemplateResponse(
request, "login.html",
{"auth": None, "csrf": token,
# A provider sign-in that failed comes back here, and the reason
# has nowhere else to go: the whole attempt happened off-site.
"error": error or None,
# Said BEFORE they try, because the failure it describes looks
# like a wrong password and would otherwise be retried forever.
"insecure": auth.insecure_login_problem(request, settings),
# Sent here by the idle overlay. Saying why beats a login page
# that appears for no reason someone can see.
"timed_out": timeout == "1",
"passkeys_usable": _passkeys_usable(request),
**_oidc_context()},
)
_set_pre_auth_csrf(resp, token)
return resp
@app.post("/login")
async def login_submit(
request: Request, email: str = Form(...), password: str = Form(...)
):
ip = auth.client_ip(request)
email_l = email.strip().lower()
with anon() as db:
def _fail(message: str):
token = auth.ensure_pre_auth_csrf(request)
resp = templates.TemplateResponse(
request, "login.html",
{"auth": None, "csrf": token, "error": message,
"passkeys_usable": _passkeys_usable(request),
**_oidc_context()}
)
_set_pre_auth_csrf(resp, token)
return resp
# A stale token hands back a fresh login form, not a 403: presentation
# only, not a weaker check — decisions.md #43.
try:
await auth.verify_csrf(request, db)
except HTTPException as exc:
if exc.status_code != 403:
raise
log.info("stale login CSRF from %s — re-issuing the form", ip)
return _fail("This page had been open a while. Try again.")
if not auth.email_fits(email_l):
return _fail("Invalid email or password.")
if auth.is_locked(db, email_l, settings):
return _fail("Too many attempts. Try again shortly.")
user = db.scalar(select(User).where(User.email == email_l))
stored = user.password_hash if user else None
# Same message either way — no account enumeration. Argon2 in a thread,
# here and wherever a handler checks or makes a hash: tens of
# milliseconds by design, unknown addresses included, and on the event
# loop every other request waited for them.
ok = (await asyncio.to_thread(auth.verify_password, stored, password)
and user is not None and user.is_active)
auth.record_attempt(db, email_l, ip, success=ok)
if not ok:
return _fail("Invalid email or password.")
if auth.needs_rehash(user.password_hash):
user.password_hash = await asyncio.to_thread(auth.hash_password, password)
landing = _landing_portfolio(db, user, request)
# With a second factor on, the password only buys a half-authenticated
# session: `auth.load_session` refuses it everywhere, so the cookie is
# useless until the code step flips the flag.
pending = twofactor.is_enabled(user)
raw = auth.create_session(
db, user, settings, active_portfolio_id=landing, awaiting_totp=pending,
ip=ip, user_agent=request.headers.get("user-agent"),
)
if pending:
resp = _redirect("/login/code")
_set_session_cookie(resp, raw)
return resp
resp = _redirect("/profile" if user.must_change_password else "/")
_set_session_cookie(resp, raw)
_remember_portfolio(resp, landing)
return resp
def _portfolio_cookie() -> str:
"""The last portfolio opened, on this browser: named after the session's
cookie, so two apps on one host do not share it."""
return f"{settings.auth.cookie_name}_portfolio"
def _remember_portfolio(resp: Response, portfolio_id: int | None) -> None:
"""Note which portfolio is open, for the next sign-in on this browser.
The id and nothing else — no name, no person — and a hint, never a
credential: `_landing_portfolio` uses it only for a portfolio the person is
a member of. It outlives the session on purpose; that is what it is for.
Set only once a sign-in is complete, never by a password still waiting on
its code."""
if portfolio_id is None:
return
resp.set_cookie(_portfolio_cookie(), str(portfolio_id), max_age=365 * 86400,
httponly=True, samesite="lax", secure=settings.auth.cookie_secure,
path="/")
def _remembered_portfolio(request: Request) -> int | None:
return row_id(request.cookies.get(_portfolio_cookie(), ""))
def _landing_portfolio(db: DbSession, user: User, request: Request) -> int | None:
"""Where a sign-in lands. Shared by every way in, so they cannot drift.
The portfolio chosen in the profile; else the last one opened on this
browser; else one they own before one merely shared with them — otherwise
the lowest id wins and someone invited into an older portfolio opens
somebody else's holdings instead of their own. The first two count only
while the person is still a member: the cookie is whatever the browser
sends, and a default can outlast a membership.
"""
mine = set(db.scalars(select(PortfolioMember.portfolio_id)
.where(PortfolioMember.user_id == user.id)))
for wanted in (user.default_portfolio_id, _remembered_portfolio(request)):
if wanted in mine:
return wanted
return db.scalar(
select(PortfolioMember.portfolio_id)
.where(PortfolioMember.user_id == user.id)
.order_by(
(PortfolioMember.role != "owner"), # owners first
PortfolioMember.portfolio_id,
)
)
def _code_form(request: Request, error: str | None = None):
token = auth.ensure_pre_auth_csrf(request)
resp = templates.TemplateResponse(
request,
"login_code.html",
{"auth": None, "csrf": token, "error": error},
)
_set_pre_auth_csrf(resp, token)
return resp
# --------------------------------------------------------------------------- #
# Forgotten password: a recovery code stands in for it — but never for the
# second factor, which is what stops a stolen code list being the whole
# account. decisions.md #36.
# --------------------------------------------------------------------------- #
def _recover_form(request: Request, error: str | None = None) -> HTMLResponse:
token = auth.ensure_pre_auth_csrf(request)
resp = templates.TemplateResponse(
request, "recover.html", {"auth": None, "csrf": token, "error": error},
)
_set_pre_auth_csrf(resp, token)
return resp
@app.get("/login/recover", response_class=HTMLResponse)
def recover_form(request: Request):
return _recover_form(request)
@app.post("/login/recover")
async def recover_submit(request: Request, email: str = Form(""), code: str = Form("")):
"""Spend a recovery code to get past a forgotten password.
Every failure returns the SAME page with the SAME message. An
unauthenticated endpoint that takes an email address is an enumeration
oracle the moment "no such account" and "wrong code" look different.
"""
ip = auth.client_ip(request)
email_l = email.strip().lower()
with anon() as db:
await auth.verify_pre_auth_csrf(request)
# Throttled. NIST stops requiring this once a look-up secret clears 64
# bits — ours do — but it is the difference between one wrong guess and
# an unbounded stream of them, and it already exists.
if not auth.email_fits(email_l):
return _recover_form(
request, "That email and recovery code do not match.")
if auth.is_locked(db, email_l, settings):
return _recover_form(request, "Too many attempts. Try again shortly.")
user = db.scalar(select(User).where(User.email == email_l))
ok = False
if user is not None and user.is_active:
ok = twofactor.consume_recovery_code(db, user, code)
auth.record_attempt(db, email_l, ip, success=ok)
if not ok:
return _recover_form(
request, "That email and recovery code do not match.")
# The code proved possession of the list; it did NOT prove the second
# factor. If one is enrolled, this session stays half-authenticated
# exactly as a password login would.
landing = _landing_portfolio(db, user, request)
pending = twofactor.is_enabled(user)
# Whatever password they could not remember is now unusable to anyone
# who might have learned it, and they are made to choose a new one
# before the app is reachable.
user.must_change_password = True
raw = auth.create_session(
db, user, settings, active_portfolio_id=landing, awaiting_totp=pending,
ip=ip, user_agent=request.headers.get("user-agent"),
recovery_spent=True,
)
left = twofactor.remaining_recovery_codes(db, user)
# The code itself is never logged — it would be a live credential in a
# log file until it was spent.
log.warning("user %d signed in with a recovery code (%d left)", user.id, left)
resp = _redirect("/login/code" if pending else "/profile")
_set_session_cookie(resp, raw)
if not pending:
_remember_portfolio(resp, landing)
return resp
@app.get("/login/code", response_class=HTMLResponse)
def login_code_form(request: Request):
with anon() as db:
pending = auth.load_pending_session(
db, request.cookies.get(settings.auth.cookie_name), settings)
if pending is None:
# No half-authenticated session: either they never gave a password,
# or they already finished. Either way this page is not for them.
return _redirect("/login")
return _code_form(request)
@app.post("/login/code")
async def login_code_submit(request: Request, code: str = Form(...)):
ip = auth.client_ip(request)
with anon() as db:
await auth.verify_csrf(request, db)
pending = auth.load_pending_session(
db, request.cookies.get(settings.auth.cookie_name), settings)
if pending is None:
return _redirect("/login")
user = db.get(User, pending.user_id)
if user is None: # account deleted between the two steps
auth.destroy_session(db, request.cookies.get(settings.auth.cookie_name))
return _redirect("/login")
# Lockout covers this stage too. Without it the code is a six-digit
# secret that can be brute-forced at will by anyone holding a password
# — which is precisely the situation the second factor exists for.
if auth.is_locked(db, user.email, settings):
return _code_form(request, "Too many attempts. Try again shortly.")
used_recovery = False
if twofactor.accept_code(db, user, code):
ok = True
elif pending.recovery_spent:
# ONE recovery code per sign-in (decisions.md #45), refused
# indistinguishably from a wrong code (#51).
ok = False
else:
ok = used_recovery = twofactor.consume_recovery_code(db, user, code)
auth.record_attempt(db, user.email, ip, success=ok)
if not ok:
return _code_form(request, "That code isn't right.")
pending.awaiting_totp = False
db.flush()
if used_recovery:
left = twofactor.remaining_recovery_codes(db, user)
log.info("user %s signed in with a recovery code (%d left)", user.id, left)
resp = _redirect("/profile" if user.must_change_password else "/")
_remember_portfolio(resp, pending.active_portfolio_id)
return resp
# --------------------------------------------------------------------------- #
# Metrics
# --------------------------------------------------------------------------- #
@app.get("/metrics", include_in_schema=False)
def prometheus_metrics():
"""Health for a scraper. See `app/metrics.py` for what may go in here.
**404 when disabled, not 403.** A 403 would confirm the endpoint exists and
is merely shut, which is a small piece of free reconnaissance for no
benefit — a caller who is allowed to know already reads the config.
This is in `PUBLIC_PATHS`, so the check below is the *only* thing standing
between an unconfigured install and an anonymous database read. It stays
the first statement in the function.
"""
if not settings.metrics.enabled:
raise HTTPException(status_code=404)
if not database.is_ready():
# No database yet (the wizard has not been finished). Still a valid
# scrape — `stocktake_database_ready 0` is the useful answer, and a
# 500 here would show up as a scrape failure rather than as the state
# it actually is.
return PlainTextResponse(metrics.render(None), media_type=metrics.CONTENT_TYPE)
with anon() as db:
return PlainTextResponse(metrics.render(db), media_type=metrics.CONTENT_TYPE)
# --------------------------------------------------------------------------- #
# Session lifetime: the countdown, and the one endpoint that resets it
# --------------------------------------------------------------------------- #
@app.get("/session/status")
def session_status(request: Request):
"""Seconds left before this session dies, for the idle overlay.
In `auth.SLIDING_EXEMPT`, so asking must NOT extend the thing being asked
about — otherwise the overlay's own polling would keep the session alive
and the timeout would never fire.
The server is the authority here. The browser runs its own timer for a
smooth countdown, but re-syncs from this so a clock skew, a sleeping laptop
or a second tab can't make it disagree with reality.
"""
with anon() as db:
row = auth.load_session(db, request.cookies.get(settings.auth.cookie_name),
settings)
if row is None:
return {"authenticated": False, "seconds_left": 0}
idle_deadline, hard_deadline = auth.session_deadlines(row, settings)
deadline = min(idle_deadline, hard_deadline, ensure_utc(row.expires_at))
left = (deadline - dt.datetime.now(dt.timezone.utc)).total_seconds()
return {
"authenticated": True,
"seconds_left": max(0, int(left)),
"warn_at": settings.auth.idle_warning_seconds,
}
@app.post("/session/keepalive")
async def session_keepalive(request: Request):
""""Stay signed in" — the ONE endpoint that deliberately slides the window.
A button press is a person, which is exactly what the status poll is not.
It still can't push past the absolute cap; `load_auth` clamps to it.
"""
with anon() as db:
await auth.verify_csrf(request, db)
ctx = auth.load_auth(request, db) # this path slides, by design
if ctx is None:
raise HTTPException(401, "session expired")
idle_deadline, hard_deadline = auth.session_deadlines(ctx.session, settings)
left = (min(idle_deadline, hard_deadline)
- dt.datetime.now(dt.timezone.utc)).total_seconds()
return {"authenticated": True, "seconds_left": max(0, int(left))}
@app.post("/logout")
async def logout(request: Request, next: str = ""):
provider_logout = None
with anon() as db:
await auth.verify_csrf(request, db)
token = request.cookies.get(settings.auth.cookie_name)
row = auth.load_session(db, token, settings)
# Read BEFORE the session is destroyed, and only for a session that
# began at the provider. Failing to reach the provider must not stop
# the local sign-out, so this is worked out first and used after.
if row is not None and row.via_oidc and federation.configured(settings):
try:
provider_logout = oidc.end_session_url(federation.provider(settings))
except oidc.OidcError:
provider_logout = None
auth.destroy_session(db, token)
# `next` exists for one caller: the idle overlay, which sends people to
# /login?timeout=1 so the page explains itself. Only same-site paths are
# honoured — see `_safe_path`, which is also what the trade editor's
# `?return=` uses.
target = _safe_path(next, "/login")
resp = _redirect(provider_logout or target)
resp.delete_cookie(settings.auth.cookie_name, path="/")
return resp
@app.get("/profile", response_class=HTMLResponse)
def account_page(request: Request, saved: str = "", error: str = ""):
"""`saved` names WHAT was saved: the page has two forms and one message
for both told you your password had changed when you picked a colour."""
with scoped(request) as (ctx, db):
return _render(request, ctx, "account.html",
_account_context(request, ctx, db, saved=saved, error=error))
def _account_context(request: Request, ctx, db: DbSession, *, saved: str = "",
error: str = "") -> dict:
return {
"saved": saved if saved in ("password", "appearance", "2fa-off", "oidc",
"session-revoked", "sessions-revoked",
"key-revoked") else "",
# Names the profile control as the active one, the same way a page
# names its nav entry.
"error": error, "active_nav": "account",
"hideable_nav": navigation.hideable_for(ctx, settings),
# Passkeys need a secure context, so the page says whether this
# connection is one rather than claiming they are unavailable.
"secure": auth.is_secure_request(request, settings),
# The theme designer: what may be set, what this person has set,
# what the built-in value is (shown as the placeholder, so an
# empty box still tells you what you would get), and any choice
# that will be hard to see.
"swatches": theming.SWATCHES,
"modes": theming.MODES,
"colors": ctx.user.theme_colors or {},
"defaults": theming.DEFAULTS,
"color_warnings": theming.warnings(ctx.user.theme_colors),
**_twofactor_context(db, ctx.user),
**_passkey_context(request, db, ctx.user),
"identities": [
{"id": row.id, "issuer": row.issuer,
"created": ensure_utc(row.created_at)}
for row in federation.identities_for(db, ctx.user)
],
"oidc_reason": federation.unavailable_reason(settings),
**_oidc_context(),
**_sessions_context(db, ctx),
**_keys_context(db, ctx),
}
def _keys_context(db: DbSession, ctx) -> dict:
"""This person's API keys, each with what it can do in every portfolio it
was given (auth.key_reach). Without the second half, a key that stopped
reaching a portfolio when they left it would look fine here while every
request it made there failed."""
keys = db.scalars(select(ApiKey).where(ApiKey.created_by == ctx.user.id)
.order_by(ApiKey.id.desc())).all()
return {"api_keys": [{"key": k, "reach": auth.key_reach(db, k)} for k in keys],
"write_roles": WRITE_ROLES}
def _describe_agent(ua: str | None) -> str:
"""A user agent as something a person recognises.
Deliberately crude. The goal is "is this the laptop or the phone?", and a
real UA-parsing library would be a dependency, a data file to keep current,
and a much longer answer than the question deserves.
"""
if not ua:
return "Unknown device"
browser = next(
(name for token, name in (
("Edg/", "Edge"), ("OPR/", "Opera"), ("Firefox/", "Firefox"),
("Chrome/", "Chrome"), ("Safari/", "Safari"),
) if token in ua),
"Browser",
)
platform = next(
(name for token, name in (
("iPhone", "iPhone"), ("iPad", "iPad"), ("Android", "Android"),
("Macintosh", "macOS"), ("Windows", "Windows"), ("Linux", "Linux"),
) if token in ua),
None,
)
return f"{browser} on {platform}" if platform else browser
def _sessions_context(db: DbSession, ctx) -> dict:
rows = db.scalars(
select(UserSession)
.where(UserSession.user_id == ctx.user.id)
.order_by(UserSession.last_seen_at.desc())
).all()
return {
"sessions": [
{
"id": row.id,
"device": _describe_agent(row.user_agent),
"ip": row.ip or "—",
"last_seen": ensure_utc(row.last_seen_at),
"started": ensure_utc(row.created_at),
# Compared on the hash, never on anything from the request —
# the cookie itself never needs to be looked at here.
"current": row.token_hash == ctx.session.token_hash,
}
for row in rows
]
}
@app.post("/profile/sessions/{session_id}/revoke")
async def revoke_session(request: Request, session_id: RowId):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
row = db.get(UserSession, session_id)
# Own sessions only. Without the ownership check this would be a way to
# sign anybody out by guessing a small integer.
if row is None or row.user_id != ctx.user.id:
raise HTTPException(404, "no such session")
if row.token_hash == ctx.session.token_hash:
return _redirect("/profile?error=current-session")
db.delete(row)
db.flush()
return _redirect("/profile?saved=session-revoked")
@app.post("/profile/sessions/revoke-others")
async def revoke_other_sessions(request: Request):
"""Sign out everywhere else — the thing you want after losing a laptop."""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
removed = 0
for row in db.scalars(
select(UserSession).where(UserSession.user_id == ctx.user.id)
).all():
if row.token_hash != ctx.session.token_hash:
db.delete(row)
removed += 1
db.flush()
log.info("user %s revoked %d other sessions", ctx.user.id, removed)
return _redirect("/profile?saved=sessions-revoked")
def _enrolment(db: DbSession, user: User) -> dict:
"""The pending secret and its QR, for the enrolment form.
The secret is held on the user row until a code confirms it, so returning
here shows the SAME QR. It used to be generated per render, which meant a
refresh silently invalidated the code somebody had just scanned.
"""
secret = twofactor.begin_enrolment(user)
db.flush()
return {"secret": secret,
"qr": twofactor.qr_svg(
twofactor.provisioning_uri(secret, user.email, branding.NAME))}
def _twofactor_context(db: DbSession, user: User) -> dict:
return {
"totp_enabled": twofactor.is_enabled(user),
"recovery_left": (
twofactor.remaining_recovery_codes(db, user)
if twofactor.is_enabled(user) else 0
),
# The dialog on the profile page needs it up front; None once it is on,
# which is what hides the "Set up" dialog entirely.
"enrol": None if twofactor.is_enabled(user) else _enrolment(db, user),
"new_codes": None, # shown exactly once, right after enabling
}
# --------------------------------------------------------------------------- #
# Two-factor authentication. Enrolment is two requests and the unconfirmed
# secret never reaches the user row — decisions.md #39.
# --------------------------------------------------------------------------- #
@app.get("/profile/2fa", response_class=HTMLResponse)
def twofactor_setup(request: Request, error: str = ""):
with scoped(request) as (ctx, db):
if twofactor.is_enabled(ctx.user):
return _redirect("/profile")
return _render(
request, ctx, "twofactor_setup.html",
{"error": error, "active_nav": "account",
"enrol": _enrolment(db, ctx.user)},
)
@app.post("/profile/2fa/enable")
async def twofactor_enable(request: Request, code: str = Form(...)):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
if twofactor.is_enabled(ctx.user):
return _redirect("/profile")
# The PENDING secret, not one posted with the form: a caller that
# supplied its own would be enrolling an authenticator this account
# never scanned. The code proves the app holds it — skipping that check
# is how people lock themselves out.
secret = ctx.user.totp_secret
if not twofactor.accept_code(db, ctx.user, code):
return _redirect("/profile/2fa?error=" + quote_plus(
"That code isn't right — check the app and try again."))
codes = twofactor.enable(db, ctx.user, secret)
log.info("2FA enabled for user %s", ctx.user.id)
return _render(
request,
ctx,
"account.html",
{"saved": "", "error": None, "active_nav": "",
**_twofactor_context(db, ctx.user),
"new_codes": codes},
)
# --------------------------------------------------------------------------- #
# Passkeys. Two ceremonies, each two requests: the browser asks for options,
# talks to the authenticator, and posts back what it signed. The challenge
# lives in a row between the two — app/passkeys.py.
# --------------------------------------------------------------------------- #
def _passkey_context(request: Request, db: DbSession, user: User) -> dict:
secure = auth.is_secure_request(request, settings)
return {
# Dates normalised here rather than in the template: SQLite hands back
# naive datetimes, and a template cannot tell that from UTC.
"passkeys": [
{"id": row.id, "name": row.name or "Passkey",
"created": ensure_utc(row.created_at),
"last_used": ensure_utc(row.last_used_at) if row.last_used_at else None}
for row in passkeys.for_user(db, user)
],
"passkeys_usable": passkeys.configured(settings) and secure,
# The reason is shown rather than the control hidden: every one of them
# is something the reader can act on.
"passkeys_reason": passkeys.unavailable_reason(settings, secure=secure),
}
@app.get("/profile/passkeys", response_class=HTMLResponse)
def passkeys_page(request: Request, error: str = "", saved: str = ""):
with scoped(request) as (ctx, db):
return _render(request, ctx, "passkeys.html", {
"saved": saved, "error": error or None, "active_nav": "account",
"totp_enabled": twofactor.is_enabled(ctx.user),
**_passkey_context(request, db, ctx.user),
})
@app.post("/profile/passkeys/options")
async def passkeys_options(request: Request):
"""Step one of enrolment: what the browser should ask the authenticator."""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
if not passkeys.configured(settings) or not auth.is_secure_request(
request, settings):
raise HTTPException(400, "Passkeys are not available here")
options, token = passkeys.registration_options(db, ctx.user, settings)
return JSONResponse({"options": json.loads(options), "token": token})
@app.post("/profile/passkeys")
async def passkeys_register(request: Request):
"""Step two: store what the authenticator signed."""
form = await request.form()
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
try:
passkeys.verify_registration(
db, ctx.user, str(form.get("credential") or ""),
str(form.get("token") or ""), settings,
name=str(form.get("name") or ""),
)
except passkeys.PasskeyError as exc:
return _redirect("/profile/passkeys?error=" + quote_plus(exc.message))
log.info("passkey registered for user %s", ctx.user.id)
return _redirect("/profile/passkeys?saved=added")
@app.post("/profile/passkeys/{credential_id}/remove")
async def passkeys_remove(request: Request, credential_id: RowId):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
removed = passkeys.remove(db, ctx.user, credential_id)
if removed:
log.info("passkey removed for user %s", ctx.user.id)
return _redirect("/profile/passkeys?saved=" + ("removed" if removed else ""))
@app.post("/login/passkey/options")
async def login_passkey_options(request: Request):
"""Sign-in options. No email is asked for and none is confirmed — the
ceremony is discoverable, so the authenticator names the account."""
await auth.verify_pre_auth_csrf(request)
if not passkeys.configured(settings) or not auth.is_secure_request(
request, settings):
raise HTTPException(400, "Passkeys are not available here")
with anon() as db:
options, token = passkeys.authentication_options(db, settings)
return JSONResponse({"options": json.loads(options), "token": token})
@app.post("/login/passkey")
async def login_passkey(request: Request):
"""A passkey signs in on its own — no password, and no code step.
That is the point of one rather than a shortcut around the second factor:
the ceremony requires user verification, so it already proves possession of
the authenticator AND a biometric or PIN. decisions.md #113.
"""
await auth.verify_pre_auth_csrf(request)
form = await request.form()
ip = auth.client_ip(request)
with anon() as db:
try:
user = passkeys.verify_authentication(
db, str(form.get("credential") or ""),
str(form.get("token") or ""), settings)
except passkeys.PasskeyError as exc:
auth.record_attempt(db, "", ip, success=False)
return JSONResponse({"error": exc.message}, status_code=400)
landing = _landing_portfolio(db, user, request)
raw = auth.create_session(
db, user, settings, active_portfolio_id=landing,
ip=ip, user_agent=request.headers.get("user-agent"),
)
log.info("passkey sign-in for user %s", user.id)
target = "/profile" if user.must_change_password else "/"
resp = JSONResponse({"next": target})
_set_session_cookie(resp, raw)
_remember_portfolio(resp, landing)
return resp
# --------------------------------------------------------------------------- #
# Signing in through an identity provider. Two requests: out to the provider,
# and back with a code. app/federation.py decides whose account it is.
# --------------------------------------------------------------------------- #
@app.get("/login/oidc")
def login_oidc(request: Request, invite: str = ""):
"""Start a sign-in. `invite` carries an invitation across the round trip."""
with anon() as db:
held = None
if invite:
try:
held = invites.lookup(db, invite)
except invites.InviteError as exc:
return _redirect("/login?error=" + quote_plus(str(exc)))
try:
url, token = federation.begin(db, settings, held)
except (federation.FederationError, oidc.OidcError) as exc:
return _redirect("/login?error=" + quote_plus(str(exc)))
resp = _redirect(url)
_set_oidc_cookie(resp, token)
return resp
@app.get("/login/oidc/callback")
def login_oidc_callback(request: Request, code: str = "", state: str = "",
error: str = ""):
"""Where the provider sends the browser back.
A GET with no CSRF token of its own: `state` is the defence, and it is
checked against the row this attempt created before anything else happens.
"""
if error:
return _redirect("/login?error=" + quote_plus(
f"Your identity provider refused the sign-in: {error}"))
ip = auth.client_ip(request)
with anon() as db:
try:
attempt = federation.take_state(
db, request.cookies.get(OIDC_STATE_COOKIE), state)
pending = {"state": attempt.state, "nonce": attempt.nonce,
"verifier": attempt.verifier}
held = db.get(PortfolioInvite, attempt.invite_id) if attempt.invite_id else None
# Checked when the sign-in started, and again now: withdrawn, used
# or expired while the browser was at the provider, it is no invite
# at all (#116). Only checked then, it still made the account.
if held is not None and invites.spent(held):
held = None
identity = oidc.complete(federation.provider(settings), code=code,
pending=pending)
if attempt.link_user_id is not None:
# Started from the profile page: attach, do not sign in.
user = db.get(User, attempt.link_user_id)
if user is None:
raise federation.FederationError("That account is gone.")
federation.link(db, user, identity)
else:
user = federation.resolve(db, settings, identity, held)
if held is not None:
invites.accept_row(db, held, user)
# Spent only once everything above succeeded: a provider error must
# not burn the attempt and force a second trip.
db.delete(attempt)
db.flush()
except (federation.FederationError, oidc.OidcError) as exc:
# Nothing a refused sign-in started is kept: an account made for an
# identity that then could not be linked would have no way in.
db.rollback()
auth.record_attempt(db, "", ip, success=False)
return _redirect("/login?error=" + quote_plus(str(exc)))
if user.must_change_password:
target = "/profile"
else:
target = "/"
landing = _landing_portfolio(db, user, request)
raw = auth.create_session(
db, user, settings, active_portfolio_id=landing,
ip=ip, user_agent=request.headers.get("user-agent"),
)
# So signing out can end the provider's session too — and only for the
# sessions that started there. `oidc_sid` is what a back-channel logout
# matches on; None when the provider issues no `sid`, which leaves only
# the blunter subject-keyed path.
row = auth.load_session(db, raw, settings)
row.via_oidc = True
row.oidc_sid = federation.session_id(identity)
db.flush()
log.info("oidc sign-in for user %s", user.id)
resp = _redirect(target)
_set_session_cookie(resp, raw)
_remember_portfolio(resp, landing)
resp.delete_cookie(OIDC_STATE_COOKIE, path="/")
return resp
@app.post("/oidc/backchannel-logout")
async def oidc_backchannel_logout(request: Request,
logout_token: str = Form("")):
"""The provider telling us one of its sessions has ended — issue #28.
**Unauthenticated on purpose.** No cookie, no CSRF token, no API key: a
provider has none of those, and the spec is explicit that the signature on
the token is what proves the request came from a legitimate party. That is
also what lets a PUBLIC client use this — there is no client secret to
send. `appcore.oidc.verify_logout_token` is the whole gate, and it refuses
an ID token dressed as a logout among other things.
**Session propagation, not revocation.** This ends sessions and never
touches `is_active`: letting a provider disable an account is a far larger
authority to hand over, and it would break the promise that local accounts
keep working when the provider is down. Somebody with a password signs
straight back in, which is the intended behaviour — decisions.md #127.
Always 200 once the token verifies, whether or not it matched a session.
A provider retrying because we said 404 about a session that had already
expired would be noise about nothing.
"""
if not federation.configured(settings):
# Not "bad request" — there is no such endpoint on an installation
# with no provider, and saying otherwise invites probing.
raise HTTPException(404, "not found")
if not logout_token.strip():
raise HTTPException(400, "logout_token is required")
with anon() as db:
try:
# In a thread: verifying it can fetch the provider's configuration
# and keys, ten seconds' timeout each.
notice = await asyncio.to_thread(
oidc.verify_logout_token, federation.provider(settings), logout_token)
except oidc.OidcError as exc:
log.warning("back-channel logout refused: %s", exc)
raise HTTPException(400, "that logout could not be verified") from exc
ended = federation.end_sessions_for(db, notice)
log.info("back-channel logout from %s ended %s session(s)",
notice.issuer, ended)
return PlainTextResponse("", status_code=200)
@app.post("/profile/oidc/link")
async def profile_oidc_link(request: Request):
"""Link the account already signed in to a provider identity.
Deliberately a signed-in action: it is how an existing local account gains
a provider without email matching ever being involved — decisions.md #119.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
try:
url, token = federation.begin(db, settings, link_to=ctx.user)
except (federation.FederationError, oidc.OidcError) as exc:
return _redirect("/profile?error=" + quote_plus(str(exc)))
resp = _redirect(url)
_set_oidc_cookie(resp, token)
return resp
@app.post("/profile/oidc/{identity_id}/unlink")
async def profile_oidc_unlink(request: Request, identity_id: RowId):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
try:
federation.unlink(db, ctx.user, identity_id)
except federation.FederationError as exc:
return _redirect("/profile?error=" + quote_plus(str(exc)))
return _redirect("/profile?saved=oidc")
@app.post("/profile/recovery/later")
async def account_recovery_later(request: Request):
"""Put the banner away until the next sign-in.
On the session rather than the user: an account with no saved codes is one
forgotten password from being unreachable, so "not now" must not mean
"never". A new session asks again.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
ctx.session.codes_banner_hidden = True
db.flush()
return _redirect(_back_to_referer(request))
@app.post("/profile/recovery")
async def account_recovery_codes(request: Request):
"""Issue a fresh set of recovery codes and show them once.
Always a REGENERATION, never a re-display. They are stored hashed, so the
originals genuinely cannot be shown again — and a page that offered to
"view" them would be promising something it cannot do. The consequence is
real and the page says it: any list already written down stops working.
A POST rather than a GET because it changes state, and because a link that
silently invalidated somebody's printed codes on hover-preview would be a
poor thing to have built.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
codes = twofactor.generate_recovery_codes(db, ctx.user)
ctx.user.recovery_codes_seen_at = dt.datetime.now(dt.timezone.utc)
db.flush()
log.info("recovery codes reissued for user %d", ctx.user.id)
return _render(
request, ctx, "recovery_codes.html",
{"active_nav": "", "codes": codes,
"totp_on": twofactor.is_enabled(ctx.user)},
)
@app.post("/profile/2fa/disable")
async def twofactor_disable(
request: Request, password: str = Form(...), code: str = Form("")
):
"""Turning it off needs the password AND a current code.
Both, deliberately: the password alone would let someone who walked up to
an unlocked screen remove the factor, and a code alone would let anyone
holding the phone do it.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
if not twofactor.is_enabled(ctx.user):
return _redirect("/profile")
if not await asyncio.to_thread(auth.verify_password, ctx.user.password_hash, password):
return _redirect("/profile?error=password")
if not (twofactor.accept_code(db, ctx.user, code)
or twofactor.consume_recovery_code(db, ctx.user, code)):
return _redirect("/profile?error=code")
twofactor.disable(db, ctx.user)
log.info("2FA disabled for user %s", ctx.user.id)
return _redirect("/profile?saved=2fa-off")
@app.post("/users/{user_id}/2fa/clear")
async def users_clear_2fa(request: Request, user_id: RowId):
"""Admin escape hatch for someone who lost their phone AND their codes.
Sits beside the existing password reset because it is the same kind of act:
an administrator vouching for someone out of band. Logged loudly — this
removes a security control from another person's account.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_admin(ctx)
user = db.get(User, user_id)
if user is None:
raise HTTPException(404, "no such user")
twofactor.disable(db, user)
# Any half-authenticated sessions become unusable; full ones would
# survive, and shouldn't — the account's security just changed.
auth.destroy_sessions_for(db, user.id)
log.warning("admin %s cleared 2FA for user %s", ctx.user.id, user.id)
return _redirect("/users?reset=2fa")
@app.post("/profile/password")
async def account_password(
request: Request,
current_password: str = Form(...),
password: str = Form(...),
confirm: str = Form(...),
):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
error = None
if not await asyncio.to_thread(auth.verify_password, ctx.user.password_hash,
current_password):
error = "Current password is incorrect."
elif password != confirm:
error = "The new passwords don't match."
else:
error = auth.password_problem(password)
if error:
return _render(
request, ctx, "account.html",
{"saved": "", "error": error, "active_nav": ""},
)
ctx.user.password_hash = await asyncio.to_thread(auth.hash_password, password)
ctx.user.must_change_password = False
db.flush()
# Every other session for this account is invalidated; keep this one.
this_token = ctx.session.token_hash
auth.destroy_sessions_for(db, ctx.user.id)
raw = auth.create_session(
db, ctx.user, settings, active_portfolio_id=ctx.active_portfolio_id,
ip=auth.client_ip(request), user_agent=request.headers.get("user-agent"),
)
log.info("password changed for user %d (was token %s…)", ctx.user.id, this_token[:8])
resp = _redirect("/profile?saved=password")
_set_session_cookie(resp, raw)
return resp
# --------------------------------------------------------------------------- #
# User administration (admin only)
# --------------------------------------------------------------------------- #
def _kick_feed() -> bool:
"""Nudge the price feed after a new instrument appears, so its history
shows up without waiting for the daily run. True if it dispatched.
Gated on `price_feed.enabled`, which `lifespan` applies to the scheduled
LOOPS only. Without it here, adding an instrument reached a price provider
on an install that had deliberately switched the feed off — and "only
ticker symbols leave the machine, and only if you let them" is the headline
promise (decisions.md #9).
Returns a bool so both sides of that gate are observable: with no running
loop the dispatch is skipped anyway, so a caller cannot otherwise tell
"refused" from "nothing to hand it to".
"""
if not settings.price_feed.enabled or feed_status["running"]:
return False
try:
asyncio.get_running_loop().run_in_executor(None, _run_feed)
return True
except RuntimeError: # no loop (CLI/tests) — the schedule will catch it
return False
# Jurisdictions worth offering. Deliberately short: the financial-year and
# capital-gains logic in `fyreport` is Australia's, so a 250-entry country list
# would imply support that does not exist. Adding a code here is the cheap half
# of supporting a jurisdiction — see T28b for the other half.
# How many lines the console shows on load. Enough to cover the last restart
# and a failed import; small enough that the panel is a window, not the page.
CONSOLE_LINES = 100
COUNTRIES = [
("AU", "Australia"),
("NZ", "New Zealand"),
("GB", "United Kingdom"),
("US", "United States"),
("CA", "Canada"),
("IE", "Ireland"),
]
def _require_admin(ctx):
"""App-wide account administration."""
if not ctx.is_admin:
raise HTTPException(403, "Admins only")
def _require_write(ctx):
"""Anything that changes this portfolio. Viewers are read-only."""
if not ctx.can_write:
raise HTTPException(403, "You have read-only access to this portfolio")
def _invite_url(request: Request, token: str) -> str:
"""The full link to send.
Built from the request, because the external URL is not stored anywhere —
the wizard only uses it to decide `cookie_secure`. Behind a proxy that
means the link is right only when `trusted_proxies` is set, which is the
same condition every other absolute URL here depends on.
"""
if not token:
return ""
return f"{str(request.base_url).rstrip('/')}/invite/{token}"
def _require_owner(ctx):
"""This portfolio's settings, and who is in it: members and invitations.
A key is anyone's to make for themselves."""
if not ctx.is_owner:
raise HTTPException(403, "Only the portfolio owner can do that")
def _sole_memberships(db) -> dict[int, list[str]]:
"""Portfolios each person is the ONLY member of.
Disabling them takes those out of everybody's view: membership is the only
route to a portfolio, so one with no member who can sign in is one nobody
can open — decisions.md #123. Worth saying before the click, not after.
"""
counts = (
select(PortfolioMember.portfolio_id)
.group_by(PortfolioMember.portfolio_id)
.having(func.count() == 1)
.subquery()
)
rows = db.execute(
select(PortfolioMember.user_id, Portfolio.name)
.join(Portfolio, Portfolio.id == PortfolioMember.portfolio_id)
.join(counts, counts.c.portfolio_id == PortfolioMember.portfolio_id)
).all()
out: dict[int, list[str]] = {}
for user_id, name in rows:
out.setdefault(user_id, []).append(name)
return out
def _users_context(db, *, created: str = "", reset: str = "",
temp_password: str = "", error: str | None = None) -> dict:
return {
"active_nav": "admin",
"users": db.scalars(select(User).order_by(User.id)).all(),
# Who signs in through a provider, so an administrator deciding who to
# disable can see which accounts the directory does not control.
"federated": {row.user_id for row in db.scalars(select(ExternalIdentity))},
"sole_of": _sole_memberships(db),
"created": created,
"reset": reset,
"temp_password": temp_password,
"error": error,
}
@app.get("/admin/accounts", response_class=HTMLResponse)
@app.get("/users", response_class=HTMLResponse)
def users_page(request: Request):
with scoped(request) as (ctx, db):
_require_admin(ctx)
return _render(request, ctx, "users.html", _users_context(db))
@app.post("/users/add")
async def users_add(
request: Request,
name: str = Form(...),
email: str = Form(...),
password: str = Form(...),
is_admin: str = Form(""),
):
with scoped(request) as (ctx, db):
_require_admin(ctx)
await auth.verify_csrf(request, db)
email_l = email.strip().lower()
error = (auth.email_problem(email_l) or auth.password_problem(password)
or textfield.problem(name, User.name, "Name"))
if not error and db.scalar(select(User).where(User.email == email_l)):
error = "That email already has an account."
if error:
return _render(request, ctx, "users.html", _users_context(db, error=error))
new_user = User(
email=email_l,
name=_name_or_email(name, email_l),
password_hash=await asyncio.to_thread(auth.hash_password, password),
is_admin=bool(is_admin),
# They set their own on first login; this one is only a handover.
must_change_password=True,
)
db.add(new_user)
db.flush()
# Everyone starts with a portfolio of their own; sharing is opt-in from
# the Members page of whichever portfolio wants them.
own = Portfolio(name=f"{new_user.name}'s portfolio"[:80])
db.add(own)
db.flush()
db.add(PortfolioMember(portfolio_id=own.id, user_id=new_user.id, role="owner"))
# Issued now, but deliberately NOT shown: these are the new user's
# recovery codes, and an administrator is not who they belong to.
# `recovery_codes_seen_at` stays NULL, so the banner asks them to
# generate a set of their own the first time they sign in.
twofactor.ensure_recovery_codes(db, new_user)
db.flush()
# Rendered, not redirected: a temporary password in a query string
# would be written to the access log, the browser history and every
# proxy on the way. Refreshing re-submits and fails harmlessly with
# "that email already has an account".
return _render(request, ctx, "users.html", _users_context(
db, created=email_l, temp_password=password))
@app.post("/users/{user_id}/active")
async def users_toggle_active(request: Request, user_id: RowId):
with scoped(request) as (ctx, db):
_require_admin(ctx)
await auth.verify_csrf(request, db)
user = db.get(User, user_id)
if user is None:
raise HTTPException(404, "no such user")
if user.id == ctx.user.id:
raise HTTPException(400, "you can't deactivate yourself")
user.is_active = not user.is_active
if not user.is_active:
auth.destroy_sessions_for(db, user.id)
return _redirect("/users")
@app.post("/users/{user_id}/reset")
async def users_reset(request: Request, user_id: RowId, password: str = Form(...)):
with scoped(request) as (ctx, db):
_require_admin(ctx)
await auth.verify_csrf(request, db)
error = auth.password_problem(password)
if error:
raise HTTPException(400, error)
user = db.get(User, user_id)
if user is None:
raise HTTPException(404, "no such user")
user.password_hash = await asyncio.to_thread(auth.hash_password, password)
user.must_change_password = True
auth.destroy_sessions_for(db, user.id)
db.flush()
return _render(request, ctx, "users.html", _users_context(
db, reset=user.email, temp_password=password))
# --------------------------------------------------------------------------- #
# Application settings (admin only)
# --------------------------------------------------------------------------- #
def _settings_context(saved: str = "", error: str = "", pending: list[str] | None = None) -> dict:
"""Everything the settings page renders from.
Values come from the LIVE settings object rather than the file, so the page
shows what the application is actually using — an environment variable
beats the file, and a page showing the file would quietly disagree with
reality.
"""
# The console's opening window. Rendered server-side so it is there with
# scripting off; `console.js` tails from the last seq after that.
values = {}
for option in configfile.OPTIONS:
value = configfile.shown(settings, option)
# Lists reach the template as lists and are joined there; a date has to
# be a string for `<input type="date">` to select it.
values[option.name] = value.isoformat() if isinstance(value, dt.date) else value
db = settings.database
return {
"active_nav": "admin",
"options": configfile.OPTIONS,
"values": values,
"writable": configfile.writability(),
"config_path": str(configfile.config_path()),
"from_environment": {
o.name: f"APP_{'__'.join(p.upper() for p in o.path)}"
for o in configfile.OPTIONS
if configfile.overridden_by_environment(o)
},
"fixed": {
"app_name": settings.app_name,
"database": (f"sqlite at {db.path}" if db.type == "sqlite"
else f"postgres {db.name} on {db.host}"),
"feed_timezone": settings.price_feed.timezone,
"brokers": ", ".join(sorted(settings.imports.brokers_available())),
},
"saved": saved,
"pending": pending or [],
"error": error,
"supervision": lifecycle.supervision(),
# The console's opening window. Rendered server-side so the panel is
# useful with scripting off; console.js tails from the last seq.
"log_lines": logbuffer.recent(CONSOLE_LINES),
}
@app.get("/admin/settings", response_class=HTMLResponse)
def settings_page(request: Request, saved: str = ""):
with scoped(request) as (ctx, db):
_require_admin(ctx)
return _render(request, ctx, "settings.html", _settings_context(saved=saved))
@app.post("/admin/settings")
async def settings_save(request: Request):
"""Write the submitted values back to config.yaml.
Refused outright when the file is read-only: the page hides the controls in
that case, so reaching here means the request was made some other way.
"""
with scoped(request) as (ctx, db):
_require_admin(ctx)
await auth.verify_csrf(request, db)
state = configfile.writability()
if not state.writable:
raise HTTPException(409, state.reason)
form = await request.form()
values, problems = {}, []
for option in configfile.OPTIONS:
raw = form.get(option.name)
# An unchecked checkbox submits nothing at all; every other field
# that is genuinely absent is left alone rather than blanked.
if raw is None and option.kind != "bool":
continue
try:
values[option.name] = configfile.coerce(option, raw if isinstance(raw, str) else "")
except ValueError as exc:
problems.append(str(exc))
if problems:
return _render(request, ctx, "settings.html",
_settings_context(error=" ".join(problems)))
configfile.save(values)
pending = configfile.apply_live(settings, values)
log.info("settings updated by user %d: %s", ctx.user.id, ", ".join(sorted(values)))
if form.get("restart"):
lifecycle.request_stop(f"settings saved by user {ctx.user.id}")
return _render(request, ctx, "restarting.html",
{"active_nav": "admin", "saved": True,
"supervision": lifecycle.supervision()})
return _render(request, ctx, "settings.html",
_settings_context(saved="1", pending=pending))
@app.post("/admin/settings/restart")
async def settings_restart(request: Request):
"""Stop the application, and let whatever started it start it again.
Offered whether or not the settings are writable — a restart is useful on
its own (to pick up a ConfigMap changed from outside, or to clear a wedged
price feed), and a read-only config file is exactly the deployment where
changes arrive from elsewhere and need one.
"""
with scoped(request) as (ctx, db):
_require_admin(ctx)
await auth.verify_csrf(request, db)
lifecycle.request_stop(f"requested by user {ctx.user.id}")
return _render(request, ctx, "restarting.html",
{"active_nav": "admin", "saved": False,
"supervision": lifecycle.supervision()})
# --------------------------------------------------------------------------- #
# Portfolios: switching, members, API keys
# --------------------------------------------------------------------------- #
@app.post("/portfolio/switch")
async def portfolio_switch(request: Request, portfolio_id: Annotated[RowId, Form()]):
"""Change which portfolio this session acts in.
The id is validated against the user's own memberships — a forged form
value can't move the session into someone else's portfolio.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
if portfolio_id not in {m.portfolio_id for m in ctx.memberships}:
raise HTTPException(403, "you're not a member of that portfolio")
ctx.session.active_portfolio_id = portfolio_id
db.flush()
resp = _redirect(_back_to_referer(request))
_remember_portfolio(resp, portfolio_id)
return resp
@app.get("/admin/logs.json")
def admin_logs(request: Request, after: int = 0):
"""Log lines newer than `after`, for the Settings page console to tail.
`after=0` means "the initial window"; the page then sends the highest seq
it has and gets only what arrived since. That is what lets it keep the
lines it was first shown instead of replacing them on every poll.
"""
with scoped(request) as (ctx, _db):
_require_admin(ctx)
lines = logbuffer.since(after) if after else logbuffer.recent(CONSOLE_LINES)
return JSONResponse({"lines": [
{"seq": line.seq, "when": line.when, "level": line.level,
"logger": line.logger, "message": line.message}
for line in lines
]})
_FEE_COLUMNS = ("brokerage_flat", "brokerage_percent", "brokerage_minimum",
"foreign_brokerage_flat", "foreign_brokerage_percent",
"foreign_brokerage_minimum")
def _portfolio_settings(typed: dict) -> tuple[str, str, str, dict]:
"""The settings form's name, currency, country, fees and tax rates,
checked; a 400 saying which is wrong otherwise."""
cleaned = typed["name"].strip()
if not cleaned:
raise HTTPException(400, "A portfolio needs a name.")
currency = typed["reporting_currency"].strip().upper()
if currency not in money.SYMBOLS:
raise HTTPException(400, "That currency isn't one this app knows.")
place = typed["jurisdiction"].strip().upper()
if len(place) != 2 or not place.isalpha():
raise HTTPException(400, "Country has to be a two-letter code.")
fees = {}
for column in _FEE_COLUMNS:
text = typed[column]
# Blank is "not set", which is not the same as free.
if not text.strip():
fees[column] = None
continue
percent = column.endswith("percent")
label = "Percentage" if percent else "Brokerage"
try:
figure = money.parse(text, getattr(Portfolio, column), label)
except money.FigureError as exc:
raise HTTPException(400, f"{exc}.") from None
except ValueError:
raise HTTPException(400, f"{label} must be a number.") from None
if figure < 0:
raise HTTPException(400, f"{label} can't be negative.")
if percent and figure > 100:
raise HTTPException(400, "Percentage can't be more than 100.")
fees[column] = figure
for column, label in _TAX_RATES:
fees[column] = _tax_rate(typed[column], label)
fees["income_support"] = bool(typed["income_support"])
return cleaned, currency, place, fees
_TAX_RATES = (("income_tax_rate", "Income tax rate"), ("medicare_levy", "Medicare levy"))
def _tax_rate(text: str, label: str) -> Decimal | None:
"""A tax rate as typed, in percent. Blank is not set."""
if not text.strip():
return None
try:
figure = money.parse(text, Portfolio.income_tax_rate, label)
except money.FigureError as exc:
raise HTTPException(400, f"{exc}.") from None
if figure < 0:
raise HTTPException(400, f"{label} can't be negative.")
if figure > 100:
raise HTTPException(400, f"{label} can't be more than 100.")
return figure
@app.post("/portfolio/settings")
async def portfolio_settings(
request: Request,
name: str = Form(...),
reporting_currency: str = Form("AUD"),
jurisdiction: str = Form("AU"),
brokerage_flat: str = Form(""),
brokerage_percent: str = Form(""),
brokerage_minimum: str = Form(""),
foreign_brokerage_flat: str = Form(""),
foreign_brokerage_percent: str = Form(""),
foreign_brokerage_minimum: str = Form(""),
income_tax_rate: str = Form(""),
medicare_levy: str = Form(""),
income_support: str = Form(""),
):
"""Rename this portfolio, set the currency and jurisdiction it reports in,
what its broker charges (app/brokerage.py), and the tax rates the
sale simulator uses (decisions.md #147), and whether the minimum tax
applies (#151).
**Both are stored and only partly wired.** Totals are still computed in
`money.REPORTING` and the financial-year logic is still AU's — T28b is the
task that unpicks the ~110 places that assume it. The page says so rather
than implying a change here re-denominates anything, because a setting that
silently does nothing is worse than no setting.
The field is `jurisdiction` in the code and **"Country" on the page**. That
is deliberate: what the column decides is which tax rules and which words
apply, which follows tax residency — but "country" is the word a person
recognises for it. See `Portfolio.jurisdiction`.
"""
typed = {"name": name, "reporting_currency": reporting_currency,
"jurisdiction": jurisdiction,
"brokerage_flat": brokerage_flat, "brokerage_percent": brokerage_percent,
"brokerage_minimum": brokerage_minimum,
"foreign_brokerage_flat": foreign_brokerage_flat,
"foreign_brokerage_percent": foreign_brokerage_percent,
"foreign_brokerage_minimum": foreign_brokerage_minimum,
"income_tax_rate": income_tax_rate, "medicare_levy": medicare_levy,
"income_support": income_support}
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_owner(ctx)
try:
cleaned, currency, place, fees = _portfolio_settings(typed)
except HTTPException as exc:
# The page again, holding what was typed, with the reason on it.
return _members(request, ctx, db, error=exc.detail, typed=typed,
status_code=exc.status_code)
portfolio = db.get(Portfolio, ctx.active_portfolio_id)
portfolio.name = cleaned[:80]
portfolio.reporting_currency = currency
portfolio.jurisdiction = place
for column, figure in fees.items():
setattr(portfolio, column, figure)
db.commit()
return _redirect("/members")
@app.post("/portfolio/new")
async def portfolio_new(request: Request, name: str = Form(...)):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
label = name.strip()[:80]
if not label:
raise HTTPException(400, "name is required")
portfolio = Portfolio(name=label)
db.add(portfolio)
db.flush()
db.add(
PortfolioMember(
portfolio_id=portfolio.id, user_id=ctx.user.id, role="owner"
)
)
ctx.session.active_portfolio_id = portfolio.id
db.flush()
resp = _redirect("/members")
_remember_portfolio(resp, portfolio.id)
return resp
@app.get("/admin", response_class=HTMLResponse)
@app.get("/members", response_class=HTMLResponse)
def members_page(request: Request, error: str = ""):
with scoped(request) as (ctx, db):
return _members(request, ctx, db, error=error)
def _members(request: Request, ctx, db: DbSession, *, error: str = "",
typed: dict | None = None, status_code: int = 200,
new_invite: str = "") -> HTMLResponse:
"""The portfolio's admin page. `typed` is a refused settings form's own
values, shown in place of the stored ones; `new_invite` the token just
issued, the one time it can be shown."""
rows = db.execute(
select(PortfolioMember, User)
.join(User, User.id == PortfolioMember.user_id)
.where(PortfolioMember.portfolio_id == ctx.active_portfolio_id)
.order_by(PortfolioMember.role, User.name)
).all()
# Only people with an account can be added; accounts are created by an
# app admin on /users.
member_ids = {u.id for (_m, u) in rows}
candidates = [
u
for u in db.scalars(select(User).where(User.is_active.is_(True)).order_by(User.name))
if u.id not in member_ids
]
return _render(
request,
ctx,
"members.html",
{
# The portfolio's own settings, editable by an admin.
"portfolio": db.get(Portfolio, ctx.active_portfolio_id),
"currencies": sorted(money.SYMBOLS),
"countries": COUNTRIES,
"active_nav": "admin",
"rows": rows,
"candidates": candidates,
"roles": MEMBER_ROLES,
"role_blurbs": ROLE_BLURBS,
"invitations": [
{"id": i.id, "role": i.role, "state": invites.state(i),
"created": ensure_utc(i.created_at)}
for i in invites.outstanding(db, ctx.active_portfolio_id)
],
# Shown once, straight after issuing: it is a credential, and
# the database keeps only its hash.
"new_invite": _invite_url(request, new_invite),
"error": error,
"typed": typed,
},
status_code=status_code,
)
@app.post("/members/add")
async def members_add(request: Request, user_id: Annotated[RowId, Form()], role: str = Form("member")):
with scoped(request) as (ctx, db):
_require_owner(ctx)
await auth.verify_csrf(request, db)
if role not in MEMBER_ROLES:
raise HTTPException(400, "unknown role")
if db.get(User, user_id) is None:
raise HTTPException(404, "no such account")
existing = db.scalar(
select(PortfolioMember).where(
PortfolioMember.portfolio_id == ctx.active_portfolio_id,
PortfolioMember.user_id == user_id,
)
)
if existing is not None:
return _redirect("/members?error=Already+a+member")
db.add(
PortfolioMember(
portfolio_id=ctx.active_portfolio_id, user_id=user_id, role=role
)
)
db.flush()
return _redirect("/members")
@app.post("/members/invite")
async def members_invite(request: Request, role: str = Form("member")):
"""Issue a link. Shown once, in the response, because it is a credential."""
with scoped(request) as (ctx, db):
_require_owner(ctx)
await auth.verify_csrf(request, db)
try:
token = invites.create(db, portfolio_id=ctx.active_portfolio_id,
role=role, created_by=ctx.user.id)
except invites.InviteError as exc:
return _redirect("/members?error=" + quote_plus(str(exc)))
log.info("invite issued for portfolio %s by user %s",
ctx.active_portfolio_id, ctx.user.id)
# In THIS response, never an address: in a redirect's query string the
# link went into browser history and the proxy's access log (#128).
return _members(request, ctx, db, new_invite=token)
@app.post("/members/invite/{invite_id}/revoke")
async def members_invite_revoke(request: Request, invite_id: RowId):
with scoped(request) as (ctx, db):
_require_owner(ctx)
await auth.verify_csrf(request, db)
invites.revoke(db, invite_id=invite_id,
portfolio_id=ctx.active_portfolio_id)
return _redirect("/members")
# --------------------------------------------------------------------------- #
# Accepting an invitation. Public, because the whole point is somebody who has
# no account yet — the token in the URL is the only authorisation, and it is
# checked on every one of these.
# --------------------------------------------------------------------------- #
async def _confirm_problem(request: Request, password: str) -> str | None:
"""A mismatch with the confirm field, when the form has one — the account
forms do. Read from the raw form because an empty field would otherwise
arrive as "not sent", and an empty confirm is a mismatch."""
form = await request.form()
if "confirm" in form and form.get("confirm") != password:
return "The passwords don't match."
return None
def _invite_page(request: Request, token: str, *, error: str = "",
ctx=None, typed: dict | None = None) -> HTMLResponse:
with anon() as db:
try:
invite = invites.lookup(db, token)
except invites.InviteError as exc:
return templates.TemplateResponse(
request, "invite.html",
{"auth": None, "csrf": auth.ensure_pre_auth_csrf(request),
"refused": str(exc), "invite": None, "token": token},
status_code=410,
)
portfolio = db.get(Portfolio, invite.portfolio_id)
detail = {"portfolio": portfolio.name if portfolio else "a portfolio",
"role": invite.role}
csrf = ctx.session.csrf_token if ctx else auth.ensure_pre_auth_csrf(request)
resp = templates.TemplateResponse(
request, "invite.html",
{"auth": ctx, "csrf": csrf, "refused": None, "invite": detail,
"token": token, "error": error or None, "typed": typed, **_oidc_context()},
)
if ctx is None:
_set_pre_auth_csrf(resp, csrf)
return resp
@app.get("/invite/{token}", response_class=HTMLResponse)
def invite_page(request: Request, token: str):
with anon() as db:
ctx = auth.load_auth(request, db)
return _invite_page(request, token, ctx=ctx)
@app.post("/invite/{token}/accept")
async def invite_accept(request: Request, token: str):
"""Join with the account already signed in."""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
try:
invite = invites.accept(db, token, ctx.user)
except invites.InviteError as exc:
return _invite_page(request, token, error=str(exc), ctx=ctx)
ctx.session.active_portfolio_id = invite.portfolio_id
db.flush()
log.info("invite accepted by existing user %s", ctx.user.id)
joined = invite.portfolio_id
resp = _redirect("/")
_remember_portfolio(resp, joined)
return resp
@app.post("/invite/{token}/signup")
async def invite_signup(
request: Request,
token: str,
name: str = Form(...),
email: str = Form(...),
password: str = Form(...),
):
"""Create an account and join, in one step.
The invite is the whole of the authorisation: nobody may sign up here
without one, which is what stops a public instance growing accounts.
"""
await auth.verify_pre_auth_csrf(request)
problem = (auth.password_problem(password) or await _confirm_problem(request, password)
or auth.email_problem(email) or textfield.problem(name, User.name, "Name"))
if problem:
return _invite_page(request, token, error=problem,
typed={"name": name, "email": email})
with anon() as db:
try:
invites.lookup(db, token)
except invites.InviteError as exc:
return _invite_page(request, token, error=str(exc))
email_l = email.strip().lower()
if db.scalar(select(User).where(User.email == email_l)) is not None:
return _invite_page(request, token, error=(
"That email already has an account. Sign in, then open this "
"link again."))
user = User(email=email_l, name=_name_or_email(name, email_l),
password_hash=await asyncio.to_thread(auth.hash_password, password))
db.add(user)
db.flush()
# Issued but NOT shown: the banner asks them to save a set of their own
# on first sign-in, exactly as for an admin-created account.
twofactor.ensure_recovery_codes(db, user)
invite = invites.accept(db, token, user)
raw = auth.create_session(db, user, settings,
active_portfolio_id=invite.portfolio_id,
ip=auth.client_ip(request),
user_agent=request.headers.get("user-agent"))
log.info("invite accepted by new user %s", user.id)
resp = _redirect("/")
_set_session_cookie(resp, raw)
return resp
@app.post("/members/{member_id}/role")
async def members_role(request: Request, member_id: RowId, role: str = Form(...)):
with scoped(request) as (ctx, db):
_require_owner(ctx)
await auth.verify_csrf(request, db)
if role not in MEMBER_ROLES:
raise HTTPException(400, "unknown role")
member = db.get(PortfolioMember, member_id)
if member is None or member.portfolio_id != ctx.active_portfolio_id:
raise HTTPException(404, "no such member")
if member.user_id == ctx.user.id and role != "owner":
# Losing your own ownership could orphan the portfolio.
return _redirect("/members?error=Hand+ownership+over+first")
member.role = role
db.flush()
return _redirect("/members")
@app.post("/members/{member_id}/remove")
async def members_remove(request: Request, member_id: RowId,
delete_portfolio: str = Form("")):
"""Remove somebody, or — when they are the last one — delete the portfolio.
The two are one action because they are one decision. Removing the only
member would leave a portfolio nobody can reach: membership is the only
route to one, so it would still hold every trade and be invisible to
everybody, admins included. Offering the deletion here is what stops that
state being reachable by accident — decisions.md #123.
"""
with scoped(request) as (ctx, db):
_require_owner(ctx)
await auth.verify_csrf(request, db)
member = db.get(PortfolioMember, member_id)
if member is None or member.portfolio_id != ctx.active_portfolio_id:
raise HTTPException(404, "no such member")
counts = db.execute(
select(func.count(),
func.count().filter(PortfolioMember.role == "owner"))
.select_from(PortfolioMember)
.where(PortfolioMember.portfolio_id == ctx.active_portfolio_id)
).one()
members, owners = counts
if members <= 1:
if delete_portfolio.lower() not in ("1", "true", "on", "yes"):
return _redirect("/members?error=A+portfolio+needs+an+owner")
portfolio = db.get(Portfolio, ctx.active_portfolio_id)
log.info("portfolio %s deleted by user %s",
ctx.active_portfolio_id, ctx.user.id)
# Everything it holds goes with it: trades, dividends, plans,
# charts, keys, invites. The session is left alone — `load_auth`
# already falls back when its active portfolio is not one of your
# memberships, which is the same code path as being removed from
# one, and it runs on every request rather than only this one.
db.delete(portfolio)
db.flush()
return _redirect("/")
if member.role == "owner" and owners <= 1:
return _redirect("/members?error=A+portfolio+needs+an+owner")
db.delete(member)
db.flush()
return _redirect("/members")
@app.post("/keys/new")
async def keys_new(request: Request, name: str = Form(...), scopes: str = Form("read"),
portfolios: Annotated[list[RowId], Form()] = []):
"""Issue an API key for yourself, reaching portfolios you choose from your own.
Anyone with access may, at any role: a key never does more in a portfolio
than its creator can do there now (auth.key_reach), and a viewer can
already read and export everything a read key would show.
The raw value is shown once, in THIS response. It used to travel in a
redirect's query string, which put a live key in browser history and in
the proxy's access log.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
if scopes not in ("read", "read,write"):
raise HTTPException(400, "scopes must be 'read' or 'read,write'")
mine = {m.portfolio_id for m in ctx.memberships}
chosen = sorted(set(portfolios))
if not set(chosen) <= mine:
raise HTTPException(400, "a key can only reach portfolios you belong to")
if not chosen:
return _render(request, ctx, "account.html", _account_context(
request, ctx, db, error="Choose at least one portfolio for the key."))
raw, key_hash, prefix = auth.new_api_key()
db.add(ApiKey(
name=name.strip()[:80] or "unnamed",
key_hash=key_hash,
prefix=prefix,
scopes=scopes,
created_by=ctx.user.id,
portfolios=list(db.scalars(select(Portfolio).where(Portfolio.id.in_(chosen)))),
))
db.flush()
log.info("api key %s issued by user %d for portfolios %s",
prefix, ctx.user.id, chosen)
return _render(request, ctx, "account.html",
{**_account_context(request, ctx, db), "new_key": raw})
@app.post("/keys/{key_id}/revoke")
async def keys_revoke(request: Request, key_id: RowId):
"""Your own keys only. Somebody else's stops when their access does."""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
key = db.get(ApiKey, key_id)
if key is None or key.created_by != ctx.user.id:
raise HTTPException(404, "no such key")
key.revoked_at = dt.datetime.now(dt.timezone.utc)
db.flush()
return _redirect("/profile?saved=key-revoked#keys")
# Only a query string, and only characters a query string is made of. The
# chooser posts the page's current query back so that saving columns does not
# throw away the sort you were reading under — and this is what stops that
# convenience from becoming an open redirect. No scheme, no host, no path: the
# value can only ever be appended to "/".
_QUERY_ONLY = re.compile(r"^\?[A-Za-z0-9_=&%.,+:~-]*$")
def _safe_query(back: str | None) -> str:
return back if back and _QUERY_ONLY.match(back) else ""
# What a back button says, for the paths one can point at. The label has to be
# decided WITH the path and not in the template: `?return=` is an arbitrary
# path, and a template turning one into a name is guessing. Anything not named
# here falls back to something the caller knows — the ticker, usually.
BACK_LABELS = navigation.BACK_LABELS
def _back_label(path: str, fallback: str) -> str:
"""The name of the page `path` leads to."""
# A financial year is the one destination whose name is in the path rather
# than in the table — there is one per year and they are generated.
if path.startswith("/?fy="):
year = path.removeprefix("/?fy=")
if re.fullmatch(r"[0-9]{4}", year):
return fyreport.fy_label(int(year))
return navigation.back_label(path, fallback)
# Both live in `navigation` so the imports router can reach them too; a guard
# that is copied is a guard that will eventually be copied slightly wrong.
_safe_path = navigation.safe_path
def _name_or_email(name: str, email: str) -> str:
"""The name somebody typed, or their email when they left it blank.
The fallback is a value the app picks, not text a person typed, so it is cut
to the column (an email holds 320 characters, `User.name` 120). Typed text
is refused instead, by `textfield`: decisions.md #131.
"""
return (name.strip() or email.strip())[:User.name.type.length]
def _back_to_referer(request: Request, fallback: str = "/") -> str:
"""Where the request came from, or `fallback`.
The PATH of the referer, never the whole header: it is client-supplied, and
an absolute URL somebody else chose would be an open redirect. The path goes
through `safe_path` because a referer of `https://host//evil.test` has the
path `//evil.test`, which a browser follows off this site.
"""
path = urllib.parse.urlparse(request.headers.get("referer") or "").path
return _safe_path(path, fallback)
@app.post("/profile/columns")
async def save_columns(request: Request):
"""Which holdings columns this person wants.
A plain form post rather than anything cleverer: the page has to re-render
to show the new table anyway, so a fetch would buy nothing but a code path
that works differently with JavaScript off.
New columns are appended rather than slotted into declaration order: the
order is the user's, and re-sorting it here would undo every move
they had made the moment they ticked one more box.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
form = await request.form()
wanted = columns.clean(form.getlist("column"))
existing = [k for k in (ctx.user.dashboard_columns or []) if k in wanted]
ctx.user.dashboard_columns = existing + [k for k in wanted if k not in existing]
db.flush()
return _redirect("/" + _safe_query(form.get("back")))
@app.post("/profile/columns/order")
async def order_columns(request: Request):
"""The order the columns are rendered in.
One route for two interactions, deliberately. The form carries the whole
current order as hidden inputs; the ↑/↓ buttons add `move=<key>:<direction>`
and the server does the swap, while the drag enhancement rewrites the hidden
inputs and submits with no `move` at all. Same code path either way, so the
keyboard route and the mouse route cannot drift — and the keyboard one works
with no JavaScript, which is the requirement drag cannot meet on its own.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
form = await request.form()
order = columns.clean(form.getlist("order"))
move = form.get("move") or ""
if ":" in move:
key, _, direction = move.partition(":")
order = columns.move(order, key, direction)
ctx.user.dashboard_columns = order
db.flush()
return _redirect("/" + _safe_query(form.get("back")))
@app.post("/profile/columns/reset")
async def reset_columns(request: Request):
"""Back to the defaults, and back to *tracking* them — storing NULL rather
than a copy of the current list means a later change to what the defaults
are reaches this person too."""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
ctx.user.dashboard_columns = None
db.flush()
return _redirect("/")
@app.post("/profile/colors")
async def account_colors(request: Request):
"""Save this person's colour overrides.
A blank field means "back to the default", which is stored as an ABSENT
key rather than an empty string — so the stylesheet's own value shows
through and a later change to it still reaches them. An all-blank form
therefore stores NULL, which is the same state a fresh account is in.
Validation lives in `theming.clean`: anything that is not a hex colour is
dropped rather than saved, because a value the `<input type="color">`
beside it cannot round-trip is worse than being told no.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
chosen = theming.clean(await request.form())
ctx.user.theme_colors = chosen or None
db.commit()
return _redirect("/profile")
@app.post("/profile/colors/reset")
async def account_colors_reset(request: Request):
"""Back to the built-in palette.
Stores NULL, not a copy of the defaults — the same distinction the columns
reset makes. A stored copy would freeze today's colours into the account
and quietly opt them out of every future change to them.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
ctx.user.theme_colors = None
db.commit()
return _redirect("/profile")
@app.post("/profile/appearance")
async def account_appearance(
request: Request,
theme: str = Form("auto"),
nav_style: str = Form("both"),
times_in: str = Form("market"),
):
"""Theme, menu style and which nav entries to show.
**Accent is not set here** — the Theme colours section does it, to any
colour, and two controls for one value is how they end up disagreeing.
`user.accent` is left alone rather than reset, so an account keeps its
choice until it picks a colour, and the column still drives the per-mode
accent defaults.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
if (theme not in THEMES or nav_style not in NAV_STYLES
or times_in not in TIME_ZONES_SHOWN):
raise HTTPException(400, "unknown appearance option")
ctx.user.theme = theme
ctx.user.nav_style = nav_style
ctx.user.times_in = times_in
form = await request.form()
# Only offered with more than one portfolio, so only changed when sent.
if "default_portfolio" in form:
wanted = str(form.get("default_portfolio") or "")
chosen = row_id(wanted)
if not wanted:
ctx.user.default_portfolio_id = None # the last one opened
elif chosen in {m.portfolio_id for m in ctx.memberships}:
ctx.user.default_portfolio_id = chosen
else:
raise HTTPException(400, "not one of your portfolios")
# Checkboxes say what to SHOW; what gets stored is what to hide. An
# unticked box submits nothing, so "shown" cannot be read from absence —
# and storing the hidden set means a nav entry added to the app later
# appears for everybody rather than staying invisible until they find
# this page.
shown = set(form.getlist("nav_show"))
hidden = [key for key in navigation.HIDEABLE if key not in shown]
ctx.user.nav_hidden = hidden or None
db.flush()
return _redirect("/profile?saved=appearance")
# --------------------------------------------------------------------------- #
# Dashboard / charts / calendar
# --------------------------------------------------------------------------- #
def _one_currency(holdings) -> str | None:
"""The single currency these holdings trade in, or None if there are
several. None is not "unknown" — it is "this column cannot name one", and
the header says `(native)` instead."""
found = {h.instrument.currency for h in holdings}
return found.pop() if len(found) == 1 else None
def _instrument_options(db, instruments):
"""What each option in the instrument picker needs to carry.
The currency, because it decides whether the FX field applies and that
should not cost a round trip. NOT a price or a rate: those used to be here
as each instrument's latest close, which is wrong for any trade not dated
today. `/holdings/price` answers for the form's date — decisions.md #124.
"""
return [{
"id": inst.id,
"ticker": inst.ticker,
"exchange": inst.exchange,
"name": inst.name,
"asset_class": inst.asset_class,
"currency": inst.currency,
} for inst in instruments]
def _brokerage_context(db, ctx) -> dict:
"""What the trade and plan forms need to suggest a brokerage: each kind's
fee if the portfolio set one, and each kind's last trade for when it did
not. The value a form starts at is the local fee on nothing — its flat
part, or its minimum — or else the last local trade's."""
portfolio = db.get(Portfolio, ctx.active_portfolio_id)
local, foreign = brokerage.local(portfolio), brokerage.foreign(portfolio)
last_local = queries.last_brokerage(db)
last_foreign = queries.last_brokerage(db, foreign=True)
def cents(v):
return f"{v:.2f}" if v is not None else None
start = brokerage.for_trade(portfolio, money.REPORTING, Decimal(0), last_local)
return {
"brokerage_fees": {"local": local.as_json(), "foreign": foreign.as_json(),
"last": {"local": cents(last_local),
"foreign": cents(last_foreign)}},
"initial_brokerage": cents(start) or "",
}
def _dca_prefill(db):
"""Today's scheduled buy, as form values — or None.
Skipped entirely when the DCA schedule feature is switched off: an install
that turned it off should not get its trade form quietly filled in from a
schedule it does not have.
"""
if not features.enabled(settings, "dca_schedule"):
return None
return plans.prefill(db)
@app.get("/", response_class=HTMLResponse)
def dashboard(request: Request, fy: int | None = None):
"""Today's position, or a past financial year.
The FY tabs replace the separate Reports page: a financial year is another
view of the same portfolio, not a different feature.
"""
with scoped(request) as (ctx, db):
fys = fyreport.available_fys(db)
query = dict(request.query_params)
common = {
"active_nav": "portfolio",
"fys": fys,
"fy_labels": {y: fyreport.fy_label(y) for y in fys},
"fy": fy,
"feed": feed_status,
"latest_price_date": db.execute(select(func.max(Price.date))).scalar(),
# Whether that latest figure is a live price for a session still
# running, rather than a close. The label has to say which: "prices
# to today" reads as settled, and quietly meaning something else is
# how a number stops being trustworthy.
# What the chooser posts back so that saving columns keeps whatever
# sort you were reading under. `_safe_query` refuses anything that
# is not purely a query string.
"back": ("?" + str(request.url.query)) if request.url.query else "",
}
if fy is not None:
if fy not in fys:
raise HTTPException(404, "no such financial year")
report = fyreport.fy_report(db, fy)
# Three tables, three independent sorts. A shared `?sort=` would
# order all of them at once. `fyreport` owns both the column names
# and where the rows live, because it owns the row shapes.
sorts = {
table: sorting.read(query, table, keys)
for table, keys in fyreport.SORTABLE.items()
}
fyreport.sort_tables(report, sorts)
return _render(request, ctx, "dashboard_fy.html",
{**common, "r": report, "sorts": sorts,
# Only the snapshot's Price is in an instrument's
# own currency; everything else on this page is a
# tax figure and therefore in the reporting one.
# A dict, not an object — Jinja's `r.snapshot`
# hides that, Python does not.
"native_ccy": _one_currency(report["snapshot"])})
holdings = queries.all_holdings(db)
open_positions, _closed = queries.split_positions(holdings)
cols = columns.chosen(ctx.user.dashboard_columns)
sort = sorting.read(query, "holdings", columns.sortable_keys())
if sort.key:
# The allow-list is every column that EXISTS, not every column
# showing: otherwise a bookmarked link would work or not depending
# on whose preferences were loaded.
open_positions = sorting.apply(
open_positions, columns.sort_key(columns.BY_KEY[sort.key]),
descending=sort.descending,
)
return _render(
request,
ctx,
"dashboard.html",
{
**common,
"totals": queries.totals(open_positions),
# Built from the positions being shown, not from the instrument
# table — see `_live_exchanges` for the zero-unit BTC row that
# kept this lit all night.
"live_exchanges": _live_exchanges(open_positions),
"open_positions": open_positions,
# Nothing has ever been recorded here, which is a different
# state from "everything has been sold" — one wants an
# invitation, the other wants five zeros and a closed-position
# history. `holdings` is every instrument this portfolio has,
# so an empty list is the genuinely-new case.
"first_run": not holdings,
# The Record trade dialog renders the same include `/trade/new`
# does, so it needs the same two things. Cheap: one indexed
# query, and the dialog is on every dashboard load anyway.
"instruments": queries.portfolio_instruments(db),
"today": clock.today().isoformat(),
# The performance chart, below the year tabs. Built from the
# same daily series every other chart uses, so it cannot
# disagree with the charts page about how the portfolio did.
"summary": charts_build.summary_chart(
db, clock.today(),
fyreport.fy_bounds(fyreport.current_fy(clock.today()))[0]),
"prefill": _dca_prefill(db),
"options": _instrument_options(db, queries.portfolio_instruments(db)),
**_brokerage_context(db, ctx),
"cols": cols,
# The one currency these holdings trade in, or None when they
# trade in several — which is what lets a native-currency
# column name its currency in the header instead of repeating
# a symbol down every cell. See `Column.heading`.
"native_ccy": _one_currency(open_positions),
# True when every figure in the table is in one currency, so
# the table names it once instead of seven headers each
# repeating it.
"one_ccy": columns.one_currency_everywhere(
_one_currency(open_positions)),
"col_groups": columns.chooser_groups(
columns.one_currency_everywhere(_one_currency(open_positions)),
{c.key for c in cols}),
"chosen_keys": {c.key for c in cols},
"chosen_order": [c.key for c in cols],
"sort": sort,
},
)
@app.post("/refresh")
async def refresh(request: Request):
"""Manual price-feed run, for when the scheduled one missed (pod restart,
Yahoo hiccup). Fire-and-forget: the dashboard polls itself while running."""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
if not feed_status["running"]:
asyncio.get_running_loop().run_in_executor(None, _run_feed)
return _redirect(_back_to_referer(request))
@app.get("/charts", response_class=HTMLResponse)
def charts(request: Request):
with scoped(request) as (ctx, db):
_seed_charts(db, ctx)
saved = db.scalars(
select(SavedChart).order_by(SavedChart.position, SavedChart.id)
).all()
drawn, broken = [], []
for chart in saved:
try:
drawn.append(
{
"id": chart.id,
"name": chart.name,
"width": chart.width,
"data": charts_build.run(db, chart.spec),
}
)
except ValueError as exc:
# A spec the vocabulary no longer supports: say so on the card
# rather than dropping the chart without explanation.
broken.append({"id": chart.id, "name": chart.name, "why": str(exc)})
return _render(
request,
ctx,
"charts.html",
{
"active_nav": "charts",
"charts": drawn,
"broken": broken,
"chart_json": {c["id"]: c["data"] for c in drawn},
# Holdings left out of every AUD figure on this page because no
# exchange rate is stored for their currency. Said once at the
# top rather than per card: it applies to all of them equally,
# and the same disclosure is on the dashboard.
"excluded": queries.cached_portfolio_series(db).get("excluded", []),
},
)
@app.get("/simulate", response_class=HTMLResponse)
def simulate_page(request: Request):
"""What selling some of each holding today would bring home after tax
(decisions.md #148). It records nothing, so it is a GET whose query is the
form: anyone who can see the portfolio can use it, and a refusal comes back
holding what was typed."""
with scoped(request) as (ctx, db):
portfolio = db.get(Portfolio, ctx.active_portfolio_id)
holdings = queries.split_positions(queries.all_holdings(db))[0]
typed = dict(request.query_params)
for column, _label in _TAX_RATES:
if column not in typed:
saved = getattr(portfolio, column)
typed[column] = "" if saved is None else f"{saved.normalize():f}"
error, status = "", 200
try:
rates, units, fees = _sale_asked(typed, holdings)
except HTTPException as exc:
(rates, units, fees), error, status = ({}, {}, {}), exc.detail, exc.status_code
sale = simulate.sale(db, units, brokerage=fees, holdings=holdings, **rates)
return _render(request, ctx, "simulate.html", {
"active_nav": "portfolio",
"sale": sale,
"typed": typed,
"error": error,
"fy_label": fyreport.fy_label(fyreport.current_fy(clock.today())),
}, status_code=status)
def _sale_asked(typed: dict, holdings) -> tuple[dict, dict, dict]:
"""The Simulate sale form's tax rates, units and brokerage, checked; a 400
saying what is wrong otherwise. Fill and Sell all write the units they work
out into `typed`, so the form shows them."""
rates = {column: _tax_rate(typed.get(column, ""), label) for column, label in _TAX_RATES}
if typed.get("fill") in ("all", "percent"):
percent = Decimal(100)
if typed["fill"] == "percent":
percent = _sale_figure(typed.get("percent", "") or "0",
Portfolio.income_tax_rate, "Percentage")
if not 0 < percent <= 100:
raise HTTPException(400, "Sell a percentage above 0 and up to 100.")
for inst_id, figure in simulate.fill(holdings, percent).items():
typed[f"units_{inst_id}"] = f"{figure.normalize():f}"
units, fees = {}, {}
for holding in holdings:
inst = holding.instrument
text = typed.get(f"units_{inst.id}", "").strip()
if text:
figure = _sale_figure(text, Trade.quantity, inst.ticker)
if figure < 0:
raise HTTPException(400, f"{inst.ticker}: units can't be negative.")
if figure > holding.units:
raise HTTPException(400, f"{inst.ticker}: {figure.normalize():f} is more than "
f"the {holding.units.normalize():f} units held.")
units[inst.id] = figure
text = typed.get(f"brokerage_{inst.id}", "").strip()
if text:
figure = _sale_figure(text, Trade.brokerage, f"{inst.ticker} brokerage")
if figure < 0:
raise HTTPException(400, f"{inst.ticker} brokerage can't be negative.")
fees[inst.id] = figure
return rates, units, fees
def _sale_figure(text: str, column, name: str) -> Decimal:
try:
return money.parse(text, column, name)
except money.FigureError as exc:
raise HTTPException(400, f"{exc}.") from None
def _seed_charts(db: DbSession, ctx) -> None:
"""Lay out the default charts for someone who has never seen the page.
Guarded by `charts_seeded` rather than "has no charts", so deleting the lot
is a decision that sticks instead of being undone on the next page load.
"""
if ctx.user.charts_seeded:
return
for position, key in enumerate(chart_templates.DEFAULT_KEYS):
template = chart_templates.BY_KEY[key]
db.add(
tenancy.owned(
db,
SavedChart(
name=template["name"],
spec=dict(template["spec"]),
template_key=key,
position=position,
width=template["width"],
),
)
)
ctx.user.charts_seeded = True
db.flush()
log.info("seeded %d default charts for user %d", len(chart_templates.DEFAULT_KEYS), ctx.user.id)
@app.post("/charts/order")
async def charts_order(request: Request):
"""Persist the grid order after a drag. Ids not owned by this user are
ignored rather than trusted."""
body = await request.json()
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
mine = {c.id: c for c in db.scalars(select(SavedChart)).all()}
for position, raw in enumerate(body.get("order") or []):
try:
chart = mine.get(int(raw))
except (TypeError, ValueError):
continue # not an id we issued; ignore rather than 500
if chart is not None:
chart.position = position
db.flush()
return {"ok": True}
@app.post("/charts/{chart_id}/move")
async def chart_move(request: Request, chart_id: RowId, direction: str = Form(...)):
"""One place earlier or later. Dragging is mouse-only: HTML drag fires for
neither a finger nor a keyboard (decisions.md #101), and this is also the
way with scripting off. chartgrid.js does the same move in place."""
step = {"earlier": -1, "later": 1}.get(direction)
if step is None:
raise HTTPException(400, "A chart moves earlier or later.")
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
charts = list(db.scalars(select(SavedChart)
.order_by(SavedChart.position, SavedChart.id)))
at = next((i for i, c in enumerate(charts) if c.id == chart_id), None)
if at is None:
raise HTTPException(404, "no such chart")
if 0 <= at + step < len(charts):
charts[at], charts[at + step] = charts[at + step], charts[at]
for position, chart in enumerate(charts):
chart.position = position
return _redirect("/charts")
# --------------------------------------------------------------------------- #
# Chart builder
# --------------------------------------------------------------------------- #
@app.get("/charts/build", response_class=HTMLResponse)
def chart_builder(request: Request, edit: RowId | None = None):
"""Drag fields onto shelves, see the chart as you go."""
with scoped(request) as (ctx, db):
existing = None
if edit:
existing = db.get(SavedChart, edit)
# Charts belong to the USER and follow them between portfolios, so
# ownership is the user id — checking the portfolio instead would
# 404 someone's own chart while they had another portfolio active.
if existing is None or existing.user_id != ctx.user.id:
raise HTTPException(404, "no such chart")
return _render(
request,
ctx,
"chart_builder.html",
{
"active_nav": "charts",
"catalogue": fields.catalogue(),
"existing": existing,
"existing_spec": existing.spec if existing else None,
"filter_values": _filter_values(db),
"templates": chart_templates.TEMPLATES,
"templates_json": {
t["key"]: {"name": t["name"], "spec": t["spec"], "width": t["width"]}
for t in chart_templates.TEMPLATES
},
},
)
def _filter_values(db: DbSession) -> dict:
"""What the filter shelf can offer — drawn from this portfolio, so nobody
is invited to filter on a ticker they don't hold."""
held = {h.instrument for h in queries.cached_holdings(db) if h.instrument.trades}
return {
"ticker": sorted({i.ticker for i in held}),
"asset_class": sorted({i.asset_class for i in held}),
"currency": sorted({i.currency for i in held}),
"exchange": sorted({i.exchange for i in held}),
}
@app.post("/charts/preview")
async def chart_preview(request: Request):
"""Data for a spec that hasn't been saved — the live preview.
Same engine as a saved chart, so what you see while dragging is what you
get afterwards.
"""
spec = await request.json()
with scoped(request) as (ctx, db):
# Checked though it writes nothing: SameSite is per site, so a page on
# another port of this host could post here, and the body is read as
# JSON whatever its type says. Each one runs the whole chart engine.
await auth.verify_csrf(request, db)
try:
# In a thread, as the charts page runs it: this is the whole
# engine, asked again every time the builder is dragged.
return await asyncio.to_thread(charts_build.run, db, spec)
except ValueError as exc:
raise HTTPException(400, str(exc))
class _ChartIn(BaseModel):
"""What the chart builder posts. Read by hand before this, so a body of the
wrong shape reached `.strip()` and `int()` and was a 500."""
name: str = ""
spec: dict = Field(default_factory=dict)
width: str = "half"
id: RowId | None = None
template_key: str | None = None
@app.post("/charts/save")
async def chart_save(request: Request):
# No write check: a chart is its owner's, not the portfolio's, so a viewer
# keeping a page of their own changes nothing of the portfolio's.
with scoped(request) as (ctx, db):
try:
body = _ChartIn.model_validate(await request.json())
except ValueError: # not JSON, or not the builder's shape
raise HTTPException(400, "That is not a chart.") from None
await auth.verify_csrf(request, db)
spec = body.spec
name = body.name.strip()[:80]
if not name:
raise HTTPException(400, "give the chart a name")
problem = fields.validate(spec)
if problem:
raise HTTPException(400, problem)
# Which default it came from: one of ours or none, never free text.
if body.template_key is not None and body.template_key not in chart_templates.BY_KEY:
raise HTTPException(400, "That is not one of the chart templates.")
width = "full" if body.width == "full" else "half"
if body.id:
chart = db.get(SavedChart, body.id)
if chart is None or chart.user_id != ctx.user.id:
raise HTTPException(404, "no such chart")
chart.name, chart.spec, chart.width = name, spec, width
else:
last = db.scalar(select(func.max(SavedChart.position))) or 0
chart = tenancy.owned(
db,
SavedChart(
name=name,
spec=spec,
width=width,
position=last + 1,
template_key=body.template_key,
),
)
db.add(chart)
db.flush()
return {"id": chart.id, "name": chart.name}
@app.post("/charts/{chart_id}/delete")
async def chart_delete(request: Request, chart_id: RowId):
with scoped(request) as (ctx, db): # the owner's, as chart_save
await auth.verify_csrf(request, db)
chart = db.get(SavedChart, chart_id)
if chart is None or chart.user_id != ctx.user.id:
raise HTTPException(404, "no such chart")
db.delete(chart)
db.flush()
return _redirect("/charts")
@app.get("/calendar")
def calendar_legacy_redirect(year: int | None = None, month: int | None = None):
"""The calendar lives on /schedule now (merged 2026-08-01)."""
suffix = f"?year={year}&month={month}" if year and month else ""
return _redirect(f"/schedule{suffix}")
@app.get("/reports")
@app.get("/reports/fy")
@app.get("/fy")
def fy_legacy_redirect(year: int | None = None):
"""FY reporting moved onto the portfolio page as tabs (2026-08-01)."""
return _redirect(f"/?fy={year}" if year else "/")
# --------------------------------------------------------------------------- #
# Investment plan (the rotation formerly known as DCA)
# --------------------------------------------------------------------------- #
@app.get("/dca")
def dca_legacy_redirect():
return _redirect("/schedule")
@app.get("/schedule", response_class=HTMLResponse)
def plan_page(
request: Request,
year: Annotated[int, Field(ge=dt.MINYEAR, le=dt.MAXYEAR)] | None = None,
month: int | None = None,
edit: int = 0,
):
"""Calendar and the rotation on one page — what's coming, then what drives it."""
today = clock.today()
year = year or today.year
month = month or today.month
if not 1 <= month <= 12:
raise HTTPException(400, "month must be 1-12")
with scoped(request) as (ctx, db):
# 24 rather than 6: these are the Preview plan's rows, and a preview
# you cannot scroll is a list. Roughly two years of a monthly plan.
sched = plans.schedule(db, upcoming=24)
instruments = sorted(queries.portfolio_instruments(db), key=lambda i: i.ticker)
return _render(
request,
ctx,
"plan.html",
{
"active_nav": "plan",
"sched": sched,
"plan": sched["plan"],
"instruments": instruments,
**_brokerage_context(db, ctx),
# Only needed when there is no plan to show; cheap either way.
"suggested_rotation": plans.suggested_rotation(db),
"edit": bool(edit),
**calendarview.month_grid(db, year, month),
},
)
@app.get("/feed/status")
def feed_status_json(request: Request):
"""Whether the price feed is running, for the page to watch.
Lets the page ask, and reload once, when there is something new — rather
than reloading on a blind timer whether or not anything has finished.
"""
with scoped(request) as (_ctx, _db):
return JSONResponse({
"running": bool(feed_status["running"]),
"ok": feed_status["ok"],
"finished": feed_status["finished"].isoformat()
if feed_status["finished"] else None,
# The live-price poll writes new numbers into the database every few
# minutes, and a page rendered before that has no way to know. This
# is what an open dashboard compares against what it was rendered
# with — without it the only way to see a fresh price was to reload
# by hand, which makes the refresh look broken when it is working.
"quoted": feed_status["quoted"].isoformat()
if feed_status["quoted"] else None,
})
# How far ahead a plan may start. A start in 9999 was saved happily and then
# made /schedule fail on every load, overflowing the date arithmetic that
# draws it: the page where the plan would be fixed was the one that broke.
PLAN_START_YEARS_AHEAD = 10
def _plan_start(raw: str) -> tuple[dt.date | None, str | None]:
"""A plan's start date from the form (None when blank), and the problem to
show instead when it is not one a plan can have.
A problem string rather than an exception, like the other `*_problem`
checks: the message is ours, and nothing raised flows into a response.
"""
if not raw.strip():
return None, None
try:
start = dt.date.fromisoformat(raw.strip())
except ValueError:
return None, "start date must be YYYY-MM-DD"
if start.year > clock.today().year + PLAN_START_YEARS_AHEAD:
return None, f"A plan has to start within the next {PLAN_START_YEARS_AHEAD} years."
return start, None
def _unknown_tickers(unknown) -> str:
names = sorted(set(unknown))
return f"Not in this portfolio: {', '.join(names)}"
def _rotation(db: DbSession, tickers: str) -> tuple[list[Instrument], list[str], list[Instrument]]:
"""The holdings a rotation names, in order; the names that are no holding;
and, for a bare ticker held on two exchanges, its listings. Such a ticker
names its listing, as BHP:NYSE."""
active = queries.portfolio_instruments(db)
chosen, unknown = [], []
for token in (t.strip().upper() for t in tickers.replace("\n", ",").split(",")):
if not token:
continue
code, _, exchange = token.partition(":")
found = [i for i in active if i.ticker == code and (not exchange or i.exchange == exchange)]
if len(found) > 1:
return [], [], found
if found:
chosen.append(found[0])
else:
unknown.append(token)
return chosen, unknown, []
@app.post("/schedule/preview")
async def plan_preview(
request: Request,
interval_days: int = Form(...),
start_date: str = Form(""),
tickers: str = Form(""),
):
"""The dates a DRAFT rotation would produce. Saves nothing.
Live, so rearranging the rotation shows what it produces without having to
save first. Live by asking the SERVER rather than projecting dates in
JavaScript: `plans.project` is the same arithmetic the real schedule uses,
so the preview cannot drift from it.
A POST because it carries a draft and must not be cached or bookmarked —
not because it writes. Nothing here touches the database except to read
where the rotation currently stands.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
if interval_days < 1 or interval_days > 365:
return JSONResponse({"rows": [], "why": "interval must be 1–365 days"})
start, problem = _plan_start(start_date)
if problem:
return JSONResponse({"rows": [], "why": problem})
# What save would refuse, the preview says, rather than drawing dates
# for a plan that cannot be saved.
chosen, unknown, twice = _rotation(db, tickers)
if unknown:
return JSONResponse(
{"rows": [], "why": _unknown_tickers(unknown)})
if twice:
return JSONResponse({"rows": [], "why": queries.held_twice(twice)})
ambiguous = queries.ambiguous_tickers(db)
rows = plans.preview(db, [(i.id, queries.listing_ref(i.ticker, i.exchange, ambiguous))
for i in chosen], interval_days, start)
return JSONResponse(
{"rows": [{"date": due.isoformat(), "ticker": ticker} for due, ticker in rows],
"why": ""})
@app.post("/schedule/save")
async def plan_save(
request: Request,
name: str = Form(...),
interval_days: int = Form(...),
amount: str = Form(""),
brokerage: str = Form(""),
start_date: str = Form(""),
tickers: str = Form(""),
):
"""Create or update the plan, including its whole rotation.
The rotation arrives as an ordered comma/newline separated ticker list —
the same shape as the config.yaml it replaces, and easy to paste.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
if interval_days < 1 or interval_days > 365:
raise HTTPException(400, "interval must be between 1 and 365 days")
try:
amount_val = (money.parse(amount, InvestmentPlan.amount, "Amount")
if amount.strip() else None)
brokerage_val = money.parse(brokerage or "0", InvestmentPlan.brokerage,
"Brokerage")
except money.FigureError as exc:
raise HTTPException(400, str(exc))
start, problem = _plan_start(start_date)
if problem:
raise HTTPException(400, problem)
chosen, unknown, twice = _rotation(db, tickers)
if unknown:
raise HTTPException(
400, f"{_unknown_tickers(unknown)}. Add {'it' if len(set(unknown)) == 1 else 'them'} first."
)
if twice:
raise HTTPException(400, queries.held_twice(twice))
try:
plan_name = textfield.fit(name, InvestmentPlan.name, "Name") or "My plan"
except textfield.TextError as exc:
raise HTTPException(400, str(exc))
plan = plans.active_plan(db)
if plan is None:
plan = tenancy.owned(db, InvestmentPlan(name=plan_name))
db.add(plan)
db.flush()
plan.name = plan_name
plan.interval_days = interval_days
plan.amount = amount_val
plan.brokerage = brokerage_val
plan.start_date = start
plans.set_entries(db, plan, [i.id for i in chosen])
return _redirect("/schedule")
def _verify_next(db: DbSession, ticker: str, due_date: str) -> plans.UpcomingBuy:
"""Guard: the submitted entry must still be the computed next (double
submit, or another buy recorded since the form was opened)."""
nxt = plans.next_buy(db)
if nxt is None:
raise HTTPException(404, "no rotation configured")
if nxt.ticker != ticker or nxt.due_date.isoformat() != due_date:
raise HTTPException(409, f"next buy is now {nxt.ticker} ({nxt.due_date}) — reload")
return nxt
@app.get("/schedule/complete", response_class=HTMLResponse)
def plan_complete_form(request: Request):
with scoped(request) as (ctx, db):
_require_write(ctx) # as /trade/new: a form there is no saving
nxt = plans.next_buy(db)
if nxt is None:
return _redirect("/schedule")
inst = db.get(Instrument, nxt.instrument_id) if nxt.instrument_id else None
latest = queries._latest_price(db, inst.id) if inst else None
plan = plans.active_plan(db)
return _render(
request,
ctx,
"plan_complete.html",
{
"active_nav": "plan",
"nxt": nxt,
"price": latest[0] if latest else None,
"today": clock.today().isoformat(),
"brokerage": plan.brokerage if plan else (queries.last_brokerage(db) or ""),
},
)
@app.post("/schedule/complete")
async def plan_complete(
request: Request,
ticker: str = Form(...),
due_date: str = Form(...),
trade_date: str = Form(...),
quantity: str = Form(...),
unit_price: str = Form(...),
brokerage: str = Form(""),
):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
try:
date = dt.date.fromisoformat(trade_date)
qty = money.parse(quantity, Trade.quantity, "Units")
price = money.parse(unit_price, Trade.unit_price, "Price")
brk = money.parse(brokerage or "0", Trade.brokerage, "Brokerage")
except money.FigureError as exc:
raise HTTPException(400, str(exc))
except ValueError:
raise HTTPException(400, "bad date/quantity/price/brokerage")
if qty <= 0 or price < 0: # see the trade form: zero is a real price
raise HTTPException(400, "quantity must be positive, price not negative")
nxt = _verify_next(db, ticker, due_date)
inst = db.get(Instrument, nxt.instrument_id)
if inst is None:
raise HTTPException(400, f"instrument {nxt.ticker} not found")
trade = tenancy.owned(
db,
Trade(
instrument_id=inst.id,
date=date,
type="buy",
quantity=qty,
unit_price=price,
brokerage=brk,
fx_rate=Decimal(1) if inst.currency == "AUD" else None,
note=f"planned buy (scheduled {nxt.due_date})",
),
)
db.add(trade)
db.flush() # trade needs its id before we can link it
db.add(
tenancy.owned(
db,
PlannedPurchase(
due_date=nxt.due_date,
instrument_id=inst.id,
plan_entry_id=nxt.entry_id,
status="done",
trade_id=trade.id,
),
)
)
return _redirect(_holding_page(db, inst))
@app.post("/schedule/skip")
async def plan_skip(request: Request, ticker: str = Form(...), due_date: str = Form(...)):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
nxt = _verify_next(db, ticker, due_date)
db.add(
tenancy.owned(
db,
PlannedPurchase(
due_date=nxt.due_date,
instrument_id=nxt.instrument_id,
plan_entry_id=nxt.entry_id,
status="skipped",
),
)
)
return _redirect("/schedule")
@app.post("/schedule/delete")
async def plan_delete(request: Request):
"""Remove the plan and its rotation, keeping what was already bought.
`investment_plan_entry` is `delete-orphan` off the plan, so the rotation
goes with it. `planned_purchase.plan_entry_id` is `ON DELETE SET NULL`, so
the record of what was bought or skipped survives with its slot nulled —
the purchase happened whatever became of the schedule that suggested it,
and tidying a plan away must not rewrite the ledger's history.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
plan = db.scalars(select(InvestmentPlan)).first()
if plan is None:
raise HTTPException(404, "no plan to delete")
db.delete(plan)
db.flush()
log.info("investment plan %s deleted", plan.id)
return _redirect("/schedule")
# --------------------------------------------------------------------------- #
# Exports
# --------------------------------------------------------------------------- #
@app.get("/export")
def export_legacy_redirect():
"""Exports live on the imports page now (merged 2026-08-01)."""
return _redirect("/imports-exports#exports")
def _filename(stem: str, fy: int | None, ticker: str | None, ext: str) -> str:
parts = ["portfolio", stem]
if ticker:
parts.append(ticker.lower())
if fy:
parts.append(f"fy{fy}")
parts.append(clock.today().isoformat())
return "-".join(parts) + "." + ext
@app.get("/export/download")
def export_download(
request: Request,
report: str = "transactions",
fmt: str = "csv",
fy: str = "",
ticker: str = "",
):
"""One report as CSV, or every selected report as sheets in one workbook.
`fy` arrives as a string because the form's "All years" option submits an
empty value; typing it as int|None made FastAPI reject every download with
a parse error before the handler ran.
"""
if fmt not in ("csv", "xlsx"):
raise HTTPException(400, "format must be csv or xlsx")
year: int | None = None
if fy.strip():
# Four ASCII digits: isdigit() alone takes 5000 of them, which int()
# then refuses, and superscript digits, which it cannot parse at all.
if not re.fullmatch(r"[0-9]{4}", fy.strip()):
raise HTTPException(400, "financial year must be a year, e.g. 2026")
year = int(fy)
wanted = [report] if report != "all" else list(exports.TITLES)
unknown = [r for r in wanted if r not in exports.TITLES]
if unknown:
raise HTTPException(400, f"Unknown {'report' if len(unknown) == 1 else 'reports'}: "
f"{', '.join(unknown)}.")
if report == "all" and fmt == "csv":
raise HTTPException(400, "choose Excel for the combined workbook")
# The filter becomes part of the download's filename, in a header, so it
# has to be a ticker: anything else matches nothing anyway.
if ticker.strip() and ticker_problem(ticker):
raise HTTPException(400, "that is not a ticker")
ticker = ticker.strip()
with scoped(request) as (ctx, db):
try:
sheets = exports.build(db, wanted, fy=year, ticker=ticker.upper() or None)
except fyreport.OversoldError as err:
raise HTTPException(409, f"{err.sale.sentence()} Correct it or record the "
"missing buy, then export again.") from None
stem = "everything" if report == "all" else report
if fmt == "csv":
headers, rows = next(iter(sheets.values()))
body = exports.to_csv(headers, rows)
media = "text/csv; charset=utf-8"
else:
body = exports.to_xlsx(sheets, note=exports.ABOUT)
media = (
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
)
name = _filename(stem, year, ticker or None, fmt)
log.info("export %s (%s) by user %d", stem, fmt, ctx.user.id)
return Response(
content=body,
media_type=media,
headers={"content-disposition": f'attachment; filename="{name}"'},
)
# --------------------------------------------------------------------------- #
# Ad-hoc trades (anything outside the plan — the individual shares, one-offs)
# --------------------------------------------------------------------------- #
def _parse_time(set_time: str, trade_time: str) -> dt.time | None:
"""The optional intra-day time, or MARKET_OPEN when the box isn't ticked.
Returns None only when the box IS ticked and what was typed isn't a time,
which the callers turn into a validation message.
"""
if not set_time:
return MARKET_OPEN
try:
return dt.time.fromisoformat(trade_time)
except ValueError:
return None
def _trade_figures(type: str, trade_date: str, quantity: str, unit_price: str, brokerage: str,
fx_rate: str, note: str, set_time: str, trade_time: str) -> tuple:
"""The trade form's fields, checked: (date, units, price, brokerage, rate,
note, time), or the problem to show instead. The add and the edit form
both use it, so neither accepts what the other refuses."""
if type not in ("buy", "sell"):
return None, "Choose buy or sell."
try:
date = dt.date.fromisoformat(trade_date)
qty = money.parse(quantity, Trade.quantity, "Units")
price = money.parse(unit_price, Trade.unit_price, "Price")
brk = money.parse(brokerage or "0", Trade.brokerage, "Brokerage")
fx = (money.parse(fx_rate, Trade.fx_rate, "FX rate")
if fx_rate.strip() else None)
except money.FigureError as exc:
return None, f"{exc}."
except ValueError:
return None, "Date, units, price, brokerage and FX must be numbers (date as YYYY-MM-DD)."
if qty <= 0:
return None, "Units must be greater than zero."
# Zero is a real price: a bonus issue, a demerger allocation, an
# employer's reward-plan grant. Negative is not.
if price < 0:
return None, "Price can't be negative."
if brk < 0 or (fx is not None and fx <= 0):
return None, "Brokerage can't be negative and FX must be positive."
try:
note_text = textfield.fit(note, Trade.note, "Note")
except textfield.TextError as exc:
return None, f"{exc}."
if date > clock.today():
return None, "That date is in the future."
when = _parse_time(set_time, trade_time)
if when is None:
return None, "That time isn't a time — use HH:MM, like 14:30."
return (date, qty, price, brk, fx, note_text, when), None
def _held_trades(db: DbSession, instrument_id: int) -> list[Trade]:
return list(db.scalars(select(Trade).where(Trade.instrument_id == instrument_id)).all())
def _held_actions(db: DbSession, instrument_id: int) -> list[CorporateAction]:
return list(db.scalars(
select(CorporateAction).where(CorporateAction.instrument_id == instrument_id)).all())
def _breach(
db: DbSession, inst: Instrument, candidate: Trade | None, *, exclude_ids=frozenset(),
actions: list[CorporateAction] | None = None,
) -> str | None:
"""Validate the WHOLE timeline for this instrument, and say what breaks.
Every write path goes through here — create, edit, delete and the API —
because each of them can strand a trade that isn't the one being touched.
Cutting an old buy's quantity, moving its date later, or deleting it
outright all break a sell that was fine a moment ago.
"""
found = queries.balance_breach(
_held_trades(db, inst.id), candidate, exclude_ids=exclude_ids,
actions=_held_actions(db, inst.id) if actions is None else actions,
)
return None if found is None else queries.breach_message(inst.ticker, found)
@app.get("/trade/new", response_class=HTMLResponse)
def trade_form(request: Request, ticker: str = "", error: str = "", exchange: str = ""):
with scoped(request) as (ctx, db):
# A form there is no saving: refused before it is filled in, not after.
_require_write(ctx)
instruments = queries.portfolio_instruments(db)
return _render(
request,
ctx,
"trade_new.html",
{
"active_nav": "trade",
"instruments": instruments,
"ticker": ticker.upper(),
"exchange": exchange.upper(),
"today": clock.today().isoformat(),
"prefill": _dca_prefill(db),
"options": _instrument_options(db, instruments),
**_brokerage_context(db, ctx),
"error": error,
"form": {},
"edit": None,
# Reached from the dashboard's Record trade button and from a
# holding's ledger. The `?ticker=` in the URL is what tells the
# two apart — state, not a guess at the referrer, which is why
# the button can name where it lands.
"return_to": (queries.holding_path(ticker.upper(), exchange.upper(),
queries.ambiguous_tickers(db))
if ticker else "/"),
"back_label": ticker.upper() if ticker else "Portfolio",
},
)
@app.post("/trade/new")
async def trade_create(
request: Request,
instrument_id: str = Form(...),
type: str = Form(...),
trade_date: str = Form(...),
set_time: str = Form(""),
trade_time: str = Form(""),
quantity: str = Form(...),
unit_price: str = Form(...),
brokerage: str = Form("0"),
fx_rate: str = Form(""),
note: str = Form(""),
# Only used when instrument_id == "new" (added inline from this page).
new_ticker: str = Form(""),
new_name: str = Form(""),
new_exchange: str = Form("ASX"),
new_asset_class: str = Form(""),
new_currency: str = Form("AUD"),
new_yahoo: str = Form(""),
new_drp: str = Form(""),
submission: str = Form(""),
):
# One trade per form, however many times Save is pressed: a repeat gets
# the first one's answer. app/submissions.py.
owned = submission if submissions.valid(submission) else None
if owned:
answer = await submissions.claim(owned)
if answer is not None:
return _redirect(answer)
try:
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
instruments = queries.portfolio_instruments(db)
def _reject(message: str):
return _render(
request,
ctx,
"trade_new.html",
{
"instruments": instruments,
"options": _instrument_options(db, instruments),
**_brokerage_context(db, ctx),
"ticker": "",
"today": clock.today().isoformat(),
"error": message,
"edit": None,
# A rejected submission has no ticker to go back to — the
# instrument may be the thing that was wrong.
"return_to": "/",
"back_label": "Portfolio",
"form": {
"instrument_id": instrument_id,
"type": type,
"trade_date": trade_date,
"set_time": set_time,
"trade_time": trade_time,
"quantity": quantity,
"unit_price": unit_price,
"brokerage": brokerage,
"fx_rate": fx_rate,
"note": note,
"new_ticker": new_ticker,
"new_name": new_name,
"new_exchange": new_exchange,
"new_asset_class": new_asset_class,
"new_currency": new_currency,
"new_yahoo": new_yahoo,
"new_drp": new_drp,
},
},
)
figures, problem = _trade_figures(type, trade_date, quantity, unit_price, brokerage,
fx_rate, note, set_time, trade_time)
if problem:
return _reject(problem)
date, qty, price, brk, fx, note_text, when = figures
# "new" means the fields below the picker were filled in: create the
# instrument and record the trade in one go, rather than sending someone
# to another page and losing what they typed. Nothing is written until
# the trade is accepted, so a refused one adds no instrument either.
adding = instrument_id == "new"
if adding:
ticker = new_ticker.strip().upper()
typed_currency = new_currency.strip().upper()
problem = (ticker_problem(ticker) or exchange_problem(new_exchange or "ASX")
or (typed_currency and currency_problem(typed_currency))
or textfield.problem(new_name, Instrument.name, "Name")
or textfield.problem(new_yahoo, Instrument.yahoo_symbol,
"Yahoo symbol"))
if problem:
return _reject(f"{problem}." if problem[-1] not in ".?" else problem)
exchange = (new_exchange or "ASX").strip().upper()
inst = db.scalar(
select(Instrument).where(
Instrument.exchange == exchange, Instrument.ticker == ticker
)
)
if inst is not None and inst in instruments:
# Listed already. Reusing it here dropped what was typed
# for the "new" one, its class above all, with nothing
# said; the form switches to it before getting this far.
return _reject(f"{ticker} is already recorded. Pick it from the list.")
if inst is None:
if new_asset_class not in ("etf", "share", "crypto"):
return _reject("Pick an asset class for the new instrument.")
# No call to the provider here: the form asked it as the
# ticker was typed and carries the answer, and a save that
# waited on it again was slow enough to be pressed twice.
inst = Instrument(
ticker=ticker,
exchange=exchange,
name=new_name.strip() or None,
currency=typed_currency or "AUD",
asset_class=new_asset_class,
drp=bool(new_drp),
active=True,
yahoo_symbol=new_yahoo.strip()
or pricefeed.yahoo_symbol_for(ticker, exchange),
)
else:
try:
wanted = int(instrument_id)
except ValueError:
wanted = 0
# Parsed here rather than typed as RowId because "new" is valid too,
# so the bounds RowId would apply are applied by hand.
inst = db.get(Instrument, wanted) if 1 <= wanted <= ROW_ID_MAX else None
if inst not in instruments:
inst = None # the catalogue's, not this portfolio's
if inst is None:
return _reject("Pick an instrument.")
trade = tenancy.owned(
db,
Trade(
date=date,
time=when,
type=type,
quantity=qty,
unit_price=price,
brokerage=brk,
# AUD is 1:1; for anything else an unset rate is backfilled from
# the stored FX close for that date by the price feed.
fx_rate=Decimal(1) if inst.currency == "AUD" else fx,
note=note_text,
),
)
# An instrument being added has no trades here yet, so this walks
# the new trade alone.
problem = _breach(db, inst, trade)
if problem:
return _reject(problem)
if adding:
# Not in this portfolio's list: new, or another portfolio added
# it, or it was removed. A shared row comes back with its
# history and its class.
inst.active = True
db.add(inst)
db.flush()
if queries.prefs_by_instrument(db).get(inst.id) is None:
db.add(tenancy.owned(
db, HoldingPref(instrument_id=inst.id, drp=bool(new_drp))))
trade.instrument_id = inst.id
db.add(trade)
db.flush()
if adding:
_kick_feed() # its price history arrives with the next run
target = _holding_page(db, inst)
if owned:
submissions.finish(owned, target)
return _redirect(target)
finally:
if owned:
submissions.release(owned)
# --------------------------------------------------------------------------- #
# Editing and removing what's already recorded — decisions.md #91.
# --------------------------------------------------------------------------- #
def _owning_dividend(db: DbSession, trade_id: int) -> Dividend | None:
return db.scalar(select(Dividend).where(Dividend.reinvest_trade_id == trade_id))
def _own_instrument(db: DbSession, instrument_id: int) -> Instrument:
"""An instrument this portfolio has, or a 404. One only another portfolio
has is not this one's to change or price, and a setting saved on it would
put it on this portfolio's list."""
inst = db.get(Instrument, instrument_id)
if inst is None or inst.id not in queries.portfolio_instrument_ids(db):
raise HTTPException(404, "no such instrument")
return inst
def _back_to_holding(db: DbSession, inst: Instrument, path: str | None = None) -> str:
"""`path`, by default the holding's page. Once the last row this portfolio
had for the instrument has gone, that page is another portfolio's or
nobody's, so the portfolio's own page is where to land instead."""
page = _holding_page(db, inst)
path = path or page
if (path.split("?")[0] == page.split("?")[0]
and inst.id not in queries.portfolio_instrument_ids(db)):
return "/"
return path
def _trade_for_move(db: DbSession, trade_id: int) -> Trade:
"""The trade the move dialog was opened from.
Unlike `_trade_for_edit` this ACCEPTS a reinvested distribution. Editing
half of a pair would leave the units and the cash disagreeing, but moving
it does not: `moves.expand` carries both halves across together, so the
pair stays whole and simply lives somewhere else.
"""
trade = db.get(Trade, trade_id)
# `db.get` can answer from the identity map without going near the tenancy
# filter, so ownership is checked explicitly — same reason as below.
if trade is None or trade.portfolio_id != tenancy.current_portfolio_id(db):
raise HTTPException(404, "no such trade")
return trade
def _trade_for_edit(db: DbSession, trade_id: int) -> Trade:
"""The trade, or a 404/409 explaining why it can't be edited directly."""
trade = db.get(Trade, trade_id)
# `db.get` bypasses the tenancy filter (it can return from the identity
# map), so the ownership check is explicit here rather than assumed.
if trade is None or trade.portfolio_id != tenancy.current_portfolio_id(db):
raise HTTPException(404, "no such trade")
if _owning_dividend(db, trade.id) is not None:
raise HTTPException(
409,
"That's a reinvested distribution — edit it through its dividend so "
"the units and the cash stay in step.",
)
return trade
def _log_change(kind: str, row_id: int, before: dict, after: dict) -> None:
"""What changed, at INFO. `user_id` stays the recorder, not the last
editor, so this log is the only record of who touched what."""
changed = {k: (before[k], after[k]) for k in before if before[k] != after[k]}
if changed:
log.info("%s %s edited: %s", kind, row_id, changed)
def _trade_snapshot(t: Trade) -> dict:
return {
"date": t.date,
"time": t.time,
"type": t.type,
"quantity": t.quantity,
"unit_price": t.unit_price,
"brokerage": t.brokerage,
"fx_rate": t.fx_rate,
"note": t.note,
}
@app.get("/trade/{trade_id}/edit", response_class=HTMLResponse)
def trade_edit_form(request: Request, trade_id: RowId, error: str = ""):
with scoped(request) as (ctx, db):
_require_write(ctx)
trade = _trade_for_edit(db, trade_id)
inst = db.get(Instrument, trade.instrument_id)
# Where to go back to once this trade is dealt with. Arrives as
# `?return=`, from whatever linked here — the holdings page's "what is
# blocking this" panel is the first caller. Validated rather than
# trusted: it ends up in a Location header.
back_to = _safe_path(request.query_params.get("return"), _holding_page(db, inst))
# The move dialog's contents, prepared here so the page can carry it —
# `move_targets` empty means the control is not drawn at all.
move = _move_context(db, ctx, trade, inst)
return _render(
request,
ctx,
"trade_new.html",
{
"active_nav": "holdings",
"instruments": [inst],
"ticker": inst.ticker,
"today": clock.today().isoformat(),
"error": error,
"edit": trade,
"inst": inst,
"return_to": back_to,
"move_rows": move["rows"],
"move_targets": move["targets"],
"move_ticked": move["ticked"],
"move_pulled_in": move["pulled_in"],
# The back button goes wherever the delete button already
# goes. Two controls on one page that both mean "I am finished
# here" must not land in different places.
"back_label": _back_label(back_to, inst.ticker),
"form": {
"instrument_id": str(inst.id),
"type": trade.type,
"trade_date": trade.date.isoformat(),
# Ticked only when the time carries information — a trade
# left at the open shouldn't come back looking deliberate.
"set_time": "1" if trade.time not in (None, MARKET_OPEN) else "",
"trade_time": (trade.time or MARKET_OPEN).strftime("%H:%M"),
"quantity": f"{trade.quantity.normalize():f}",
"unit_price": f"{trade.unit_price.normalize():f}",
"brokerage": f"{trade.brokerage:f}",
"fx_rate": "" if trade.fx_rate is None else f"{trade.fx_rate.normalize():f}",
"note": trade.note or "",
},
},
)
@app.post("/trade/{trade_id}/edit")
async def trade_edit(
request: Request,
trade_id: RowId,
type: str = Form(...),
trade_date: str = Form(...),
set_time: str = Form(""),
trade_time: str = Form(""),
quantity: str = Form(...),
unit_price: str = Form(...),
brokerage: str = Form("0"),
fx_rate: str = Form(""),
note: str = Form(""),
):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
trade = _trade_for_edit(db, trade_id)
inst = db.get(Instrument, trade.instrument_id)
def _reject(message: str):
return _redirect(
f"/trade/{trade_id}/edit?error={quote_plus(message)}"
)
figures, problem = _trade_figures(type, trade_date, quantity, unit_price, brokerage,
fx_rate, note, set_time, trade_time)
if problem:
return _reject(problem)
date, qty, price, brk, fx, note_text, when = figures
# Check the edit against the timeline WITHOUT the original row: the
# candidate replaces it. Validating the new values alone would miss a
# quantity cut or a date move that strands a later sell.
candidate = Trade(
instrument_id=inst.id,
date=date,
time=when,
type=type,
quantity=qty,
unit_price=price,
brokerage=brk,
fx_rate=Decimal(1) if inst.currency == "AUD" else fx,
)
# The candidate stands in for a SAVED row, so it keeps that row's place
# in the day rather than sorting to the end as a new entry would.
candidate.id = trade.id
problem = _breach(db, inst, candidate, exclude_ids={trade.id})
if problem:
return _reject(problem)
before = _trade_snapshot(trade)
trade.date = date
trade.time = when
trade.type = type
trade.quantity = qty
trade.unit_price = price
trade.brokerage = brk
trade.fx_rate = Decimal(1) if inst.currency == "AUD" else fx
trade.note = note_text
_log_change("trade", trade.id, before, _trade_snapshot(trade))
db.flush()
return _redirect(_holding_page(db, inst))
def _release_planned_purchases(db: DbSession, trade_id: int) -> None:
"""Keep the plan history, drop the link to a trade that no longer exists.
The purchase happened — deleting the trade doesn't unschedule it — so the
row stays and says what became of it.
"""
for purchase in db.scalars(
select(PlannedPurchase).where(PlannedPurchase.trade_id == trade_id)
).all():
purchase.trade_id = None
note = " ".join(filter(None, [purchase.note, "(trade removed)"]))
purchase.note = note[:200] # the column's width, not a guess
# --------------------------------------------------------------------------- #
# Moving a holding's rows to another portfolio — decisions.md #125, issue #36.
# --------------------------------------------------------------------------- #
def _move_targets(ctx) -> list:
"""Portfolios this person could move something INTO.
Their own memberships, minus the one they are in, minus the ones they can
only read — a move writes at the far end. The POST checks this list again
rather than trusting the posted id.
"""
return [
m for m in ctx.memberships
if m.portfolio_id != ctx.active_portfolio_id and m.role in WRITE_ROLES
]
def _parse_rows(values: list[str]) -> tuple[set[int], set[int]]:
"""`trade:12` and `dividend:5` tickboxes into two sets of ids.
One checkbox name for both tables, because on screen they are one list.
Anything unparseable is dropped rather than raising: the ids are re-checked
against this portfolio's own rows before anything moves, so junk in the
form cannot reach a row, and a 500 would be a worse answer than ignoring it.
"""
trades: set[int] = set()
dividends: set[int] = set()
for value in values:
kind, _, raw = value.partition(":")
ident = row_id(raw)
if ident is None:
continue
if kind == "trade":
trades.add(ident)
elif kind == "dividend":
dividends.add(ident)
return trades, dividends
def _release_for_move(db: DbSession, trade_ids: set[int]) -> None:
"""A plan step whose trade has left this portfolio goes back to planned.
Deliberately different from `_release_planned_purchases`, which keeps the
step at "done" because a deleted trade still happened. A MOVED trade did
not happen *here* — the source portfolio has no record of that buy any
more, so the slot is genuinely unfilled and the schedule should offer it
again. Leaving it "done" would show a purchase backed by nothing.
"""
if not trade_ids:
return
for purchase in db.scalars(
select(PlannedPurchase).where(PlannedPurchase.trade_id.in_(trade_ids))
).all():
purchase.trade_id = None
purchase.status = "planned"
note = " ".join(filter(None, [purchase.note, "(trade moved out)"]))
purchase.note = note[:200] # the column's width, not a guess
def _move_context(db: DbSession, ctx, trade, inst, *, ticked=None, error=""):
"""What the dialog needs, or just enough to know not to draw it.
The edit page asks for this on every load, and for the ordinary install —
one portfolio — there is nowhere to move anything. So an empty `targets`
returns early rather than walking the instrument's whole timeline to build
a list no page will render.
"""
targets = _move_targets(ctx)
if not targets:
return {"active_nav": "holdings", "inst": inst, "edit": trade,
"rows": [], "targets": [], "ticked": set(), "pulled_in": 0,
"error": error, "return_to": _holding_page(db, inst)}
rows = moves.movable_rows(db, inst.id)
# `pull_in=True` on purpose, including when re-rendering after a refusal:
# the form arrives with the fix applied and the message above it saying
# why. The POST path is the one that must NOT pull in — there the ticks are
# somebody's answer, not a starting point.
chosen = moves.expand(
db, inst.id,
trades={trade.id} if ticked is None else ticked[0],
dividends=set() if ticked is None else ticked[1],
pull_in=True,
)
return {
"active_nav": "holdings",
"inst": inst,
"edit": trade,
"rows": rows,
"targets": targets,
# What arrives ticked: what was asked for, plus whatever `expand` had
# to add. The dialog says so rather than moving rows nobody chose.
"ticked": ({f"trade:{i}" for i in chosen.trades}
| {f"dividend:{i}" for i in chosen.dividends}),
# Trades only: a coupled distribution has no tickbox of its own, so
# counting it would announce a row nobody can see — see _move_form.html.
"pulled_in": sum(1 for kind, _ in chosen.added if kind == "trade"),
"error": error,
"return_to": _holding_page(db, inst),
}
@app.get("/trade/{trade_id}/move", response_class=HTMLResponse)
def trade_move_form(request: Request, trade_id: RowId, error: str = ""):
"""The selection page, and the same include the dialog shows.
A real route so the control works without scripting and a refusal has
somewhere to land — the pattern Record trade uses.
"""
with scoped(request) as (ctx, db):
_require_write(ctx)
trade = _trade_for_move(db, trade_id)
inst = db.get(Instrument, trade.instrument_id)
context = _move_context(db, ctx, trade, inst, error=error)
if not context["targets"]:
# Nowhere to move anything, so there is no such page — rather than
# a form whose only control is an empty dropdown.
raise HTTPException(404, "no other portfolio to move this into")
return _render(request, ctx, "trade_move.html", context)
@app.post("/trade/{trade_id}/move")
async def trade_move(
request: Request,
trade_id: RowId,
target: Annotated[RowId, Form()],
row: list[str] = Form(default=[]),
):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
trade = _trade_for_move(db, trade_id)
inst = db.get(Instrument, trade.instrument_id)
def _reject(message: str):
return _render(
request, ctx, "trade_move.html",
_move_context(db, ctx, trade, inst,
ticked=_parse_rows(row), error=message),
)
if target not in {m.portfolio_id for m in _move_targets(ctx)}:
raise HTTPException(403, "you can't move anything into that portfolio")
asked_trades, asked_dividends = _parse_rows(row)
# Only rows this portfolio actually holds for this instrument. The form
# is user input, and without this a posted id could reach a row the
# dialog never offered.
offered = moves.movable_rows(db, inst.id)
allowed_trades = {r.row_id for r in offered if r.is_trade}
allowed_dividends = ({r.row_id for r in offered if not r.is_trade}
| {r.pairs_with for r in offered if r.pairs_with})
asked_trades &= allowed_trades
asked_dividends &= allowed_dividends
if not (asked_trades or asked_dividends):
return _reject("Nothing was ticked, so nothing moved.")
# `pull_in=False`: the ticks are the answer. The dialog pre-ticks the
# rows that cannot be left behind, and unticking one is a judgement
# somebody is allowed to make — so it is honoured and then refused,
# rather than silently put back. Pairs are still coupled.
chosen = moves.expand(db, inst.id, trades=asked_trades,
dividends=asked_dividends, pull_in=False)
problem = (
moves.source_problem(db, inst.id, trades=chosen.trades)
or moves.target_problem(
db, inst.id, target, user_id=ctx.user.id,
trades=chosen.trades, dividends=chosen.dividends,
)
)
if problem:
return _reject(problem)
_release_for_move(db, chosen.trades)
moves.perform(db, target, user_id=ctx.user.id,
trades=chosen.trades, dividends=chosen.dividends)
log.info("moved %s: trades=%s dividends=%s -> portfolio %s",
inst.ticker, sorted(chosen.trades), sorted(chosen.dividends),
target)
db.flush()
return _redirect(_back_to_holding(db, inst))
@app.post("/trade/{trade_id}/delete")
async def trade_delete(request: Request, trade_id: RowId, return_to: str = Form("")):
"""Delete, then go back where you came from.
Deleting a trade returns to the window you came from, wherever that was.
Carried explicitly through the form rather than read from `Referer` — that
header is absent under several privacy settings, and the one time it is
missing is the one time somebody is lost.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
trade = _trade_for_edit(db, trade_id)
inst = db.get(Instrument, trade.instrument_id)
# A path on this site or the instrument's own page — never the
# dashboard, which is further from where they were than the fallback.
back = _safe_path(return_to, _holding_page(db, inst))
# Removing a buy can strand a sell that depended on its units.
problem = _breach(db, inst, None, exclude_ids={trade.id})
if problem:
return _redirect(_holding_page(db, inst, error=problem))
_release_planned_purchases(db, trade.id)
log.info("trade %s deleted: %s", trade.id, _trade_snapshot(trade))
db.delete(trade)
db.flush()
return _redirect(_back_to_holding(db, inst, back))
def _dividend_for_edit(db: DbSession, dividend_id: int) -> Dividend:
dividend = db.get(Dividend, dividend_id)
if dividend is None or dividend.portfolio_id != tenancy.current_portfolio_id(db):
raise HTTPException(404, "no such dividend")
return dividend
@app.get("/dividend/{dividend_id}/edit", response_class=HTMLResponse)
def dividend_edit_form(request: Request, dividend_id: RowId, error: str = ""):
with scoped(request) as (ctx, db):
_require_write(ctx)
dividend = _dividend_for_edit(db, dividend_id)
inst = db.get(Instrument, dividend.instrument_id)
drp = dividend.reinvest_trade
return _render(
request,
ctx,
"dividend_edit.html",
{
"active_nav": "holdings",
"inst": inst,
"dividend": dividend,
"drp": drp,
"today": clock.today().isoformat(),
"error": error,
},
)
@app.post("/dividend/{dividend_id}/edit")
async def dividend_edit(
request: Request,
dividend_id: RowId,
div_date: str = Form(...),
cash_amount: str = Form(...),
franking_credits: str = Form(""),
withholding_tax: str = Form(""),
period_end: str = Form(""),
quantity: str = Form(""),
unit_price: str = Form(""),
note: str = Form(""),
):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
dividend = _dividend_for_edit(db, dividend_id)
inst = db.get(Instrument, dividend.instrument_id)
drp = dividend.reinvest_trade
def _reject(message: str):
return _redirect(f"/dividend/{dividend_id}/edit?error={quote_plus(message)}")
try:
date = dt.date.fromisoformat(div_date)
cash = money.parse(cash_amount, Dividend.cash_amount, "Cash")
franking = (money.parse(franking_credits, Dividend.franking_credits,
"Franking credits")
if franking_credits.strip() else None)
withheld = (money.parse(withholding_tax, Dividend.withholding_tax,
"Foreign tax withheld")
if withholding_tax.strip() else None)
ended = dt.date.fromisoformat(period_end.strip()) if period_end.strip() else None
except money.FigureError as exc:
return _reject(f"{exc}.")
except ValueError:
return _reject("Date, cash and franking must be numbers (date as YYYY-MM-DD).")
problem = fyreport.period_problem(ended, date)
if problem:
return _reject(problem)
if cash <= 0:
return _reject("A distribution has to be more than zero.")
if franking is not None and franking < 0:
return _reject("Franking credits can't be negative.")
if withheld is not None and withheld < 0:
return _reject("Foreign tax withheld can't be negative.")
try:
note_text = textfield.fit(note, Dividend.note, "Note")
except textfield.TextError as exc:
return _reject(f"{exc}.")
if date > clock.today():
return _reject("That date is in the future.")
# A reinvested distribution carries units too — the pair moves together
# or not at all, so both halves are validated before either is written.
if drp is not None:
try:
qty = money.parse(quantity, Trade.quantity, "Units")
price = money.parse(unit_price, Trade.unit_price, "Price")
except money.FigureError as exc:
return _reject(f"{exc}.")
except ValueError:
return _reject("Units and price must be numbers.")
if qty <= 0:
return _reject("Units must be greater than zero.")
if price < 0: # see the trade form: zero is a real price
return _reject("Price can't be negative.")
candidate = Trade(
instrument_id=inst.id,
date=date,
time=drp.time,
type="drp",
quantity=qty,
unit_price=price,
brokerage=drp.brokerage,
fx_rate=drp.fx_rate,
)
candidate.id = drp.id
problem = _breach(db, inst, candidate, exclude_ids={drp.id})
if problem:
return _reject(problem)
before = {
"date": dividend.date,
"cash_amount": dividend.cash_amount,
"franking_credits": dividend.franking_credits,
"withholding_tax": dividend.withholding_tax,
"period_end": dividend.period_end,
"note": dividend.note,
}
dividend.date = date
dividend.cash_amount = cash
dividend.franking_credits = franking
dividend.withholding_tax = withheld
dividend.period_end = ended
dividend.note = note_text
if drp is not None:
drp.date = date
drp.quantity = qty
drp.unit_price = price
_log_change(
"dividend",
dividend.id,
before,
{
"date": dividend.date,
"cash_amount": dividend.cash_amount,
"franking_credits": dividend.franking_credits,
"withholding_tax": dividend.withholding_tax,
"period_end": dividend.period_end,
"note": dividend.note,
},
)
db.flush()
return _redirect(_holding_page(db, inst))
@app.post("/dividend/{dividend_id}/delete")
async def dividend_delete(request: Request, dividend_id: RowId):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
dividend = _dividend_for_edit(db, dividend_id)
inst = db.get(Instrument, dividend.instrument_id)
drp = dividend.reinvest_trade
if drp is not None:
# The units go with the cash: they were the same event. Check the
# timeline survives losing them before removing either.
problem = _breach(db, inst, None, exclude_ids={drp.id})
if problem:
return _redirect(_holding_page(db, inst, error=problem))
log.info(
"dividend %s deleted: %s on %s%s",
dividend.id,
dividend.cash_amount,
dividend.date,
f" (with DRP trade {drp.id})" if drp is not None else "",
)
# The FK is RESTRICT, so the dividend's reference has to go first.
dividend.reinvest_trade = None
db.delete(dividend)
if drp is not None:
_release_planned_purchases(db, drp.id)
db.delete(drp)
db.flush()
return _redirect(_back_to_holding(db, inst))
# --------------------------------------------------------------------------- #
# Instruments
# --------------------------------------------------------------------------- #
def _blocking_trades(db, instruments) -> dict[int, dict]:
"""What stands in the way of removing each instrument from this portfolio:
its own trades, listed with links so each is one click from being fixed,
and its own distributions, counted. Its splits and capital changes are
counted for the delete confirm, which takes them too. Another portfolio's
never block it — removing drops the instrument from this portfolio, not
from the shared catalogue — and are never mentioned, which would say what
they hold."""
ids = [i.id for i in instruments]
if not ids:
return {}
# A reinvested distribution's units are edited through the distribution,
# so its link goes there: the trade's own page refuses it (#91).
owners = dict(db.execute(
select(Dividend.reinvest_trade_id, Dividend.id).where(
Dividend.instrument_id.in_(ids), Dividend.reinvest_trade_id.is_not(None))
).all())
mine: dict[int, list[tuple[Trade, str]]] = {}
for trade in db.scalars(
select(Trade).where(Trade.instrument_id.in_(ids)).order_by(Trade.date.desc())
):
edit = (f"/dividend/{owners[trade.id]}/edit" if trade.id in owners
else f"/trade/{trade.id}/edit")
mine.setdefault(trade.instrument_id, []).append((trade, f"{edit}?return=/holdings"))
dividends = dict(db.execute(
select(Dividend.instrument_id, func.count())
.where(Dividend.instrument_id.in_(ids)).group_by(Dividend.instrument_id)
).all())
actions = dict(db.execute(
select(CorporateAction.instrument_id, func.count())
.where(CorporateAction.instrument_id.in_(ids)).group_by(CorporateAction.instrument_id)
).all())
return {
iid: {"mine": mine.get(iid, []),
"total": len(mine.get(iid, [])) + dividends.get(iid, 0),
"actions": actions.get(iid, 0)}
for iid in ids
}
# --------------------------------------------------------------------------- #
# Corporate actions: splits, consolidations and capital changes (#146)
# --------------------------------------------------------------------------- #
# A ratio is whole units; past a million for one, it is not a split anyone made.
_RATIO = re.compile(r"[0-9]{1,7}")
_RATIO_MAX = 1_000_000
def _action_for_edit(db: DbSession, action_id: int) -> CorporateAction:
action = db.get(CorporateAction, action_id)
if action is None or action.portfolio_id != tenancy.current_portfolio_id(db):
raise HTTPException(404, "no such action")
return action
def _action_typed(action: CorporateAction) -> dict:
"""An action as its form shows it."""
amount = action.amount
return {"kind": action.kind, "action_date": action.date.isoformat(),
"ratio_new": str(action.ratio_new or ""), "ratio_old": str(action.ratio_old or ""),
"amount": "" if amount is None else f"{abs(amount).normalize():f}",
"direction": "decrease" if amount is not None and amount < 0 else "increase",
"fx_rate": "" if action.fx_rate is None else f"{action.fx_rate.normalize():f}",
"note": action.note or ""}
def _action_from(db: DbSession, inst: Instrument, typed: dict,
editing: CorporateAction | None = None) -> CorporateAction | str:
"""The action `typed` describes, or why it cannot be. Like a trade, it is
checked against the whole timeline: something has to be held on its day,
and a consolidation must leave every later sale covered."""
kind = typed["kind"]
if kind not in ACTION_KINDS:
return "Pick what happened."
try:
date = dt.date.fromisoformat(typed["action_date"].strip())
except ValueError:
return "The date has to be a date, like 2026-03-02."
if date > clock.today():
return "That date is in the future."
ratio_new = ratio_old = amount = None
if kind == "split":
raw = (typed["ratio_new"].strip(), typed["ratio_old"].strip())
if not all(_RATIO.fullmatch(r) and 1 <= int(r) <= _RATIO_MAX for r in raw):
return f"New and old units are whole numbers from 1 to {_RATIO_MAX}."
ratio_new, ratio_old = int(raw[0]), int(raw[1])
if ratio_new == ratio_old:
return "A split changes the number of units: new and old can't be the same."
else:
try:
amount = money.parse(typed["amount"], CorporateAction.amount, "Amount")
except money.FigureError as exc:
return f"{exc}."
if amount <= 0:
return "An amount has to be more than zero."
if kind == "cost_base" and typed["direction"] == "decrease":
amount = -amount
fx = Decimal(1) if inst.currency == money.REPORTING else None
if fx is None and typed["fx_rate"].strip():
try:
fx = money.parse(typed["fx_rate"], CorporateAction.fx_rate, "FX rate")
except money.FigureError as exc:
return f"{exc}."
if fx <= 0:
return "An FX rate has to be more than zero."
try:
note = textfield.fit(typed["note"], CorporateAction.note, "Note")
except textfield.TextError as exc:
return f"{exc}."
candidate = CorporateAction(instrument_id=inst.id, date=date, kind=kind,
ratio_new=ratio_new, ratio_old=ratio_old, amount=amount,
fx_rate=fx, note=note or None)
candidate.id = editing.id if editing is not None else None
others = [a for a in _held_actions(db, inst.id) if editing is None or a.id != editing.id]
before = queries.units_through(
r for r in queries.timeline(_held_trades(db, inst.id), others)
if queries.moment(r) < queries.moment(candidate))
if before <= 0:
return f"Nothing was held on {date}."
return _breach(db, inst, None, actions=[*others, candidate]) or candidate
def _action_page(request: Request, ctx, db: DbSession, inst: Instrument, typed: dict,
*, action: CorporateAction | None = None, error: str = "",
status_code: int = 200) -> HTMLResponse:
return _render(request, ctx, "action.html", {
"active_nav": "holdings", "inst": inst, "action": action, "form": typed,
"error": error, "today": clock.today().isoformat(),
"back": _holding_page(db, inst)}, status_code=status_code)
_ACTION_BLANK = {"kind": "split", "action_date": "", "ratio_new": "", "ratio_old": "",
"amount": "", "direction": "increase", "fx_rate": "", "note": ""}
@app.get("/action/new", response_class=HTMLResponse)
def action_new_form(request: Request, ticker: str = "", exchange: str = ""):
with scoped(request) as (ctx, db):
_require_write(ctx)
found = queries.held_as(db, ticker, exchange or None)
if len(found) != 1:
raise HTTPException(404, "no such holding")
return _action_page(request, ctx, db, found[0], dict(_ACTION_BLANK))
@app.post("/action/new")
async def action_new(
request: Request,
instrument_id: Annotated[RowId, Form()],
kind: str = Form(""),
action_date: str = Form(""),
ratio_new: str = Form(""),
ratio_old: str = Form(""),
amount: str = Form(""),
direction: str = Form("increase"),
fx_rate: str = Form(""),
note: str = Form(""),
):
typed = {"kind": kind, "action_date": action_date, "ratio_new": ratio_new,
"ratio_old": ratio_old, "amount": amount, "direction": direction,
"fx_rate": fx_rate, "note": note}
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
inst = _own_instrument(db, instrument_id)
found = _action_from(db, inst, typed)
if isinstance(found, str):
return _action_page(request, ctx, db, inst, typed, error=found, status_code=400)
db.add(tenancy.owned(db, found))
db.flush()
log.info("corporate action %s recorded on %s: %s", found.id, inst.ticker,
queries.action_label(found))
return _redirect(_holding_page(db, inst))
@app.get("/action/{action_id}/edit", response_class=HTMLResponse)
def action_edit_form(request: Request, action_id: RowId):
with scoped(request) as (ctx, db):
_require_write(ctx)
action = _action_for_edit(db, action_id)
inst = db.get(Instrument, action.instrument_id)
return _action_page(request, ctx, db, inst, _action_typed(action), action=action)
@app.post("/action/{action_id}/edit")
async def action_edit(
request: Request,
action_id: RowId,
kind: str = Form(""),
action_date: str = Form(""),
ratio_new: str = Form(""),
ratio_old: str = Form(""),
amount: str = Form(""),
direction: str = Form("increase"),
fx_rate: str = Form(""),
note: str = Form(""),
):
typed = {"kind": kind, "action_date": action_date, "ratio_new": ratio_new,
"ratio_old": ratio_old, "amount": amount, "direction": direction,
"fx_rate": fx_rate, "note": note}
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
action = _action_for_edit(db, action_id)
inst = db.get(Instrument, action.instrument_id)
found = _action_from(db, inst, typed, editing=action)
if isinstance(found, str):
return _action_page(request, ctx, db, inst, typed, action=action, error=found,
status_code=400)
for column in ("date", "kind", "ratio_new", "ratio_old", "amount", "fx_rate", "note"):
setattr(action, column, getattr(found, column))
db.flush()
log.info("corporate action %s on %s changed: %s", action.id, inst.ticker,
queries.action_label(action))
return _redirect(_holding_page(db, inst))
@app.post("/action/{action_id}/delete")
async def action_delete(request: Request, action_id: RowId):
"""Refused when a later sale needs it: deleting a split takes away the
units it made."""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
action = _action_for_edit(db, action_id)
inst = db.get(Instrument, action.instrument_id)
others = [a for a in _held_actions(db, inst.id) if a.id != action.id]
problem = _breach(db, inst, None, actions=others)
if problem:
return _redirect(_holding_page(db, inst, error=problem))
log.info("corporate action %s on %s deleted: %s", action.id, inst.ticker,
queries.action_label(action))
db.delete(action)
db.flush()
return _redirect(_back_to_holding(db, inst))
@app.get("/holdings", response_class=HTMLResponse)
def instruments_page(request: Request, error: str = "", added: str = "",
removed: str = "", unplanned: bool = False):
with scoped(request) as (ctx, db):
# This portfolio's, not the whole shared catalogue: listing that showed
# every portfolio the tickers the others had added.
instruments = queries.portfolio_instruments(db)
prefs = queries.prefs_by_instrument(db)
held = {
iid
for iid in db.scalars(select(Trade.instrument_id).distinct()).all()
}
return _render(
request,
ctx,
"instruments.html",
{
"active_nav": "holdings",
"instruments": instruments,
"prefs": prefs,
"held": held,
"blocking": _blocking_trades(db, instruments),
# The latest stored close per instrument. Shown even for one
# no longer held: a position you exited is still one you may be
# watching, and the dashboard deliberately hides it.
"prices": queries.latest_prices(db, [i.id for i in instruments]),
"error": error,
"added": added,
"removed": removed,
"unplanned": unplanned,
"editable": _catalogue_editable(ctx, [i.id for i in instruments]),
# Only an instance admin can edit one another portfolio has
# too, and a new currency then resets that portfolio's rates.
"shared": (_shared_elsewhere([i.id for i in instruments],
ctx.active_portfolio_id)
if ctx.user.is_admin else set()),
},
)
@app.post("/holdings/{instrument_id}/remove")
async def instrument_remove(request: Request, instrument_id: RowId):
"""Take an instrument off this portfolio.
Refused while this portfolio has trades or distributions for it: those
would refer to nothing here. Otherwise this portfolio's setting for it
goes, and so do its places in the plan, which closes up around them
(`plans.remove_instrument`). The instrument stays in the shared catalogue
with its price history. Only once no portfolio has it at all does the feed
stop fetching it — **deactivated, not deleted**, so adding it again brings
it back.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
inst = _own_instrument(db, instrument_id)
blocking = _blocking_trades(db, [inst]).get(instrument_id, {})
if blocking.get("total"):
raise HTTPException(
409,
f"{inst.ticker} still has "
f"{wording.plural(blocking['total'], 'trade or distribution', 'trades or distributions')} "
"recorded against it. Delete those first.",
)
unplanned = plans.remove_instrument(db, instrument_id)
for pref in db.scalars(
select(HoldingPref).where(HoldingPref.instrument_id == instrument_id)
).all():
db.delete(pref)
# With no trades left, an action has nothing to act on (#146).
for action in _held_actions(db, instrument_id):
db.delete(action)
# Committed before asking who else has it: the question goes through a
# session of its own, which cannot see this one's uncommitted delete.
db.commit()
if not _anyone_has(instrument_id):
inst.active = False
log.info("instrument %s removed from portfolio %s by user %d",
inst.ticker, ctx.active_portfolio_id, ctx.user.id)
return _redirect("/holdings?removed=" + quote_plus(inst.ticker)
+ ("&unplanned=1" if unplanned else ""))
def _anyone_has(instrument_id: int) -> bool:
"""Whether any portfolio still has this instrument, counted across all of
them — the one question here that is not this portfolio's to answer
alone, because the feed fetches for everybody."""
with tenancy.unscoped_session(SessionLocal) as everyone:
for model in (HoldingPref, Trade, Dividend, CorporateAction, InvestmentPlanEntry):
if everyone.scalar(select(model.id).where(
model.instrument_id == instrument_id).limit(1)) is not None:
return True
return False
def _shared_elsewhere(instrument_ids, portfolio_id: int) -> set[int]:
"""Which of these instruments another portfolio has too. Asked across all
portfolios, like `_anyone_has`, and answered with ids alone."""
ids = list(instrument_ids)
if not ids:
return set()
with tenancy.unscoped_session(SessionLocal) as everyone:
found: set[int] = set()
for model in (HoldingPref, Trade, Dividend, CorporateAction):
found |= set(everyone.scalars(select(model.instrument_id).where(
model.instrument_id.in_(ids), model.portfolio_id != portfolio_id)))
found |= set(everyone.scalars(
select(InvestmentPlanEntry.instrument_id)
.join(InvestmentPlan, InvestmentPlanEntry.plan_id == InvestmentPlan.id)
.where(InvestmentPlanEntry.instrument_id.in_(ids),
InvestmentPlan.portfolio_id != portfolio_id)))
return found
def _catalogue_editable(ctx, instrument_ids) -> set[int]:
"""Which instruments' shared details this person may change: the name,
class, currency and price symbol every portfolio holding it sees. Those this
portfolio alone has, so the edit reaches nobody else's; an instance admin's,
all of them (decisions.md #143)."""
ids = set(instrument_ids)
if not ctx.can_write:
return set()
if ctx.user.is_admin:
return ids
return ids - _shared_elsewhere(ids, ctx.active_portfolio_id)
@app.post("/holdings/{instrument_id}/delete-trades")
async def instrument_delete_trades(request: Request, instrument_id: RowId):
"""Every trade and distribution this portfolio has for an instrument, and
its corporate actions, in one go, so that it can then be removed. Another
portfolio's are untouched.
The same steps as deleting each in turn: a reinvested distribution takes
its trade with it, and a planned buy keeps its history with the link to
the trade dropped (`_release_planned_purchases`). Nothing can be left
stranded — with every trade gone there is no sell to outrun its buys.
"""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
inst = _own_instrument(db, instrument_id)
dividends = db.scalars(
select(Dividend).where(Dividend.instrument_id == instrument_id)).all()
for dividend in dividends:
dividend.reinvest_trade = None
db.delete(dividend)
# A split or a capital change means nothing with no trades to act on.
for action in _held_actions(db, instrument_id):
db.delete(action)
db.flush()
trades = db.scalars(select(Trade).where(Trade.instrument_id == instrument_id)).all()
for trade in trades:
_release_planned_purchases(db, trade.id)
db.delete(trade)
# Still listed afterwards, so removing it is a step of its own. A
# holding known only through its trades would otherwise vanish here.
if queries.prefs_by_instrument(db).get(instrument_id) is None:
db.add(tenancy.owned(db, HoldingPref(instrument_id=instrument_id)))
db.flush()
log.info("all of portfolio %s's %s deleted: %d trades, %d dividends, by user %d",
ctx.active_portfolio_id, inst.ticker, len(trades), len(dividends),
ctx.user.id)
return _redirect("/holdings")
@app.get("/holdings/lookup")
def instruments_lookup(request: Request, ticker: str = "", exchange: str = "ASX"):
"""What Yahoo knows about a ticker — name, currency, price symbol.
Backs the add-instrument forms so nobody has to know that ASX codes carry
a .AX suffix or that NOVA prices in USD.
"""
with scoped(request) as (ctx, _db):
if not ticker.strip():
raise HTTPException(400, "ticker is required")
if not settings.price_feed.enabled:
# Nobody asked: the feed governs every outbound request (#9). The
# symbol is a guess made here, and needs nobody.
return {"symbol": pricefeed.yahoo_symbol_for(ticker, exchange), "name": None,
"currency": None, "asset_class": None, "found": False, "off": True}
return pricefeed.lookup(ticker, exchange)
def _provider_close(symbol: str, on: dt.date):
"""The provider's close for a symbol on or before `on`, or None.
For a ticker with no catalogue row yet. The symbol has already gone to the
provider by the time this is asked — the name and currency on the same form
came from `pricefeed.lookup` — so this is no new disclosure, but it is a
network call and `price_feed.enabled` governs those.
A fortnight's window rather than a day: one close is all the form wants,
and reaching back is only for a weekend, a holiday or a suspension. Any
failure is an empty field. A provider being unreachable must not turn into
a 500 on somebody's trade form.
"""
if not settings.price_feed.enabled:
return None
try:
result = providers.fetch(providers.kind_for("ASX"), symbol,
on - dt.timedelta(days=14), on)
except Exception: # noqa: BLE001 - a provider is allowed to be broken
log.warning("price lookup failed for %s", symbol, exc_info=True)
return None
if not result.ok:
return None
# Provider rows are (date, close); the caller wants (close, date) — the
# same shape `queries.close_on_or_before` returns, so the two are
# interchangeable at the call site.
usable = [(close, when) for when, close in result.rows if when <= on]
return max(usable, key=lambda row: row[1]) if usable else None
@app.get("/holdings/price")
def instruments_price_on(request: Request, date: str = "",
instrument: Annotated[int, Field(ge=0, le=ROW_ID_MAX)] = 0,
symbol: str = ""):
"""The close and FX rate for one instrument on one date.
Backs the trade form, which re-asks whenever the date changes. Nulls mean
nothing is known for that date and the form should stay empty rather than
offer the nearest figure to hand — decisions.md #124.
`as_at` is the date the figure is actually for — the Friday close for a
trade dated on a Saturday.
Two ways in, because the form has two states. `instrument` reads what is
STORED, which is the normal case and costs nothing. `symbol` is for an
instrument that does not exist yet — the inline "something not listed"
fieldset — where there is no history to read and the alternative is an
empty field on the one trade somebody is most likely to be typing from a
contract note. It asks the provider, so it is gated on the feed being on.
"""
with scoped(request) as (ctx, db):
try:
on = dt.date.fromisoformat(date)
except ValueError:
raise HTTPException(400, "date must be YYYY-MM-DD")
if not instrument and not symbol.strip():
raise HTTPException(400, "instrument or symbol is required")
if instrument:
inst = _own_instrument(db, instrument)
close = queries.close_on_or_before(db, inst.id, on)
rate = queries.fx_on_or_before(db, inst.currency, on)
else:
close, rate = _provider_close(symbol.strip(), on), None
return {
"price": f"{close[0].normalize():f}" if close else None,
"as_at": close[1].isoformat() if close else None,
"fx": f"{rate[0].normalize():f}" if rate else None,
"fx_as_at": rate[1].isoformat() if rate else None,
}
def _yahoo_currency(found: dict) -> str | None:
"""Yahoo's currency for a new instrument, if it is a code. A suggestion,
held to the same rule as typing one (decisions.md #130)."""
code = found.get("currency")
return code if code and not currency_problem(code) else None
@app.post("/holdings/add")
async def instruments_add(
request: Request,
ticker: str = Form(...),
name: str = Form(""),
exchange: str = Form("ASX"),
asset_class: str = Form("etf"),
# "" not "AUD": an empty field takes this default, and "AUD" here beat
# the lookup below, so a USD stock was added as AUD.
currency: str = Form(""),
yahoo_symbol: str = Form(""),
drp: str = Form(""),
):
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
ticker = ticker.strip().upper()
exchange = exchange.strip().upper()
currency = currency.strip().upper()
problem = (ticker_problem(ticker) or exchange_problem(exchange)
or (currency and currency_problem(currency))
or textfield.problem(name, Instrument.name, "Name")
or textfield.problem(yahoo_symbol, Instrument.yahoo_symbol,
"Yahoo symbol"))
if problem:
return _redirect(f"/holdings?error={quote_plus(problem)}")
if asset_class not in ("etf", "share", "crypto"):
return _redirect("/holdings?error=Unknown+asset+class")
existing = db.scalar(
select(Instrument).where(
Instrument.exchange == exchange, Instrument.ticker == ticker
)
)
if existing is None:
# Anything left blank is filled from Yahoo; what was typed wins.
found = (pricefeed.lookup(ticker, exchange)
if settings.price_feed.enabled and not (name.strip() and currency) else {})
symbol = yahoo_symbol.strip() or found.get("symbol") or pricefeed.yahoo_symbol_for(ticker, exchange)
existing = Instrument(
ticker=ticker,
exchange=exchange,
name=(name.strip() or found.get("name") or None),
currency=currency or _yahoo_currency(found) or "AUD",
asset_class=asset_class,
drp=bool(drp),
active=True,
yahoo_symbol=symbol,
)
db.add(existing)
db.flush()
elif not existing.active:
# Every portfolio removed it, which only deactivated it: back it
# comes, with its price history.
existing.active = True
# Adding it to MY list is the per-user half: a pref row is what makes
# the instrument mine to track (the catalogue entry itself is shared).
pref = queries.prefs_by_instrument(db).get(existing.id)
if pref is None:
db.add(
tenancy.owned(
db, HoldingPref(instrument_id=existing.id, drp=bool(drp))
)
)
else:
pref.drp = bool(drp)
db.flush()
_kick_feed() # prices for a brand-new instrument arrive with the next run
# The same answer whether or not it was new to the shared catalogue:
# saying which would tell this portfolio what another one had added.
return _redirect(f"/holdings?added={ticker}")
def _reach(reset: dict[tuple[int, str], int]) -> str:
"""'portfolio 1 (2 trades), portfolio 4 (1 trade, 1 distribution)'."""
parts = []
for pid in sorted({pid for pid, _kind in reset}):
counts = [f"{n} {kind}{'' if n == 1 else 's'}" for kind in ("trade", "distribution")
if (n := reset.get((pid, kind)))]
parts.append(f"portfolio {pid} ({', '.join(counts)})")
return ", ".join(parts) or "nothing recorded"
@app.post("/holdings/{instrument_id}/pref")
async def instrument_pref(
request: Request,
instrument_id: RowId,
drp: str = Form(""),
note: str = Form(""),
catalogue: str = Form(""),
name: str = Form(""),
asset_class: str = Form(""),
currency: str = Form(""),
yahoo_symbol: str = Form(""),
):
"""A row of Manage holdings: this portfolio's DRP and note, and, from a row
that offers them (`catalogue`), the shared details."""
with scoped(request) as (ctx, db):
await auth.verify_csrf(request, db)
_require_write(ctx)
inst = _own_instrument(db, instrument_id)
try:
note_text = textfield.fit(note, HoldingPref.note, "Note")
except textfield.TextError as exc:
raise HTTPException(400, str(exc))
fetch = False
if catalogue:
if instrument_id not in _catalogue_editable(ctx, [instrument_id]):
raise HTTPException(403, "another portfolio has this instrument too")
code = currency.strip().upper()
problem = (currency_problem(code)
or textfield.problem(name, Instrument.name, "Name")
or textfield.problem(yahoo_symbol, Instrument.yahoo_symbol,
"Yahoo symbol"))
if problem:
raise HTTPException(400, problem)
if asset_class not in ("etf", "share", "crypto"):
raise HTTPException(400, "Unknown asset class")
# Blank is the guess, as when adding. A new symbol keeps the prices
# already stored and fetches its own from the next run.
symbol = yahoo_symbol.strip() or pricefeed.yahoo_symbol_for(inst.ticker,
inst.exchange)
fetch = symbol != inst.yahoo_symbol
if code != inst.currency:
# Every recorded rate was for the old currency: an AUD row's 1
# would book US dollars as Australian ones. AUD needs none, and
# the feed fills a foreign one from the stored series.
rate = Decimal(1) if code == money.REPORTING else None
reset: dict[tuple[int, str], int] = {}
for model, kind in ((Trade, "trade"), (Dividend, "distribution")):
for pid in db.scalars(update(model).where(model.instrument_id == inst.id)
.values(fx_rate=rate).returning(model.portfolio_id)):
reset[(pid, kind)] = reset.get((pid, kind), 0) + 1
# Every portfolio's rows, not only this one's (#143): the one
# bulk update on a request path to cross portfolios says so.
log.info("%s on %s: currency %s to %s by user %d, resetting the rates of %s",
inst.ticker, inst.exchange, inst.currency, code, ctx.user.id,
_reach(reset))
fetch = fetch or rate is None
inst.name = textfield.fit(name, Instrument.name)
inst.asset_class = asset_class
inst.currency = code
inst.yahoo_symbol = symbol
pref = queries.prefs_by_instrument(db).get(instrument_id)
if pref is None:
pref = tenancy.owned(db, HoldingPref(instrument_id=instrument_id))
db.add(pref)
pref.drp = bool(drp)
pref.note = note_text
if fetch:
_kick_feed()
return _redirect(_back_to_referer(request, "/holdings"))
def _distribution_form(db: DbSession, inst: Instrument) -> dict:
choices = statements.choices(db)
value = next((v for v, _label in choices if v in (inst.ticker, f"{inst.ticker}:{inst.exchange}")),
"")
return {"dist": statements.ParsedStatement(text_sample="", ticker=value),
"dist_tickers": choices}
@app.get("/holding/{ticker}", response_class=HTMLResponse)
def instrument(request: Request, ticker: str, error: str = "", exchange: str = ""):
with scoped(request) as (ctx, db):
result = queries.ledger(db, ticker.upper(), exchange or None)
if result is None:
raise HTTPException(404, f"unknown instrument {ticker!r}")
inst, holding, events = result
# The same ticker held here on another exchange, a click away.
others = [i for i in queries.held_as(db, inst.ticker) if i.id != inst.id]
series = queries.instrument_series(db, inst)
back_to = _safe_path(request.query_params.get("return"), "/")
return _render(
request,
ctx,
"instrument.html",
{
"active_nav": "holdings",
"inst": inst,
"others": others,
"holding": holding,
"events": events,
"series_json": series,
"has_series": bool(series["dates"]),
# Record a distribution: the statement form, this holding
# chosen in its list.
**(_distribution_form(db, inst) if ctx.can_write else {}),
# Where the back button goes comes from `?return=`, not from
# `Referer` or `history.back()` — decisions.md #40.
"return_to": back_to,
"back_label": _back_label(back_to, "Portfolio"),
# A refused delete comes back here — the ledger is where you
# need to be to act on what the message says.
"error": error,
},
)