diff --git a/alembic.ini b/alembic.ini new file mode 100644 index 0000000..d6b82fa --- /dev/null +++ b/alembic.ini @@ -0,0 +1,147 @@ +# A generic, single database configuration. + +[alembic] +# path to migration scripts. +# this is typically a path given in POSIX (e.g. forward slashes) +# format, relative to the token %(here)s which refers to the location of this +# ini file +script_location = %(here)s/alembic + +# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s +# Uncomment the line below if you want the files to be prepended with date and time +# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file +# for all available tokens +# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s + +# sys.path path, will be prepended to sys.path if present. +# defaults to the current working directory. for multiple paths, the path separator +# is defined by "path_separator" below. +prepend_sys_path = . + + +# timezone to use when rendering the date within the migration file +# as well as the filename. +# If specified, requires the python>=3.9 or backports.zoneinfo library and tzdata library. +# Any required deps can installed by adding `alembic[tz]` to the pip requirements +# string value is passed to ZoneInfo() +# leave blank for localtime +# timezone = + +# max length of characters to apply to the "slug" field +# truncate_slug_length = 40 + +# set to 'true' to run the environment during +# the 'revision' command, regardless of autogenerate +# revision_environment = false + +# set to 'true' to allow .pyc and .pyo files without +# a source .py file to be detected as revisions in the +# versions/ directory +# sourceless = false + +# version location specification; This defaults +# to /versions. When using multiple version +# directories, initial revisions must be specified with --version-path. +# The path separator used here should be the separator specified by "path_separator" +# below. +# version_locations = %(here)s/bar:%(here)s/bat:%(here)s/alembic/versions + +# path_separator; This indicates what character is used to split lists of file +# paths, including version_locations and prepend_sys_path within configparser +# files such as alembic.ini. +# The default rendered in new alembic.ini files is "os", which uses os.pathsep +# to provide os-dependent path splitting. +# +# Note that in order to support legacy alembic.ini files, this default does NOT +# take place if path_separator is not present in alembic.ini. If this +# option is omitted entirely, fallback logic is as follows: +# +# 1. Parsing of the version_locations option falls back to using the legacy +# "version_path_separator" key, which if absent then falls back to the legacy +# behavior of splitting on spaces and/or commas. +# 2. Parsing of the prepend_sys_path option falls back to the legacy +# behavior of splitting on spaces, commas, or colons. +# +# Valid values for path_separator are: +# +# path_separator = : +# path_separator = ; +# path_separator = space +# path_separator = newline +# +# Use os.pathsep. Default configuration used for new projects. +path_separator = os + +# set to 'true' to search source files recursively +# in each "version_locations" directory +# new in Alembic version 1.10 +# recursive_version_locations = false + +# the output encoding used when revision files +# are written from script.py.mako +# output_encoding = utf-8 + +# database URL. This is consumed by the user-maintained env.py script only. +# other means of configuring database URLs may be customized within the env.py +# file. +sqlalchemy.url = + + +[post_write_hooks] +# post_write_hooks defines scripts or Python functions that are run +# on newly generated revision scripts. See the documentation for further +# detail and examples + +# format using "black" - use the console_scripts runner, against the "black" entrypoint +# hooks = black +# black.type = console_scripts +# black.entrypoint = black +# black.options = -l 79 REVISION_SCRIPT_FILENAME + +# lint with attempts to fix using "ruff" - use the module runner, against the "ruff" module +# hooks = ruff +# ruff.type = module +# ruff.module = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Alternatively, use the exec runner to execute a binary found on your PATH +# hooks = ruff +# ruff.type = exec +# ruff.executable = ruff +# ruff.options = check --fix REVISION_SCRIPT_FILENAME + +# Logging configuration. This is also consumed by the user-maintained +# env.py script only. +[loggers] +keys = root,sqlalchemy,alembic + +[handlers] +keys = console + +[formatters] +keys = generic + +[logger_root] +level = WARNING +handlers = console +qualname = + +[logger_sqlalchemy] +level = WARNING +handlers = +qualname = sqlalchemy.engine + +[logger_alembic] +level = INFO +handlers = +qualname = alembic + +[handler_console] +class = StreamHandler +args = (sys.stderr,) +level = NOTSET +formatter = generic + +[formatter_generic] +format = %(levelname)-5.5s [%(name)s] %(message)s +datefmt = %H:%M:%S diff --git a/alembic/README b/alembic/README new file mode 100644 index 0000000..98e4f9c --- /dev/null +++ b/alembic/README @@ -0,0 +1 @@ +Generic single-database configuration. \ No newline at end of file diff --git a/alembic/env.py b/alembic/env.py new file mode 100644 index 0000000..e24f909 --- /dev/null +++ b/alembic/env.py @@ -0,0 +1,66 @@ +"""Alembic environment configuration for async SQLAlchemy.""" + +import asyncio +from logging.config import fileConfig + +from sqlalchemy import pool +from sqlalchemy.engine import Connection +from sqlalchemy.ext.asyncio import async_engine_from_config, create_async_engine + +from alembic import context + +# Import all models so that Base.metadata is fully populated +from app.core.database import Base, engine +from app.models import financial, error_log, request_log, fred_data, filing, finra_short_volume, alpaca_price +from app.models import etf # ETF models + +# this is the Alembic Config object +config = context.config + +# Interpret the config file for Python logging. +if config.config_file_name is not None: + fileConfig(config.config_file_name) + +target_metadata = Base.metadata + + +def run_migrations_offline() -> None: + """Run migrations in 'offline' mode.""" + from app.core.config import settings + url = settings.DATABASE_URL + context.configure( + url=url, + target_metadata=target_metadata, + literal_binds=True, + dialect_opts={"paramstyle": "named"}, + ) + + with context.begin_transaction(): + context.run_migrations() + + +def do_run_migrations(connection: Connection) -> None: + context.configure(connection=connection, target_metadata=target_metadata) + with context.begin_transaction(): + context.run_migrations() + + +async def run_async_migrations() -> None: + """Run migrations in 'online' mode using the app's async engine.""" + connectable = engine + + async with connectable.connect() as connection: + await connection.run_sync(do_run_migrations) + + await connectable.dispose() + + +def run_migrations_online() -> None: + """Run migrations in 'online' mode.""" + asyncio.run(run_async_migrations()) + + +if context.is_offline_mode(): + run_migrations_offline() +else: + run_migrations_online() diff --git a/alembic/script.py.mako b/alembic/script.py.mako new file mode 100644 index 0000000..1101630 --- /dev/null +++ b/alembic/script.py.mako @@ -0,0 +1,28 @@ +"""${message} + +Revision ID: ${up_revision} +Revises: ${down_revision | comma,n} +Create Date: ${create_date} + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +${imports if imports else ""} + +# revision identifiers, used by Alembic. +revision: str = ${repr(up_revision)} +down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)} +branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)} +depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)} + + +def upgrade() -> None: + """Upgrade schema.""" + ${upgrades if upgrades else "pass"} + + +def downgrade() -> None: + """Downgrade schema.""" + ${downgrades if downgrades else "pass"} diff --git a/alembic/versions/5c3d0565afcc_initial_schema.py b/alembic/versions/5c3d0565afcc_initial_schema.py new file mode 100644 index 0000000..f76009d --- /dev/null +++ b/alembic/versions/5c3d0565afcc_initial_schema.py @@ -0,0 +1,34 @@ +"""Initial schema + +Revision ID: 5c3d0565afcc +Revises: +Create Date: 2026-03-12 16:24:02.052399 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = '5c3d0565afcc' +down_revision: Union[str, Sequence[str], None] = None +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema. + + This is a placeholder initial migration. The actual tables are created + by ``Base.metadata.create_all()`` during application startup. Future + migrations will be autogenerated against the live database once it is + running on PostgreSQL. + """ + pass + + +def downgrade() -> None: + """Downgrade schema.""" + pass diff --git a/app/api/v1/endpoints/alpaca.py b/app/api/v1/endpoints/alpaca.py index 326a682..63119ed 100644 --- a/app/api/v1/endpoints/alpaca.py +++ b/app/api/v1/endpoints/alpaca.py @@ -17,11 +17,13 @@ from app.models.alpaca_price import AlpacaPriceData from app.schemas.financial import ( PriceDataResponse, AlpacaPriceDataPoint, + AlpacaBarsResponse, + AlpacaIntradayResponse, ErrorType, ) from app.services.alpaca_client import AlpacaClient from app.services.alpaca_price_service import AlpacaPriceService -from app.utils.cache import build_cache_key, get_cached_response, set_cached_response +from app.utils.cache import build_cache_key, get_cached_response, set_cached_response, with_cache router = APIRouter() @@ -65,9 +67,11 @@ async def alpaca_status(): @router.get( "/bars/{ticker}", + response_model=AlpacaBarsResponse, summary="Get Alpaca bars (raw, no DB)", description="Fetch historical bars directly from Alpaca without storing in DB.", ) +@with_cache(namespace="alpaca:bars", ttl=None, key_params=["ticker", "interval", "start_date", "end_date", "limit"]) async def get_alpaca_bars( ticker: str, response: Response, @@ -79,18 +83,6 @@ async def get_alpaca_bars( ): svc = _require_alpaca() - # Check cache - cache_key = build_cache_key("alpaca:bars", ticker.upper(), interval, start_date, end_date, limit) - if not force_refresh: - cached = await get_cached_response(cache_key) - if cached: - cached_body, etag = cached - response.headers["X-Cache"] = "HIT" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "redis-cache" - return cached_body - start_dt = datetime.combine(start_date, datetime.min.time()).replace(tzinfo=timezone.utc) if start_date else None end_dt = datetime.combine(end_date, datetime.min.time()).replace(tzinfo=timezone.utc) if end_date else None @@ -109,11 +101,6 @@ async def get_alpaca_bars( "bars": bars, } - etag = await set_cached_response(cache_key, body_dict, ttl_seconds=settings.CACHE_TTL) - response.headers["X-Cache"] = "MISS" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "alpaca-api" return body_dict except Exception as e: raise HTTPException(status_code=502, detail=f"Alpaca API error: {e}") @@ -138,6 +125,7 @@ async def get_alpaca_bars( - Supports: 1m, 5m, 15m, 1h, 1d, 1w, 1mo intervals """, ) +@with_cache(namespace="alpaca:data", ttl=None, key_params=["ticker", "interval", "start_date", "end_date"]) async def get_alpaca_price_data( ticker: str, response: Response, @@ -149,18 +137,6 @@ async def get_alpaca_price_data( ): svc = _require_alpaca() - # Check cache - cache_key = build_cache_key("alpaca:data", ticker.upper(), interval, start_date, end_date) - if not force_refresh: - cached = await get_cached_response(cache_key) - if cached: - cached_body, etag = cached - response.headers["X-Cache"] = "HIT" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "redis-cache" - return cached_body - start_dt = datetime.combine(start_date, datetime.min.time()).replace(tzinfo=timezone.utc) end_dt = datetime.combine(end_date, datetime.min.time()).replace(tzinfo=timezone.utc) @@ -198,7 +174,7 @@ async def get_alpaca_price_data( low=p.low, close=p.close, volume=p.volume, - adjusted_close=p.vwap, # Map vwap β†’ adjusted_close for compatibility + adjusted_close=p.vwap, # Map vwap -> adjusted_close for compatibility data_source=p.data_source, ) for p in alpaca_points @@ -223,12 +199,6 @@ async def get_alpaca_price_data( }, ) - body_dict = body.model_dump() - etag = await set_cached_response(cache_key, body_dict, ttl_seconds=settings.CACHE_TTL) - response.headers["X-Cache"] = "MISS" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "alpaca-api" return body except HTTPException: raise @@ -244,9 +214,11 @@ async def get_alpaca_price_data( @router.get( "/intraday/{ticker}", + response_model=AlpacaIntradayResponse, summary="Get intraday candles from Alpaca", description="Fetch intraday bars (1m, 5m, 15m, 1h) directly from Alpaca. Not stored in DB.", ) +@with_cache(namespace="alpaca:intraday", ttl=300, key_params=["ticker", "interval", "start_date", "end_date"]) async def get_alpaca_intraday( ticker: str, response: Response, @@ -258,18 +230,6 @@ async def get_alpaca_intraday( ): svc = _require_alpaca() - # Check cache (short TTL for intraday) - cache_key = build_cache_key("alpaca:intraday", ticker.upper(), interval, start_date, end_date) - if not force_refresh: - cached = await get_cached_response(cache_key) - if cached: - cached_body, etag = cached - response.headers["X-Cache"] = "HIT" - response.headers["Cache-Control"] = f"public, max-age={INTRADAY_CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "redis-cache" - return cached_body - start_dt = datetime.combine(start_date, datetime.min.time()).replace(tzinfo=timezone.utc) if start_date else None end_dt = datetime.combine(end_date, datetime.min.time()).replace(tzinfo=timezone.utc) if end_date else None @@ -289,11 +249,6 @@ async def get_alpaca_intraday( "candles": bars, } - etag = await set_cached_response(cache_key, body_dict, ttl_seconds=INTRADAY_CACHE_TTL) - response.headers["X-Cache"] = "MISS" - response.headers["Cache-Control"] = f"public, max-age={INTRADAY_CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "alpaca-api" return body_dict except Exception as e: raise HTTPException(status_code=502, detail=f"Alpaca API error: {e}") diff --git a/app/api/v1/endpoints/etf.py b/app/api/v1/endpoints/etf.py index 7e88710..c68ec4d 100644 --- a/app/api/v1/endpoints/etf.py +++ b/app/api/v1/endpoints/etf.py @@ -5,13 +5,14 @@ ETF endpoints (clean and correctly indented) from datetime import datetime, timezone from typing import Optional -from fastapi import APIRouter, HTTPException, Query, Depends +from fastapi import APIRouter, HTTPException, Query, Depends, Response import asyncio import logging from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db +from app.utils.cache import with_cache from app.services.etf_loader_service import etf_loader_service from app.services.etf_holdings_fetcher import etf_holdings_fetcher from app.models.etf import CusipMap, ETFCIKMap, ETFSeriesMap @@ -33,11 +34,14 @@ class ETFHoldingsOut(BaseModel): @router.get("/holdings/{ticker}", response_model=ETFHoldingsOut) +@with_cache(namespace="etf:holdings", ttl=3600, key_params=["ticker", "as_of_date", "top_n", "top_percentage"]) async def get_etf_holdings( ticker: str, + response: Response, as_of_date: Optional[str] = Query(None, description="YYYY-MM-DD"), top_n: Optional[int] = Query(None, description="Return top N holdings by weight/value (mutually exclusive with top_percentage)"), top_percentage: Optional[float] = Query(None, description="Return minimal set covering X percent (e.g., 0.5 or 50 for 50%). Mutually exclusive with top_n"), + force_refresh: bool = Query(False, description="Bypass cache"), db: AsyncSession = Depends(get_db), ): target_dt: Optional[datetime] = None diff --git a/app/api/v1/endpoints/filings.py b/app/api/v1/endpoints/filings.py index d7394eb..73a68d8 100644 --- a/app/api/v1/endpoints/filings.py +++ b/app/api/v1/endpoints/filings.py @@ -6,7 +6,7 @@ import logging from datetime import datetime, timezone from typing import Optional -from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi import APIRouter, Depends, HTTPException, Query, Response from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db @@ -18,14 +18,17 @@ from app.schemas.filing import ( FilingSummary, ) from app.services.sec_filings_service import sec_filings_service +from app.utils.cache import with_cache router = APIRouter() logger = logging.getLogger("app.api.v1.filings") @router.get("/search/{ticker}", response_model=FilingSearchResponse) +@with_cache(namespace="filings:search", ttl=3600, key_params=["ticker", "form_type", "start_date", "end_date", "limit", "offset"]) async def search_filings( ticker: str, + response: Response, form_type: Optional[str] = Query( None, description="Comma-separated form types (e.g. '8-K,6-K'). Default: all supported.", diff --git a/app/api/v1/endpoints/finra.py b/app/api/v1/endpoints/finra.py index 5c90c19..dfc255c 100644 --- a/app/api/v1/endpoints/finra.py +++ b/app/api/v1/endpoints/finra.py @@ -19,7 +19,7 @@ from app.schemas.finra import ( IngestResponse, ) from app.services.finra_short_volume_service import FinraShortVolumeService -from app.utils.cache import build_cache_key, get_cached_response, set_cached_response +from app.utils.cache import build_cache_key, get_cached_response, set_cached_response, with_cache router = APIRouter() @@ -30,6 +30,7 @@ router = APIRouter() summary="Get short volume data for a symbol", description="Query FINRA RegSHO short sale volume. Auto-ingests if data is missing.", ) +@with_cache(namespace="finra:short-volume", ttl=None, key_params=["symbol", "days", "limit"]) async def get_short_volume( symbol: str, response: Response, @@ -38,18 +39,6 @@ async def get_short_volume( force_refresh: bool = Query(False, description="Bypass cache"), db: AsyncSession = Depends(get_db), ): - # Check cache - cache_key = build_cache_key("finra:short-volume", symbol.upper(), days, limit) - if not force_refresh: - cached = await get_cached_response(cache_key) - if cached: - cached_body, etag = cached - response.headers["X-Cache"] = "HIT" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "redis-cache" - return cached_body - svc = FinraShortVolumeService() end = datetime.now(timezone.utc).date() start = end - timedelta(days=days) @@ -71,12 +60,6 @@ async def get_short_volume( }, ) - body_dict = body.model_dump() - etag = await set_cached_response(cache_key, body_dict, ttl_seconds=settings.CACHE_TTL) - response.headers["X-Cache"] = "MISS" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "finra-db" return body @@ -86,6 +69,7 @@ async def get_short_volume( summary="Get short ratio history for a symbol", description="Return daily short_ratio (aggregated across markets) for the last N days.", ) +@with_cache(namespace="finra:short-ratio", ttl=None, key_params=["symbol", "days"]) async def get_short_ratio( symbol: str, response: Response, @@ -93,18 +77,6 @@ async def get_short_ratio( force_refresh: bool = Query(False, description="Bypass cache"), db: AsyncSession = Depends(get_db), ): - # Check cache - cache_key = build_cache_key("finra:short-ratio", symbol.upper(), days) - if not force_refresh: - cached = await get_cached_response(cache_key) - if cached: - cached_body, etag = cached - response.headers["X-Cache"] = "HIT" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "redis-cache" - return cached_body - svc = FinraShortVolumeService() history = await svc.get_short_ratio_history(db, symbol=symbol, days=days) @@ -120,12 +92,6 @@ async def get_short_ratio( metadata={"days_requested": days, "data_points": len(points)}, ) - body_dict = body.model_dump() - etag = await set_cached_response(cache_key, body_dict, ttl_seconds=settings.CACHE_TTL) - response.headers["X-Cache"] = "MISS" - response.headers["Cache-Control"] = f"public, max-age={settings.CACHE_TTL}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "finra-db" return body diff --git a/app/api/v1/endpoints/health.py b/app/api/v1/endpoints/health.py index d6c0509..9b20f7b 100644 --- a/app/api/v1/endpoints/health.py +++ b/app/api/v1/endpoints/health.py @@ -2,7 +2,8 @@ Health check endpoint """ -from datetime import datetime +import logging +from datetime import datetime, timezone from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import text @@ -12,6 +13,8 @@ from app.core.database import get_db from app.core.config import settings from app.schemas.financial import HealthCheckResponse +logger = logging.getLogger(__name__) + router = APIRouter() @router.get( @@ -22,16 +25,16 @@ router = APIRouter() ) async def health_check(db: AsyncSession = Depends(get_db)): """Health check endpoint""" - + # Check database db_status = "unhealthy" try: result = await db.execute(text("SELECT 1")) if result.scalar(): db_status = "healthy" - except Exception: - pass - + except Exception as e: + logger.warning("Database health check failed: %s", e) + # Check Redis cache cache_status = "unhealthy" try: @@ -39,22 +42,22 @@ async def health_check(db: AsyncSession = Depends(get_db)): await r.ping() cache_status = "healthy" await r.close() - except Exception: - pass - + except Exception as e: + logger.warning("Redis health check failed: %s", e) + # Check SEC data availability sec_available = True # Simplified for now - + # Overall status overall_status = "healthy" if db_status != "healthy" or cache_status != "healthy": overall_status = "degraded" - + return HealthCheckResponse( status=overall_status, version=settings.APP_VERSION, database=db_status, cache=cache_status, sec_data_available=sec_available, - timestamp=datetime.utcnow() + timestamp=datetime.now(timezone.utc) ) \ No newline at end of file diff --git a/app/api/v1/endpoints/news.py b/app/api/v1/endpoints/news.py index 630ae15..0575ef7 100644 --- a/app/api/v1/endpoints/news.py +++ b/app/api/v1/endpoints/news.py @@ -6,10 +6,12 @@ from datetime import datetime from typing import Optional, Dict, Any, List import logging -from fastapi import APIRouter, HTTPException, Query +from fastapi import APIRouter, HTTPException, Query, Response from pydantic import BaseModel, Field +from app.schemas.financial import NewsOnlyResponse, SocialOnlyResponse from app.services.news_social_service import news_social_service +from app.utils.cache import with_cache logger = logging.getLogger(__name__) @@ -74,12 +76,15 @@ class NewsSocialResponse(BaseModel): @router.get("/{ticker}", response_model=NewsSocialResponse) +@with_cache(namespace="news:full", ttl=600, key_params=["ticker", "days_back", "max_articles", "max_social_posts", "include_social"]) async def get_ticker_news_and_social( ticker: str, + response: Response, days_back: int = Query(7, ge=1, le=30, description="Number of days to look back for articles (1-30)"), max_articles: int = Query(20, ge=1, le=100, description="Maximum number of news articles to return (1-100)"), max_social_posts: int = Query(15, ge=0, le=50, description="Maximum number of social media posts to return (0-50)"), - include_social: bool = Query(True, description="Whether to include social media data") + include_social: bool = Query(True, description="Whether to include social media data"), + force_refresh: bool = Query(False, description="Bypass cache and fetch fresh data"), ): """ Get comprehensive news and social media data for a ticker @@ -143,11 +148,14 @@ async def get_ticker_news_and_social( ) -@router.get("/{ticker}/news-only", response_model=Dict[str, Any]) +@router.get("/{ticker}/news-only", response_model=NewsOnlyResponse) +@with_cache(namespace="news:news-only", ttl=600, key_params=["ticker", "days_back", "max_articles"]) async def get_ticker_news_only( ticker: str, + response: Response, days_back: int = Query(7, ge=1, le=30, description="Number of days to look back for articles (1-30)"), - max_articles: int = Query(30, ge=1, le=100, description="Maximum number of news articles to return (1-100)") + max_articles: int = Query(30, ge=1, le=100, description="Maximum number of news articles to return (1-100)"), + force_refresh: bool = Query(False, description="Bypass cache and fetch fresh data"), ): """ Get only news articles for a ticker (faster endpoint without social media data) @@ -209,11 +217,14 @@ async def get_ticker_news_only( ) -@router.get("/{ticker}/social-only", response_model=Dict[str, Any]) +@router.get("/{ticker}/social-only", response_model=SocialOnlyResponse) +@with_cache(namespace="news:social-only", ttl=600, key_params=["ticker", "days_back", "max_social_posts"]) async def get_ticker_social_only( ticker: str, + response: Response, days_back: int = Query(7, ge=1, le=30, description="Number of days to look back for posts (1-30)"), - max_social_posts: int = Query(20, ge=1, le=50, description="Maximum number of social media posts to return (1-50)") + max_social_posts: int = Query(20, ge=1, le=50, description="Maximum number of social media posts to return (1-50)"), + force_refresh: bool = Query(False, description="Bypass cache and fetch fresh data"), ): """ Get only social media posts for a ticker diff --git a/app/api/v1/endpoints/stocks.py b/app/api/v1/endpoints/stocks.py index 35da68c..3393b2c 100644 --- a/app/api/v1/endpoints/stocks.py +++ b/app/api/v1/endpoints/stocks.py @@ -11,14 +11,14 @@ from datetime import datetime from app.services.yahoo_most_active_service import yahoo_most_active_service from app.services.yahoo_52week_gainers_service import yahoo_52week_gainers_service -from app.utils.cache import build_cache_key, get_cached_response, set_cached_response -from app.core.config import settings +from app.utils.cache import with_cache router = APIRouter() logger = logging.getLogger("app.api.v1.stocks") @router.get("/most-active") +@with_cache(namespace="stocks:most-active", ttl=3600, key_params=["limit"]) async def get_most_active_stocks( response: Response, limit: Optional[int] = Query(None, ge=1, le=500, description="Maximum number of stocks to return (1-500). If not specified, returns all available stocks."), @@ -79,23 +79,6 @@ async def get_most_active_stocks( - Uses advanced rate limiting bypass techniques (curl_cffi + Chrome impersonation) """ try: - # Build cache key by limit parameter - cache_key = build_cache_key( - "stocks:most-active", - f"limit={limit}" if limit is not None else "limit=all" - ) - - # Try cache (skip if force_refresh) - if not force_refresh: - cached = await get_cached_response(cache_key) - if cached: - cached_body, etag = cached - response.headers["X-Cache"] = "HIT" - response.headers["Cache-Control"] = f"public, max-age={3600}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "redis-cache" - return cached_body - # Fetch fresh data if limit is None: logger.info("πŸ“Š Getting ALL most active stocks (no limit specified)") @@ -121,13 +104,6 @@ async def get_most_active_stocks( **result } - # Set cache after successful fetch - etag = await set_cached_response(cache_key, response_body, ttl_seconds=3600) - response.headers["X-Cache"] = "MISS" if not force_refresh else "BYPASS" - response.headers["Cache-Control"] = f"public, max-age={3600}" - response.headers["ETag"] = etag - response.headers["X-Data-Source"] = "scraper" - return response_body except HTTPException: diff --git a/app/models/financial.py b/app/models/financial.py index 450374c..8e534fd 100644 --- a/app/models/financial.py +++ b/app/models/financial.py @@ -67,6 +67,7 @@ class FinancialData(Base): UniqueConstraint('ticker', 'period_date', 'period_type', name='uq_financial_data'), Index('idx_financial_ticker_date', 'ticker', 'period_date'), Index('idx_financial_period', 'period_date', 'period_type'), + Index('idx_financial_data_source', 'data_source'), ) class CalculatedMetrics(Base): @@ -153,6 +154,7 @@ class PriceData(Base): __table_args__ = ( UniqueConstraint('ticker', 'date', name='uq_price_data'), Index('idx_price_ticker_date', 'ticker', 'date'), + Index('idx_price_data_source', 'data_source'), ) class DataUpdateLog(Base): diff --git a/app/schemas/error_log.py b/app/schemas/error_log.py index 2f97e01..85dd79f 100644 --- a/app/schemas/error_log.py +++ b/app/schemas/error_log.py @@ -4,7 +4,7 @@ Pydantic schemas for error logs from datetime import datetime from typing import Optional, Dict, List, Any -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, ConfigDict class ErrorLogBase(BaseModel): @@ -34,8 +34,7 @@ class ErrorLogResponse(ErrorLogBase): resolution_notes: Optional[str] = None created_at: str - class Config: - from_attributes = True + model_config = ConfigDict(from_attributes=True) class ErrorLogListResponse(BaseModel): diff --git a/app/schemas/financial.py b/app/schemas/financial.py index 0adbf7e..2428fdd 100644 --- a/app/schemas/financial.py +++ b/app/schemas/financial.py @@ -2,17 +2,17 @@ Pydantic schemas for API requests and responses """ -from datetime import datetime, date +from datetime import datetime, date, timezone from typing import Optional, Dict, List, Any -from pydantic import BaseModel, Field, ConfigDict, validator +from pydantic import BaseModel, Field, ConfigDict, field_validator from enum import Enum -import uuid import re from .validators import ( - validate_period_field, - validate_quarters_field, - validate_time_approaches, - validate_end_date_field + validate_period_field, + validate_quarters_field, + validate_time_approaches, + validate_end_date_field, + validate_interval_field, ) class PeriodType(str, Enum): @@ -74,21 +74,25 @@ class FinancialDataRequest(BaseModel): include_metrics: bool = Field(True, description="Include calculated metrics in response") force_refresh: bool = Field(False, description="Force refresh data from SEC") - @validator('period') + @field_validator('period') + @classmethod def validate_period(cls, v): return validate_period_field(cls, v) - - @validator('quarters') + + @field_validator('quarters') + @classmethod def validate_quarters(cls, v): return validate_quarters_field(cls, v) - - @validator('start_date') - def validate_time_approaches(cls, v, values): - return validate_time_approaches(cls, v, values) - - @validator('end_date') - def validate_end_date(cls, v, values): - return validate_end_date_field(cls, v, values) + + @field_validator('start_date') + @classmethod + def validate_time_approaches(cls, v, info): + return validate_time_approaches(cls, v, info.data) + + @field_validator('end_date') + @classmethod + def validate_end_date(cls, v, info): + return validate_end_date_field(cls, v, info.data) class BulkFinancialDataRequest(BaseModel): tickers: List[str] = Field(..., min_items=1, max_items=500, description="List of stock ticker symbols (max 500 for efficient bulk processing)") @@ -115,31 +119,29 @@ class BulkFinancialDataRequest(BaseModel): include_metrics: bool = Field(True, description="Include calculated metrics in response") force_refresh: bool = Field(False, description="Force refresh data from SEC") - @validator('quarters') + @field_validator('quarters') + @classmethod def validate_quarters(cls, v): - """Validate quarter format""" - if v: - for quarter in v: - if not re.match(r'^\d{4}Q[1-4]$', quarter): - raise ValueError(f"Invalid quarter format: {quarter}. Expected format: YYYYQN (e.g., 2020Q1)") - return v - - @validator('start_date') - def validate_dates_or_quarters(cls, v, values): + return validate_quarters_field(cls, v) + + @field_validator('start_date') + @classmethod + def validate_dates_or_quarters(cls, v, info): """Ensure either dates or quarters are provided""" - quarters = values.get('quarters') + quarters = info.data.get('quarters') if not v and not quarters: raise ValueError("Either start_date/end_date or quarters must be provided") if v and quarters: raise ValueError("Cannot specify both date range and quarters - use one or the other") return v - - @validator('end_date') - def validate_end_date(cls, v, values): + + @field_validator('end_date') + @classmethod + def validate_end_date(cls, v, info): """Validate end_date if using date-based approach""" - start_date = values.get('start_date') - quarters = values.get('quarters') - + start_date = info.data.get('start_date') + quarters = info.data.get('quarters') + if not quarters: # Using date-based approach if not v: raise ValueError("end_date is required when not using quarters") @@ -182,39 +184,34 @@ class PriceDataRequest(BaseModel): interval: str = Field("1d", description="Data interval: 1d, 1w, 1m, 5d, 1h, etc.") force_refresh: bool = Field(False, description="Force refresh data from Yahoo Finance") - @validator('interval') + @field_validator('interval') + @classmethod def validate_interval(cls, v): - """Validate interval format""" - valid_intervals = ['1m', '2m', '5m', '15m', '30m', '60m', '90m', '1h', '1d', '5d', '1w', '1mo', '3mo'] - if v not in valid_intervals: - raise ValueError(f"Invalid interval: {v}. Valid intervals: {', '.join(valid_intervals)}") - return v - - @validator('quarters') + return validate_interval_field(cls, v) + + @field_validator('quarters') + @classmethod def validate_quarters(cls, v): - """Validate quarter format""" - if v: - for quarter in v: - if not re.match(r'^\d{4}Q[1-4]$', quarter): - raise ValueError(f"Invalid quarter format: {quarter}. Expected format: YYYYQN (e.g., 2020Q1)") - return v - - @validator('start_date') - def validate_dates_or_quarters(cls, v, values): + return validate_quarters_field(cls, v) + + @field_validator('start_date') + @classmethod + def validate_dates_or_quarters(cls, v, info): """Ensure either dates or quarters are provided""" - quarters = values.get('quarters') + quarters = info.data.get('quarters') if not v and not quarters: raise ValueError("Either start_date/end_date or quarters must be provided") if v and quarters: raise ValueError("Cannot specify both date range and quarters - use one or the other") return v - - @validator('end_date') - def validate_end_date(cls, v, values): + + @field_validator('end_date') + @classmethod + def validate_end_date(cls, v, info): """Validate end_date if using date-based approach""" - start_date = values.get('start_date') - quarters = values.get('quarters') - + start_date = info.data.get('start_date') + quarters = info.data.get('quarters') + if not quarters: # Using date-based approach if not v: raise ValueError("end_date is required when not using quarters") @@ -246,39 +243,34 @@ class BulkPriceDataRequest(BaseModel): interval: str = Field("1d", description="Data interval: 1d, 1w, 1m, 5d, 1h, etc.") force_refresh: bool = Field(False, description="Force refresh data from Yahoo Finance") - @validator('interval') + @field_validator('interval') + @classmethod def validate_interval(cls, v): - """Validate interval format""" - valid_intervals = ['1m', '2m', '5m', '15m', '30m', '60m', '90m', '1h', '1d', '5d', '1w', '1mo', '3mo'] - if v not in valid_intervals: - raise ValueError(f"Invalid interval: {v}. Valid intervals: {', '.join(valid_intervals)}") - return v - - @validator('quarters') + return validate_interval_field(cls, v) + + @field_validator('quarters') + @classmethod def validate_quarters(cls, v): - """Validate quarter format""" - if v: - for quarter in v: - if not re.match(r'^\d{4}Q[1-4]$', quarter): - raise ValueError(f"Invalid quarter format: {quarter}. Expected format: YYYYQN (e.g., 2020Q1)") - return v - - @validator('start_date') - def validate_dates_or_quarters(cls, v, values): + return validate_quarters_field(cls, v) + + @field_validator('start_date') + @classmethod + def validate_dates_or_quarters(cls, v, info): """Ensure either dates or quarters are provided""" - quarters = values.get('quarters') + quarters = info.data.get('quarters') if not v and not quarters: raise ValueError("Either start_date/end_date or quarters must be provided") if v and quarters: raise ValueError("Cannot specify both date range and quarters - use one or the other") return v - - @validator('end_date') - def validate_end_date(cls, v, values): + + @field_validator('end_date') + @classmethod + def validate_end_date(cls, v, info): """Validate end_date if using date-based approach""" - start_date = values.get('start_date') - quarters = values.get('quarters') - + start_date = info.data.get('start_date') + quarters = info.data.get('quarters') + if not quarters: # Using date-based approach if not v: raise ValueError("end_date is required when not using quarters") @@ -384,7 +376,8 @@ class PriceDataPoint(BaseModel): adjusted_close: Optional[float] = None data_source: str - @validator('date', pre=True) + @field_validator('date', mode='before') + @classmethod def convert_datetime_to_date(cls, v): """Convert datetime to date if needed""" if isinstance(v, datetime): @@ -405,7 +398,8 @@ class AlpacaPriceDataPoint(BaseModel): trade_count: Optional[int] = None data_source: str = "ALPACA" - @validator('date', pre=True) + @field_validator('date', mode='before') + @classmethod def convert_datetime_to_date(cls, v): if isinstance(v, datetime): return v.date() @@ -448,7 +442,7 @@ class ErrorResponse(BaseModel): error_type: ErrorType message: str detail: Optional[Dict[str, Any]] = None - timestamp: datetime = Field(default_factory=datetime.utcnow) + timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) # New schemas for quote/intraday/today endpoints @@ -525,4 +519,33 @@ class MigrationResponse(BaseModel): migrated_records: int failed_records: int errors: List[Dict[str, Any]] = Field(default_factory=list) - duration_seconds: float \ No newline at end of file + duration_seconds: float + + +class AlpacaBarsResponse(BaseModel): + ticker: str + interval: str + count: int + bars: List[Dict[str, Any]] + + +class AlpacaIntradayResponse(BaseModel): + ticker: str + interval: str + source: str = "ALPACA" + count: int + candles: List[Dict[str, Any]] + + +class NewsOnlyResponse(BaseModel): + ticker: str + retrieved_at: str + news: Dict[str, Any] + summary: Dict[str, Any] + + +class SocialOnlyResponse(BaseModel): + ticker: str + retrieved_at: str + social_media: Dict[str, Any] + summary: Dict[str, Any] \ No newline at end of file diff --git a/app/schemas/validators.py b/app/schemas/validators.py index 8c08c38..7f54867 100644 --- a/app/schemas/validators.py +++ b/app/schemas/validators.py @@ -2,9 +2,17 @@ Additional validators for schema validation """ -from pydantic import validator import re +VALID_INTERVALS = ['1m', '2m', '5m', '15m', '30m', '60m', '90m', '1h', '1d', '5d', '1w', '1mo', '3mo'] + + +def validate_interval_field(cls, v): + """Validate interval format""" + if v not in VALID_INTERVALS: + raise ValueError(f"Invalid interval: {v}. Valid intervals: {', '.join(VALID_INTERVALS)}") + return v + def validate_period_field(cls, v): """Validate period format""" diff --git a/app/services/fred_proxy_service.py b/app/services/fred_proxy_service.py index c0b18fc..29ed5f7 100644 --- a/app/services/fred_proxy_service.py +++ b/app/services/fred_proxy_service.py @@ -25,6 +25,17 @@ class FredProxyService: self.base_url = "https://api.stlouisfed.org/fred" self.daily_limit = 1000 self.cache_duration_hours = 24 # 24μ‹œκ°„ μΊμ‹œ + self._client: Optional[httpx.AsyncClient] = None + + async def _get_client(self) -> httpx.AsyncClient: + if self._client is None or self._client.is_closed: + self._client = httpx.AsyncClient(timeout=30.0) + return self._client + + async def close(self) -> None: + if self._client and not self._client.is_closed: + await self._client.aclose() + self._client = None async def _check_daily_limit(self, db: AsyncSession) -> Tuple[bool, int, int]: """ @@ -275,35 +286,35 @@ class FredProxyService: params['file_type'] = 'json' try: - async with httpx.AsyncClient(timeout=30.0) as client: - response = await client.get(url, params=params) - response.raise_for_status() - - data = response.json() - - # 응닡 크기 μΆ”μ • - response_size = 0 - if 'seriess' in data: - response_size = len(data.get('seriess', [])) - elif 'observations' in data: - response_size = len(data.get('observations', [])) - elif 'categories' in data: - response_size = len(data.get('categories', [])) - elif 'sources' in data: - response_size = len(data.get('sources', [])) - elif 'releases' in data: - response_size = len(data.get('releases', [])) - elif 'tags' in data: - response_size = len(data.get('tags', [])) - else: - response_size = 1 - - logger.info(f"βœ… FRED API success: {endpoint} -> {response_size} records") - return { - 'data': data, - 'response_size': response_size - } - + client = await self._get_client() + response = await client.get(url, params=params) + response.raise_for_status() + + data = response.json() + + # 응닡 크기 μΆ”μ • + response_size = 0 + if 'seriess' in data: + response_size = len(data.get('seriess', [])) + elif 'observations' in data: + response_size = len(data.get('observations', [])) + elif 'categories' in data: + response_size = len(data.get('categories', [])) + elif 'sources' in data: + response_size = len(data.get('sources', [])) + elif 'releases' in data: + response_size = len(data.get('releases', [])) + elif 'tags' in data: + response_size = len(data.get('tags', [])) + else: + response_size = 1 + + logger.info(f"βœ… FRED API success: {endpoint} -> {response_size} records") + return { + 'data': data, + 'response_size': response_size + } + except httpx.HTTPError as e: logger.error(f"❌ FRED API error: {endpoint} -> {e}") return None diff --git a/app/services/sec_http_client.py b/app/services/sec_http_client.py index 8a286de..8fdb20a 100644 --- a/app/services/sec_http_client.py +++ b/app/services/sec_http_client.py @@ -59,6 +59,22 @@ class SECHttpClient: pass self._user_agent = f"{user_agent_name} ({settings.SEC_EMAIL})" self._deadline: Optional[float] = None + self._session: Optional[aiohttp.ClientSession] = None + + async def _get_session(self) -> aiohttp.ClientSession: + """Get or create a persistent session.""" + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession( + timeout=self.http_timeout, + headers={"User-Agent": self._user_agent}, + ) + return self._session + + async def close(self) -> None: + """Close the persistent session.""" + if self._session and not self._session.closed: + await self._session.close() + self._session = None # ------------------------------------------------------------------ # Deadline management @@ -155,51 +171,45 @@ class SECHttpClient: ) async with self._req_sem: try: - async with aiohttp.ClientSession( - timeout=req_timeout, - headers={ - "User-Agent": self._user_agent, - "Accept": "application/json", - }, - ) as session: - async with session.get(url, timeout=req_timeout) as resp: - if resp.status == 429: - retry_after = resp.headers.get("Retry-After") - delay = ( - float(retry_after) - if retry_after and retry_after.isdigit() - else backoff - ) - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) - ) - backoff *= 1.8 - continue - if 500 <= resp.status < 600: - delay = backoff - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) - ) - backoff *= 1.8 - continue - resp.raise_for_status() - data = await resp.json() - # Cache successful response - try: - with open(self._cache_path(url) + ".json", "w", encoding="utf-8") as f: - json.dump(data, f) - except Exception: - pass - self._json_cache[url] = data - return data + session = await self._get_session() + async with session.get(url, timeout=req_timeout, headers={"Accept": "application/json"}) as resp: + if resp.status == 429: + retry_after = resp.headers.get("Retry-After") + delay = ( + float(retry_after) + if retry_after and retry_after.isdigit() + else backoff + ) + if self._deadline is not None: + remaining = max(0.0, self._deadline - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + if 500 <= resp.status < 600: + delay = backoff + if self._deadline is not None: + remaining = max(0.0, self._deadline - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + resp.raise_for_status() + data = await resp.json() + # Cache successful response + try: + with open(self._cache_path(url) + ".json", "w", encoding="utf-8") as f: + json.dump(data, f) + except Exception: + pass + self._json_cache[url] = data + return data except Exception as e: last_exc = e delay = backoff @@ -255,63 +265,57 @@ class SECHttpClient: ) async with self._req_sem: try: - async with aiohttp.ClientSession( - timeout=req_timeout, - headers={ - "User-Agent": self._user_agent, - "Accept": accept, - }, - ) as session: - async with session.get(url, timeout=req_timeout) as resp: - if resp.status == 429: - retry_after = resp.headers.get("Retry-After") - delay = ( - float(retry_after) - if retry_after and retry_after.isdigit() - else backoff - ) - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) - ) - backoff *= 1.8 - continue - if 500 <= resp.status < 600: - delay = backoff - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) - ) - backoff *= 1.8 - continue - resp.raise_for_status() - text = await resp.text() - if _is_sec_block_page(text): - last_exc = RuntimeError("SEC_BLOCKED") - delay = backoff * 2.0 - if self._deadline is not None: - remaining = max(0.0, self._deadline - _time.monotonic()) - delay = min(delay, max(0.0, remaining - 0.05)) - await asyncio.sleep( - max(0.0, delay) - + random.uniform(0.0, delay * 0.5 if delay > 0 else 0.0) - ) - backoff *= 2.0 - continue - # Cache successful response - self._text_cache[url] = text - try: - with open(self._cache_path(url) + ".txt", "w", encoding="utf-8") as f: - f.write(text) - except Exception: - pass - return text + session = await self._get_session() + async with session.get(url, timeout=req_timeout, headers={"Accept": accept}) as resp: + if resp.status == 429: + retry_after = resp.headers.get("Retry-After") + delay = ( + float(retry_after) + if retry_after and retry_after.isdigit() + else backoff + ) + if self._deadline is not None: + remaining = max(0.0, self._deadline - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + if 500 <= resp.status < 600: + delay = backoff + if self._deadline is not None: + remaining = max(0.0, self._deadline - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.25 if delay > 0 else 0.0) + ) + backoff *= 1.8 + continue + resp.raise_for_status() + text = await resp.text() + if _is_sec_block_page(text): + last_exc = RuntimeError("SEC_BLOCKED") + delay = backoff * 2.0 + if self._deadline is not None: + remaining = max(0.0, self._deadline - _time.monotonic()) + delay = min(delay, max(0.0, remaining - 0.05)) + await asyncio.sleep( + max(0.0, delay) + + random.uniform(0.0, delay * 0.5 if delay > 0 else 0.0) + ) + backoff *= 2.0 + continue + # Cache successful response + self._text_cache[url] = text + try: + with open(self._cache_path(url) + ".txt", "w", encoding="utf-8") as f: + f.write(text) + except Exception: + pass + return text except Exception as e: last_exc = e delay = backoff diff --git a/app/utils/cache.py b/app/utils/cache.py index cb971d1..2f3e96a 100644 --- a/app/utils/cache.py +++ b/app/utils/cache.py @@ -10,6 +10,8 @@ Design goals: from __future__ import annotations +import functools +import inspect import logging from typing import Any, Dict, Optional, Tuple import hashlib @@ -134,5 +136,69 @@ async def set_cached_response(key: str, body: Dict[str, Any], ttl_seconds: Optio return etag - +def with_cache(namespace: str, ttl: int = None, key_params: list[str] = None): + """ + FastAPI endpoint caching decorator. + - Detects force_refresh parameter automatically + - Sets X-Cache/Cache-Control/ETag/X-Data-Source headers on Response + - key_params: list of function argument names to include in cache key + """ + def decorator(func): + @functools.wraps(func) + async def wrapper(*args, **kwargs): + # Extract bound arguments + sig = inspect.signature(func) + bound = sig.bind(*args, **kwargs) + bound.apply_defaults() + arguments = bound.arguments + + # Build cache key from key_params + parts = [] + for p in (key_params or []): + val = arguments.get(p) + if val is not None: + parts.append(str(val)) + cache_key = build_cache_key(namespace, *parts) + + # Check force_refresh + force_refresh = arguments.get("force_refresh", False) + + # Get Response object for headers + response = arguments.get("response") + + effective_ttl = ttl if ttl is not None else max(60, int(getattr(settings, "CACHE_TTL", 3600))) + + # Try cache + if not force_refresh: + cached = await get_cached_response(cache_key) + if cached: + cached_body, etag = cached + if response is not None: + response.headers["X-Cache"] = "HIT" + response.headers["Cache-Control"] = f"public, max-age={effective_ttl}" + response.headers["ETag"] = etag + response.headers["X-Data-Source"] = "redis-cache" + return cached_body + + # Cache miss β€” call original function + result = await func(*args, **kwargs) + + # Cache the result + if isinstance(result, dict): + body_dict = result + elif hasattr(result, "model_dump"): + body_dict = result.model_dump() + else: + body_dict = result + + etag = await set_cached_response(cache_key, body_dict, ttl_seconds=effective_ttl) + if response is not None: + response.headers["X-Cache"] = "MISS" + response.headers["Cache-Control"] = f"public, max-age={effective_ttl}" + response.headers["ETag"] = etag + + return result + + return wrapper + return decorator