FastAPI Background Tasks and Async Endpoints

from fastapi import FastAPI, BackgroundTasks, status
from pydantic import BaseModel, EmailStr
import logging
import time

app = FastAPI()
log = logging.getLogger(__name__)


class RegistrationRequest(BaseModel):
    email: EmailStr
    username: str


def send_welcome_email(email: str, username: str) -> None:
    """
    Regular def — FastAPI runs this in the thread pool after the response is sent.
    The event loop is not involved. time.sleep() is safe here because this
    runs in a worker thread, not on the event loop thread.

    Critical: wrap everything in try/except. FastAPI silently swallows
    exceptions raised inside background functions. Without this block,
    a failed SMTP connection produces no log entry and no alert — the
    user simply never receives the email and no one knows.
    """
    try:
        log.info(
            "welcome_email_started",
            extra={"email": email, "username": username}
        )
        # Simulate SMTP handshake — safe in a thread pool, catastrophic on event loop
        time.sleep(2)
        # In production: use smtplib here, or aiosmtplib in an async background task
        log.info(
            "welcome_email_sent",
            extra={"email": email, "username": username}
        )
    except Exception:
        # Log with exc_info=True to capture the full traceback
        # Also increment a metric counter here in production:
        # metrics.counter("forge.bg.email.failure").increment()
        log.error(
            "welcome_email_failed",
            extra={"email": email},
            exc_info=True
        )


def write_registration_audit_log(email: str, username: str) -> None:
    """
    Second background task — runs sequentially after send_welcome_email.
    Tasks registered with add_task() execute in registration order, not in parallel.
    """
    try:
        log.info(
            "registration_audit_written",
            extra={"email": email, "username": username, "event": "user_registered"}
        )
    except Exception:
        log.error("audit_log_failed", extra={"email": email}, exc_info=True)


@app.post('/forge/register', status_code=status.HTTP_201_CREATED)
async def register_user(payload: RegistrationRequest, tasks: BackgroundTasks):
    """
    Registers a user and enqueues post-response work.

    Execution order:
      1. Validate payload (Pydantic)
      2. Write user to database (not shown)
      3. Send HTTP 201 to client  <-- client connection returns here
      4. send_welcome_email() runs in thread pool
      5. write_registration_audit_log() runs in thread pool

    The client's measured latency includes only steps 1-3.
    Steps 4-5 are invisible to the client and independent of client connectivity.
    """
    # Step 1: In production, write to database here
    # await db.users.insert({"email": payload.email, "username": payload.username})

    # Step 2: Register background tasks — they do not run yet
    tasks.add_task(send_welcome_email, payload.email, payload.username)
    tasks.add_task(write_registration_audit_log, payload.email, payload.username)

    # Step 3: Return response — background tasks run after this is sent
    return {
        "status": "created",
        "detail": "Registration complete. Welcome email is on its way.",
    }


# GET /forge/register POST {"email": "dev@thecodeforge.io", "username": "forge-dev"}
# Client receives immediately:
# -> {"status": "created", "detail": "Registration complete. Welcome email is on its way."}
#
# Server logs after response (invisible to client):
# INFO  welcome_email_started    email=dev@thecodeforge.io
# INFO  welcome_email_sent       email=dev@thecodeforge.io
# INFO  registration_audit_written email=dev@thecodeforge.io

from fastapi import FastAPI
import asyncio
import httpx
import hashlib
import time

app = FastAPI()


# ─────────────────────────────────────────────────────────────────────────────
# CASE 1: async def — the correct choice for async I/O
# ─────────────────────────────────────────────────────────────────────────────
@app.get('/forge/weather/{city}')
async def get_weather(city: str):
    """
    Uses httpx.AsyncClient — a truly async HTTP client.
    Every network operation yields to the event loop via await.
    While waiting for the external API response, FastAPI handles other requests.
    This endpoint scales to thousands of concurrent requests on a single worker.
    """
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(
            f'https://api.thecodeforge.io/v1/weather/{city}'
        )
        response.raise_for_status()
        return response.json()


# ─────────────────────────────────────────────────────────────────────────────
# CASE 2: regular def — the correct choice for synchronous or CPU-bound work
# ─────────────────────────────────────────────────────────────────────────────
@app.get('/forge/hash')
def compute_password_hash(password: str):
    """
    CPU-bound operation using hashlib — synchronous and compute-intensive.
    FastAPI detects regular def and submits this to the thread pool.
    The event loop is free to handle other requests while this runs.
    Thread pool default: min(32, os.cpu_count() + 4).
    """
    # Simulate bcrypt-level work with high iteration SHA-256
    digest = hashlib.pbkdf2_hmac(
        'sha256',
        password.encode(),
        b'forge-salt',
        iterations=100_000
    )
    return {"hash": digest.hex()}


# ─────────────────────────────────────────────────────────────────────────────
# CASE 3: async def with mixed sync/async — use asyncio.to_thread()
# ─────────────────────────────────────────────────────────────────────────────
@app.post('/forge/report')
async def generate_report(report_id: str):
    """
    Needs both async (fetch data from async DB) and sync (write to filesystem).
    Wrapping the synchronous file write in asyncio.to_thread() delegates it
    to the thread pool, keeping the event loop free during file I/O.
    """
    # Async database call — yields to event loop
    # data = await db.reports.find_one({"id": report_id})
    data = {"report_id": report_id, "rows": 1500}  # Simulated

    # Synchronous file I/O — must NOT run on event loop directly
    # Wrap with asyncio.to_thread() to delegate to thread pool
    def write_to_disk(payload: dict) -> str:
        path = f"/tmp/forge-report-{payload['report_id']}.json"
        import json
        with open(path, 'w') as f:
            json.dump(payload, f)
        return path

    output_path = await asyncio.to_thread(write_to_disk, data)
    return {"report_id": report_id, "output": output_path}


# ─────────────────────────────────────────────────────────────────────────────
# CASE 4: THE LOOP KILLER — async def with synchronous blocking call
# This is the pattern behind most FastAPI production incidents.
# ─────────────────────────────────────────────────────────────────────────────
#
# @app.get('/forge/disaster')
# async def event_loop_hostage():
#     time.sleep(10)  # BLOCKS THE ENTIRE SERVER FOR 10 SECONDS
#                     # Every concurrent request queues behind this.
#                     # CPU shows nearly 0% — the server appears idle.
#                     # Health checks time out. Kubernetes kills the pod.
#                     # There is no error. There is no warning.
#                     # There is just silence and then a wall of timeouts.


# ─────────────────────────────────────────────────────────────────────────────
# CASE 5: asyncio.sleep() — the correct non-blocking delay in async context
# ─────────────────────────────────────────────────────────────────────────────
@app.get('/forge/delayed-response')
async def delayed_response():
    """
    asyncio.sleep() yields control to the event loop for the duration.
    During this 2-second wait, FastAPI processes thousands of other requests.
    time.sleep(2) here would freeze all of them.
    """
    await asyncio.sleep(2)
    return {"status": "ready", "source": "thecodeforge"}


# Summary of behavior:
# Case 1: async def + httpx.AsyncClient -> event loop, non-blocking, scales to 10k+ concurrent
# Case 2: regular def + hashlib        -> thread pool, safe for sync/CPU work
# Case 3: async def + to_thread()      -> event loop for async parts, thread pool for sync parts
# Case 4: async def + time.sleep()     -> event loop BLOCKED, all other requests frozen
# Case 5: async def + asyncio.sleep()  -> event loop free, correct non-blocking delay

# io.thecodeforge.durable_tasks
from fastapi import BackgroundTasks, FastAPI
from redis import Redis
from rq import Queue

app = FastAPI()
redis = Redis()
task_queue = Queue(connection=redis)

def send_email_persistent(user_id: str):
    # This runs in a separate worker process
    print(f"Sending email to user {user_id} at {time.time()}")

@app.post("/signup")
async def signup(user_id: str, background_tasks: BackgroundTasks):
    # Use BackgroundTasks for disposable logging
    background_tasks.add_task(lambda: print(f"Audit: {user_id} signed up"))
    
    # Use proper queue for durable work
    task_queue.enqueue(send_email_persistent, user_id)
    return {"status": "queued", "user_id": user_id}

# Run: rq worker --with-scheduler

# io.thecodeforge.async_io_tasks
import asyncio
import time

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def blocking_file_write(data: str):
    time.sleep(1)  # Simulates blocking I/O
    print(f"Written: {data}")

async def async_http_call(url: str):
    await asyncio.sleep(1)  # Non-blocking
    print(f"Called: {url}")

@app.post("/process")
async def process(background_tasks: BackgroundTasks):
    # This blocks a thread pool worker
    background_tasks.add_task(blocking_file_write, "data")
    
    # This runs on the event loop — don't block inside it
    background_tasks.add_task(async_http_call, "https://example.com")
    
    return {"status": "ok"}