notifier/backend-api
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feature/windows notifications #9

Merged
florian merged 3 commits from feature/windows-notifications into main 2026-10-13 08:14:52 +00:00
Conversation 0 Commits 3 Files Changed 7
+131 -50
7 changed files with 131 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Dockerfile
+1 -1
View File
@@ -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
pyproject.toml
+28
View File
@@ -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",
]
requirements-dev.txt
+1
View File
@@ -0,0 +1 @@
ruff==0.17.0
requirements.txt
+3
View File
@@ -24,3 +24,6 @@ starlette==0.48.0
typing-inspection==0.4.2
typing_extensions==4.15.0
uvicorn==0.37.0
certifi==2026.7.22
httpcore==1.0.9
httpx==0.28.1
src/main.py
+94 -47
View File
@@ -1,26 +1,33 @@
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
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
from validator import is_valid_platform, is_valid_token, verify_api_key
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,13 +47,28 @@ class TokenRequest(BaseModel):
async def lifespan(app: FastAPI):
logger.info("Starting application...")
logger.info("Creating MySQL connection pool...")
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")
try:
logger.info("[FastAPI] Creating MySQL connection pool...")
create_connection_pool("ApiPool")
start_healthcheck_thread()
logger.info("MySQL healthcheck thread started.")
logger.info("[FastAPI] MySQL healthcheck thread started.")
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"
@@ -54,9 +76,14 @@ async def lifespan(app: FastAPI):
cleanup_secret_files(SECRET_PATHS)
yield
logger.info("Closing MySQL connection pool...")
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(
title="Device Token Management",
@@ -69,30 +96,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 +129,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 +157,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,15 +172,15 @@ 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()
WHERE user_id=%s AND hashed_token=%s
""", (request_data.platform,
""",(request_data.platform,
request_data.app_ver,
request_data.locale,
request_data.topics,
@@ -164,7 +189,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,10 +206,10 @@ 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}")
raise HTTPException(status_code=500, detail=str(e))
logger.exception(f"[FastAPI] Failed to register token for user_id={user_id}: {e}")
raise HTTPException(status_code=500, detail=str(e)) from e
return {"status":"registered"}
@@ -195,9 +220,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 +231,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}")
raise HTTPException(status_code=500, detail=str(e))
logger.info(f"Success: Unregistering token for user_id={user_id}, platform={request_data.platform}")
logger.exception(f"[FastAPI] Failed to unregister token for user_id={user_id}: {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"}
@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") 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") from e
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 +267,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())
src/metrics_server.py
+1 -1
View File
@@ -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")
src/validator.py
+3 -1
View File
@@ -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__)