DEV Community: Ravi Gupta

I Thought JWTs Were Stateless. Turns Out Logout Made Me Build a Stateful Layer Anyway.

Ravi Gupta — Mon, 06 Apr 2026 03:15:00 +0000

This is Part 3 of a 4-part series on building AuthShield - a production-ready standalone authentication microservice. This post covers JWT design, the logout problem, Redis blacklisting, the two-token strategy, and refresh token rotation with reuse detection.
Part 1 is here: Why I Stopped Writing Auth Code for Every Project and Built AuthShield
Part 2 is here: I Thought OAuth Was Just Adding a Google Button. Turns Out It's a CSRF Problem Disguised as a Feature

The selling point of JWTs is that they are stateless.

The server issues a token, signs it, and forgets about it. Every subsequent request includes the token. The server verifies the signature, reads the claims, and knows everything it needs - who the user is, what roles they have, when the token expires. No database lookup. No session store. No shared state between instances.

For an auth microservice like AuthShield that needs to work across multiple downstream services, this is exactly the right foundation. Any service with the same signing secret can verify any token independently.

Then I implemented logout.

The Logout Problem

When a user logs out, what actually happens to their JWT?

Nothing. It keeps working until it expires.

A JWT is a signed string. It is not stored anywhere on the server. There is nothing to delete. If a user logs out and someone else - an attacker, a browser tab left open on a shared computer, anyone - has that token, it is still valid for the remainder of its lifetime.

That is not logout. That is the appearance of logout.

The solution is a blacklist. When a user logs out, you store the token's unique identifier in a fast lookup store and check every incoming token against it. If the token is on the blacklist, reject it regardless of the signature being valid.

AuthShield uses Redis for this, and the specific identifier used is the jti claim - JWT ID - a unique value embedded in every token at creation time.

import jwt
import uuid
from datetime import datetime, timedelta, timezone
from redis.asyncio import Redis

def create_access_token(user_id: str, email: str, roles: list[str]) -> tuple[str, str]:
    jti = str(uuid.uuid4())  # Unique token ID - used for blacklisting

    now = datetime.now(timezone.utc)
    expires_at = now + timedelta(minutes=15)

    payload = {
        "sub": user_id,
        "email": email,
        "roles": roles,
        "jti": jti,           # The blacklist key
        "type": "access",
        "iat": now,
        "exp": expires_at,
    }

    token = jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256")
    return token, jti  # Return both - jti needed for blacklisting on logout


async def blacklist_token(redis: Redis, jti: str, expires_at: datetime) -> None:
    # Calculate remaining lifetime of the token
    now = datetime.now(timezone.utc)
    remaining_seconds = int((expires_at - now).total_seconds())

    if remaining_seconds > 0:
        # Store in Redis with TTL matching token's remaining life
        # When the token would have expired naturally, Redis removes it automatically
        # No cleanup job needed — it is self-managing
        await redis.setex(f"blacklist:{jti}", remaining_seconds, "1")


async def is_token_blacklisted(redis: Redis, jti: str) -> bool:
    return await redis.exists(f"blacklist:{jti}") > 0

The TTL on the Redis key is important. There is no point keeping a blacklisted token in Redis after it would have expired naturally - it is already invalid. Setting the TTL to match the remaining token lifetime means the blacklist is self-cleaning. Old entries disappear automatically without any maintenance.

Every protected request checks the blacklist before doing anything else:

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    redis: Redis = Depends(get_redis),
    db: AsyncSession = Depends(get_db)
) -> User:
    token = credentials.credentials

    try:
        payload = jwt.decode(
            token,
            settings.SECRET_KEY,
            algorithms=["HS256"]  # Always specify explicitly — never trust the token's own alg header
        )
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

    # Check blacklist before anything else
    jti = payload.get("jti")
    if await is_token_blacklisted(redis, jti):
        raise HTTPException(status_code=401, detail="Token has been revoked")

    # Token is valid and not blacklisted — proceed
    user_id = payload.get("sub")
    return await user_repo.get_by_id(db, user_id)

One detail worth calling out: algorithms=["HS256"] is explicit and non-negotiable. Early JWT libraries had a vulnerability where if the token header specified "alg": "none", some libraries would skip signature verification entirely. An attacker could forge any payload they wanted. Always specify exactly which algorithms you accept and reject everything else.

Now logout works. But solving logout uncovered the next problem.

The Two-Token Problem

If the logout blacklist works, why not just use a single long-lived token? Issue it on login, blacklist it on logout, done.

Because if that token is stolen, the attacker has access for the entire duration of its lifetime. A 7-day token stolen on day one gives an attacker 7 days of undetected access. A 30-day token is worse. The longer the lifetime, the larger the window of damage from theft.

So make it short-lived. 15 minutes. Now a stolen token is only useful for 15 minutes.

But now the user has to log in every 15 minutes. That is not a user experience - that is punishment.

The solution is two tokens with different purposes and different lifetimes.

An access token is short-lived - 15 minutes in AuthShield. It is sent with every API request. If it is stolen, the damage window is 15 minutes maximum.

A refresh token is long-lived - 7 days. It is used for exactly one purpose: getting a new access token when the old one expires. It is never sent to API endpoints. It never touches the application layer. It only ever goes to the /auth/refresh endpoint.

The user experience is seamless. The access token expires silently, the client uses the refresh token to get a new one, and the user never knows it happened. Security and UX are no longer in conflict.

import secrets
import hashlib

def generate_refresh_token() -> tuple[str, str]:
    # Generate a cryptographically random token
    raw_token = secrets.token_urlsafe(64)

    # Hash it before storing — same reason we hash passwords
    # If the database is breached, raw tokens are not exposed
    token_hash = hashlib.sha256(raw_token.encode()).hexdigest()

    return raw_token, token_hash  # Return raw to client, store hash in DB


async def store_refresh_token(
    db: AsyncSession,
    user_id: str,
    session_id: str,
    token_hash: str,
    family_id: str  # More on this shortly
) -> RefreshToken:
    refresh_token = RefreshToken(
        user_id=user_id,
        session_id=session_id,
        token_hash=token_hash,
        family_id=family_id,
        is_used=False,
        is_revoked=False,
        expires_at=datetime.now(timezone.utc) + timedelta(days=7),
    )
    db.add(refresh_token)
    await db.commit()
    return refresh_token

Refresh tokens are not JWTs. They are opaque random strings, stored hashed in PostgreSQL. There is no benefit to making them JWTs because they are always verified against the database anyway - a database lookup is required regardless, so the self-contained nature of JWTs adds nothing here.

Storing them hashed mirrors the same principle as password hashing. If the database is compromised, an attacker gets hashes, not usable tokens.

Two tokens solved the UX problem. But it introduced another one: what happens if the refresh token is stolen?

Refresh Token Rotation and Reuse Detection

A 7-day refresh token is a valuable target. If an attacker gets hold of one, they can keep generating new access tokens for 7 days without the user's password. The user might log out, the access token gets blacklisted, but the attacker still has the refresh token and can get a new access token immediately.

The first defence is rotation. Every time a refresh token is used, it is immediately invalidated and a new one is issued. The client must always use the latest token. Old tokens stop working the moment they are used.

async def rotate_refresh_token(
    db: AsyncSession,
    old_token_hash: str,
    user_id: str,
    session_id: str,
    family_id: str
) -> tuple[str, str]:
    # Mark the old token as used
    old_token = await token_repo.get_by_hash(db, old_token_hash)
    old_token.is_used = True
    await db.flush()

    # Issue a new token in the same family
    new_raw_token, new_token_hash = generate_refresh_token()
    await store_refresh_token(
        db,
        user_id=user_id,
        session_id=session_id,
        token_hash=new_token_hash,
        family_id=family_id  # Same family as the old token
    )
    await db.commit()

    return new_raw_token, new_token_hash

Rotation limits the damage window. But it does not fully solve the theft problem. If an attacker steals the refresh token and uses it before the real user does, the attacker gets a new valid token and the real user's next refresh attempt fails - their token was already rotated.

The real user gets a confusing error. The attacker continues undetected.

This is where token families solve the problem completely.

Every refresh token belongs to a family - a UUID assigned at login time and shared by all tokens in a rotation chain. When a token is rotated, the new token carries the same family ID. The chain is trackable.

The key insight: if a token that has already been marked as used is presented again, it means one of two things. Either there is a bug in the client — unlikely. Or the token was stolen and used by a second party - the real user rotated it, then the attacker tried to use the original.

When reuse is detected, the correct response is not just to reject the request. It is to revoke the entire family - every token descended from that login. Both the attacker's current token and the real user's current token become invalid simultaneously.

async def handle_refresh_token(
    db: AsyncSession,
    redis: Redis,
    raw_token: str
) -> tuple[str, str]:
    # Hash the incoming token to look it up
    token_hash = hashlib.sha256(raw_token.encode()).hexdigest()
    token_record = await token_repo.get_by_hash(db, token_hash)

    if not token_record:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    if token_record.is_revoked:
        raise HTTPException(status_code=401, detail="Refresh token has been revoked")

    # REUSE DETECTION — this is the critical check
    if token_record.is_used:
        # This token was already rotated — someone is using an old token
        # Could be an attacker who stole it before the real user rotated it
        # Revoke the entire family — both attacker and real user are logged out
        await token_repo.revoke_family(db, token_record.family_id)
        await db.commit()
        raise HTTPException(
            status_code=401,
            detail="Token reuse detected. All sessions have been revoked."
        )

    # Token is valid — rotate it
    if token_record.expires_at < datetime.now(timezone.utc):
        raise HTTPException(status_code=401, detail="Refresh token expired")

    new_raw_token, _ = await rotate_refresh_token(
        db,
        old_token_hash=token_hash,
        user_id=token_record.user_id,
        session_id=token_record.session_id,
        family_id=token_record.family_id
    )

    # Issue a new access token
    user = await user_repo.get_by_id(db, token_record.user_id)
    new_access_token, _ = create_access_token(
        user_id=str(user.id),
        email=user.email,
        roles=[role.name for role in user.roles]
    )

    return new_access_token, new_raw_token

The outcome of reuse detection is uncomfortable for the real user - they get logged out and have to re-authenticate. But that is the correct response. Something is wrong. Either the token was stolen, or the client has a bug. Either way, forcing re-authentication is the right call. The real user is inconvenienced for 30 seconds. The attacker loses access entirely.

What Lives Where and Why

After working through all of this, the storage decisions become clear.

Access tokens live in client memory - never in localStorage, which is accessible to any JavaScript on the page including third-party scripts. Memory is cleared when the tab closes, which is acceptable given the 15-minute TTL.

Refresh tokens live in PostgreSQL, stored as SHA-256 hashes. The raw token is given to the client once and never stored on the server. If the database is breached, the hashes are useless without the raw tokens.

The blacklist lives in Redis, keyed by JTI with TTLs matching each token's remaining lifetime. Self-cleaning, fast, and adds under 1 millisecond to every protected request.

What JWT Security Actually Is

Working through all four of these problems changed how I think about token security.

The stateless property of JWTs is real and valuable - it is what makes AuthShield's tokens verifiable by any downstream service without calling back to AuthShield. But statelessness and revocability are fundamentally in conflict. The solution is not to pick one. It is to understand which layer handles which concern.

JWTs handle identity and authorization - who the user is and what they can do. Redis handles revocation - making logout immediate. PostgreSQL handles refresh token lifecycle - rotation, family tracking, and theft detection. Each layer does one thing.

Understanding that separation is what made the implementation make sense.

What Is Next

Next week: rate limiting, integration testing against real infrastructure, Docker, and what I found out about the gap between working locally and running in production.

I thought finishing the code meant I was done. It did not.

AuthShield on GitHub: AuthShield-Repository

Always learning, always observing.

I Thought OAuth Was Just Adding a Google Button. Turns Out It's a CSRF Problem Disguised as a Feature.

Ravi Gupta — Mon, 30 Mar 2026 03:15:00 +0000

This is Part 2 of a 4-part series on building AuthShield - a production-ready standalone authentication microservice. This post covers the OAuth 2.0 implementation: Authorization Code Flow, PKCE, CSRF protection, and account linking.
Part 1 is here: Why I Stopped Writing Auth Code for Every Project and Built AuthShield

When I started implementing OAuth 2.0 in AuthShield, I thought it was going to be one of the easier parts.

Add a Google button. Redirect the user. Get their email back. Done.

I was wrong - not because OAuth is complicated, but because I did not understand what it actually is. I thought it was a login mechanism. It is not. OAuth 2.0 is an authorization framework, and almost every step in the flow exists to defend against a specific attack.

Once I understood that, everything clicked. But getting there took more time than I expected.

What OAuth 2.0 Actually Is

Most engineers encounter OAuth for the first time through social login. Click "Sign in with Google," get redirected, come back logged in. The experience is smooth and the implementation feels like it should match that smoothness.

But OAuth 2.0 was not designed for authentication. It was designed for authorization - specifically, for allowing a third-party application to access resources on a user's behalf without ever seeing the user's password.

The "Sign in with Google" use case is built on top of OAuth 2.0 using OpenID Connect, which adds an identity layer. When you implement Google login, you are using both, whether you realise it or not.

This distinction matters because it changes how you think about the flow. You are not just "getting the user." You are asking Google to vouch for the user's identity and hand you a verified email address. Every step in the flow is about ensuring that handoff cannot be tampered with.

The Authorization Code Flow

AuthShield uses the Authorization Code Flow, which is the most secure OAuth flow for server-side applications. Here is how it works:

The user clicks "Sign in with Google"
AuthShield redirects the user to Google's authorization server with a set of parameters — client ID, requested scopes, redirect URI, and two security parameters we will come back to
The user logs into Google and grants permission
Google redirects the user back to AuthShield's callback URL with a short-lived authorization code
AuthShield exchanges that code for an access token by making a server-to-server request to Google — this is the key step, because the exchange happens on the backend, never in the browser
AuthShield uses the access token to fetch the user's profile from Google
AuthShield creates or finds the user in its own database and issues its own JWT

Step 5 is what makes this flow secure. The authorization code is short-lived and single-use. Even if someone intercepts the redirect in step 4, they cannot exchange the code - they do not have the client secret required for the server-to-server exchange.

The implicit flow, which used to be recommended for single-page applications, skipped step 5 and returned tokens directly in the URL fragment. That meant tokens were exposed in browser history, server logs, and referrer headers. It is deprecated now. Do not use it.

The State Parameter: The CSRF Problem I Almost Skipped

The first security parameter in the authorization request is the state parameter, and I almost did not implement it properly because it looked optional in the documentation.

It is not optional.

Here is the attack it prevents. Without a state parameter, an attacker can initiate an OAuth flow on your site and craft a malicious link to the callback URL. If they trick a victim into clicking that link — through a phishing email, a malicious website, anything - the victim's browser completes the OAuth callback and their account gets linked to the attacker's identity. The attacker now has access to the victim's account.

This is a Cross-Site Request Forgery attack applied to an OAuth flow. The state parameter prevents it by creating a binding between the authorization request and the callback.

Here is how AuthShield implements it:

import secrets
import json
from redis.asyncio import Redis

async def generate_oauth_state(redis: Redis) -> str:
    # Generate a cryptographically random state token
    state = secrets.token_urlsafe(32)

    # Store in Redis with a short TTL — 10 minutes is enough
    # After that, the state is invalid and the flow must restart
    await redis.setex(
        f"oauth_state:{state}",
        600,  # 10 minutes in seconds
        "1"
    )

    return state


async def verify_oauth_state(redis: Redis, state: str) -> bool:
    # Check if the state exists in Redis
    key = f"oauth_state:{state}"
    exists = await redis.exists(key)

    if not exists:
        return False

    # Delete immediately — state tokens are single-use
    # A replayed callback with the same state will fail
    await redis.delete(key)

    return True

A few things worth noting here. The state is stored in Redis, not the database. It is short-lived and single-use - once it is verified in the callback, it is deleted immediately. If an attacker somehow captures the state and tries to replay the callback, the token is already gone.

The callback handler verifies the state before doing anything else:

async def google_callback(
    code: str,
    state: str,
    redis: Redis = Depends(get_redis),
    db: AsyncSession = Depends(get_db)
):
    # First thing — verify the state
    # If this fails, reject the request entirely
    is_valid = await verify_oauth_state(redis, state)
    if not is_valid:
        raise HTTPException(
            status_code=400,
            detail="Invalid or expired state parameter"
        )

    # Only proceed if state is valid
    # Exchange code for token, fetch user info, etc.
    ...

PKCE: The Part That Took Me the Longest

The second security parameter is PKCE - Proof Key for Code Exchange, pronounced "pixie." This is the part of the OAuth implementation that surprised me the most and took the longest to understand properly.

PKCE was originally designed for mobile and single-page applications that cannot securely store a client secret. But it is now recommended for all OAuth flows regardless of application type.

Here is the problem it solves. In the Authorization Code Flow, the authorization code is passed through the browser in a redirect. If an attacker can intercept that redirect - through a malicious app on the same device, a compromised browser extension, or a misconfigured redirect URI - they have the code. And if they also have the client secret, they can exchange it for tokens.

PKCE removes the client secret from the equation for the code exchange. Instead, the client proves it initiated the original request by solving a cryptographic challenge.

Here is how it works:

import hashlib
import base64
import secrets

def generate_pkce_pair() -> tuple[str, str]:
    # Step 1: Generate a random code verifier
    # This is a long random string that only the client knows
    code_verifier = secrets.token_urlsafe(64)

    # Step 2: Hash the verifier to create the code challenge
    # SHA-256 hash, then base64url encode it
    digest = hashlib.sha256(code_verifier.encode()).digest()
    code_challenge = base64.urlsafe_b64encode(digest).rstrip(b'=').decode()

    return code_verifier, code_challenge

The flow with PKCE works like this:

AuthShield generates the code verifier and code challenge before redirecting to Google
The code challenge is sent to Google in the authorization request
The code verifier is stored temporarily - in AuthShield's case, in Redis alongside the state token
When Google redirects back with the authorization code, AuthShield sends both the code and the original code verifier to exchange for tokens
Google hashes the verifier, compares it to the challenge it stored from step 2, and only issues tokens if they match

An attacker who intercepts the authorization code cannot exchange it because they do not have the code verifier. It never left the server.

async def initiate_google_oauth(redis: Redis) -> str:
    # Generate state and PKCE pair
    state = await generate_oauth_state(redis)
    code_verifier, code_challenge = generate_pkce_pair()

    # Store the code verifier in Redis linked to the state
    # So we can retrieve it in the callback
    await redis.setex(
        f"pkce_verifier:{state}",
        600,
        code_verifier
    )

    # Build the Google authorization URL
    params = {
        "client_id": settings.GOOGLE_CLIENT_ID,
        "redirect_uri": settings.GOOGLE_REDIRECT_URI,
        "response_type": "code",
        "scope": "openid email profile",
        "state": state,
        "code_challenge": code_challenge,
        "code_challenge_method": "S256",
    }

    base_url = "https://accounts.google.com/o/oauth2/v2/auth"
    return f"{base_url}?{'&'.join(f'{k}={v}' for k, v in params.items())}"


async def exchange_code_for_token(
    code: str,
    state: str,
    redis: Redis
) -> dict:
    # Retrieve the code verifier stored during initiation
    code_verifier = await redis.get(f"pkce_verifier:{state}")
    if not code_verifier:
        raise HTTPException(status_code=400, detail="PKCE verifier not found")

    # Clean up
    await redis.delete(f"pkce_verifier:{state}")

    # Exchange code for token — include the verifier, not the challenge
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://oauth2.googleapis.com/token",
            data={
                "client_id": settings.GOOGLE_CLIENT_ID,
                "client_secret": settings.GOOGLE_CLIENT_SECRET,
                "code": code,
                "redirect_uri": settings.GOOGLE_REDIRECT_URI,
                "grant_type": "authorization_code",
                "code_verifier": code_verifier,  # This is what proves we initiated the request
            }
        )

    return response.json()

The reason this took me time to understand was not the code itself - it is not complicated once you see it. It was understanding why each step exists and what happens if you skip it. The docs describe PKCE as an enhancement. In practice, for any public-facing OAuth flow, it is a requirement.

Account Linking: The Edge Case Nobody Thinks About

Once the OAuth flow works, there is an edge case that is easy to miss: what happens when a user who already has an email and password account tries to sign in with Google using the same email?

There are two options. Reject them with an error - "this email is already registered, please log in with your password." Or link the accounts - recognise that it is the same person and give them access.

Rejecting them is the wrong call. It creates a confusing user experience and forces the user to remember which method they used to sign up. Linking them is the right call, but it requires careful handling.

async def find_or_create_oauth_user(
    email: str,
    full_name: str,
    oauth_provider: str,
    oauth_id: str,
    avatar_url: str,
    db: AsyncSession
) -> User:
    # First, check if a user with this OAuth provider ID already exists
    # This handles returning OAuth users
    existing_oauth_user = await user_repo.get_by_oauth(
        db, provider=oauth_provider, oauth_id=oauth_id
    )
    if existing_oauth_user:
        return existing_oauth_user

    # Check if a user with this email exists (registered with password)
    existing_email_user = await user_repo.get_by_email(db, email=email)

    if existing_email_user:
        # Link the OAuth account to the existing user
        # They can now sign in with either method
        existing_email_user.oauth_provider = oauth_provider
        existing_email_user.oauth_id = oauth_id
        existing_email_user.avatar_url = avatar_url
        # Mark as verified — the OAuth provider already verified the email
        existing_email_user.is_verified = True
        await db.commit()
        return existing_email_user

    # New user — create them
    # No password hash needed — they authenticate via OAuth
    # Mark as verified immediately — OAuth provider verified the email
    new_user = User(
        email=email,
        full_name=full_name,
        password_hash=None,  # Nullable — OAuth users have no password
        oauth_provider=oauth_provider,
        oauth_id=oauth_id,
        avatar_url=avatar_url,
        is_verified=True,  # Google already verified it
        is_active=True,
    )
    db.add(new_user)
    await db.commit()
    return new_user

Two things worth noting in this implementation.

First, the lookup order matters. We check by OAuth provider ID first, then by email. This handles the case where a user has already linked their OAuth account - we find them immediately without touching the email lookup.

Second, OAuth users are marked as verified immediately. Google has already verified the email address. Requiring them to go through email verification again would be both redundant and confusing. The password_hash field in the database is nullable specifically to accommodate users who authenticate only through OAuth and have no password.

What OAuth Actually Is

After implementing all of this, my understanding of OAuth shifted.

It is not a login feature. It is a set of security boundaries, each one defending against a specific attack. The state parameter defends against CSRF. PKCE defends against authorization code interception. The server-side code exchange defends against token exposure in the browser. Account linking defends against a fragmented user experience that pushes users toward insecure workarounds.

When I thought of it as "adding a Google button," I was looking at the 1% of the implementation that is visible. The other 99% is the security reasoning underneath it.

Understanding that reasoning is the difference between OAuth that works and OAuth that holds up.

What Is Next

Next week: JWTs, logout, and the conflict between stateless tokens and immediate revocation.

I thought JWTs were stateless and that was their whole point. Then I tried to implement logout. The solution - Redis blacklisting, token families, and reuse detection - turned out to be the most interesting engineering problem in the entire project.

AuthShield on GitHub: AuthShield-Repository

Always learning, always observing.

Why I Stopped Writing Auth Code for Every Project and Built AuthShield

Ravi Gupta — Mon, 23 Mar 2026 03:27:46 +0000

This is Part 1 of a 4-part series on building AuthShield - a production-ready standalone authentication microservice. This post is the origin story. No code, just the thinking.

Every backend project I have built starts the same way.

New repo. Fresh database. And then before writing a single line of product code, auth.

Registration. Email verification. Login. JWT tokens. Password reset. OAuth. Roles. Sessions.

The first time I built it, I learned a lot. I understood JWTs, token expiry, password hashing, OAuth flows. It was genuinely valuable.

Then I started the next project. And built it again.

Then the next one. And built it again.

At some point I stopped and looked at what I was doing. Every project needs authentication. The requirements are almost identical across all of them. The security concerns are exactly the same. Yet I was treating it as a fresh problem every single time.

That is not engineering. That is repetition.

So I asked myself a simple question: why am I solving the same problem again?

The Real Problem With Copying Auth Code

The obvious solution is to copy it from the previous project. Most engineers do this. I did it too.

But copying auth code carries a problem that only shows up later. Auth logic gets tangled with the application it was written for. It references that project's database models, that project's config structure, that project's way of handling errors. You copy it, spend a day untangling it, make adjustments to fit the new project, and ship it.

Two months later you find a security issue in the original. You patch it there. The copy still has the old version. You forget. It happens.

The deeper issue is that every time you touch auth code to make it fit a new project, you introduce risk. Security code is not the place for casual modifications. A small change in how tokens are validated, how refresh tokens are stored, or how OAuth state is handled can create a vulnerability that looks completely normal in a code review.

Copying is not reusing. It is duplicating - and duplicating security-critical code is one of the quieter ways things go wrong in backend engineering.

The Idea: A Microservice Built Once, Used Everywhere

The question was not how to write better auth code. The question was how to write it once and never write it again.

Authentication is infrastructure. It is not a feature of any specific application — it is the layer that makes features possible. And like any infrastructure, it should live in one place, be maintained in one place, and be consumed by everything that needs it.

That is what microservices are for. We separate payment processing into its own service. We separate email sending into its own service. Auth deserves the same treatment, arguably more so, because the consequences of getting it wrong are more severe than a failed payment webhook.

The vision was straightforward: build a standalone auth microservice that any backend project can plug into. Not import as a library. Not copy code from. Actually plug into - send it an HTTP request, get back a JWT, validate that JWT locally, done. The application never touches passwords, tokens, or OAuth flows directly.

One service. Any number of consumers. One place to maintain security. One place to patch vulnerabilities.

That is AuthShield.

What AuthShield Actually Handles

AuthShield is a deployable FastAPI service backed by PostgreSQL and Redis. Here is what it covers out of the box:

Authentication - Email and password registration with email verification, login returning JWT access and refresh tokens, Google and GitHub OAuth 2.0 via Authorization Code Flow, and TOTP-based two-factor authentication compatible with Google Authenticator and Authy.

Authorisation - Role-based access control with three built-in roles: user, moderator, and admin. Roles are embedded in the JWT so downstream services check permissions without a single database call.

Token security - 15-minute access tokens, 7-day refresh tokens with rotation on every use, Redis-based blacklisting so logout takes effect on the very next request, and refresh token reuse detection that revokes an entire token family the moment theft is suspected.

Session management - Every login creates a tracked session recording IP address and device info. Users can list all active sessions and revoke any of them individually.

Rate limiting - Redis sliding-window rate limits on every sensitive endpoint. Not fixed-window — sliding-window, which means there is no clock boundary reset to exploit.

Production infrastructure - Dockerised with a multi-stage build, Nginx for TLS termination and security headers on every response, structured JSON logging, and 48 integration tests running against real PostgreSQL and Redis.

The integration pattern is the part that makes this actually useful as a reusable service. Any downstream project needs exactly one file and one shared environment variable. The file is 40 lines of Python. It validates the JWT locally - no runtime call to AuthShield on every request - extracts the user ID and roles from the token, and makes them available to every protected route. That is it. The application never thinks about auth again.

Why Not Auth0, Clerk, or Any Existing Solution

This is the question worth answering honestly.

Managed auth services are good products. Auth0, Clerk, Supabase Auth - they are mature, well-documented, and genuinely save time. For many teams and many products, they are the right call.

But I had two specific reasons for not going that route.

The first is understanding. When you use a managed auth service, you are trusting a black box with the most security-critical layer of your application. That is a reasonable tradeoff for speed. But it means that when something goes wrong - and in auth, something eventually always goes wrong — you are dependent on someone else's documentation, someone else's support, and someone else's timeline to fix it.

I wanted to understand every decision in my auth layer. Why the token TTL is what it is. Why refresh tokens are stored hashed rather than plain. Why the OAuth state parameter exists and what happens if it is missing. Why sliding-window rate limiting behaves differently from fixed-window at the boundary. Not because I enjoy complexity, but because understanding your security layer is the difference between debugging a problem and being surprised by one.

The second is control. Security decisions should belong to the engineer shipping the system. What hashing algorithm, what token strategy, what revocation mechanism, what rate limit thresholds - these are decisions that have real consequences. Delegating them entirely to a third-party service means accepting their defaults without always knowing what those defaults are or why they were chosen.

AuthShield gives full control over every one of those decisions. Every choice is visible, documented, and mine to change.

Neither of these reasons means managed auth services are wrong. They mean they were wrong for what I was trying to build.

What This Series Covers

This post is the origin story. The next three go into the parts that actually made me think.

Next week - OAuth and why it is more than just adding a Google button

Week after - JWTs, logout, and why stateless tokens create a stateful problem.

Final week - Rate limiting, testing, and the gap between working locally and running in production.

The GitHub repo is linked below. The README has a full integration guide — how to plug AuthShield into any backend project in under 30 minutes. Contributions and feedback are welcome.

One Last Thought

I built AuthShield because I saw a problem I kept solving repeatedly and decided to solve it once properly.

But there is something else that came out of building it that I did not expect. The process of designing a security-focused microservice from scratch - making every decision deliberately, documenting every tradeoff, writing tests that actually run against real infrastructure - changed how I think about backend engineering in general.

Not every project needs a custom auth microservice. But every engineer who has built one has a different relationship with the auth layer in every project they touch after that.

That, more than anything, is why I would recommend the experience.

AuthShield on GitHub: https://github.com/ravigupta97/authshield

Always learning, always observing.

The Faster We Build with AI, the More Dangerous Bad Auth Becomes - And the Rarer Good Auth Becomes

Ravi Gupta — Mon, 16 Mar 2026 03:15:00 +0000

The Faster We Build with AI, the More Dangerous Bad Auth Becomes - And the Rarer Good Auth Becomes

While everyone races to ship with AI, understanding authentication and authorization is quietly becoming the most valuable skill in backend engineering.

Let me set a scene that probably sounds familiar.

You're building a new backend service. You open Cursor, type a prompt - "implement OAuth 2.0 login with JWT tokens" - and in about 8 seconds you have 150 lines of clean, readable, production-looking code. You skim it. It looks right. You move on.

I've done this. Most engineers I know have done this.

And here's what nobody talks about: that code is often almost correct. Not obviously broken - just subtly, silently wrong in ways that don't surface until something goes wrong in production. A token that should expire doesn't. A user who should lose access still has it. An endpoint that should be protected isn't.

The AI didn't fail you. You just didn't know what to check.

That's what this post is about.

First, Let's Get the Basics Right - Because Most People Still Conflate These

Before we go deeper, one thing I need to address: Authentication and Authorization are not the same thing, and confusing them is the root cause of more bugs than I can count.

Authentication answers: Who are you?
It's the process of verifying identity. When you log in with a username and password, or sign in with Google - that's authentication.

Authorization answers: What are you allowed to do?
It's the process of determining permissions. After the system knows who you are, authorization decides what resources you can access and what actions you can take.

Here's a simple way to remember it:

Authentication is the bouncer checking your ID at the door.
Authorization is the VIP list that decides which rooms you can enter.

You need both. In that order. Always.

Most AI-generated auth code handles authentication passably. Authorization is where things quietly go wrong — missing scope checks, over-permissioned tokens, endpoints that assume "logged in" means "allowed."

Keep this distinction in your head for the rest of this post.

Why This Matters More Right Now Than It Ever Has

A few years ago, building a backend with proper auth took meaningful effort. You had to understand the flow, write the code, debug it, test it. That friction was annoying - but it also meant you spent time with the code. You understood it.

Today, that friction is largely gone.

Copilot, ChatGPT, Claude, Cursor - these tools can scaffold an entire authentication system in minutes. That's genuinely incredible. It's also genuinely risky when the engineer shipping that code doesn't understand what's under the hood.

Think about what's happening at scale:

Thousands of developers are shipping AI-generated auth code every day
Many of them are newer engineers who haven't been burned by auth bugs yet
The code looks correct - it has the right function names, the right libraries, the right structure
But subtle mistakes - wrong grant type, missing token validation, no refresh rotation - slip through

The attack surface of the internet is growing faster than collective security knowledge. That's the uncomfortable reality.

And here's the flip side - the opportunity: engineers who genuinely understand auth, who can look at AI-generated OAuth code and immediately spot what's missing or wrong, are increasingly rare and increasingly valuable.

Let's make sure you're one of them.

OAuth 2.0: What It Actually Is (And What People Get Wrong About It)

OAuth 2.0 is an authorization framework - not an authentication protocol. This trips people up constantly. OAuth 2.0 on its own does not tell you who the user is. It tells you what a client application is allowed to do on behalf of a user.

Authentication on top of OAuth 2.0 is what OpenID Connect (OIDC) adds. If you're using OAuth to log users in, you're almost certainly using OIDC on top — whether you know it or not.

OAuth 2.0 works through flows, also called grant types. Different situations call for different flows. This is where I see AI-generated code go wrong most often - it picks a flow, but not necessarily the right one for the context.

The Four Flows You Need to Know

1. Authorization Code Flow + PKCE
Use this for: web apps, mobile apps, SPAs - anything where a user is present

This is the most common and most secure flow for user-facing applications. Here's how it works:

Your app redirects the user to the authorization server (e.g., Google, Auth0)
The user logs in and grants permission
The authorization server redirects back to your app with a short-lived authorization code
Your app exchanges that code for an access token (and optionally a refresh token)

The code is single-use and expires quickly. Even if someone intercepts the redirect, they can't do much with just the code.

PKCE (Proof Key for Code Exchange - pronounced "pixie") is the security enhancement that makes this flow safe for public clients like SPAs and mobile apps that can't securely store a client secret.

How PKCE works:

Your app generates a random code verifier (a long random string)
It hashes that verifier to create a code challenge
It sends the code challenge with the authorization request
When exchanging the code for tokens, it sends the original code verifier
The authorization server verifies the verifier matches the challenge

This means even if an attacker intercepts the authorization code, they can't exchange it - they don't have the code verifier.

AI-generated code mistake I see here: Omitting PKCE entirely for SPAs, or implementing the Authorization Code flow without PKCE for mobile apps. It looks fine. It isn't.

2. Client Credentials Flow
Use this for: machine-to-machine (M2M), service-to-service, backend APIs

No user involved here. This is for when one service needs to talk to another service. Your backend requests a token directly from the authorization server using its client ID and client secret.

Backend Service → Authorization Server: "Here's my client ID and secret"
Authorization Server → Backend Service: "Here's your access token"
Backend Service → Resource API: "Here's my access token"

Simple. But the mistakes here are costly:

Storing the client secret in version control (it happens more than you think — and AI-generated setup instructions sometimes put secrets in example config files that get committed)
Not rotating secrets - a secret that never changes is a liability
Over-scoping the token - requesting all scopes "just in case" instead of the minimum required
Not handling token expiry - the service breaks at 2am because nobody implemented token refresh logic

3. Implicit Flow
Status: Deprecated. Do not use.

I'm mentioning this because AI tools sometimes still generate it - especially if trained on older code. The Implicit Flow was designed for SPAs before PKCE existed. It returned access tokens directly in the URL fragment, which is a significant security issue. PKCE-enhanced Authorization Code Flow replaced it entirely.

If you see Implicit Flow in AI-generated code, replace it.

4. Resource Owner Password Credentials (ROPC)
Status: Highly discouraged. Almost never appropriate.

This flow has the user send their username and password directly to your application, which then exchanges them for tokens. It defeats the entire point of OAuth - the user should never give their credentials to a third-party app.

The only time this might be acceptable is for highly trusted first-party applications, and even then, there are usually better alternatives. AI tools sometimes generate this for "simplicity." It isn't worth it.

Token Security: Where Most Bugs Live

Let's talk about the actual tokens - because this is where subtle mistakes cause the most damage.

JWT: The Good, The Bad, and The Dangerous

JSON Web Tokens (JWTs) are everywhere. An access token in most OAuth implementations is a JWT. Here's what one looks like under the hood:

header.payload.signature

Header: algorithm used to sign the token (RS256, HS256, etc.)
Payload: the claims - user ID, scopes, expiry, issuer
Signature: cryptographic proof the token hasn't been tampered with

The most dangerous JWT mistake: accepting alg: none

Early JWT libraries had a flaw - if the header said "alg": "none", some libraries would skip signature verification entirely. An attacker could forge any token they wanted.

The fix: always explicitly specify which algorithms your server accepts. Reject anything else.

import jwt

# Bad - trusting the token's own algorithm header
# Some libraries will skip verification if alg is "none"
payload = jwt.decode(token, secret)

# Better - explicitly specify accepted algorithms
payload = jwt.decode(
    token,
    secret,
    algorithms=["RS256"]  # reject anything else
)

Other JWT mistakes I've observed:

No expiry (exp claim missing): A token that never expires is a permanent backdoor. Always set short expiry - 15 minutes for access tokens is common.
Sensitive data in the payload: The payload is base64-encoded, not encrypted. Anyone can decode it. Don't put passwords, PII, or sensitive business logic in JWT claims.
Not validating the issuer (iss) and audience (aud): A token issued for Service A should not be accepted by Service B. Always validate these claims.
Weak signing secrets: If you're using HMAC (HS256), your secret needs to be long and random. secret123 is not a secret.

Refresh Tokens: The Part Everyone Forgets to Secure

Access tokens are short-lived. Refresh tokens are long-lived - they're used to get new access tokens without the user re-authenticating. That makes them valuable targets.

Refresh Token Rotation is the security practice you should be using:

Every time a refresh token is used, the authorization server issues a new refresh token and invalidates the old one. If an attacker steals a refresh token and tries to use it after the legitimate user already has, the server detects the reuse and can invalidate the entire session.

AI-generated code often implements refresh token logic without rotation. It works - it just doesn't protect against token theft.

Refresh token storage matters too:

In web apps: HttpOnly, Secure, SameSite=Strict cookies - not localStorage (XSS can read localStorage)
In mobile apps: the device's secure keychain/keystore
In backend services: encrypted at rest

Token Revocation: The Feature Nobody Implements Until They Need It

What happens when a user reports their account was compromised? Or when an employee leaves the company? Or when you detect suspicious activity?

You need to revoke tokens - immediately invalidate them before they expire naturally.

JWTs are stateless by design, which means the server doesn't track them. A revoked JWT is still cryptographically valid until it expires. There are two practical approaches:

Token blocklist: Maintain a fast store (Redis works well) of revoked token IDs (jti claim). Check every incoming token against this list. Overhead is low if done right.
Short expiry + refresh rotation: Keep access tokens extremely short-lived (5-15 minutes). Revoke the refresh token at the authorization server. The user's access naturally expires within minutes.

Most AI-generated code implements neither. It's not a bug that shows up in testing — only in an incident.

Common Vulnerabilities: A Quick-Reference List

These are the ones I check for when reviewing auth code — AI-generated or otherwise:

Scope Creep: Requesting more OAuth scopes than needed. Always use minimum required scopes. If your app only needs to read a user's email, don't request write access to their entire account.

Open Redirect: OAuth flows use redirect URIs. If your server doesn't strictly validate the redirect URI against a whitelist, an attacker can redirect tokens to their own server. Always validate exact URI matches — not just domain matches.

State Parameter Missing: The state parameter in OAuth flows is CSRF protection. It's a random value your app sends with the authorization request and expects back in the callback. Without it, an attacker can trick a user into completing an OAuth flow the attacker initiated. AI-generated code frequently omits this.

Logging Tokens: Access tokens and refresh tokens should never appear in logs. One misconfigured logger that records request headers or bodies, and your tokens are in your logging system - accessible to anyone with log access. Scrub tokens from logs explicitly.

Trusting Client-Provided User IDs: If your API accepts a user_id in the request body and acts on it without verifying it matches the authenticated token's subject claim - that's a broken authorization vulnerability. Always derive the user identity from the validated token, never from user-provided input.

What This Means for Your Career

Here's what I've been observing and thinking about.

AI tools are genuinely compressing the time it takes to build things. That's not going away. More backends will be built, by more people, faster than ever before.

But auth is not a problem that's been solved by AI generation. It's a domain where:

Mistakes are invisible until they're exploited
"Looks correct" and "is correct" are dangerously different things
The consequences of getting it wrong fall on real users

Engineers who can look at AI-generated auth code and immediately know what to look for, why certain patterns are dangerous, and what questions to ask - those engineers are not being replaced by AI tools. They're the ones who can actually use those tools responsibly.

That's a real differentiator. Not just technically, but professionally.

The Practical Checklist: What to Review in Every Auth Implementation

Whether the code is AI-generated or hand-written, here's what I run through:

[ ] Are we using the right OAuth flow for this use case?
[ ] Is PKCE implemented for public clients?
[ ] Are JWTs validated with explicit algorithm specification?
[ ] Do access tokens have short expiry (exp claim set)?
[ ] Is the iss (issuer) and aud (audience) being validated?
[ ] Is refresh token rotation implemented?
[ ] Are refresh tokens stored securely (not in localStorage)?
[ ] Is there a token revocation strategy?
[ ] Are redirect URIs strictly validated against a whitelist?
[ ] Is the state parameter used to prevent CSRF?
[ ] Are tokens being excluded from logs?
[ ] Does each OAuth scope request follow the principle of least privilege?
[ ] Is user identity derived from the token, not from request input?

Print it. Bookmark it. Run through it before every auth PR you review.

Final Thought

AI didn't make auth easier to get right. It made it easier to get almost right - which in security, is the most dangerous place to be.

The engineers who will stand out in the next few years aren't the ones who can prompt the fastest. They're the ones who understand what the prompt produced deeply enough to know when to trust it, when to question it, and when to rewrite it.

Auth is one of those areas where that judgment is everything.

Go build - but build like you know what's under the hood.

Found this useful? A backend engineer - always learning, always observing, and writing about what I pick up along the way. If this resonated,
let's connect on LinkedIn | GitHub.

#Authentication #Authorization #OAuth #Security #BackendDevelopment #SoftwareEngineering

Building a Production-Ready Task Management API with FastAPI: Testing, Deployment & Production (Part 3)

Ravi Gupta — Mon, 09 Mar 2026 03:19:27 +0000

Building a Production-Ready Task Management API with FastAPI: Testing, Deployment & Production (Part 3)

Part 3 of 3: After the architecture and development phases, I thought deployment would be straightforward. I was wrong. This is the final chapter of my journey from local code to live production system.

The API worked perfectly on my laptop.

50+ endpoints responding.
Authentication solid.
Clean architecture.
All tests passing locally.

I was ready to deploy.

Then I clicked "Deploy."

That's when everything broke.

CORS errors I'd never seen.
Environment variables mysteriously missing.
Database migrations failing for no reason.
Rate limiting crashing the entire app.
Cold starts taking 30 seconds.

This article covers Phase 3: Testing, Deployment & Production - where theory meets reality, and "working" becomes "production-ready."

Not just what worked.
What broke, what surprised me, and what I learned debugging a live system at midnight.

Quick Links:

📚 What You'll Learn

Testing async FastAPI with pytest (the setup that actually works)
Docker optimization strategies (1.2GB → 380MB)
Free-tier deployment with Render + Neon PostgreSQL
5 production bugs that only happen after deployment
Real metrics from 30 days of uptime
What I'd do differently if I started over
Cost breakdown of running production API ($0!)

Reading time: ~15 minutes

GitHub Repository: https://github.com/ravigupta97/task_management_api

Live API: https://task-management-api-a775.onrender.com/docs

🎯 The Journey So Far

Phase 1: Spent days designing the perfect architecture.
Phase 2: Built features, fought async bugs, implemented JWT from scratch.
Phase 3: Time to deploy. How hard could it be?

Spoiler: Very hard.

In Part 1, I designed a clean architecture:

FastAPI + PostgreSQL
Repository pattern
Service layer separation
Clean project structure

In Part 2, I built the actual system:

JWT authentication
50+ endpoints
Rate limiting
Advanced features

On paper, everything looked solid.

In development, everything worked.

But there's a gap between "works on my machine" and "works in production."

A massive gap.

This phase was about:

Writing tests that catch real bugs (not just make CI green)
Containerizing with Docker (and learning why 1.2GB is unacceptable)
Deploying to actual infrastructure (free tier!)
Debugging production issues at midnight
Measuring real performance (not synthetic benchmarks)

The honest truth: This phase took almost as long as building the features.

Because deployment isn't just "push to Render."

It's environment variables you forgot. CORS configurations that work locally but fail remotely. Database connection pools that exhaust mysteriously. Cold starts that make your API feel broken. Migrations that succeed locally but fail in production.

Every bug taught me something tutorials never mentioned.

This is where you stop being a tutorial follower and start being an engineer.

🧪 Testing Strategy

The Wake-Up Call

I thought I could skip comprehensive testing.

"I manually tested everything. It works."

Then I added a new feature.

Broke 3 existing endpoints.

Didn't notice until I tried to create a task and got a 500 error.

That's when I learned: Manual testing doesn't scale.

You need automated tests. Not because they're trendy.

Because your memory isn't perfect.

The Goal: 85%+ Coverage

Target: 85%+ test coverage

Not arbitrary.

85% forces you to test:

Happy paths (obviously)
Error cases (what happens when things fail)
Edge cases (empty strings, null values, boundary conditions)
Integration points (do layers work together?)

Below 85%? You're probably skipping important scenarios.

Above 90%? Diminishing returns (testing getters/setters, framework code).

My target: 85-87%. Test what matters.

The Challenge: Async Testing

Coming from Spring Boot, testing was familiar:

@Test
public void testGetTasks() {
    List<Task> tasks = taskService.getTasks();
    assertEquals(5, tasks.size());
}

Synchronous. Simple. Works.

In FastAPI with async?

Nothing worked initially.

# This failed
def test_get_tasks():
    response = client.get("/tasks")  # Error: can't await

# This also failed
async def test_get_tasks():
    response = await client.get("/tasks")  # Error: client not async

Async testing requires:

Async test client (not regular TestClient)
Async fixtures
Event loop management
Database session handling

It's not intuitive.

But once you understand it, it's powerful.

The Setup That Finally Worked

After hours of trial and error, here's the pytest configuration that actually works:

# conftest.py
import pytest
import asyncio
from httpx import AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker

from app.main import app
from app.database import get_db, Base

# Test database URL
TEST_DATABASE_URL = "postgresql+asyncpg://test_user:test_pass@localhost/test_db"

# Create test engine
test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)

# Create test session factory
TestingSessionLocal = sessionmaker(
    test_engine,
    class_=AsyncSession,
    expire_on_commit=False
)

@pytest.fixture(scope="session")
def event_loop():
    """Create event loop for async tests."""
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="function")
async def db_session():
    """Create fresh test database for each test."""
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    async with TestingSessionLocal() as session:
        yield session

    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)

@pytest.fixture(scope="function")
async def client(db_session):
    """Create test client with database override."""
    async def override_get_db():
        yield db_session

    app.dependency_overrides[get_db] = override_get_db

    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

    app.dependency_overrides.clear()

Why this setup works:

Separate test database - Never pollute your dev database
Fresh database per test - Isolation prevents flaky tests
Async fixtures - Match your application's async nature
Dependency override - FastAPI's killer feature for testing

Time to figure this out: 4 hours of Stack Overflow and trial and error.

Worth it? Absolutely. Tests are now fast, isolated, and reliable.

Testing Authentication: The Fixture Pattern

Challenge: How do you test protected endpoints without logging in every time?

Solution: Create authenticated test client fixture.

# conftest.py
import pytest
from app.core.security import create_access_token
from app.models.user import User

@pytest.fixture
async def test_user(db_session):
    """Create a test user."""
    user = User(
        email="test@example.com",
        username="testuser",
        hashed_password="$2b$12$...",  # hashed "testpassword"
        is_active=True,
        is_verified=True
    )
    db_session.add(user)
    await db_session.commit()
    await db_session.refresh(user)
    return user

@pytest.fixture
async def authenticated_client(client, test_user):
    """Create authenticated test client."""
    access_token = create_access_token(data={"sub": str(test_user.id)})
    client.headers = {
        **client.headers,
        "Authorization": f"Bearer {access_token}"
    }
    return client

Now testing is clean:

@pytest.mark.asyncio
async def test_create_task(authenticated_client):
    """Test creating a task."""
    response = await authenticated_client.post(
        "/api/v1/tasks/",
        json={
            "title": "Test Task",
            "description": "Test Description",
            "status": "TODO",
            "priority": "HIGH"
        }
    )

    assert response.status_code == 201
    data = response.json()
    assert data["title"] == "Test Task"
    assert data["priority"] == "HIGH"

No manual login. No token management. Just test the business logic.

This pattern saved me hundreds of lines of repetitive setup code.

Integration Tests: The Full Journey

Integration tests verify the entire flow works together.

@pytest.mark.asyncio
async def test_complete_task_lifecycle(authenticated_client):
    """Test full CRUD flow for tasks."""

    # Create
    create_response = await authenticated_client.post(
        "/api/v1/tasks/",
        json={"title": "Integration Test", "status": "TODO", "priority": "LOW"}
    )
    assert create_response.status_code == 201
    task_id = create_response.json()["id"]

    # Read
    get_response = await authenticated_client.get(f"/api/v1/tasks/{task_id}")
    assert get_response.status_code == 200
    assert get_response.json()["title"] == "Integration Test"

    # Update
    update_response = await authenticated_client.put(
        f"/api/v1/tasks/{task_id}",
        json={"status": "COMPLETED"}
    )
    assert update_response.status_code == 200
    assert update_response.json()["status"] == "COMPLETED"

    # Delete
    delete_response = await authenticated_client.delete(f"/api/v1/tasks/{task_id}")
    assert delete_response.status_code == 204

    # Verify deletion
    verify_response = await authenticated_client.get(f"/api/v1/tasks/{task_id}")
    assert verify_response.status_code == 404

What this caught:

Forgot to add CASCADE DELETE (got constraint violation errors)
Token expiry during long test runs
Session closure issues in UPDATE operations

Integration tests = real-world scenarios = real bugs found.

Test Coverage: The Results

After writing tests for all critical paths:

$ pytest --cov=app tests/

---------- coverage: platform linux, python 3.11.16 -----------
Name                              Stmts   Miss  Cover
-----------------------------------------------------
app/main.py                          45      2    96%
app/api/v1/auth.py                   89      8    91%
app/api/v1/tasks.py                 102     10    90%
app/services/task_service.py        124     15    88%
app/core/security.py                 52      3    94%
-----------------------------------------------------
TOTAL                              1247    158    87%

87% coverage achieved.

What I didn't test (the remaining 13%):

Some error handling edge cases
Specific database constraint failures
Rate limiter timing dependencies

Why that's okay: 87% covers all critical paths. The remaining 13% is mostly defensive code and edge cases that are hard to reliably test.

Time invested in testing: 1 week

Bugs caught before production: 15+

Worth it? Absolutely.

🐳 Docker: From 1.2GB to 380MB

My First Docker Image Was Embarrassing

1.2GB.

For a Python API.

I pushed it to Docker Hub, proud of myself.

Then I tried to deploy it to Render.

"Build timeout: Image too large"

Panic.

That's when I learned: Docker image size matters.

Not just for build time.
For deployment speed.
For resource usage.
For everything.

Time to optimize.

The Naive Approach

My first Dockerfile was... simple:

FROM python:3.11
COPY . /app
WORKDIR /app
RUN pip install -r requirements.txt
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Image size: 1.2GB

Problems:

Includes full Python image (300MB+ of unnecessary tools)
Build dependencies stay in final image
pip cache included
Unnecessary files copied

Build time: 8 minutes

Deployment: Timed out on Render's free tier

Not acceptable.

Multi-Stage Builds: The Solution

After researching Docker best practices, I discovered multi-stage builds.

The concept: Build in one image, run in another.

# Stage 1: Builder - Install dependencies
FROM python:3.11-slim as builder

WORKDIR /app

# Install build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    postgresql-client \
    && rm -rf /var/lib/apt/lists/*

# Copy and install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# Stage 2: Runtime - Small, clean image
FROM python:3.11-slim

WORKDIR /app

# Install only runtime dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    libpq5 \
    && rm -rf /var/lib/apt/lists/*

# Copy installed packages from builder
COPY --from=builder /root/.local /root/.local

# Copy application code
COPY ./app ./app
COPY alembic.ini .
COPY ./alembic ./alembic

# Ensure pip packages are in PATH
ENV PATH=/root/.local/bin:$PATH

EXPOSE 8000

# Run migrations and start server
CMD alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000

Final image size: 380MB

Reduction: 68% smaller!

Key optimizations:

Multi-stage build - Builder stage separate from runtime
Slim base image - python:3.11-slim instead of full python:3.11
No cache in pip - --no-cache-dir flag
Clean apt lists - rm -rf /var/lib/apt/lists/* after installs
Only runtime dependencies - gcc and build tools left in builder stage

Build time: 3 minutes (down from 8!)

Deployment: Successful on Render free tier

That felt good.

Local Development: docker-compose

For local development, I needed:

PostgreSQL database
API server
Hot reload
Easy setup

docker-compose made it simple:

version: '3.8'

services:
  db:
    image: postgres:16
    environment:
      POSTGRES_USER: taskapi
      POSTGRES_PASSWORD: taskapi
      POSTGRES_DB: taskapi_dev
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  api:
    build: .
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
    volumes:
      - ./app:/app/app  # Hot reload
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql+asyncpg://taskapi:taskapi@db:5432/taskapi_dev
      SECRET_KEY: dev-secret-key
    depends_on:
      - db

volumes:
  postgres_data:

One command to start everything:

docker-compose up

Benefits:

Consistent environment across team
No "works on my machine" issues
Easy onboarding for new developers
Matches production setup

This decision saved hours of environment debugging.

🚀 Deployment: Free Tier Reality

Platform Choice: The $0 Constraint

I had a strict budget for infrastructure.

$0.

Not "limited." Zero dollars.

This eliminated most options immediately.

The Contenders:

Platform	Free Tier	PostgreSQL	Catches
Render	✅ Yes	✅ Neon integration	Sleeps after 15min
Heroku	❌ No longer free	✅ Yes	Killed free tier
AWS	⚠️ 12 months only	✅ Yes	Complex setup
Railway	✅ Limited hours	✅ Yes	$5 credit/month
Fly.io	✅ Yes	✅ Yes	Credit system

Decision: Render

Why?

Truly free forever (not trial)
Docker support
Auto-deploy from GitHub
Simple configuration
Good documentation

Trade-off: Cold starts after 15 minutes of inactivity.

Acceptable for a learning project.

Database: Neon PostgreSQL

The Problem: Most free PostgreSQL offerings have catches.

Heroku: Limited to 10K rows (not enough)
Supabase: Pauses after 7 days inactivity (annoying)
AWS RDS: Only 12 months free (not sustainable)

Neon's free tier:

✅ Actually free forever
✅ 3GB storage
✅ Serverless (auto-scales)
✅ Branch-based development (like git!)
✅ No sleep/pause

Setup time: 5 minutes

Create Neon account
Create database
Copy connection string
Add to Render environment variables

Done.

This was the easiest part of deployment.

The rest? Not so much.

First Deployment: Everything Broke

I pushed to GitHub.

Render detected the Dockerfile.

Build started.

Then:

Error: Environment variable DATABASE_URL not found
Error: SECRET_KEY is None
Error: Application failed to start

What I forgot:

Environment variables don't automatically transfer from .env file to Render.

You have to set them manually in Render's dashboard.

Lesson learned: Environment configuration is separate from code.

Never assume.

The Health Check Endpoint

Critical for free tier deployments.

Render's free tier sleeps after 15 minutes of inactivity.

Health checks can:

Wake it up automatically
Verify it's actually running
Test database connectivity

# app/main.py
from fastapi import FastAPI, status
from sqlalchemy import text

@app.get("/health", status_code=status.HTTP_200_OK)
async def health_check(db: AsyncSession = Depends(get_db)):
    """
    Health check endpoint.
    Tests database connectivity.
    """
    try:
        await db.execute(text("SELECT 1"))
        return {
            "status": "healthy",
            "database": "connected",
            "environment": settings.ENVIRONMENT
        }
    except Exception as e:
        return {
            "status": "unhealthy",
            "database": "disconnected",
            "error": str(e)
        }

Benefits:

Confirms app is running
Tests database connection
Used by monitoring services
Simple debugging tool

This endpoint became my best friend during deployment debugging.

🚧 Production Bugs: The 2 AM Stories

Bug #1: The 30-Second Cold Start

The Problem:

I sent the API link to a friend.

"Check out my live project!"

I waited.

30 seconds passed.

Still loading.

I refreshed. 30 more seconds.

Internal panic.

"Is it down? Did I break something??"

Checked Render logs: "Spinning up service from sleep..."

Oh.

Free tier sleeps after 15 minutes.

Every. Single. Time.

The Solution:

Can't eliminate cold starts on free tier.

But I could handle the UX:

Health check endpoint - Wake it up faster
Frontend loading state - "Server waking up, please wait..."
Set expectations - Document it's free tier

Better solution: Upgrade to paid tier ($7/month for always-on).

But this is a learning project. Free tier it is.

Lesson learned: Constraints force creative solutions. Or at least, creative messaging.

Bug #2: The Case-Sensitive Environment Variable

The Problem:

API deployed successfully.

Opened the docs page.

Everything loaded.

Tried to login.

{
  "detail": "Internal Server Error"
}

Checked logs:

AttributeError: 'NoneType' object has no attribute 'decode'

What?

Spent an hour debugging.

Checked JWT code. Looked fine.
Checked database connection. Working.
Checked everything. Nothing made sense.

Then I saw it.

Environment variable in Render dashboard:

SECRECT_KEY instead of SECRET_KEY

I misspelled "SECRET" as "SECRECT".

JWT library tried to decode with None as the key.

Crashed.

The Fix:

Fixed the typo
Added startup validation:

# app/main.py
@app.on_event("startup")
async def validate_config():
    """Validate required configuration on startup."""
    required = ["SECRET_KEY", "REFRESH_SECRET_KEY", "DATABASE_URL"]

    missing = [key for key in required if not getattr(settings, key, None)]

    if missing:
        raise ValueError(f"Missing required config: {missing}")

    logger.info("✅ Configuration validated")

Now if a required variable is missing, the app fails immediately on startup.

Not after the first request.

Lesson learned: Fail fast. Validate early. Save yourself hours of debugging.

Bug #3: Database Connection Pool Exhaustion

The Problem:

API ran fine for a few hours.

Then started throwing errors:

asyncpg.exceptions.TooManyConnectionsError: too many connections

What?

I wasn't doing anything different.

No traffic spike.

Just... ran out of connections.

The Root Cause:

Some error paths weren't closing database sessions.

Session leaked.
Connection leaked.
Pool exhausted.

The Fix:

Ensure session cleanup in dependency:

# app/database.py
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()  # Always close, even on error

Configure connection pool limits:

# app/database.py
engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=5,          # Max connections
    max_overflow=10,      # Overflow allowed
    pool_pre_ping=True,   # Verify before use
    pool_recycle=3600     # Recycle after 1 hour
)

After the fix: No more connection errors.

Lesson learned: Connection pools are finite resources. Manage them carefully.

Bug #4: Alembic Migration Conflicts

The Problem:

Deployed new code with database migration.

Render build failed:

alembic.util.exc.CommandError: 
Can't locate revision identified by 'abc123'

Why?

I created migrations locally.
Pushed code.
But Render's database didn't have the previous migrations.

The Fix:

Run migrations as part of deployment:

# In Dockerfile CMD
CMD alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000

This ensures:

Migrations run before app starts
Database is always up to date
No manual migration steps

Also added:

# Before creating new migration locally
alembic current  # Check current state
alembic upgrade head  # Apply all pending

Lesson learned: Migrations are sequential. Treat them like git commits. Linear history matters.

📊 Performance: 30 Days of Real Data

The Numbers

After 30 days of production traffic, here's how it actually performed:

Response Times:

Endpoint	Average	P95	P99
Health check	45ms	89ms	142ms
Login	156ms	278ms	421ms
Get tasks	87ms	156ms	289ms
Create task	112ms	198ms	334ms

Overall average: ~190ms

My initial target: 100ms

Reality: Almost 2x slower.

Why?

Neon serverless adds latency (~20ms)
Free tier CPU is shared
No caching layer
Some queries not optimized

Is 190ms acceptable?

For a free-tier learning project? Yes.

For a production SaaS? No. I'd need caching and optimization.

Uptime Tracking

Tool: UptimeRobot (free tier)

30-day results:

✅ 99.2% uptime
❌ 3 outages (total: 5 hours 47 minutes)
⏱️ Average response: 180ms
📈 Checks performed: 43,200

Outage breakdown:

Render platform maintenance (2 hours)
Neon database scaling issue (30 minutes)
My deployment bug (15 minutes - the ENV variable typo)

99.2% uptime on $0 infrastructure?

Honestly better than I expected.

For comparison, AWS promises 99.99% (that's 4 minutes downtime per month).

I had ~5 hours downtime in 30 days.

Not bad for free.

Database Query Performance

Biggest optimization: Fixing N+1 queries.

Before:

# Get tasks - N+1 problem
tasks = await task_repository.get_by_user(user_id)
for task in tasks:
    print(task.category.name)  # Separate query for each!

After:

# Eager load relationships
from sqlalchemy.orm import selectinload

tasks = await db.execute(
    select(Task)
    .options(selectinload(Task.category))
    .where(Task.user_id == user_id)
)

Performance improvement: 3x faster for lists with categories.

Lesson learned: Async doesn't make bad queries good. Optimize your database access.

💰 Cost Analysis: The $0 Stack

Monthly Infrastructure Costs

Service	Plan	Cost
Render	Free tier	$0
Neon PostgreSQL	Free tier	$0
Domain (optional)	Namecheap	~$1/month
Monitoring	UptimeRobot free	$0
Total		$0/month

Yearly cost: $0 (or $12 if you buy a domain)

Free Tier Constraints

Render Free Tier:

✅ 750 hours/month (enough for one service)
✅ Auto-deploy from Git
✅ SSL certificate included
❌ Sleeps after 15 min inactivity
❌ Limited to 512MB RAM
❌ Shared CPU

Neon Free Tier:

✅ 3GB storage
✅ Unlimited databases
✅ Serverless auto-scaling
❌ Limited compute units
❌ Shared resources

When to Upgrade?

Stay on free tier if:

Learning project
Portfolio piece
Low traffic (<1000 requests/day)
Can tolerate cold starts
No SLA requirements

Upgrade to paid ($7-10/month) if:

Real users depending on it
Need consistent performance
Can't afford downtime
Traffic grows
Professional project

For this project: Free tier is perfect.

It does everything I need for a portfolio piece.

💡 Lessons Learned

1. Tests Save More Time Than They Take

My initial thought: "Tests slow me down. I'll skip them."

Reality: Tests caught 15+ bugs before production.

Time spent writing tests: 1 week
Time saved debugging production: Probably 2+ weeks

ROI: Massively positive.

Tests aren't overhead. They're insurance.

2. Docker Isn't Optional

Why Docker matters:

Reason 1: "Works on my machine" becomes "works everywhere"
Reason 2: Deployment is just "push image"
Reason 3: Forces you to think about dependencies

Time invested learning Docker: 2 days
Time saved in deployment issues: Countless hours

Every modern backend project should use Docker.

Not optional.

3. Monitoring Isn't Optional Either

Without monitoring, you're blind.

Questions you can't answer:

Is the API down?
Why is it slow?
Where are errors happening?
What's the actual performance?

With basic monitoring:

Health check endpoint
Logging to stdout
UptimeRobot for uptime
Manual log review

Cost: $0
Value: Priceless

You can't fix what you can't measure.

4. Free Tier Forces Better Decisions

Constraints I faced:

No Redis → In-memory rate limiting
Cold starts → UX handling + health checks
Limited resources → Query optimization

Result: Better architecture.

Paid tiers let you throw money at problems.

Free tiers force you to solve them properly.

The best learning happens under constraints.

5. Production Is the Real Teacher

Local development teaches:

How to write code
How to structure projects
How to use frameworks

Production teaches:

How systems fail
How to debug without IDE
How to handle real constraints
How to think about users

The gap between these two is where engineers are made.

🔄 What I'd Do Differently

1. Write Tests Earlier (TDD)

What I did: Built features, then wrote tests.

Problem: Had to refactor code to make it testable.

Better approach: Test-Driven Development

Write test first
Implement feature
Refactor

Why: Forces good design from the start.

2. Set Up CI/CD Immediately

What I did: Manually ran tests before deploying.

Problem: Forgot twice. Deployed broken code.

Better approach: GitHub Actions

# Simple workflow
name: Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - uses: actions/setup-python@v2
      with:
        python-version: 3.11
    - run: pip install -r requirements.txt
    - run: pytest --cov=app tests/

Benefit: Tests run automatically. Can't merge broken code.

3. Add Caching Earlier

What I did: Direct database queries for everything.

Problem: Repeated queries for same data.

Better approach: Redis caching for:

User sessions
Frequently accessed data
API rate limit counters

Why I didn't: Free tier doesn't include Redis.

Workaround: functools.lru_cache for config/static data.

4. Implement Database Backups

What I did: Relied on Neon's automatic backups.

Problem: No control over backup schedule or retention.

Better approach:

Scheduled pg_dump to S3
Version-controlled schema
Tested restore process

Why it matters: Data loss is unacceptable. Even for learning projects.

5. Use Feature Flags

What I did: Deploy features directly to production.

Problem: Can't disable buggy features without rollback.

Better approach:

# Simple feature flags
class FeatureFlags:
    ENABLE_EMAIL_VERIFICATION = os.getenv("FEATURE_EMAIL", "true") == "true"
    ENABLE_ADVANCED_SEARCH = os.getenv("FEATURE_SEARCH", "false") == "true"

# Usage
if FeatureFlags.ENABLE_ADVANCED_SEARCH:
    # New feature logic
else:
    # Old feature logic

Benefit: Enable/disable features without deploying.

🎯 Final Thoughts

What This Project Taught Me

1. Architecture matters — but implementation teaches you why

2. Tests aren't optional — they're the difference between hobby and professional

3. Production is different — cold starts, CORS, connection pools, migrations... none of this shows up in tutorials

4. Free tier is enough — to learn, to build portfolio, to prove you can ship

5. Building in public works — documenting this journey kept me accountable

The Numbers

What I built:

✅ 50+ API endpoints
✅ JWT authentication
✅ 87% test coverage
✅ Clean architecture
✅ Dockerized
✅ Deployed (99.2% uptime)
✅ $0 infrastructure cost

What I learned:

Way more than any tutorial could teach
The gap between "working" and "production-ready"
How to debug systems you can't see
How to make architectural decisions under constraints
How to ship something real

Was It Worth It?

Absolutely.

Not just for the portfolio.

Not just for the resume.

For the learning.

There's a massive difference between:

Following a tutorial
Building from scratch
Deploying to production

This project forced me through all three.

Every bug taught me something.
Every constraint forced creativity.
Every deployment taught me production.

That's the real value.

🚀 Try It Yourself

Live API: https://task-management-api-a775.onrender.com/docs

(First request might take 30 seconds if it's sleeping - free tier!)

GitHub: https://github.com/ravigupta97/task_management_api

Clone and run:

git clone https://github.com/ravigupta97/task_management_api
cd task_management_api
docker-compose up

Open http://localhost:8000/docs

Everything you need is in the README.

💬 Discussion

Questions for you:

What's your testing strategy for async APIs?
How do you handle cold starts on free tiers?
What was your biggest production surprise?
What free tier tools do you use?

Let's learn together - comment below! 👇

This completes the 3-part series on building a production-ready FastAPI application.

Part 1: Architecture & Design
Part 2: Development & Implementation
Part 3: Testing, Deployment & Production (you just read it)

What's next for me?

Follow along: LinkedIn | GitHub

#FastAPI #Python #Testing #Docker #Deployment #DevOps #BackendDevelopment #BuildInPublic #SoftwareEngineering

Building a Production-Ready Task Management API with FastAPI: Development & Implementation (Part 2)

Ravi Gupta — Mon, 02 Mar 2026 03:10:00 +0000

Building a Production-Ready Task Management API with FastAPI: Development & Implementation (Part 2)

Part 2 of 3: Architecture is comfortable. Implementation is humbling.

In Part 1, everything looked clean.

The layers were well separated.

The database schema made sense.

The architecture diagram looked impressive.

And then I started writing code.
That’s when things stopped being theoretical.
Not just what worked — but what broke, what surprised me, and that forced me to level up.
This article covers JWT authentication, CRUD operations, advanced features, and the real challenges of building production-ready APIs.

Quick Links:

📋 Table of Contents

Introduction & Recap
JWT Authentication Implementation
Repository Pattern in Practice
Service Layer Development
CRUD Operations
Advanced Features
Real Challenges & Solutions
Code Examples & Patterns
Lessons Learned
What's Next

Reading time: ~12 minutes

🎯 Introduction & Recap

In Part 1, I designed the architecture for a production-ready Task Management API using FastAPI and PostgreSQL.

What we covered:

Technology stack selection (FastAPI, PostgreSQL, SQLAlchemy 2.0)
Database schema design (Users, Tasks, Categories)
Clean architecture layers (API → Service → Repository → Database)
Project structure organization

Phase 1 Result: A solid foundation ready for implementation.
On paper, it was solid.

But architecture doesn’t test your assumptions, Implementation does.

Phase 2 Goal: Build the actual features while maintaining architectural integrity.
And discovering that “working” is not the same as “production-ready.”

This article covers the development journey - the features I built, patterns I implemented, and challenges I faced turning architecture diagrams into working code.

🔐 JWT Authentication Implementation

The Challenge

Coming from Spring Boot, I was used to this:

@Configuration
@EnableWebSecurity
public class SecurityConfig {
    // Spring Security handles everything
}

In FastAPI, there's no magic @EnableWebSecurity.

You build authentication from the ground up.

What I Built

Complete authentication system with:

User Registration
- Email validation
- Password strength requirements
- Duplicate email/username checking
- Automatic password hashing
Login Flow
- Credential verification
- Access token generation (15 min expiry)
- Refresh token generation (7 days expiry)
- Token storage strategy
Token Management
- Token validation middleware
- Token refresh endpoint
- Automatic token expiry handling
- Logout mechanism
Security Features
- Email verification flow
- Password reset mechanism
- Account activation
- Rate limiting on auth endpoints

Implementation Deep Dive

1. Password Hashing with bcrypt

# app/core/security.py
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    """Hash a password using bcrypt."""
    return pwd_context.hash(password)

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """Verify a password against a hashed password."""
    return pwd_context.verify(plain_password, hashed_password)

Why bcrypt?

Industry standard for password hashing
Built-in salt generation
Adaptive cost factor (future-proof against hardware improvements)
Slow by design (prevents brute force)

This was my first mindset shift:

FastAPI gives flexibility.
Security becomes your responsibility.

Spring Boot comparison: Similar to BCryptPasswordEncoder, but more explicit configuration.

2. JWT Token Generation

# app/core/security.py
from datetime import datetime, timedelta
from jose import jwt
from app.config import settings

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    """Create JWT access token."""
    to_encode = data.copy()

    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)

    to_encode.update({"exp": expire, "type": "access"})
    encoded_jwt = jwt.encode(
        to_encode, 
        settings.SECRET_KEY, 
        algorithm=settings.ALGORITHM
    )
    return encoded_jwt

def create_refresh_token(data: dict) -> str:
    """Create JWT refresh token."""
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=7)
    to_encode.update({"exp": expire, "type": "refresh"})

    encoded_jwt = jwt.encode(
        to_encode,
        settings.REFRESH_SECRET_KEY,
        algorithm=settings.ALGORITHM
    )
    return encoded_jwt

Design decisions here matter.

JWT is just a token format.

Security comes from validation.

Key decisions:

Two token types:
- Access tokens (short-lived, 15 min)
- Refresh tokens (long-lived, 7 days)
Separate secrets:
- Different keys for access vs refresh tokens
- Compromised access token ≠ compromised refresh token
Token payload:

   {
     "sub": "user_id",
     "exp": 1234567890,
     "type": "access"
   }

3. Token Validation Dependency

How it works:

Extract JWT from Authorization: Bearer <token> header
Decode and validate token signature
Check token type (must be "access")
Fetch user from database
Verify user is active
Return user object

Usage in routes:

@router.get("/me", response_model=UserResponse)
async def get_current_user_profile(
    current_user: User = Depends(get_current_user)
):
    """Get current user's profile."""
    return current_user

Clean. Type-safe. Reusable.

4. Login Endpoint Implementation

# app/api/v1/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from app.schemas.token import TokenResponse
from app.schemas.user import UserLogin

router = APIRouter(prefix="/auth", tags=["Authentication"])

@router.post("/login", response_model=TokenResponse)
async def login(
    credentials: UserLogin,
    db: AsyncSession = Depends(get_db)
):
    """
    Authenticate user and return access + refresh tokens.
    """
    # Get user by email
    user = await user_repository.get_by_email(db, credentials.email)

    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )

    # Verify password
    if not verify_password(credentials.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password"
        )

    # Check if user is active
    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="User account is inactive"
        )

    # Generate tokens
    access_token = create_access_token(data={"sub": str(user.id)})
    refresh_token = create_refresh_token(data={"sub": str(user.id)})

    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_token,
        token_type="bearer"
    )

Security considerations:

✅ Generic error messages (don't reveal if email exists)
✅ Password verification before any other checks
✅ Account status validation
✅ Proper HTTP status codes

5. Token Refresh Flow

@router.post("/refresh", response_model=TokenResponse)
async def refresh_access_token(
    refresh_token: str,
    db: AsyncSession = Depends(get_db)
):
    """
    Generate new access token using refresh token.
    """
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate refresh token"
    )

    try:
        # Decode refresh token
        payload = jwt.decode(
            refresh_token,
            settings.REFRESH_SECRET_KEY,
            algorithms=[settings.ALGORITHM]
        )

        user_id: str = payload.get("sub")
        token_type: str = payload.get("type")

        if user_id is None or token_type != "refresh":
            raise credentials_exception

    except JWTError:
        raise credentials_exception

    # Verify user still exists and is active
    user = await user_repository.get_by_id(db, UUID(user_id))

    if not user or not user.is_active:
        raise credentials_exception

    # Generate new access token
    new_access_token = create_access_token(data={"sub": user_id})

    return TokenResponse(
        access_token=new_access_token,
        refresh_token=refresh_token,  # Reuse refresh token
        token_type="bearer"
    )

Why this approach?

Access tokens expire quickly (security)
Refresh tokens last longer (user experience)
User doesn't need to re-login every 15 minutes
Compromised access token has limited damage window

Real Challenge: Token Storage Strategy

The Question: Where do clients store JWT tokens?

Options I considered:

Storage Method	Pros	Cons	My Choice
localStorage	Simple, persists across tabs	XSS vulnerable	✅ Used (with CSP headers)
httpOnly Cookies	XSS safe	Needs CSRF protection, CORS complexity	Future improvement
Memory (state)	Most secure	Lost on refresh, UX issues	Not practical

For this project:

Documented the trade-offs in code comments
Implemented localStorage for simplicity
Added Content Security Policy headers
Noted httpOnly cookies as production improvement

Lesson learned: Perfect security doesn't exist. Understand trade-offs, document decisions, implement appropriate for context.

The Bug That Taught Me the Most

My first refresh token implementation worked.

It decoded the token.
It generated a new access token.
It returned 200 OK.

It was also insecure.

I forgot to check:

If the user still existed
If the user was still active

Meaning a banned user could still refresh tokens.

The fix:

user = await user_repository.get_by_id(db, UUID(user_id))

if not user or not user.is_active:
    raise HTTPException(status_code=401, detail="Invalid user")

That bug changed how I think about "working code."

Working ≠ Secure
Passing tests ≠ Safe

📦 Repository Pattern in Practice

In Spring Boot, repositories are generated for you.

In FastAPI, you write them.

Explicitly.

I implemented a generic base repository:

class BaseRepository(Generic[ModelType]):

    async def get_by_id(self, db: AsyncSession, id: UUID):
        result = await db.execute(
            select(self.model).where(self.model.id == id)
        )
        return result.scalar_one_or_none()

Then extended it:

class TaskRepository(BaseRepository[Task]):

    async def get_by_user(self, db: AsyncSession, user_id: UUID):
        result = await db.execute(
            select(Task)
            .where(Task.user_id == user_id)
            .order_by(Task.created_at.desc())
        )
        return result.scalars().all()

Benefits:

Centralized query logic
Cleaner services
Easier unit testing
Reusable across features

At first, it felt verbose.

Later, it felt disciplined.

Real Challenge: Async Session Management

The Problem I Hit:

# WRONG - This caused session closure errors
async def get_tasks(db: AsyncSession, user_id: UUID):
    tasks = await task_repository.get_by_user(db, user_id)
    # Session closes here
    for task in tasks:
        print(task.category.name)  # ERROR: Session closed!

Why? SQLAlchemy's async sessions close after the query completes. Accessing related objects (like task.category) requires the session to still be open.

The Solution:

Eager loading with selectinload:

from sqlalchemy.orm import selectinload

async def get_by_user_with_category(
    self,
    db: AsyncSession,
    user_id: UUID
) -> List[Task]:
    """Get tasks with categories eagerly loaded."""
    result = await db.execute(
        select(Task)
        .options(selectinload(Task.category))
        .where(Task.user_id == user_id)
    )
    return result.scalars().all()

Access relationships within session scope:

async def get_task_with_details(
    db: AsyncSession,
    task_id: UUID
) -> Optional[dict]:
    """Get task with all related data."""
    task = await task_repository.get_by_id(db, task_id)

    if not task:
        return None

    # Access relationships while session is active
    return {
        "id": task.id,
        "title": task.title,
        "category": task.category.name if task.category else None,
        "user": task.owner.username
    }

Lesson learned: Async requires explicit relationship loading. Plan your queries based on what data you'll need.

⚙️ Service Layer Development

Purpose of Service Layer

Why not put business logic in routes?

# BAD - Business logic in routes
@router.post("/tasks")
async def create_task(task_in: TaskCreate, db: AsyncSession = Depends(get_db)):
    # Validation logic here
    # Database operations here
    # Error handling here
    # All mixed together!

Better - Service layer:

# GOOD - Clean separation
@router.post("/tasks", response_model=TaskResponse)
async def create_task(
    task_in: TaskCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    return await task_service.create_task(db, task_in, current_user.id)

Benefits:

Routes handle HTTP concerns only
Services contain business logic
Easier to test services independently
Reusable business logic

Business logic doesn't belong in routes.

So I moved rules into services:

async def create_task(
    self,
    db: AsyncSession,
    task_in: TaskCreate,
    user_id: UUID
) -> Task:

    if task_in.due_date and task_in.due_date < datetime.utcnow():
        raise HTTPException(
            status_code=400,
            detail="Due date must be in the future"
        )

    task_data = task_in.dict()
    task_data["user_id"] = user_id

    return await task_repository.create(db, task_data)

Now:

Routes handle HTTP
Services enforce rules
Repositories manage data

The architecture from Part 1 finally paid off here.

🔧 CRUD Operations

Routes → Services → Repositories pattern in action.

What makes this production-ready?

✅ Clear documentation (auto-generated in Swagger)
✅ Input validation (Pydantic schemas)
✅ Authorization (user can only access their tasks)
✅ Proper HTTP status codes
✅ Type hints (IDE autocomplete, catch errors)
✅ Pagination (prevent large response payloads)
✅ Filtering & search (flexible queries)

Example:

@router.get("/", response_model=List[TaskResponse])
async def get_tasks(
    status: Optional[TaskStatus] = Query(None),
    search: Optional[str] = Query(None),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    return await task_service.get_user_tasks(
        db,
        current_user.id,
        status=status,
        search=search
    )

Underneath that simple endpoint is:

Query composition
Security enforcement
Async session handling

"Simple" rarely stays simple at scale.

🚀 Advanced Features

1. Rate Limiting with SlowAPI

Why rate limiting?

Prevent brute force attacks on auth endpoints
Protect against API abuse
Ensure fair resource usage

Configuration by endpoint type:

Auth endpoints: 5 requests/minute
Regular endpoints: 100 requests/minute
Global limit: Configurable per environment

Real challenge: SlowAPI needed custom middleware configuration on Render. Spent time debugging connection pooling and Redis integration options.

2. Advanced Filtering & Pagination

Requirements:

Filter by status, priority, category
Search in title/description
Date range queries
Pagination with metadata

Benefits:

Flexible filtering combinations
Consistent pagination structure
Metadata for building UI pagination
Type-safe responses

3. Full-Text Search

Simple but effective search implementation:

async def search_tasks(
    self,
    db: AsyncSession,
    user_id: UUID,
    search_query: str,
    skip: int = 0,
    limit: int = 100
) -> List[Task]:
    """
    Search tasks by title or description.
    Case-insensitive partial match.
    """
    search_pattern = f"%{search_query}%"

    result = await db.execute(
        select(Task)
        .where(
            and_(
                Task.user_id == user_id,
                or_(
                    Task.title.ilike(search_pattern),
                    Task.description.ilike(search_pattern)
                )
            )
        )
        .offset(skip)
        .limit(limit)
    )

    return result.scalars().all()

For future improvement:

PostgreSQL full-text search with tsvector
Search ranking/relevance scoring
Elasticsearch integration for large datasets

4. Error Handling & Logging

Centralized exception handling:

# app/core/exceptions.py
from fastapi import HTTPException

class TaskNotFoundError(HTTPException):
    def __init__(self):
        super().__init__(
            status_code=404,
            detail="Task not found"
        )

class UnauthorizedTaskAccessError(HTTPException):
    def __init__(self):
        super().__init__(
            status_code=403,
            detail="Not authorized to access this task"
        )

# app/main.py
from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    """Global exception handler for unexpected errors."""
    logger.error(f"Unhandled exception: {exc}", exc_info=True)

    return JSONResponse(
        status_code=500,
        content={
            "detail": "Internal server error",
            "path": request.url.path
        }
    )

Request/Response logging:

# app/core/monitoring.py
import time
import logging
from fastapi import Request

logger = logging.getLogger(__name__)

@app.middleware("http")
async def log_requests(request: Request, call_next):
    """Log all HTTP requests with timing."""
    start_time = time.time()

    # Log request
    logger.info(f"{request.method} {request.url.path}")

    # Process request
    response = await call_next(request)

    # Calculate duration
    duration = time.time() - start_time

    # Log response
    logger.info(
        f"{request.method} {request.url.path} "
        f"- {response.status_code} - {duration:.3f}s"
    )

    return response

Benefits:

Consistent error responses
Request tracing for debugging
Performance monitoring
Production-ready logging

🚧 Real Challenges & Solutions

Challenge 1: Async Session Closure

The Problem:

# This caused "Session is already closed" errors
async def get_task_with_category(db: AsyncSession, task_id: UUID):
    task = await task_repository.get_by_id(db, task_id)
    # Session closes after repository call
    category_name = task.category.name  # ERROR!

Why it happened:

SQLAlchemy async sessions have shorter lifecycle
Relationships not loaded by default
Accessing unloaded relationships after session closes → error

The Solution:

Eager loading:

from sqlalchemy.orm import selectinload

async def get_by_id_with_relations(
    self, 
    db: AsyncSession, 
    task_id: UUID
) -> Optional[Task]:
    result = await db.execute(
        select(Task)
        .options(
            selectinload(Task.category),
            selectinload(Task.owner)
        )
        .where(Task.id == task_id)
    )
    return result.scalar_one_or_none()

Access within session scope:

async def get_task_details(db: AsyncSession, task_id: UUID):
    task = await task_repository.get_by_id(db, task_id)

    if not task:
        return None

    # Access all relationships while session is active
    return {
        "id": task.id,
        "title": task.title,
        "category": task.category.name if task.category else None
    }

Time lost: ~3 hours debugging various "session closed" errors

Lesson learned: With async SQLAlchemy, plan your data access patterns. Decide upfront what relationships you'll need.

Challenge 2: Repository Pattern Adaptation

The Struggle:

Coming from Spring Boot's JpaRepository, I initially wrote this:

# WRONG - Trying to replicate Spring Boot directly
class TaskRepository:
    def find_by_user_id(self, user_id: UUID):  # Sync!
        # How do I make this async?

Spring Boot's repository interfaces work because of proxy magic at runtime.

FastAPI doesn't have that.

The Solution:

Accept that Python is explicit, not magic:

# RIGHT - Explicit async
class TaskRepository:
    async def get_by_user(
        self, 
        db: AsyncSession,  # Explicit session
        user_id: UUID
    ) -> List[Task]:
        result = await db.execute(
            select(Task).where(Task.user_id == user_id)
        )
        return result.scalars().all()

Lesson learned: Don't fight the language. FastAPI's explicitness is a feature, not a bug. It makes dependencies clear and testable.

Challenge 3: Token Refresh Logic

The Problem:

First implementation of token refresh had a security flaw:

What's wrong?

Doesn't check if user still exists
Doesn't check if user is still active
Could issue tokens for deleted/banned users

Time to discover: 2 days (caught during security review)

Lesson learned: JWT tokens are stateless, but security checks shouldn't be. Always validate against source of truth (database).

Challenge 4: Rate Limiting on Render

The Problem:

SlowAPI worked perfectly locally, but failed on Render:

SlowAPI: Failed to connect to Redis
Rate limiting disabled

Why?

Render's free tier doesn't include Redis.
SlowAPI uses Redis for distributed rate limiting.

The Solution:

Switch to in-memory rate limiting for free tier:

# app/core/rate_limiter.py
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.middleware import SlowAPIMiddleware

# Use in-memory storage (single instance)
limiter = Limiter(
    key_func=get_remote_address,
    storage_uri="memory://",  # In-memory instead of Redis
    enabled=True
)

# In main.py
app.state.limiter = limiter
app.add_middleware(SlowAPIMiddleware)

Trade-offs:

✅ Works on free tier
✅ No external dependencies
❌ Not distributed (single instance only)
❌ Resets on deployment

For production: Upgrade to Redis for distributed rate limiting across multiple instances.

Lesson learned: Free tiers have constraints. Design for your deployment environment, not just local development.

Challenge 5: Alembic Migration Conflicts

The Problem:

After developing features in parallel (tasks + categories), Alembic migrations conflicted:

alembic revision --autogenerate
Error: Multiple head revisions present

Why?

Created migrations from different branches without syncing.

The Solution:

# Merge heads
alembic merge heads -m "merge task and category features"

# Apply merged migration
alembic upgrade head

Prevention strategy:

Always pull latest migrations before creating new ones
Run alembic upgrade head before creating new migration
Use linear migration history (avoid parallel migrations)

Time lost: 1 hour debugging + fixing migration history

Lesson learned: Database migrations are sequential, not parallel. Coordinate with team (or yourself across branches).

💡 Lessons Learned

1. Async Requires Discipline

Spring Boot (blocking): Forgiving. Mix sync/async, it'll work (maybe slower).

FastAPI (async): Strict. One blocking call blocks the entire event loop.

Key rules:

Always await database calls
Use AsyncSession, not Session
Don't mix sync and async carelessly
Plan async from the start

2. Explicit is Better Than Magic

Spring Boot: Annotations do magic (@Autowired, @Transactional)

FastAPI: Dependency injection is explicit:

async def get_tasks(
    db: AsyncSession = Depends(get_db),
    current_user: User = Depends(get_current_user)
):
    # Dependencies explicit in function signature

Initially: Felt verbose
After building: Appreciate the clarity

Benefit: Testability. Mock dependencies easily.

3. Repository Pattern Pays Off

Centralizing database logic in repositories:

✅ Made service layer cleaner
✅ Enabled easy testing (mock repositories)
✅ Reused queries across services
✅ Single source of truth for data access

Worth the upfront effort.

4. Security is a Process

Built authentication in iterations:

Basic password check
Added JWT tokens
Added token refresh
Added rate limiting
Added email verification
Added password reset

Each iteration revealed new attack vectors to defend against.

Security isn't a feature you add once. It's continuous hardening.

5. Documentation Writes Itself

FastAPI's auto-generated docs are incredible:

@router.post("/tasks", response_model=TaskResponse)
async def create_task(
    task_in: TaskCreate,
    current_user: User = Depends(get_current_user)
):
    """
    Create a new task.

    - **title**: Required, 1-200 characters
    - **priority**: LOW, MEDIUM, HIGH, URGENT
    """
    pass

Result: Interactive Swagger UI with:

All endpoints documented
Request/response schemas
Try-it-out functionality
Type information

No separate documentation tool needed.

🔮 What's Next: Phase 3

Coming in Part 3:

Testing

pytest setup with async support
Unit tests for services
Integration tests for API endpoints
Test coverage > 85%
Testing async code patterns

Deployment

Docker multi-stage builds
Render.com deployment
Neon PostgreSQL setup
Environment variable management
Health checks & monitoring

Production Features

Performance metrics
Uptime tracking
Cost analysis ($0 free tier!)
CI/CD pipeline
Production best practices

Follow me to get notified when Part 3 drops!

📝 Complete Code & Resources

GitHub Repository:

🔗 https://github.com/ravigupta97/task_management_api
Fully documented code
Clean commit history
README with setup instructions

Live API Documentation:

🚀 https://task-management-api-a775.onrender.com/docs
Interactive Swagger UI
Try all endpoints
See request/response schemas

Key Files to Explore:

app/core/security.py - JWT implementation
app/repositories/ - Repository pattern
app/services/ - Business logic layer
app/api/v1/ - Route handlers

🙏 Acknowledgments

Tools & Resources:

FastAPI documentation (exceptional quality)
SQLAlchemy 2.0 docs (comprehensive)
AI assistants (concept clarification)
Open-source community

What helped most:

Reading production FastAPI codebases on GitHub
AI tools for understanding complex concepts
Trial and error (every bug taught something)

💬 Discussion

Questions for you:

How do you handle async session management?
What's your approach to API authentication?
Repository pattern: overkill or essential?
Biggest challenge you've faced with FastAPI?

Let's learn together - comment below! 👇

This is Part 2 of 3 in my FastAPI journey.

Connect with me:

LinkedIn: [https://www.linkedin.com/in/ravigupta97/]
GitHub: [https://github.com/ravigupta97]

#FastAPI #Python #Backend #JWT #Authentication #API #CleanArchitecture #WebDevelopment #SoftwareEngineering

Building a Production-Ready Task Management API with FastAPI: Complete Architecture Guide

Ravi Gupta — Mon, 23 Feb 2026 03:31:21 +0000

Building a Production-Ready Task Management API with FastAPI: Complete Architecture Guide

Context: After 3 years building backend services with Spring Boot, I decided to explore Python's modern web framework ecosystem. This is Part 1 of a 3-part series documenting my journey building a production-grade REST API with FastAPI.

Three weeks ago, I started a new learning project: build a production-ready Task Management API using FastAPI.

Not a tutorial-level "TODO app in 100 lines."

A real, production-grade application with authentication, proper architecture, database migrations, and deployment-ready code.

The goal: Learn modern Python web development while applying enterprise patterns I knew from Spring Boot.

The result: A fully functional API with 50+ endpoints, JWT authentication, and clean architecture - deployed on free tier infrastructure.

This article covers Phase 1: Architecture & Design - the foundation that made everything else possible.

📚 What You'll Learn

Why FastAPI over Flask/Django (from a Spring Boot developer's perspective)
Designing a clean, scalable project architecture
Database schema design with PostgreSQL
Key technology decisions and trade-offs
Mistakes I made and how I fixed them
Resources that accelerated my learning

Reading time: ~12 minutes

GitHub Repository: GitHub Repo

Live API Docs: API

🎯 The Challenge: From Tutorials to Production

Most FastAPI tutorials teach you this:

# main.py - Everything in one file
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

150 lines. No auth. SQLite. Single file.

But production applications need:

✅ 50+ endpoints organized logically
✅ User authentication & authorization
✅ Database migrations
✅ Proper error handling
✅ Rate limiting
✅ Clean, testable architecture
✅ Docker deployment
✅ API documentation

The gap between tutorials and production is massive.

This article bridges that gap.

🚀 My Background & Motivation

Experience: Around 3 years backend development with Spring Boot (Java)

Why FastAPI?

Coming from Spring Boot, I was curious about Python's modern web frameworks:

Django: Too heavy for an API-first project
Flask: Popular but felt dated (no native async)
FastAPI: Modern, async-first, great docs

The learning goal: Not just "learn FastAPI" but understand:

How to structure large Python projects
Modern async patterns in Python
Production-ready API design

Coming from Spring Boot, I wanted to see:

How Python's simplicity compares to Java's verbosity
Whether FastAPI's async is easier than Spring WebFlux
If I could apply enterprise patterns (service layer, repositories) in Python

Note: This isn't about "which is better" - both Spring Boot and FastAPI are excellent. This was about expanding my toolkit.

🔍 Phase 1 Research: FastAPI vs Flask vs Django

Time spent: A few days researching frameworks, reading docs, watching talks

Decision Matrix

I created this framework for choosing technologies:

Criteria	FastAPI	Flask	Django
Async Support	✅ Native	⚠️ Via extensions	⚠️ Limited
Performance	✅ Very fast	🟡 Moderate	🟡 Moderate
Auto Docs	✅ Swagger/ReDoc	❌ Manual	❌ DRF only
Data Validation	✅ Pydantic built-in	🟡 Libraries needed	🟡 DRF serializers
Learning Curve	🟡 Moderate	✅ Easy	🔴 Steep
Type Hints	✅ Required	🟡 Optional	🟡 Optional
API-First	✅ Yes	🟡 Flexible	❌ Full-stack focus

Why FastAPI Won

1. Native Async/Await

# FastAPI - async is first-class
@app.get("/tasks")
async def get_tasks(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Task))
    return result.scalars().all()

Clean, intuitive async. No additional configuration needed.

2. Auto-Generated API Documentation

FastAPI generates interactive Swagger UI automatically:

/docs - Swagger UI
/redoc - ReDoc
/openapi.json - OpenAPI schema

Zero configuration. Just write your code.

3. Built-in Data Validation with Pydantic

class TaskCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    priority: str = Field(pattern="^(LOW|MEDIUM|HIGH)$")

# Validation happens automatically
@app.post("/tasks")
async def create_task(task: TaskCreate):
    # task is validated, type-safe
    return task

Reminded me of Bean Validation in Spring Boot, but more Pythonic.

Final Decision

✅ FastAPI for this project because:

API-first design (perfect for backend learning)
Modern Python features (async, type hints)
Great documentation
Growing ecosystem and job market demand

🏗️ Designing Clean Architecture

Challenge: How do you structure a FastAPI project for scale?

Coming from Spring Boot

In Spring Boot, I was used to:

@RestController
public class TaskController {
    @Autowired
    private TaskService taskService;

    @GetMapping("/tasks")
    public List<Task> getTasks() {
        return taskService.findAll();
    }
}

Clear separation: Controller → Service → Repository → Entity

Could I apply the same pattern in FastAPI?

The Architecture I Designed

After researching production FastAPI projects, I landed on this structure:

task_management_api/
├── app/
│   ├── models/          # SQLAlchemy models (like @Entity)
│   ├── schemas/         # Pydantic schemas (DTOs)
│   ├── repositories/    # Data access layer
│   ├── services/        # Business logic
│   ├── api/             # Route handlers
│   │   └── v1/          # API versioning
│   ├── core/            # Security, config
│   └── main.py          # App initialization
├── tests/
├── alembic/             # Database migrations
└── requirements.txt

Why this structure?

Separation of Concerns
- Routes handle HTTP
- Services contain business logic
- Repositories handle database
Testability
- Each layer can be tested independently
- Mock repositories in service tests
- Mock services in route tests
Scalability
- Easy to add new features
- Clear place for everything
- Team can work on different layers

Layer-by-Layer Breakdown

📁 Models (app/models/)

# SQLAlchemy ORM models (database tables)
class Task(Base):
    __tablename__ = "tasks"
    id = Column(UUID, primary_key=True)
    title = Column(String(200), nullable=False)
    user_id = Column(UUID, ForeignKey("users.id"))

Like Spring's @Entity classes.

📁 Schemas (app/schemas/)

# Pydantic models (request/response validation)
class TaskCreate(BaseModel):
    title: str
    priority: TaskPriority

class TaskResponse(BaseModel):
    id: UUID
    title: str
    created_at: datetime

Like Spring's DTOs, but with automatic validation.

📁 Repositories (app/repositories/)

# Data access layer
class TaskRepository:
    async def get_by_id(self, db: AsyncSession, task_id: UUID):
        result = await db.execute(
            select(Task).where(Task.id == task_id)
        )
        return result.scalar_one_or_none()

Like Spring's @Repository classes.

📁 Services (app/services/)

# Business logic
class TaskService:
    async def create_task(self, db: AsyncSession, task_data: TaskCreate):
        # Validation logic
        # Business rules
        task = await task_repository.create(db, task_data)
        return task

Like Spring's @Service classes.

📁 API Routes (app/api/v1/)

# HTTP handlers
@router.post("/tasks", response_model=TaskResponse)
async def create_task(
    task_in: TaskCreate,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    return await task_service.create_task(db, task_in)

Like Spring's @RestController classes.

Architecture Diagram

Detailed Architecture Diagram:

┌─────────────────────────────────────────────────────┐
│                  Client Request                     │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│              API Layer (Routes)                     │
│  • Route handlers                                   │
│  • Request validation (Pydantic)                    │
│  • Response serialization                           │
│  • Dependency injection                             │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│           Service Layer (Business Logic)            │
│  • Business rules                                   │
│  • Data transformation                              │
│  • Error handling                                   │
│  • Transaction management                           │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│         Repository Layer (Data Access)              │
│  • Database queries                                 │
│  • ORM operations                                   │
│  • Data mapping                                     │
└──────────────────────┬──────────────────────────────┘
                       │
┌──────────────────────▼──────────────────────────────┐
│              Database (PostgreSQL)                  │
│  • Tables: Users, Tasks, Categories                 │
│  • Relationships & constraints                      │
└─────────────────────────────────────────────────────┘

Benefits of this architecture:
✅ Clear separation of concerns
✅ Easy to test each layer
✅ Scales with team size
✅ Familiar to developers from other frameworks

🗄️ Database Schema Design

Requirements Analysis

Before designing the schema, I listed what the API needed:

User Management:

Users can register and login
Email verification
Password reset functionality

Task Management:

Users create tasks
Tasks have status (TODO, IN_PROGRESS, COMPLETED)
Tasks have priority levels
Optional due dates
Tasks belong to users

Organization:

Users can create categories
Tasks can be assigned to categories
Categories have colors for UI

Entity-Relationship Design

After a few iterations, here's the final schema:

Entity-Relationship Diagram:

┌──────────────────────────────────────┐
│             USERS                    │
├──────────────────────────────────────┤
│ 🔑 id (UUID, PK)                     │
│ 📧 email (UNIQUE, NOT NULL)          │
│ 👤 username (UNIQUE, NOT NULL)       │
│ 🔒 hashed_password (NOT NULL)        │
│ ✅ is_active (BOOLEAN,DEFAULT TRUE)  │
│ ✉️ is_verified(BOOLEAN,DEFAULT FALSE)│
│ 📝 full_name (TEXT, NULLABLE)        │
│ 📅 created_at (TIMESTAMP)            │
│ 📅 updated_at (TIMESTAMP)            │
└──────────────┬───────────────────────┘
               │ 1
               │
               │ N
┌──────────────▼───────────────────────┐
│             TASKS                    │
├──────────────────────────────────────┤
│ 🔑 id (UUID, PK)                    │
│ 📝 title (VARCHAR(200), NOT NULL)   │
│ 📄 description (TEXT, NULLABLE)     │
│ 📊 status (ENUM, NOT NULL)          │
│    └─ TODO, IN_PROGRESS, COMPLETED  │
│ ⭐ priority (ENUM, NOT NULL)        │
│    └─ LOW, MEDIUM, HIGH, URGENT     │
│ 📆 due_date (TIMESTAMP, NULLABLE)   │
│ 🔗 user_id (UUID, FK → users.id)    │
│ 🏷️ category_id (UUID, FK→categories)│
│ 📅 created_at (TIMESTAMP)           │
│ 📅 updated_at (TIMESTAMP)           │
└──────────────┬───────────────────────┘
               │ N
               │
               │ 1
┌──────────────▼───────────────────────┐
│           CATEGORIES                 │
├──────────────────────────────────────┤
│ 🔑 id (UUID, PK)                    │
│ 🏷️ name (VARCHAR(50), NOT NULL)     │
│ 🎨 color (VARCHAR(7),DEFAULT '#...')│
│ 🔗 user_id (UUID, FK → users.id)    │
│ 📅 created_at (TIMESTAMP)           │
│ 📅 updated_at (TIMESTAMP)           │
│                                      │
│ UNIQUE CONSTRAINT (user_id, name)    │
└───────────────────────────────────────┘

Key Design Decisions

1. UUID Primary Keys

id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)

Why UUID over auto-increment integers?
✅ Better security (can't enumerate resources)
✅ Globally unique (helpful for distributed systems)
✅ Can generate client-side
❌ Slightly larger storage footprint

Trade-off accepted: Security and flexibility > minor storage increase

2. Enum Types for Status & Priority

class TaskStatus(str, Enum):
    TODO = "TODO"
    IN_PROGRESS = "IN_PROGRESS"
    COMPLETED = "COMPLETED"
    ARCHIVED = "ARCHIVED"

Benefits:

Database-level constraint validation
Type safety in Python code
Clear API documentation

3. Soft Timestamps

Every table has:

created_at - Automatically set on insert
updated_at - Automatically updated on modification

Implementation:

created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

4. Cascade Behaviors

# If user is deleted, delete their tasks
tasks = relationship("Task", back_populates="owner", cascade="all, delete-orphan")

# If category is deleted, set task category_id to NULL
category_id = Column(UUID, ForeignKey("categories.id", ondelete="SET NULL"))

Why? Data integrity while preserving user intent.

SQLAlchemy Model Example

from sqlalchemy import Column, String, Boolean, ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import relationship

class User(Base):
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email = Column(String(255), unique=True, index=True, nullable=False)
    username = Column(String(50), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    is_verified = Column(Boolean, default=False, nullable=False)
    full_name = Column(String(100), nullable=True)

    # Timestamps
    created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    # Relationships
    tasks = relationship("Task", back_populates="owner", cascade="all, delete-orphan")
    categories = relationship("Category", back_populates="owner", cascade="all, delete-orphan")

Clean, readable, type-safe.

🛠️ Complete Tech Stack & Justification

Final Stack

Component	Technology	Why?
Framework	FastAPI	Async-first, auto docs, Pydantic
Language	Python 3.11	Latest stable, performance improvements
Database	PostgreSQL 16	Advanced features, JSON support
ORM	SQLAlchemy 2.0	Async support, mature ecosystem
Migrations	Alembic	Industry standard for SQLAlchemy
Auth	JWT (python-jose)	Stateless, scalable
Validation	Pydantic v2	Built into FastAPI
Container	Docker	Consistent environments
Server	Uvicorn	ASGI server optimized for FastAPI
DB Hosting	Neon (Serverless PostgreSQL)	Free tier, excellent for learning
App Hosting	Render	Free tier, easy deployment

Decision Process for Each

PostgreSQL vs MySQL vs MongoDB

Criteria	PostgreSQL	MySQL	MongoDB
JSON Support	✅ Excellent (JSONB)	🟡 Basic	✅ Native
Transactions	✅ Full ACID	✅ Full ACID	🟡 Limited
Advanced Features	✅ Rich	🟡 Good	❌ Different paradigm
Free Hosting	✅ Neon, Supabase	✅ PlanetScale	✅ Atlas

Winner: PostgreSQL

Better for structured data (tasks, users)
JSONB for future flexibility
Excellent free tier options

SQLAlchemy vs Raw SQL vs other ORMs

SQLAlchemy 2.0 won because:

Native async support
Type hints and IDE autocomplete
Migration management (Alembic)
Mature, battle-tested

The learning curve was worth it for long-term maintainability.

Docker: Day 1 Decision

Why containerize from the start?
✅ Consistent dev/prod environments
✅ Easy onboarding for collaborators
✅ Deployment simplicity
✅ Learning industry standard practice

No regrets on this decision.

Decision Matrix Template

For each technology choice, I asked:

Does it solve my problem? (must-have features)
Is it actively maintained? (community, updates)
Can I learn it reasonably? (documentation, resources)
Does it scale? (production-ready)
Is it resume-worthy? (job market demand)

This framework helped avoid analysis paralysis.

📚 Resources That Helped Me

Official Documentation

1. FastAPI Docs

Link: https://fastapi.tiangolo.com/
⭐⭐⭐⭐⭐ (Exceptional quality)
Start here. Seriously. Best framework docs I've seen.

2. SQLAlchemy 2.0 Docs

Link: https://docs.sqlalchemy.org/
⭐⭐⭐⭐ (Comprehensive but dense)
The async section took multiple readings to understand

3. Pydantic Docs

Link: https://docs.pydantic.dev/
⭐⭐⭐⭐⭐ (Clear examples)
Great migration guide from v1 to v2

Video Courses / Tutorials

1. FastAPI - The Complete Course (YouTube)

Covers basics to advanced patterns
Helped visualize concepts

2. SQLAlchemy 2.0 Tutorial (Official)

Essential for understanding async patterns

Tools I Used Daily

Development:

VS Code + Python extension
Postman for API testing
pgAdmin for database management

Design:

dbdiagram.io - Database schema design
Excalidraw - Architecture diagrams
Markdown editors - Documentation

Learning:

ChatGPT / Claude - Explaining complex concepts
- "Explain SQLAlchemy async sessions like I'm coming from Spring Boot"
- "What's the Python equivalent of @Autowired dependency injection?"
GitHub - Studying production FastAPI projects
- fastapi/full-stack-fastapi-postgresql (official template)
- tiangolo's examples

Learning Strategy

What worked:
✅ Read docs first, code second
✅ Study production codebases
✅ Use AI for concept clarification (not copy-paste)
✅ Build incrementally, test often

What didn't work:
❌ Jumping straight to coding without planning
❌ Following outdated tutorials (SQLAlchemy 1.x vs 2.0)
❌ Trying to learn everything at once

🚧 Real Challenges I Faced

Challenge 1: Environment Setup

The Problem:

Setting up Python, PostgreSQL, virtual environments, and managing dependencies was more complex than expected.

What went wrong:

# This shouldn't have taken 2 hours, but it did
pip install asyncpg
# Error: pg_config not found

The Solution:

Documented every step for reproducibility:

# 1. Python 3.11 installation
# 2. Virtual environment
python -m venv venv
source venv/bin/activate  # or .\venv\Scripts\activate on Windows

# 3. PostgreSQL installation
brew install postgresql@16  # macOS
# Or use Neon for cloud PostgreSQL (easier)

# 4. Dependencies
pip install -r requirements.txt

Lesson learned: Good documentation saves hours of debugging later.

Challenge 2: Async/Await Debugging

The Problem:

Spent 4 hours debugging this error:

RuntimeError: Task <Task pending> attached to a different loop

What went wrong:

Mixed synchronous and asynchronous session calls:

# WRONG - mixing sync and async
def create_task(db: Session, task_data):  # Sync session
    result = await db.execute(...)  # Async call

The Solution:

Consistency is key:

# CORRECT - all async
async def create_task(db: AsyncSession, task_data):
    result = await db.execute(select(Task))
    await db.commit()

How AI helped:
I asked Claude: "Why am I getting 'attached to a different loop' error in SQLAlchemy async?"

Got a clear explanation and code examples that saved me hours.

Lesson learned: Async is all-or-nothing. Mix carefully.

Challenge 3: SQLAlchemy 2.0 Documentation Overwhelm

The Problem:

SQLAlchemy docs are comprehensive but overwhelming for beginners.

The async section felt like reading a PhD thesis.

What helped:

Used AI as a tutor
- "Explain SQLAlchemy AsyncSession vs Session"
- "Show me a complete example of async CRUD operations"
Studied working examples
- Found production codebases on GitHub
- Copied patterns, then understood them
Started simple

   # Simple async query - master this first
   async def get_user(db: AsyncSession, user_id: UUID):
       result = await db.execute(
           select(User).where(User.id == user_id)
       )
       return result.scalar_one_or_none()

Then built complexity gradually.

Lesson learned: Don't try to understand everything. Master the basics, build on them.

Challenge 4: JWT Token Security

The Problem:

Confused about where to store JWT tokens:

localStorage? (XSS vulnerable)
Cookies? (CSRF vulnerable)
Memory only? (lost on refresh)

The Solution:

Researched trade-offs:

Storage	Pros	Cons
localStorage	Simple, survives refresh	XSS vulnerable
httpOnly Cookie	XSS safe	Needs CSRF protection
Memory only	Most secure	Lost on refresh

My choice: httpOnly cookies + CSRF tokens

But for this learning project: Documented the trade-offs, implemented localStorage (simpler frontend), added strong CSP headers.

Lesson learned: Perfect security doesn't exist. Understand trade-offs, document decisions.

What I'd Do Differently

If I started over:

✅ Plan architecture BEFORE writing code
✅ Set up Docker from Day 1 (not Day 5)
✅ Write tests alongside features (not after)
✅ Study one production codebase deeply
✅ Ask for help sooner (AI, communities, docs)

Real talk: Every "mistake" taught me something. The frustration was part of learning.

✅ Phase 1 Results

After this planning phase, I had:

Documentation:
✅ Database schema designed and documented
✅ Architecture patterns decided
✅ Technology stack finalized
✅ Project structure defined

Technical Foundation:
✅ Docker environment configured
✅ PostgreSQL database set up
✅ SQLAlchemy models defined
✅ Alembic migrations initialized
✅ Project structure created

Learning Outcomes:
✅ Deep understanding of FastAPI vs other frameworks
✅ SQLAlchemy 2.0 async patterns
✅ Clean architecture principles in Python
✅ Production-ready project organization

Time invested: A few days of focused learning and planning

Was it worth it?

Absolutely. This foundation made Phase 2 (development) much smoother.

The hours spent planning saved days of refactoring later.

🚀 What's Next: Phase 2

Coming in Part 2:

⏳ JWT Authentication Implementation

Access + refresh token flow
Password hashing with bcrypt
Token validation middleware

⏳ Repository Pattern in Practice

Generic CRUD operations
User-specific queries
Transaction management

⏳ Service Layer Development

Business logic separation
Error handling strategies
Dependency injection

⏳ Real Development Challenges

Bugs I encountered
Performance optimizations
Testing strategies

Follow me to get notified when Part 2 drops!

📝 Complete Code & Resources

GitHub Repository:

🔗 https://github.com/ravigupta97/task_management_api
Fully documented code
Clean commit history
Setup instructions

Additional Resources:

Architecture diagrams (in this article)
Database schema (ER diagram above)
Technology decision matrix
Learning resources list

Have questions? Drop them in the comments below!

🙏 Acknowledgments

Resources that helped:

FastAPI documentation (exceptional quality)
SQLAlchemy docs (comprehensive)
AI tools (ChatGPT, Claude) for concept clarification
FastAPI community on Discord

Special thanks to:

Anyone reading this and learning alongside me
The open-source community making these tools free

💬 Discussion

Questions for you:

Are you considering FastAPI for your next project?
What's your biggest challenge with async Python?
Any architectural patterns you'd recommend?

Let's learn together - comment below! 👇

This is Part 1 of 3 in my FastAPI journey. Follow for:

Part 2: Development & Implementation
Part 3: Testing, Deployment & Results

Find me on:

LinkedIn: [https://www.linkedin.com/in/ravigupta97/]
GitHub: [https://github.com/ravigupta97]

#FastAPI #Python #Backend #WebDevelopment #Learning #API #SoftwareEngineering