spb/company-atlas

1"""Owner endpoints (X-CA-Owner-Token): watchlist and alerts. The owner row is created on first use; nothing here is cached."""2from __future__ import annotations34from typing import Any56from fastapi import APIRouter, Depends, HTTPException, Query, Response7from pydantic import BaseModel, Field89from companyatlas.api import queries as q10from companyatlas.api import serializers as ser11from companyatlas.api.common import NO_STORE, owner_hash12from companyatlas.db import execute, fetch_all, fetch_one, fetch_val, jsonb, transaction13from companyatlas.ids import new_id14from companyatlas.taxonomy import EventType1516ORDER = 2017router = APIRouter(prefix="/api/v1", tags=["owner"])18MAX_WATCHLIST = 20019MAX_ALERTS = 10020KNOWN_TYPES = {t.value for t in EventType}212223async def _ensure_owner(conn: Any, oh: str) -> str:24    await execute(conn, "insert into owners (token_hash) values (:h) on conflict (token_hash) do update set last_seen_at = now()", h=oh)25    wl = await fetch_val(conn, "select id from watchlists where owner_hash = :h order by created_at limit 1", h=oh)26    if wl is None:27        wl = new_id("watchlist")28        await execute(conn, "insert into watchlists (id, owner_hash) values (:id, :h)", id=wl, h=oh)29    return wl303132class WatchBody(BaseModel):33    company: str = Field(min_length=1, max_length=200)343536@router.get("/watchlist", summary="Watched companies and their recent events")37async def get_watchlist(response: Response, oh: str = Depends(owner_hash), events_limit: int = Query(30, ge=1, le=100)) -> dict[str, Any]:38    response.headers["cache-control"] = NO_STORE39    async with transaction() as conn:40        wl = await _ensure_owner(conn, oh)41        rows = await fetch_all(conn, "select company_id, added_at from watchlist_items where watchlist_id = :wl order by added_at desc limit :lim", wl=wl, lim=MAX_WATCHLIST)42        ids = [r["company_id"] for r in rows]43        cards = [ser.company_card(c) for c in await q.fetch_cards_by_ids(conn, ids, sparkline=True)]44        events: list[dict[str, Any]] = []45        if ids:46            where, params = q.event_filters()47            where.append("e.company_id = any(cast(:wl_ids as text[]))")48            params["wl_ids"] = ids49            events = [ser.event(r) for r in await q.fetch_events(conn, where, params, limit=events_limit)]50    added = {r["company_id"]: r["added_at"] for r in rows}51    for c in cards:52        c["added_at"] = added.get(c["id"])53    return {"id": wl, "items": cards, "events": events, "count": len(cards), "max": MAX_WATCHLIST}545556@router.post("/watchlist", status_code=201, summary="Add a company to the watchlist")57async def add_to_watchlist(body: WatchBody, response: Response, oh: str = Depends(owner_hash)) -> dict[str, Any]:58    response.headers["cache-control"] = NO_STORE59    async with transaction() as conn:60        wl = await _ensure_owner(conn, oh)61        c = await q.require_company(conn, body.company)62        n = await fetch_val(conn, "select count(*) from watchlist_items where watchlist_id = :wl", wl=wl)63        exists = await fetch_val(conn, "select 1 from watchlist_items where watchlist_id = :wl and company_id = :cid", wl=wl, cid=c["id"])64        if not exists and int(n or 0) >= MAX_WATCHLIST:65            raise HTTPException(status_code=409, detail=f"watchlist is full ({MAX_WATCHLIST})")66        await execute(conn, "insert into watchlist_items (watchlist_id, company_id) values (:wl, :cid) on conflict do nothing", wl=wl, cid=c["id"])67        cards = await q.fetch_cards_by_ids(conn, [c["id"]])68    return {"added": not bool(exists), "company": ser.company_card(cards[0]) if cards else None}697071@router.delete("/watchlist/{key}", summary="Remove a company from the watchlist")72async def remove_from_watchlist(key: str, response: Response, oh: str = Depends(owner_hash)) -> dict[str, Any]:73    response.headers["cache-control"] = NO_STORE74    async with transaction() as conn:75        wl = await _ensure_owner(conn, oh)76        c = await q.require_company(conn, key)77        removed = await fetch_val(conn, "with d as (delete from watchlist_items where watchlist_id = :wl and company_id = :cid returning 1) select count(*) from d",78                                  wl=wl, cid=c["id"])79    return {"removed": bool(removed), "company": c["slug"]}808182class MetricCondition(BaseModel):83    gt: float | None = None84    lt: float | None = None858687class AlertCondition(BaseModel):88    event_types: list[str] | None = None89    event_subtypes: list[str] | None = None90    min_importance: float | None = Field(None, ge=0, le=1)91    metrics: dict[str, MetricCondition] | None = None92    industries: list[str] | None = None93    countries: list[str] | None = None949596class AlertBody(BaseModel):97    name: str = Field(min_length=1, max_length=120)98    company: str | None = Field(None, max_length=200)99    condition: AlertCondition = Field(default_factory=AlertCondition)100    channel: str = Field("web", pattern="^(web|webhook)$")101    target: str | None = Field(None, max_length=500)102    enabled: bool = True103104105@router.get("/alerts", summary="Alerts of this owner")106async def list_alerts(response: Response, oh: str = Depends(owner_hash)) -> dict[str, Any]:107    response.headers["cache-control"] = NO_STORE108    async with transaction() as conn:109        await _ensure_owner(conn, oh)110        rows = await fetch_all(conn, "select a.*, c.slug as company_slug, c.display_name as company_display_name, c.canonical_domain as company_domain, "111                                     "c.country as company_country, c.logo_url as company_logo_url from alerts a left join companies c on c.id = a.company_id "112                                     "where a.owner_hash = :h order by a.created_at desc limit :lim", h=oh, lim=MAX_ALERTS)113    return {"items": [ser.alert(r) for r in rows], "count": len(rows), "max": MAX_ALERTS}114115116@router.post("/alerts", status_code=201, summary="Create an alert")117async def create_alert(body: AlertBody, response: Response, oh: str = Depends(owner_hash)) -> dict[str, Any]:118    response.headers["cache-control"] = NO_STORE119    cond = body.condition.model_dump(exclude_none=True)120    unknown = [t for t in (cond.get("event_types") or []) if t.upper() not in KNOWN_TYPES]121    if unknown:122        raise HTTPException(status_code=422, detail=f"unknown event_types: {', '.join(unknown[:5])}")123    if cond.get("event_types"):124        cond["event_types"] = [t.upper() for t in cond["event_types"]]125    if body.channel == "webhook" and (not body.target or not body.target.lower().startswith(("https://", "http://"))):126        raise HTTPException(status_code=422, detail="target: webhook alerts need an http(s) URL")127    async with transaction() as conn:128        await _ensure_owner(conn, oh)129        n = await fetch_val(conn, "select count(*) from alerts where owner_hash = :h", h=oh)130        if int(n or 0) >= MAX_ALERTS:131            raise HTTPException(status_code=409, detail=f"too many alerts ({MAX_ALERTS})")132        company_id = (await q.require_company(conn, body.company))["id"] if body.company else None133        if company_id is None and not cond:134            raise HTTPException(status_code=422, detail="an alert needs a company or at least one condition")135        aid = new_id("alert")136        await execute(conn, "insert into alerts (id, owner_hash, company_id, name, condition, channel, target, enabled) "137                            "values (:id, :h, :cid, :name, cast(:cond as jsonb), :channel, :target, :enabled)",138                      id=aid, h=oh, cid=company_id, name=body.name.strip(), cond=jsonb(cond), channel=body.channel, target=body.target, enabled=body.enabled)139        row = await fetch_one(conn, "select a.*, c.slug as company_slug, c.display_name as company_display_name, c.canonical_domain as company_domain, "140                                    "c.country as company_country, c.logo_url as company_logo_url from alerts a left join companies c on c.id = a.company_id "141                                    "where a.id = :id", id=aid)142    return ser.alert(row or {})143144145@router.delete("/alerts/{alert_id}", summary="Delete an alert")146async def delete_alert(alert_id: str, response: Response, oh: str = Depends(owner_hash)) -> dict[str, Any]:147    response.headers["cache-control"] = NO_STORE148    async with transaction() as conn:149        await _ensure_owner(conn, oh)150        removed = await fetch_val(conn, "with d as (delete from alerts where id = :id and owner_hash = :h returning 1) select count(*) from d", id=alert_id, h=oh)151    if not removed:152        raise HTTPException(status_code=404, detail="alert not found")153    return {"removed": True, "id": alert_id}154155156@router.get("/alerts/deliveries", summary="Recent alert deliveries for this owner")157async def alert_deliveries(response: Response, oh: str = Depends(owner_hash), limit: int = Query(50, ge=1, le=200)) -> dict[str, Any]:158    response.headers["cache-control"] = NO_STORE159    async with transaction() as conn:160        await _ensure_owner(conn, oh)161        rows = await fetch_all(conn, "select d.*, a.name as alert_name, e.title as event_title from alert_deliveries d join alerts a on a.id = d.alert_id "162                                     "left join events e on e.id = d.event_id where a.owner_hash = :h order by d.delivered_at desc limit :lim", h=oh, lim=limit)163    return {"items": [ser.alert_delivery(r) for r in rows]}164