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

Added pulling notifications from new backend service backend-pull-notifications

Browse Source
This commit is contained in:
florian committed 2026-10-12 11:37:04 +02:00
1 parent 34318fa9b9
commit 0506c1619d
2 files changed
+80 -32
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

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
+77 -32
View File
@@ -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,13 +45,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 +74,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 +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())