From c9ec297a5f36421cac270adc316ec41b3aefd268 Mon Sep 17 00:00:00 2001 From: Aditya Gupta Date: Fri, 3 Jul 2026 11:20:23 +0530 Subject: [PATCH] feat: Implemented observability layer and benchmarking --- README.md | 41 +-------- app/clients/exchange_rate_client.py | 4 + app/core/observability.py | 59 ++++++++++++ app/services/job_service.py | 7 ++ app/utils/csv_parser.py | 3 + app/worker.py | 136 +++++++++++++++------------- tests/services/test_benchmarker.py | 53 +++++++++++ 7 files changed, 202 insertions(+), 101 deletions(-) create mode 100644 app/core/observability.py create mode 100644 tests/services/test_benchmarker.py diff --git a/README.md b/README.md index f81f5b8..e4630f6 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ An asynchronous transactional data cleaning, validation, and LLM powered categorization pipeline. -## 🚀 Features +## Features - Reads multi format dates and handles missing transaction IDs by auto generating unique fallbacks, and parses messy amounts with currency signs and commas cleanly. - Offloads validation, spend conversion, and analysis to a Celery background worker backed by Redis. @@ -20,13 +20,11 @@ An asynchronous transactional data cleaning, validation, and LLM powered categor - Flags domestic transactions in USD currency. - Flags high value transactions and transactions with fraud indicative keywords. -## 🐳 Self Hosting & Deployment +## Self Hosting & Deployment We publish production-ready images to **GitHub Container Registry (GHCR)**. You can pull them directly or run them via Docker Compose. -### Option A: Docker Compose (Using Pre-built Images) - -Create a `docker-compose.yml` file to self-host the entire stack: +Create a `docker-compose.yml` file to host the entire stack (or just clone the repo and directly run `docker compose up -d` : ```yaml version: "3.8" @@ -104,36 +102,3 @@ Launch the stack: ```bash docker compose up -d ``` - -### Option B: Direct Pull (Manual Run) - -You can pull and run individual containers manually from GHCR: - -1. **Pull the images**: - - ```bash - docker pull ghcr.io/sortedcord/alemno-payments/web:latest - docker pull ghcr.io/sortedcord/alemno-payments/worker:latest - ``` - -2. **Run Web (FastAPI API Server)**: - - ```bash - docker run -d \ - --name alemno_web \ - -p 3000:3000 \ - -e POSTGRES_HOST=your-db-host \ - -e REDIS_HOST=your-redis-host \ - -e GEMINI_API_KEY=your-gemini-api-key \ - ghcr.io/sortedcord/alemno-payments/web:latest - ``` - -3. **Run Worker (Celery Processing Worker)**: - ```bash - docker run -d \ - --name alemno_worker \ - -e POSTGRES_HOST=your-db-host \ - -e REDIS_HOST=your-redis-host \ - -e GEMINI_API_KEY=your-gemini-api-key \ - ghcr.io/sortedcord/alemno-payments/worker:latest celery -A app.worker.celery_app worker --loglevel=info - ``` diff --git a/app/clients/exchange_rate_client.py b/app/clients/exchange_rate_client.py index dc4c34b..d358b72 100644 --- a/app/clients/exchange_rate_client.py +++ b/app/clients/exchange_rate_client.py @@ -4,6 +4,9 @@ from app.core.config import settings from app.core.logging import logger +from app.core.observability import time_it + + class ExchangeRateClient: def __init__(self): self.default_rate = 93.0 @@ -20,6 +23,7 @@ class ExchangeRateClient: logger.warning(f"Could not connect to Redis for exchange rate caching: {e}") self.redis_client = None + @time_it("Get USD to INR Exchange Rate") async def get_usd_to_inr_rate(self) -> float: """ Retrieves the USD to INR conversion rate. diff --git a/app/core/observability.py b/app/core/observability.py new file mode 100644 index 0000000..e5fcba6 --- /dev/null +++ b/app/core/observability.py @@ -0,0 +1,59 @@ +import functools +import inspect +import time +from contextlib import contextmanager +from typing import Optional +from app.core.logging import logger + + +@contextmanager +def benchmark(name: str): + """ + Context manager to benchmark a block of code. + Example: + with benchmark("Database save"): + await repo.save(data) + """ + start_time = time.perf_counter() + try: + yield + finally: + elapsed = time.perf_counter() - start_time + logger.info(f"[BENCHMARK] {name} completed in {elapsed:.4f} seconds") + + +def time_it(name: Optional[str] = None): + """ + Decorator to benchmark synchronous or asynchronous functions. + Example: + @time_it("Retrieve exchange rate") + async def get_rate(): + ... + """ + + def decorator(func): + label = name or func.__name__ + + @functools.wraps(func) + async def async_wrapper(*args, **kwargs): + start = time.perf_counter() + try: + return await func(*args, **kwargs) + finally: + elapsed = time.perf_counter() - start + logger.info(f"[BENCHMARK] {label} completed in {elapsed:.4f} seconds") + + @functools.wraps(func) + def sync_wrapper(*args, **kwargs): + start = time.perf_counter() + try: + return func(*args, **kwargs) + finally: + elapsed = time.perf_counter() - start + logger.info(f"[BENCHMARK] {label} completed in {elapsed:.4f} seconds") + + if inspect.iscoroutinefunction(func): + return async_wrapper + return sync_wrapper + + return decorator diff --git a/app/services/job_service.py b/app/services/job_service.py index e94d3a4..3fb7206 100644 --- a/app/services/job_service.py +++ b/app/services/job_service.py @@ -6,10 +6,14 @@ from app.utils.csv_parser import parse_and_validate_csv from app.worker import process_transaction_job +from app.core.observability import time_it + + class JobService: def __init__(self, repo: JobRepository): self.repo = repo + @time_it("Upload CSV Service") async def upload_csv(self, filename: str, content: bytes) -> Job: parsed_transactions = parse_and_validate_csv(content) @@ -19,17 +23,20 @@ class JobService: return job + @time_it("Get Job Status Service") async def get_job_status(self, job_id: str) -> Job: job = await self.repo.get_by_id(job_id) if not job: raise JobNotFoundException(job_id) return job + @time_it("Get Job Results Service") async def get_job_results(self, job_id: str) -> Job: job = await self.repo.get_by_id(job_id) if not job: raise JobNotFoundException(job_id) return job + @time_it("List Jobs Service") async def list_jobs(self, status: Optional[str] = None) -> List[Job]: return await self.repo.list_all(status=status) diff --git a/app/utils/csv_parser.py b/app/utils/csv_parser.py index 550c112..af8d2ed 100644 --- a/app/utils/csv_parser.py +++ b/app/utils/csv_parser.py @@ -4,9 +4,12 @@ from datetime import datetime from typing import Any, Dict, List from app.core.exceptions import InvalidCSVException +from app.core.observability import time_it + REQUIRED_COLUMNS = {"txn_id", "date", "merchant", "amount"} +@time_it("Parse and Validate CSV") def parse_and_validate_csv(content: bytes) -> List[Dict[str, Any]]: """ Parses CSV content from bytes, validates headers, and checks field types. diff --git a/app/worker.py b/app/worker.py index 6859bd1..79a7794 100644 --- a/app/worker.py +++ b/app/worker.py @@ -15,6 +15,7 @@ from app.repositories.job_repository import JobRepository from app.db.models.transaction import Transaction from app.db.models.job_summary import JobSummary from app.clients.exchange_rate_client import ExchangeRateClient +from app.core.observability import time_it, benchmark exchange_rate_client = ExchangeRateClient() @@ -53,6 +54,7 @@ class BatchClassificationResponse(BaseModel): classifications: List[TransactionClassification] +@time_it("LLM Batch Classification Request") async def classify_transactions_batch( client: genai.Client, batch_txns: List[Dict[str, Any]], @@ -160,6 +162,7 @@ class JobNarrativeSummary(BaseModel): ) +@time_it("LLM Job Narrative Summary Request") async def generate_job_summary_llm( client: genai.Client, db_transactions: List[Transaction], @@ -253,22 +256,26 @@ def process_transaction_job(job_id: str, transactions: List[Dict[str, Any]]): return run_async(_process_job_async(job_id, transactions)) +@time_it("Process Job Background Task") async def _process_job_async(job_id: str, transactions: List[Dict[str, Any]]): from app.db.session import engine - await engine.dispose() - async with AsyncSessionLocal() as db: - repo = JobRepository(db) - job = await repo.get_by_id(job_id) - if not job: - logger.error(f"Job with ID {job_id} not found in database.") - return + with benchmark("Dispose Engine"): + await engine.dispose() + + async with AsyncSessionLocal() as db: + with benchmark("Initial Job Setup"): + repo = JobRepository(db) + job = await repo.get_by_id(job_id) + if not job: + logger.error(f"Job with ID {job_id} not found in database.") + return - try: # Update status to processing await repo.update(job, status="processing") await db.commit() + try: db_transactions = [] total_spend_inr = 0.0 total_spend_usd = 0.0 @@ -277,43 +284,44 @@ async def _process_job_async(job_id: str, transactions: List[Dict[str, Any]]): USD_TO_INR = await exchange_rate_client.get_usd_to_inr_rate() - # 1. Fetch existing transaction amounts and currencies for all accounts in this job to calculate median - account_ids = { - t.get("account_id") for t in transactions if t.get("account_id") - } - existing_amounts_inr = {} - if account_ids: - stmt = select( - Transaction.account_id, Transaction.amount, Transaction.currency - ).where(Transaction.account_id.in_(account_ids)) - res = await db.execute(stmt) - for acc_id, amt, curr in res.all(): - if acc_id not in existing_amounts_inr: - existing_amounts_inr[acc_id] = [] - # Standardize to INR for median calculation - amt_inr = amt * USD_TO_INR if curr.upper() == "USD" else amt - existing_amounts_inr[acc_id].append(amt_inr) + with benchmark("Compute Account History Medians"): + # 1. Fetch existing transaction amounts and currencies for all accounts in this job to calculate median + account_ids = { + t.get("account_id") for t in transactions if t.get("account_id") + } + existing_amounts_inr = {} + if account_ids: + stmt = select( + Transaction.account_id, Transaction.amount, Transaction.currency + ).where(Transaction.account_id.in_(account_ids)) + res = await db.execute(stmt) + for acc_id, amt, curr in res.all(): + if acc_id not in existing_amounts_inr: + existing_amounts_inr[acc_id] = [] + # Standardize to INR for median calculation + amt_inr = amt * USD_TO_INR if curr.upper() == "USD" else amt + existing_amounts_inr[acc_id].append(amt_inr) - # Group incoming transaction amounts standardized to INR - incoming_amounts_inr = {} - for t in transactions: - acc_id = t.get("account_id") - if acc_id: - if acc_id not in incoming_amounts_inr: - incoming_amounts_inr[acc_id] = [] - amt = float(t["amount"]) - curr = t.get("currency", "INR").strip().upper() - amt_inr = amt * USD_TO_INR if curr == "USD" else amt - incoming_amounts_inr[acc_id].append(amt_inr) + # Group incoming transaction amounts standardized to INR + incoming_amounts_inr = {} + for t in transactions: + acc_id = t.get("account_id") + if acc_id: + if acc_id not in incoming_amounts_inr: + incoming_amounts_inr[acc_id] = [] + amt = float(t["amount"]) + curr = t.get("currency", "INR").strip().upper() + amt_inr = amt * USD_TO_INR if curr == "USD" else amt + incoming_amounts_inr[acc_id].append(amt_inr) - # Precalculate median of (existing + incoming) transactions in INR - account_medians_inr = {} - for acc_id in account_ids: - all_amts = existing_amounts_inr.get( - acc_id, [] - ) + incoming_amounts_inr.get(acc_id, []) - if all_amts: - account_medians_inr[acc_id] = statistics.median(all_amts) + # Precalculate median of (existing + incoming) transactions in INR + account_medians_inr = {} + for acc_id in account_ids: + all_amts = existing_amounts_inr.get( + acc_id, [] + ) + incoming_amounts_inr.get(acc_id, []) + if all_amts: + account_medians_inr[acc_id] = statistics.median(all_amts) # 2. Call LLM to classify transactions without a category # Find all uncategorized incoming transactions @@ -579,7 +587,8 @@ async def _process_job_async(job_id: str, transactions: List[Dict[str, Any]]): ) db_transactions.append(db_txn) - await repo.add_transactions(db_transactions) + with benchmark("Save Cleaned Transactions to DB"): + await repo.add_transactions(db_transactions) # Generate narrative and risk summary using LLM if api_key is available llm_summary = None @@ -635,26 +644,27 @@ async def _process_job_async(job_id: str, transactions: List[Dict[str, Any]]): } # Create job summary - summary = JobSummary( - job_id=job_id, - total_spend_inr=round(total_spend_inr, 2), - total_spend_usd=round(total_spend_usd, 2), - top_merchants=top_merchants_json, - anomaly_count=anomaly_count, - narrative=narrative, - risk_level=risk_level, - ) - await repo.add_summary(summary) + with benchmark("Save Summary and Complete Job"): + summary = JobSummary( + job_id=job_id, + total_spend_inr=round(total_spend_inr, 2), + total_spend_usd=round(total_spend_usd, 2), + top_merchants=top_merchants_json, + anomaly_count=anomaly_count, + narrative=narrative, + risk_level=risk_level, + ) + await repo.add_summary(summary) - # Update job state - await repo.update( - job, - status="completed", - row_count_raw=len(transactions), - row_count_clean=len(db_transactions), - completed_at=datetime.utcnow(), - ) - await db.commit() + # Update job state + await repo.update( + job, + status="completed", + row_count_raw=len(transactions), + row_count_clean=len(db_transactions), + completed_at=datetime.utcnow(), + ) + await db.commit() logger.info(f"Job {job_id} processing completed successfully.") except Exception as e: diff --git a/tests/services/test_benchmarker.py b/tests/services/test_benchmarker.py new file mode 100644 index 0000000..5d81182 --- /dev/null +++ b/tests/services/test_benchmarker.py @@ -0,0 +1,53 @@ +import pytest +import asyncio +from app.core.observability import benchmark, time_it + + +def test_benchmark_context_manager(caplog): + import logging + + with caplog.at_level(logging.INFO): + with benchmark("Test Task"): + # simulate work + pass + + assert any( + "[BENCHMARK] Test Task completed in" in record.message + for record in caplog.records + ) + + +def test_time_it_decorator_sync(caplog): + import logging + + @time_it("Sync Helper") + def my_sync_fn(x, y): + return x + y + + with caplog.at_level(logging.INFO): + res = my_sync_fn(10, 20) + + assert res == 30 + assert any( + "[BENCHMARK] Sync Helper completed in" in record.message + for record in caplog.records + ) + + +@pytest.mark.asyncio +async def test_time_it_decorator_async(caplog): + import logging + + @time_it("Async Helper") + async def my_async_fn(): + await asyncio.sleep(0.01) + return "done" + + with caplog.at_level(logging.INFO): + res = await my_async_fn() + + assert res == "done" + assert any( + "[BENCHMARK] Async Helper completed in" in record.message + for record in caplog.records + )