From db7b28ba697f971de0f2a9091ed886ad1d9ee595 Mon Sep 17 00:00:00 2001 From: Thomas Schmelzer Date: Tue, 28 Jul 2026 13:10:01 +0400 Subject: [PATCH] refactor: annotate remaining functions and arguments Rhiza v1.2.2 adds flake8-annotations (ANN) to the ruff selection and runs mypy --strict, both of which flag the same missing annotations. Manual annotations: server.py get_response(scope), health_check, not_found_handler, server_error_handler lifecycle.py __enter__ -> "ChartServer", __exit__ + its three params routes.py get_data -> DataResponse ingestion.py slice_opt(arr) -> list[Any] | None demo.py generate_ohlc -> tuple[...] Plus ten `-> None`/`-> str` returns applied by ruff's own fix, matching the pre-commit hook's --unsafe-fixes configuration. Two supporting changes: - `scope` is typed MutableMapping[str, Any] (what starlette.types.Scope aliases to) rather than importing starlette, which is not a declared dependency and would trip deptry. - JSONResponse moves to the module imports so the handler return types can reference it, removing two function-local re-imports. ANN now clean; mypy no-untyped-def/no-untyped-call drop from 13 to 0 (total mypy errors 55 -> 43, remainder tracked by #67 and #68). Closes #66 Co-Authored-By: Claude Opus 5 (1M context) --- demo.py | 9 ++++++--- src/pycharting/api/routes.py | 2 +- src/pycharting/core/lifecycle.py | 22 ++++++++++++++-------- src/pycharting/core/server.py | 20 +++++++++----------- src/pycharting/data/ingestion.py | 4 ++-- 5 files changed, 32 insertions(+), 25 deletions(-) diff --git a/demo.py b/demo.py index 16272f4..9c44cf6 100644 --- a/demo.py +++ b/demo.py @@ -7,6 +7,7 @@ """ import time +from typing import Any import numpy as np import pandas as pd @@ -42,7 +43,9 @@ def rsi_like(values: np.ndarray, period: int = 14) -> np.ndarray: return rsi -def generate_ohlc(n: int = 1000): +def generate_ohlc( + n: int = 1000, +) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray, dict[str, np.ndarray], dict[str, Any]]: """Generate synthetic OHLC data with indicators.""" base = 100.0 noise = np.random.randn(n) @@ -72,7 +75,7 @@ def generate_ohlc(n: int = 1000): return open_, high, low, close, overlays, subplots -def run_demo(choice: str): +def run_demo(choice: str) -> None: """Run the selected demo scenario identified by ``choice``.""" n = 5000 # Default size @@ -174,7 +177,7 @@ def run_demo(choice: str): print("Invalid choice.") -def main(): +def main() -> None: """Run the interactive demo menu loop.""" try: while True: diff --git a/src/pycharting/api/routes.py b/src/pycharting/api/routes.py index 9c79fea..28642e5 100644 --- a/src/pycharting/api/routes.py +++ b/src/pycharting/api/routes.py @@ -97,7 +97,7 @@ async def get_data( start_index: int = Query(0, ge=0, description="Start index for data slice"), end_index: int | None = Query(None, ge=0, description="End index for data slice"), session_id: str = Query("default", description="Session identifier for data source"), -): +) -> DataResponse: """Retrieve a specific slice of OHLC data. This endpoint is optimized for high-performance frontend rendering. Instead of sending the full dataset diff --git a/src/pycharting/core/lifecycle.py b/src/pycharting/core/lifecycle.py index ab32d56..f1bbbbd 100644 --- a/src/pycharting/core/lifecycle.py +++ b/src/pycharting/core/lifecycle.py @@ -15,6 +15,7 @@ import threading import time from datetime import datetime +from types import TracebackType from typing import Any import uvicorn @@ -45,7 +46,7 @@ def __init__( host: str = "127.0.0.1", port: int | None = None, auto_shutdown_timeout: float = 5.0, - ): + ) -> None: """Initialize the ChartServer controller. Args: @@ -70,11 +71,11 @@ def __init__( self.app = create_app() self._add_websocket_endpoint() - def _add_websocket_endpoint(self): + def _add_websocket_endpoint(self) -> None: """Add WebSocket heartbeat endpoint to the app.""" @self.app.websocket("/ws/heartbeat") - async def websocket_heartbeat(websocket: WebSocket): + async def websocket_heartbeat(websocket: WebSocket) -> None: """WebSocket endpoint for connection monitoring.""" await websocket.accept() self._websocket_connected = True @@ -96,7 +97,7 @@ async def websocket_heartbeat(websocket: WebSocket): logger.exception("WebSocket error") self._websocket_connected = False - def _monitor_connection(self): + def _monitor_connection(self) -> None: """Monitor WebSocket connection and trigger auto-shutdown if needed.""" while self._running and not self._shutdown_event.is_set(): time.sleep(1) @@ -119,7 +120,7 @@ def _monitor_connection(self): self.stop_server() break - def _run_server(self): + def _run_server(self) -> None: """Run the Uvicorn server (called in background thread).""" config = uvicorn.Config( self.app, @@ -187,7 +188,7 @@ def start_server(self) -> dict[str, Any]: "running": self._running, } - def stop_server(self): + def stop_server(self) -> None: """Gracefully stop the background server and monitor threads. This method signals the server to shut down, closes the Uvicorn loop, @@ -234,12 +235,17 @@ def server_info(self) -> dict[str, Any]: "last_heartbeat": self._last_heartbeat.isoformat() if self._last_heartbeat else None, } - def __enter__(self): + def __enter__(self) -> "ChartServer": """Context manager entry.""" self.start_server() return self - def __exit__(self, exc_type, exc_val, exc_tb): + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: TracebackType | None, + ) -> bool: """Context manager exit.""" self.stop_server() return False diff --git a/src/pycharting/core/server.py b/src/pycharting/core/server.py index a5cd1af..f096fbd 100644 --- a/src/pycharting/core/server.py +++ b/src/pycharting/core/server.py @@ -15,19 +15,21 @@ import logging import socket +from collections.abc import MutableMapping from pathlib import Path +from typing import Any import uvicorn -from fastapi import FastAPI, Response +from fastapi import FastAPI, Request, Response from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import HTMLResponse +from fastapi.responses import HTMLResponse, JSONResponse from fastapi.staticfiles import StaticFiles class NoCacheStaticFiles(StaticFiles): """Custom StaticFiles that adds no-cache headers for development.""" - async def get_response(self, path: str, scope) -> Response: + async def get_response(self, path: str, scope: MutableMapping[str, Any]) -> Response: """Return the static file response with caching disabled.""" response = await super().get_response(path, scope) # Add no-cache headers to prevent browser caching during development @@ -143,7 +145,7 @@ def create_app() -> FastAPI: # Root endpoint @app.get("/", response_class=HTMLResponse) - async def root(): + async def root() -> str: """Serve the main chart page.""" return """ @@ -204,23 +206,19 @@ async def root(): # Health check endpoint @app.get("/health") - async def health_check(): + async def health_check() -> dict[str, str]: """Health check endpoint.""" return {"status": "healthy", "service": "pycharting"} # Error handlers @app.exception_handler(404) - async def not_found_handler(request, exc): + async def not_found_handler(request: Request, exc: Exception) -> JSONResponse: """Handle 404 errors.""" - from fastapi.responses import JSONResponse - return JSONResponse(status_code=404, content={"error": "Not found", "path": str(request.url.path)}) @app.exception_handler(500) - async def server_error_handler(request, exc): # pragma: no cover + async def server_error_handler(request: Request, exc: Exception) -> JSONResponse: # pragma: no cover """Handle 500 errors.""" - from fastapi.responses import JSONResponse - logger.error(f"Server error: {exc}") return JSONResponse(status_code=500, content={"error": "Internal server error"}) diff --git a/src/pycharting/data/ingestion.py b/src/pycharting/data/ingestion.py index 448b28f..18448ef 100644 --- a/src/pycharting/data/ingestion.py +++ b/src/pycharting/data/ingestion.py @@ -246,7 +246,7 @@ def __init__( overlays: dict[str, pd.Series | np.ndarray | list] | None = None, subplots: dict[str, SubplotSpec] | None = None, trades: pd.Series | np.ndarray | list | None = None, - ): + ) -> None: """Validate the supplied series and store the normalized arrays for fast slicing.""" # Validate input and get normalized arrays validated = validate_input(index, open, high, low, close, overlays, subplots, trades) @@ -362,7 +362,7 @@ def get_chunk( index_list = index_slice.tolist() # Helper for slicing optional arrays - def slice_opt(arr): + def slice_opt(arr: np.ndarray | None) -> list[Any] | None: """Return the requested slice of ``arr`` as a list, or ``None`` if ``arr`` is ``None``.""" return arr[start_index:end_index].tolist() if arr is not None else None