Building a High-Performance Daily Ledger API: Asynchronous Concurrency & Data Validation with FastAPI

text
Daily-Ledger-API/
├── models.py      # Pydantic schemas & strict data validation contracts
├── storage.py     # Local JSON persistence & file I/O operations
├── main.py        # FastAPI routing, HTTP status handling & AsyncIO analytics
└── README.md      # Documentation & quick-start guide
##1. Strict Data Validation with Nested Pydantic Models
In production APIs, invalid payloads must never touch the core business logic or database. Using Pydantic, I established strict data contracts that enforce data types and constraints before an endpoint processes incoming JSON.

models.py

from pydantic import BaseModel, Field
from typing import List

class MealEntry(BaseModel):
    name: str = Field(..., min_length=2, description="Name of the meal")
    protein_g: float = Field(..., gt=0, description="Protein in grams (must be > 0)")
    calories: float = Field(..., gt=0, description="Calories in kcal (must be > 0)")

class ExpenseEntry(BaseModel):
    category: str = Field(..., min_length=2, description="Expense category")
    amount: float = Field(..., gt=0, description="Amount spent (must be > 0)")

class DailyLedger(BaseModel):
    date: str = Field(..., pattern=r"^\d{4}-\d{2}-\d{2}$", description="Date format: YYYY-MM-DD")
    meals: List[MealEntry] = []
    expenses: List[ExpenseEntry] = []

💡 Engineering Benefits:
Nested Validation: The DailyLedger model automatically validates every item inside the meals and expenses lists.

Automated Rejections: Any missing field, malformed date string, or negative numerical value is rejected immediately with a standardized 422 Unprocessable Entity response without executing downstream code.

##2. Safe Local Persistence Layer
To persist logs across server restarts without relying on heavy database drivers for this microservice, I developed a modular storage handler in storage.py.

storage.py

import json
import os

DB_FILE = "ledger_db.json"

def load_ledger():
    if not os.path.exists(DB_FILE):
        return []
    try:
        with open(DB_FILE, "r") as file:
            return json.load(file)
    except json.JSONDecodeError:
        return []

def save_ledger_entry(entry: dict):
    data = load_ledger()
    data.append(entry)
    with open(DB_FILE, "w") as file:
        json.dump(data, file, indent=4)

💡 Key Safeguards:
State Preservation: load_ledger() guarantees that appending a new record never overwrites existing records.

Corrupt File Guarding: Includes exception handling for JSONDecodeError and missing file checks, returning an empty list fallback instead of crashing the process.

##3. Concurrency & Performance via asyncio.gather
One of the core features of the Daily Ledger API is the /summary aggregation endpoint. Instead of calculating nutritional statistics and financial metrics sequentially (which blocks the event loop), both tasks execute concurrently using Python's asyncio.gather().

main.py (Analytics Engine)

import asyncio
from fastapi import FastAPI, HTTPException, status
from models import DailyLedger
from storage import load_ledger, save_ledger_entry

app = FastAPI(title="Daily Ledger API")

async def calculate_macro_stats(logs):
    await asyncio.sleep(0.01)  # Simulating async I/O computation
    total_calories = 0
    total_protein = 0
    total_meals_logged = 0
    for day in logs:
        for meal in day.get("meals", []):
            total_protein += meal.get("protein_g", 0)
            total_calories += meal.get("calories", 0)
            total_meals_logged += 1

    return {
        "total_meals_logged": total_meals_logged,
        "total_protein_g": total_protein,
        "total_calories": total_calories,
    }

async def calculate_expense_stats(logs):
    await asyncio.sleep(0.01)  # Simulating async I/O computation
    total_spent = 0
    total_expense_records = 0

    for day in logs:
        for expense in day.get("expenses", []):
            total_spent += expense.get("amount", 0)
            total_expense_records += 1

    return {
        "Total_Spent": total_spent,
        "Total_Expense": total_expense_records
    }

@app.get("/summary")
async def get_ledger_summary():
    all_logs = load_ledger()

    # 404 Guard: Reject aggregation if database is empty
    if not all_logs:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No ledger logs found. Please submit a log via POST /log first.",
        )

    # Concurrently execute both aggregation routines
    macro_task = calculate_macro_stats(all_logs)
    expense_task = calculate_expense_stats(all_logs)

    macro_results, expense_results = await asyncio.gather(
        macro_task, expense_task
    )

    return {
        "status": "success",
        "days_analyzed": len(all_logs),
        "nutrition_analytics": macro_results,
        "finance_analytics": expense_results,
    }
💡 Why asyncio.gather() Matters:
In a traditional synchronous architecture, tasks run one after another, adding to the total response time.

With asyncio.gather(), both calculation tasks execute in parallel on the event loop, reducing overall response latency.

##4. End-to-End Testing via Swagger UI
FastAPI automatically generates interactive OpenAPI documentation at /docs.

Through Swagger UI, I tested the full API lifecycle:

1.GET /summary (Empty State): Correctly returned 404 Not Found.

2.POST /log (Ingestion): Validated payload structure, saved the JSON record to disk, and returned 201 Created.

3.GET /logs (Data Retrieval): Fetched all historical records with count metadata.

4.GET /summary (Concurrent Processing): Returned the combined multi-domain analytics payload in a single response.


##🎯 Key Takeaways
Schema Contracts First: Using Pydantic prevents runtime bugs by failing fast on invalid inputs.

Asynchronous Scaling: Leveraging asyncio allows backend microservices to scale efficiently under I/O-bound operations.

Clean Code Structure: Separating routes, data models, and storage logic makes backend applications testable, modular, and maintainable.

🔗 GitHub Repository: https://github.com/skyatriya/Daily-Ledger-API

Feedback and contributions are always welcome!