viewer ¶

def create_app(
    db_path: Path | str,
    *,
    show_db_path: bool = True,
    cors_origins: Sequence[str] = (),
) -> FastAPI:
    """
    Builds the dashboard FastAPI app for `db_path`.

    The app runs on its own or mounted inside another FastAPI/Starlette app, either at a path
    prefix (`host.mount("/viz", create_app(db_path))`) or on its own hostname, and needs nothing
    from the host: no lifespan, no shared state. Pages and scripts build their URLs from the
    prefix the app is mounted under, so they work wherever it is served.

    All database and plot-building work runs on two dedicated worker threads
    (`app.state.runner` for real requests and `app.state.hydration_runner` for background
    prefetch, see `routes.api`), each with its own read-only `RecordStore` connection opened on
    first use. A slow request therefore never blocks the event loop, which matters when that loop
    is shared with a host site's other endpoints, and a prefetch never queues ahead of a click.
    A SQLite connection belongs to the thread that opens it, which is why each one lives
    entirely on its worker thread. The lifespan only closes those connections at shutdown; a
    mounted app doesn't get one, and its connections then close with the interpreter.

    This app is meant to run as a single process (each process would keep its own caches).

    Args:
        db_path (Path | str): path to the `.db` file to serve. Must already exist.
        show_db_path (bool): show the server-side path of `db_path` in the page header. Turn
            off when the viewer is exposed to people who shouldn't see the server's file layout.
        cors_origins (Sequence[str]): origins whose browser JavaScript may call this app's API
            from another origin (an exact list; no wildcard is added for you). Empty by
            default, which adds no CORS headers. Not needed to show the viewer in an `<iframe>`:
            the page then calls its own origin. Only needed if the host page's own scripts fetch
            from the viewer's API.

    Returns:
        (FastAPI): the configured app, ready for `uvicorn.run`, `host.mount`, or a `TestClient`.

    """
    db_path = Path(db_path)
    RecordStore.open(
        db_path, readonly=True
    ).close()  # fail now, not on the first request

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
        try:
            yield
        finally:
            close_runners(app)

    app = FastAPI(title="trendify", version=__version__, lifespan=lifespan)
    app.state.runner = HydrationRunner(db_path)
    app.state.hydration_runner = HydrationRunner(db_path)
    app.state.response_cache = OrderedDict()
    app.state.db_path = db_path
    app.state.db_mtime = None

    if cors_origins:
        app.add_middleware(
            CORSMiddleware,
            allow_origins=list(cors_origins),
            allow_methods=["GET"],
            allow_headers=["x-trendify-hydrate"],
        )

    templates = Jinja2Templates(directory=_TEMPLATES_DIR)
    # Jinja's `Environment.globals` is inferred from its default namespace (range, dict, ...),
    # so it isn't typed as the open `dict[str, Any]` it behaves as.
    env_globals = cast("dict[str, Any]", templates.env.globals)
    env_globals["current_year"] = datetime.datetime.now().year
    env_globals["app_version"] = __version__
    env_globals["db_path"] = str(db_path) if show_db_path else ""
    app.state.templates = templates

    app.mount("/static", StaticFiles(directory=_STATIC_DIR), name="static")
    app.include_router(pages.router)
    app.include_router(api.router, prefix="/api")

    return app

def create_app_from_env() -> FastAPI:
    """
    `create_app`, reading `db_path` from the `TRENDIFY_DB_PATH` environment variable instead of
    a direct argument.

    This exists solely as a uvicorn ASGI factory target (`uvicorn.run(..., factory=True)`) for
    `trendify serve --reload`: uvicorn's reload mode re-imports the app in a fresh subprocess on
    every file change, so it needs an importable `module:attribute` string rather than a live app
    instance; there is no other way to hand it a runtime value like `db_path` across that
    re-import boundary.
    """
    db_path = os.environ[_DB_PATH_ENV_VAR]
    return create_app(Path(db_path))

def build_tag_tree(store: RecordStore) -> list[TagNode]:
    """
    Walks `store.tag_tree()`'s flat, tuple-encoded tags into a nested `TagNode` hierarchy: a
    tag `("a", "b")` becomes a folder `"a"` containing a leaf `"b"`. A tuple prefix (e.g. `"a"`
    from that same tag) may never itself have been used as a real tag, so `has_records` is
    tracked separately from `children` being non-empty, since a node can be a pure folder, a
    pure leaf, or both at once.
    """
    root: dict[str, _TrieNode] = {}

    for tag in store.tag_tree():
        parts = tag_to_path_parts(tag)
        level = root
        node: _TrieNode | None = None
        for part in parts:
            node = level.setdefault(part, _TrieNode())
            level = node.children
        if node is not None:
            node.tag = tag

    sizes = store.get_tag_byte_sizes()

    def to_nodes(level: dict[str, _TrieNode]) -> list[TagNode]:
        nodes = []
        for label, node in level.items():
            tag = node.tag
            nodes.append(
                TagNode(
                    key=tag if tag is not None else label,
                    label=label,
                    children=to_nodes(node.children),
                    has_records=tag is not None,
                    record_kinds=_record_kinds(store, tag) if tag is not None else [],
                    size_bytes=sizes.get(tag, 0) if tag is not None else 0,
                )
            )
        # Two stable sorts: alphabetical first, then by category. The second sort's stability
        # preserves the alphabetical order within each category group.
        nodes.sort(key=lambda n: n.label.lower())
        nodes.sort(key=_category_rank)
        return nodes

    return to_nodes(root)