From 0506c1619d6820240afce347d7d06bbf5a1a67b9 Mon Sep 17 00:00:00 2001 From: Florian Date: Mon, 12 Oct 2026 11:37:04 +0200 Subject: [PATCH 1/3] Added pulling notifications from new backend service backend-pull-notifications --- requirements.txt | 5 +- src/main.py | 129 ++++++++++++++++++++++++++++++++--------------- 2 files changed, 91 insertions(+), 43 deletions(-) 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()) -- 2.43.0 From 8f6f65b14355524ed41b7beb08cb4d088c563603 Mon Sep 17 00:00:00 2001 From: Florian Date: Mon, 12 Oct 2026 11:37:15 +0200 Subject: [PATCH 2/3] Version bump --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 677f318..c005c50 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,7 +7,7 @@ WORKDIR /app COPY src/ /app/ # Runtime image -FROM nvcr.io/nvidia/distroless/python:3.12-v3.5.1 +FROM nvcr.io/nvidia/distroless/python:3.12-v4.1.6 COPY --from=builder /install /usr/local COPY --from=builder /app /app WORKDIR /app -- 2.43.0 From 4e088547ff205d38310d4f11df250c9aeb13b146 Mon Sep 17 00:00:00 2001 From: Florian Date: Tue, 13 Oct 2026 10:07:01 +0200 Subject: [PATCH 3/3] Added Ruff linting --- pyproject.toml | 28 +++++++++++++++++++++++++ requirements-dev.txt | 1 + src/main.py | 48 ++++++++++++++++++++++--------------------- src/metrics_server.py | 2 +- src/validator.py | 4 +++- 5 files changed, 58 insertions(+), 25 deletions(-) create mode 100644 pyproject.toml create mode 100644 requirements-dev.txt diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9f8f13d --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,28 @@ +[tool.ruff] +line-length = 120 +target-version = "py312" +[tool.ruff.lint] +select = [ + "E", "F", # pycodestyle errors, pyflakes (unused imports/variables, undefined names) + "I", # import sorting (replaces isort) + "B", # bugbear: likely bugs + "ASYNC", # async mistakes + "RUF006", # un-referenced asyncio.create_task (tasks can be garbage-collected) +] +ignore = ["E501"] + +[tool.ruff.lint.isort] +known-first-party = ["db_module", "metrics_server", "rabbitmq_handler", "secret_manager", + "simple_logger_handler", "uvicorn_logger_config", "validator"] + +[tool.ruff.format] +indent-style = "tab" + +[tool.ruff.lint.flake8-bugbear] +extend-immutable-calls = [ + "fastapi.Depends", + "fastapi.Query", + "fastapi.Header", + "fastapi.Body", + "fastapi.Security", +] \ No newline at end of file diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..92a80f4 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1 @@ +ruff==0.17.0 diff --git a/src/main.py b/src/main.py index e5950ed..b4f8856 100644 --- a/src/main.py +++ b/src/main.py @@ -1,23 +1,25 @@ -from fastapi import FastAPI, Depends, HTTPException, Request +import asyncio +import os +import uuid +from contextlib import asynccontextmanager +from hashlib import sha256 +from typing import List, Optional + +import httpx +import uvicorn +from fastapi import Depends, FastAPI, HTTPException, Request from fastapi.responses import JSONResponse from fastapi.security.api_key import APIKeyHeader -from starlette.exceptions import HTTPException as StarletteHTTPException -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, 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 -from hashlib import sha256 -import uvicorn -from contextlib import asynccontextmanager +from starlette.exceptions import HTTPException as StarletteHTTPException + +from db_module import close_connection_pool, create_connection_pool, get_db, start_healthcheck_thread from metrics_server import REQUEST_COUNTER -import asyncio +from secret_manager import cleanup_secret_files, return_credentials +from secret_manager.fernet_encryption import encrypt_token +from simple_logger_handler import LOG_LEVEL, setup_logger from uvicorn_logger_config import LOGGING_CONFIG -import httpx -import os +from validator import is_valid_platform, is_valid_token, verify_api_key logger = setup_logger(__name__) api_key_header = APIKeyHeader(name="X-API-Key") @@ -168,7 +170,7 @@ def register_token( cursor = db.cursor() cursor.execute( "SELECT * FROM device_tokens WHERE user_id=%s AND hashed_token=%s", - (user_id,hashed_token)) + (user_id,hashed_token)) existing = cursor.fetchone() logger.debug(f"[FastAPI] Existing token fetched: {existing}") @@ -178,8 +180,8 @@ def register_token( UPDATE device_tokens SET platform=%s, app_ver=%s, locale=%s, topics=%s, last_seen_at=NOW() WHERE user_id=%s AND hashed_token=%s - """, (request_data.platform, - request_data.app_ver, + """,(request_data.platform, + request_data.app_ver, request_data.locale, request_data.topics, user_id, @@ -193,7 +195,7 @@ def register_token( (token_id, user_id, platform, token, hashed_token, status, app_ver, locale, topics, created_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,NOW()) """, (token_id, - user_id, + user_id, request_data.platform, secure_token, hashed_token, @@ -207,7 +209,7 @@ def register_token( logger.info(f"[FastAPI] Success: Registering token for user_id={user_id}, platform={request_data.platform}") except Exception as e: logger.exception(f"[FastAPI] Failed to register token for user_id={user_id}: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e return {"status":"registered"} @@ -230,7 +232,7 @@ def unregister_token( """, ('expired', hashed_token)) except Exception as e: logger.exception(f"[FastAPI] Failed to unregister token for user_id={user_id}: {e}") - raise HTTPException(status_code=500, detail=str(e)) + raise HTTPException(status_code=500, detail=str(e)) from e logger.info(f"[FastAPI] Success: Unregistering token for user_id={user_id}, platform={request_data.platform}") return {"status":"unregistered"} @@ -248,10 +250,10 @@ async def pull_notifications( 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") + raise HTTPException(status_code=502, detail="Backend-API-Internal unavailable") from e 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") + raise HTTPException(status_code=502, detail="Backend-API-Internal unavailable") from e logger.info(f"[FastAPI] Returning {len(data.messages)} message(s) for user_id={user_id}, has_more={data.has_more}") return data diff --git a/src/metrics_server.py b/src/metrics_server.py index c3fdc9b..15e7fd1 100644 --- a/src/metrics_server.py +++ b/src/metrics_server.py @@ -1,5 +1,5 @@ from fastapi import FastAPI, Response -from prometheus_client import generate_latest, CONTENT_TYPE_LATEST, Counter +from prometheus_client import CONTENT_TYPE_LATEST, Counter, generate_latest metrics_api = FastAPI(title="Metrics Server", description="Prometheus metrics endpoint") diff --git a/src/validator.py b/src/validator.py index 6cfa5f9..9949b50 100644 --- a/src/validator.py +++ b/src/validator.py @@ -1,5 +1,7 @@ -from argon2 import PasswordHasher import re + +from argon2 import PasswordHasher + from simple_logger_handler import setup_logger logger = setup_logger(__name__) -- 2.43.0