dennisthiessenandClaude Opus 5
22bee28ac7
Job outcomes lived only in scheduler._job_runtime, an in-memory dict. Every
deploy wiped it, so Admin -> Jobs could report "Active" with no indication a job
had ever run or how it ended -- which is the main thing that page is for.
New job_run_state table (migration 031): one row per job, upserted on job_name.
Deliberately not history -- system_events already grows unbounded with no
retention job, and a second append-only operational table would repeat that
debt. Adding history later is purely additive.
Written from two hooks, NOT from _runtime_finish. That looked cheapest (one
function, ~40 call sites) but unit tests invoke job coroutines directly, so it
would fire detached DB writes at the real session factory throughout the suite,
and there is no testing flag to guard on.
- An APScheduler EVENT_JOB_EXECUTED/ERROR listener covers everything the
scheduler fires, including manual triggers. Its detached task is held in a
module-level set (a bare create_task result can be collected mid-flight) and
drained in the app lifespan before engine.dispose().
- _run_pipeline persists directly, and must: pipeline steps are plain
coroutine calls that emit no scheduler events, so the listener cannot see
them. The step persist sits AFTER the except that swallows step errors --
inside it, exactly the failed runs worth seeing would be skipped. The
orchestrator persists in the finally, and the disabled early-return persists
too, or "skipped" is silently dropped.
_persist_job_run never raises: a persistence failure must not break an otherwise
successful pipeline.
The API reports this as last_run_* and leaves runtime_* meaning strictly live
in-memory state. Reusing runtime_status would have been a regression, not a
no-op: JobControls drives the status chip from it (a job that errored eight days
ago would read "Last run error" forever instead of "Active") and picks the
rate-limit banner from it (a week-old rate limit would pin the banner
permanently). Tests pin the split.
The table starts empty; each job fills its row the next time it finishes. No
backfill from system_events, which records only warning/error outcomes under a
different status vocabulary and would invent successes that never happened.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
129 lines
4.6 KiB
Python
129 lines
4.6 KiB
Python
"""FastAPI application entry point with lifespan management."""
|
|
# ruff: noqa: E402
|
|
# ---------------------------------------------------------------------------
|
|
# SSL + proxy injection — MUST happen before any HTTP client imports
|
|
# ---------------------------------------------------------------------------
|
|
from app.ssl_bootstrap import bootstrap_ssl
|
|
|
|
bootstrap_ssl()
|
|
|
|
import logging
|
|
import sys
|
|
from contextlib import asynccontextmanager
|
|
from collections.abc import AsyncGenerator
|
|
|
|
from fastapi import FastAPI
|
|
from passlib.hash import bcrypt
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.database import async_session_factory, engine
|
|
from app.middleware import register_exception_handlers
|
|
from app.models.user import User
|
|
from app.scheduler import (
|
|
configure_scheduler,
|
|
flush_job_run_persists,
|
|
load_schedule_config,
|
|
scheduler,
|
|
)
|
|
from app.routers.admin import router as admin_router
|
|
from app.routers.auth import router as auth_router
|
|
from app.routers.health import router as health_router
|
|
from app.routers.ingestion import router as ingestion_router
|
|
from app.routers.ohlcv import router as ohlcv_router
|
|
from app.routers.indicators import router as indicators_router
|
|
from app.routers.fundamentals import router as fundamentals_router
|
|
from app.routers.scores import router as scores_router
|
|
from app.routers.trades import router as trades_router
|
|
from app.routers.watchlist import router as watchlist_router
|
|
from app.routers.sentiment import router as sentiment_router
|
|
from app.routers.sr_levels import router as sr_levels_router
|
|
from app.routers.tickers import router as tickers_router
|
|
from app.routers.jobs import router as jobs_router
|
|
from app.routers.market import router as market_router
|
|
from app.routers.paper_trades import router as paper_trades_router
|
|
|
|
|
|
def _configure_logging() -> None:
|
|
"""Set up structured JSON-style logging."""
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
handler.setFormatter(
|
|
logging.Formatter(
|
|
'{"time":"%(asctime)s","level":"%(levelname)s",'
|
|
'"logger":"%(name)s","message":"%(message)s"}'
|
|
)
|
|
)
|
|
root = logging.getLogger()
|
|
root.handlers.clear()
|
|
root.addHandler(handler)
|
|
root.setLevel(settings.log_level.upper())
|
|
|
|
|
|
async def _create_default_admin(session: AsyncSession) -> None:
|
|
"""Create the default admin account if no admin user exists."""
|
|
result = await session.execute(
|
|
select(User).where(User.role == "admin")
|
|
)
|
|
if result.scalar_one_or_none() is None:
|
|
admin = User(
|
|
username="admin",
|
|
password_hash=bcrypt.hash("admin"),
|
|
role="admin",
|
|
has_access=True,
|
|
)
|
|
session.add(admin)
|
|
await session.commit()
|
|
logging.getLogger(__name__).info("Default admin account created")
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
|
|
"""Manage startup and shutdown lifecycle."""
|
|
logger = logging.getLogger(__name__)
|
|
_configure_logging()
|
|
logger.info("Starting Stock Data Backend")
|
|
|
|
async with async_session_factory() as session:
|
|
await _create_default_admin(session)
|
|
schedule_config = await load_schedule_config(session)
|
|
|
|
configure_scheduler(schedule_config)
|
|
scheduler.start()
|
|
logger.info("Scheduler started")
|
|
|
|
yield
|
|
|
|
scheduler.shutdown(wait=False)
|
|
logger.info("Scheduler stopped")
|
|
# Drain detached last-run writes before the engine goes away, or a job that
|
|
# finished during shutdown loses the row it just wrote.
|
|
await flush_job_run_persists()
|
|
await engine.dispose()
|
|
logger.info("Shutting down")
|
|
|
|
|
|
app = FastAPI(
|
|
title="Stock Data Backend",
|
|
version="0.1.0",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
register_exception_handlers(app)
|
|
app.include_router(health_router, prefix="/api/v1")
|
|
app.include_router(auth_router, prefix="/api/v1")
|
|
app.include_router(admin_router, prefix="/api/v1")
|
|
app.include_router(tickers_router, prefix="/api/v1")
|
|
app.include_router(ohlcv_router, prefix="/api/v1")
|
|
app.include_router(ingestion_router, prefix="/api/v1")
|
|
app.include_router(indicators_router, prefix="/api/v1")
|
|
app.include_router(sr_levels_router, prefix="/api/v1")
|
|
app.include_router(sentiment_router, prefix="/api/v1")
|
|
app.include_router(fundamentals_router, prefix="/api/v1")
|
|
app.include_router(scores_router, prefix="/api/v1")
|
|
app.include_router(trades_router, prefix="/api/v1")
|
|
app.include_router(watchlist_router, prefix="/api/v1")
|
|
app.include_router(jobs_router, prefix="/api/v1")
|
|
app.include_router(market_router, prefix="/api/v1")
|
|
app.include_router(paper_trades_router, prefix="/api/v1")
|