diff --git a/requirements.txt b/requirements.txt index 7e208e7..f884ea8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -23,4 +23,7 @@ sniffio==1.3.1 starlette==0.48.0 typing-inspection==0.4.2 typing_extensions==4.15.0 -uvicorn==0.37.0 \ No newline at end of file +uvicorn==0.37.0 +certifi==2026.7.22 +httpcore==1.0.9 +httpx==0.28.1 diff --git a/src/main.py b/src/main.py index 2b653dd..e5950ed 100644 --- a/src/main.py +++ b/src/main.py @@ -6,7 +6,7 @@ from typing import Optional,List from pydantic import BaseModel from validator import is_valid_platform,is_valid_token,verify_api_key from secret_manager.fernet_encryption import encrypt_token -from secret_manager import cleanup_secret_files +from secret_manager import cleanup_secret_files, return_credentials from db_module import get_db, create_connection_pool, close_connection_pool, start_healthcheck_thread from simple_logger_handler import setup_logger, LOG_LEVEL import uuid @@ -16,11 +16,16 @@ from contextlib import asynccontextmanager from metrics_server import REQUEST_COUNTER import asyncio from uvicorn_logger_config import LOGGING_CONFIG +import httpx +import os logger = setup_logger(__name__) - api_key_header = APIKeyHeader(name="X-API-Key") +class PullResponse(BaseModel): + messages: list[dict] + count: int + has_more: bool def hash_token(token: str) -> str: logger.debug(f"Hashing token: {token}") @@ -40,22 +45,42 @@ class TokenRequest(BaseModel): async def lifespan(app: FastAPI): logger.info("Starting application...") - logger.info("Creating MySQL connection pool...") - create_connection_pool("ApiPool") + backend_api_internal_url = os.getenv( + "BACKEND_API_INTERNAL_URL", + "http://backend-api-internal.app-notifications.svc.cluster.local:8101", + ) + api_key_backend_api = return_credentials("/etc/secrets/api_key") - start_healthcheck_thread() - logger.info("MySQL healthcheck thread started.") + try: + logger.info("[FastAPI] Creating MySQL connection pool...") + create_connection_pool("ApiPool") - SECRET_PATHS = frozenset({ - "/etc/secrets/encryption_key", - "/etc/secrets/db_username", - "/etc/secrets/db_password" - }) - cleanup_secret_files(SECRET_PATHS) + start_healthcheck_thread() + logger.info("[FastAPI] MySQL healthcheck thread started.") - yield - logger.info("Closing MySQL connection pool...") - close_connection_pool() + app.state.http_client = httpx.AsyncClient ( + base_url=backend_api_internal_url, + headers={"X-API-Key-Internal": api_key_backend_api}, + timeout=httpx.Timeout(15.0,connect=3.0) + ) + logger.info("[FastAPI] HTTP client initialized.") + + SECRET_PATHS = frozenset({ + "/etc/secrets/api_key", + "/etc/secrets/encryption_key", + "/etc/secrets/db_username", + "/etc/secrets/db_password" + }) + cleanup_secret_files(SECRET_PATHS) + + yield + + finally: + logger.info("[FastAPI] Closing MySQL connection pool...") + close_connection_pool() + + await app.state.http_client.aclose() + logger.info("[FastAPI] Closing Http client") api = FastAPI( @@ -69,30 +94,30 @@ api = FastAPI( @api.middleware("http") async def prometheus_middleware(request: Request, call_next): status = 500 - logger.debug(f"Incoming request: {request.method} {request.url.path}") + logger.debug(f"[FastAPI] Incoming request: {request.method} {request.url.path}") try: response = await call_next(request) status = response.status_code - logger.debug(f"Request processed with status {status}") + logger.debug(f"[FastAPI] Request processed with status {status}") except Exception as e: - logger.exception(f"Exception during request processing: {e}") + logger.exception(f"[FastAPI] Exception during request processing: {e}") raise finally: REQUEST_COUNTER.labels(request.method, request.url.path, status).inc() - logger.debug(f"Prometheus counter incremented for {request.method} {request.url.path} {status}") + logger.debug(f"[FastAPI] Prometheus counter incremented for {request.method} {request.url.path} {status}") return response @api.get("/health", tags=["Health"]) def return_health(db=Depends(get_db)): - logger.debug("Health check initiated") + logger.debug("[FastAPI] Health check initiated") try: cursor = db.cursor() cursor.execute("SELECT 1") cursor.fetchone() db_status = "ok" - logger.debug("Database health check passed") + logger.debug("[FastAPI] Database health check passed") except Exception as e: - logger.error(f"Health check DB failed: {e}") + logger.error(f"[FastAPI] Health check DB failed: {e}") db_status = "error" return JSONResponse( @@ -102,19 +127,18 @@ def return_health(db=Depends(get_db)): def verify_api_key_dependency(db=Depends(get_db), api_key: str = Depends(api_key_header)) -> int: - logger.debug(f"Verifying API key: {api_key}") cursor = db.cursor() cursor.execute("SELECT user_id, api_key FROM users WHERE status = 'active'") for user_id, hashed_key in cursor.fetchall(): if verify_api_key(api_key=api_key, hashed=hashed_key): - logger.debug(f"API key verified for user_id={user_id}") + logger.debug(f"[FastAPI] API key verified for user_id={user_id}") return user_id - logger.warning("API key verification failed") - raise HTTPException(status_code=403, detail="Unauthorized here") + logger.warning("[FastAPI] API key verification failed") + raise HTTPException(status_code=403, detail="Unauthorized") @api.exception_handler(StarletteHTTPException) async def custom_http_exception_handler(request,exc): - logger.debug(f"Handling HTTP exception: {exc}") + logger.debug(f"[FastAPI] Handling HTTP exception: {exc}") if exc.status_code == 404: return JSONResponse( status_code=401, @@ -131,15 +155,14 @@ def register_token( db = Depends(get_db), user_id: int = Depends(verify_api_key_dependency) ): - logger.info(f"Registering token for user_id={user_id}, platform={request_data.platform}") - logger.debug(f"Request data: {request_data}") + logger.info(f"[FastAPI] Registering token for user_id={user_id}, platform={request_data.platform}") + logger.debug(f"[FastAPI] Request data: {request_data}") if not is_valid_platform(request_data.platform) or not is_valid_token(request_data.token): - logger.warning(f"Invalid platform or token for user_id={user_id}") + logger.warning(f"[FastAPI] Invalid platform or token for user_id={user_id}") raise HTTPException(status_code=403,detail="Unathorized") secure_token = encrypt_token(request_data.token) hashed_token = hash_token(request_data.token) - logger.debug(f"Secure token: {secure_token}, Hashed token: {hashed_token}") try: cursor = db.cursor() @@ -147,10 +170,10 @@ def register_token( "SELECT * FROM device_tokens WHERE user_id=%s AND hashed_token=%s", (user_id,hashed_token)) existing = cursor.fetchone() - logger.debug(f"Existing token fetched: {existing}") + logger.debug(f"[FastAPI] Existing token fetched: {existing}") if existing: - logger.debug(f"Updating existing token for user_id={user_id}") + logger.debug(f"[FastAPI] Updating existing token for user_id={user_id}") cursor.execute(""" UPDATE device_tokens SET platform=%s, app_ver=%s, locale=%s, topics=%s, last_seen_at=NOW() @@ -164,7 +187,7 @@ def register_token( )) else: token_id = str(uuid.uuid4()) - logger.debug(f"Creating new entry user_id={user_id}, token_id={token_id}") + logger.debug(f"[FastAPI] Creating new entry user_id={user_id}, token_id={token_id}") cursor.execute(""" INSERT INTO device_tokens (token_id, user_id, platform, token, hashed_token, status, app_ver, locale, topics, created_at) @@ -181,9 +204,9 @@ def register_token( )) db.commit() - logger.info(f"Success: Registering token for user_id={user_id}, platform={request_data.platform}") + logger.info(f"[FastAPI] Success: Registering token for user_id={user_id}, platform={request_data.platform}") except Exception as e: - logger.exception(f"Failed to register token for user_id={user_id}: {e}") + logger.exception(f"[FastAPI] Failed to register token for user_id={user_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) return {"status":"registered"} @@ -195,9 +218,9 @@ def unregister_token( db = Depends(get_db), user_id: int = Depends(verify_api_key_dependency) ): - logger.info(f"Unregistering token for user_id={user_id}, platform={request_data.platform}") + logger.info(f"[FastAPI] Unregistering token for user_id={user_id}, platform={request_data.platform}") hashed_token = hash_token(request_data.token) - logger.debug(f"Hashed token for removal: {hashed_token}") + logger.debug(f"[FastAPI] Hashed token for removal: {hashed_token}") try: cursor = db.cursor() cursor.execute(""" @@ -206,13 +229,35 @@ def unregister_token( WHERE hashed_token=%s """, ('expired', hashed_token)) except Exception as e: - logger.exception(f"Failed to unregister token for user_id={user_id}: {e}") + logger.exception(f"[FastAPI] Failed to unregister token for user_id={user_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) - logger.info(f"Success: Unregistering token for user_id={user_id}, platform={request_data.platform}") + logger.info(f"[FastAPI] Success: Unregistering token for user_id={user_id}, platform={request_data.platform}") return {"status":"unregistered"} +@api.get("/internal/pull-notifications", response_model=PullResponse) +async def pull_notifications( + request: Request, + user_id: int = Depends(verify_api_key_dependency) +): + logger.info(f"[FastAPI] User id {user_id} is requesting messages") + client: httpx.AsyncClient = request.app.state.http_client + + try: + resp = await client.post("/internal/pull-notifications",json={"user_id": user_id}) + resp.raise_for_status() + data = PullResponse.model_validate(resp.json()) + except httpx.HTTPStatusError as e: + logger.error(f"[FastAPI] Upstream returned {e.response.status_code} for user_id={user_id}") + raise HTTPException(status_code=502, detail="Backend-API-Internal unavailable") + except ValueError as e: + logger.error(f"[FastAPI] Backend-API-Internal call failed with ValueError for user_id={user_id}: {e!r}") + raise HTTPException(status_code=502, detail="Backend-API-Internal unavailable") + + logger.info(f"[FastAPI] Returning {len(data.messages)} message(s) for user_id={user_id}, has_more={data.has_more}") + return data + async def start_servers(): - logger.debug("Starting servers...") + logger.debug("[FastAPI] Starting servers...") config_main = uvicorn.Config("main:api", host="0.0.0.0", port=8100, log_level=LOG_LEVEL.lower(), log_config=LOGGING_CONFIG) config_metrics = uvicorn.Config("metrics_server:metrics_api", host="0.0.0.0", port=9000, log_level=LOG_LEVEL.lower(),log_config=LOGGING_CONFIG) @@ -220,7 +265,7 @@ async def start_servers(): server_metrics = uvicorn.Server(config_metrics) await asyncio.gather(server_main.serve(), server_metrics.serve()) - logger.debug("Servers started") + logger.debug("[FastAPI] Servers started") if __name__ == "__main__": asyncio.run(start_servers())