Python Lesson 59 of 71

Web Part 2 — Databases, Authentication & Taking an API to Production

This is Part 2 of two. In Part 1 — Flask & FastAPI REST APIs you built a real CRUD API in FastAPI: pydantic models that validate the body and return a 422 naming the bad field, a response_model that strips secrets before they ship, dependency injection with Depends, proper status codes, and free interactive docs. It ended on a deliberate cliff. The data lived in an in-memory dict, and that dict “won’t survive contact with production: four workers means four separate dicts.” Part 1 named exactly three seams it left open — databases (SQLAlchemy sessions injected through the same Depends), authentication (OAuth2, JWT, hashing passwords), and production deployment. This lesson fills all three, in that order.

Everything below targets Python 3.12+ and every output block is real — copied from a 3.12.3 run against a live server (a background uvicorn process hit with httpx, and an in-process TestClient for the offline tests), not paraphrased. The versions in play: FastAPI 0.139, Starlette 1.3, pydantic 2.13, SQLAlchemy 2.0, uvicorn 0.51, httpx 0.28, passlib 1.7.4, bcrypt 4.0.1, python-jose 3.5, python-multipart 0.0.32. One install, in a virtual environment:

python3 -m venv .venv
source .venv/bin/activate                 # Windows: .venv\Scripts\activate
python -m pip install fastapi uvicorn pydantic sqlalchemy \
    "passlib[bcrypt]" "python-jose[cryptography]" httpx python-multipart

Two of those aren’t obvious. python-multipart is required because the OAuth2 login form sends application/x-www-form-urlencoded data, and FastAPI won’t parse a form without it (you get a clear Form data requires "python-multipart" error at startup if it’s missing). And passlib[bcrypt] pulls in the bcrypt library — but ⚠️ the latest bcrypt (5.x) breaks the aging passlib 1.7.4, so this lesson pins bcrypt==4.0.1; the exact failure and fix are in the troubleshooting section, because you will hit it.


Why this matters

Part 1’s hard truth was that the request is untrusted input from a stranger. Part 2’s hard truth is worse: now there’s a database behind that input holding real users’ data, and an authentication layer deciding who gets to touch it. This is where security stops being a nice-to-have. An API that leaks a password hash, stores passwords in plaintext, or accepts a forged token isn’t “a bug to fix later” — it’s the breach that ends up in the news, and the post-mortem always reads the same way: someone took a shortcut on one of the things this lesson is about.

Three ideas carry the whole lesson, and each is a place beginners reliably get burned:

Get those three straight and the rest — OAuth2 wiring, CORS, HTTPS, Docker — is machinery. Get them wrong and no amount of machinery saves you.


Connecting a real database: SQLAlchemy behind FastAPI

You met SQLAlchemy in the databases lesson — the engine, the Session, the ORM that maps rows to objects, and the reason parameterised queries are immune to SQL injection. Here we wire it into FastAPI. The single most important design decision is how the handler gets a database session, and FastAPI’s answer is the dependency injection you already know: a session is just another Depends.

The pattern is a dependency that yields:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from typing import Annotated
from fastapi import Depends

engine = create_engine("sqlite:///./web2.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

def get_db():
    db = SessionLocal()        # open one session for THIS request
    try:
        yield db               # hand it to the handler
    finally:
        db.close()             # ALWAYS return the connection to the pool

Db = Annotated[Session, Depends(get_db)]     # name it once, reuse everywhere

A dependency with yield is a generator dependency: FastAPI runs the code up to the yield, injects the yielded value into your handler, and — crucially — runs the code after the yield once the response is finished. So the session is created before your handler, handed in, and closed after, no matter what happens (even if the handler raises). This is the same “setup then teardown” shape as a context manager, expressed as a dependency. The connect_args={"check_same_thread": False} is a SQLite-only detail: FastAPI may run def handlers in a threadpool (Part 1’s rule), and SQLite otherwise refuses to be touched from a thread other than the one that opened it. Postgres and MySQL don’t need it.

Why not a global session? It’s the intuitive first instinct — db = SessionLocal() at module level, import it everywhere — and it’s wrong on two counts, both of which produce baffling bugs:

Approach What happens Verdict
Global Session shared by all requests One transaction shared across concurrent requests; one request’s uncommitted write is visible to another; rollback in one wipes another’s work; not thread-safe ❌ Data corruption under load
Session opened per request, never closed Each request leaks a connection; the pool fills; new requests block then raise TimeoutError: QueuePool limit ... reached ❌ Falls over under load
Depends(get_db) yielding a session Each request gets its own session and transaction; always closed in finally; connection returned to the pool ✅ Correct

The middle row is the classic connection leak. A Session holds a connection from the pool for its whole life; if you never close(), that connection never goes back. SQLAlchemy’s default QueuePool (covered in the databases lesson) holds 5 connections and lends 10 more under burst — leak them and the 16th concurrent request waits, then dies. The try/finally in get_db is the whole defence, and putting it in a dependency means you write it once and every route inherits it.

Here is how the session then flows through a request, and it maps one-for-one onto the Depends diagram from Part 1 — only now the “DB session” node is a real SQLAlchemy Session instead of a stub:

Stage What FastAPI does Your code
Request arrives Sees db: Db in the handler signature
Before handler Runs get_db up to yield, opens a Session db = SessionLocal()
Handler runs Injects the session as the db argument db.scalar(select(User)...)
After response Resumes get_db past yield db.close()
On any exception Still runs the finally connection returned to pool

For a synchronous SQLite/Postgres driver like this, your handlers should be plain def (FastAPI threadpools them — Part 1’s rule), because a synchronous DB call inside async def is exactly the blocking-the-event-loop trap Part 1 measured at 5x slower. If you want async def handlers you need a genuinely async driver (asyncpg via SQLAlchemy’s async engine, aiosqlite) and AsyncSession — a real option, but not required, and easy to get subtly wrong. When unsure: def.


Models vs schemas: the ORM row is not the API

This is the idea that ties the database to Part 1’s response_model, and it’s worth stating flatly: the SQLAlchemy model and the pydantic schema are two different classes on purpose. The ORM model describes how a row is stored; the pydantic schema describes what the API accepts and returns. They overlap, but they are never the same, and the gap between them is where security lives.

from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pydantic import BaseModel, EmailStr, Field

class Base(DeclarativeBase):
    pass

class User(Base):                              # ORM MODEL — the storage shape
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String)     # a secret!
    role: Mapped[str] = mapped_column(String, default="user")

class UserCreate(BaseModel):                   # API SCHEMA — what comes IN
    email: EmailStr
    password: str = Field(min_length=8, max_length=72)       # plaintext, briefly

class UserPublic(BaseModel):                   # API SCHEMA — what goes OUT
    id: int
    email: EmailStr
    role: str
    model_config = {"from_attributes": True}   # read fields off an ORM object

Look at what each class does not have. User has hashed_password; neither UserCreate nor UserPublic does — the hash is a storage concern the client never sees. UserCreate has password (plaintext, for exactly as long as it takes to hash it); User never does. UserPublic has id and role; UserCreate doesn’t, because the client doesn’t get to set them. Three classes, three jobs:

Class Kind Has password Has hashed_password Role in the request
UserCreate pydantic (in) ✅ plaintext Validate the registration body
User SQLAlchemy (storage) ✅ the hash The row in the users table
UserPublic pydantic (out) Serialize the response — no secrets

The bridge between them is from_attributes=True (it was called orm_mode in pydantic v1 — you’ll see both). It tells pydantic “you can build this schema by reading attributes off an object,” so UserPublic.model_validate(user_row) works even though user_row is a SQLAlchemy object, not a dict. Combined with response_model=UserPublic on the route, the handler can return user (the whole ORM row, hash and all) and FastAPI serializes only id, email, role. Here it is, run for real:

class UserPublic(BaseModel):
    id: int; email: EmailStr; role: str
    model_config = {"from_attributes": True}

row = get_user_from_db()                       # an ORM User with a hashed_password
public = UserPublic.model_validate(row)        # read attributes off the ORM object
print("ORM row has     :", [a for a in vars(row) if not a.startswith("_")])
print("UserPublic emits:", public.model_dump())
print("hashed_password leaked?", "hashed_password" in public.model_dump())
ORM row has     : ['id', 'email', 'role', 'hashed_password']
UserPublic emits: {'id': 1, 'email': 'vinod@kloudvin.com', 'role': 'user'}
hashed_password leaked? False

The row carries the hash; the response doesn’t. This is Part 1’s leaky-twin lesson with real stakes: return a raw ORM object without a response_model and you ship the password hash to the client — the single most common way secrets escape a Python API. Keeping the model and the schema separate isn’t ceremony; it’s the boundary that makes the leak impossible.


CRUD backed by the database

With the session dependency and the schemas in place, a real registration endpoint writes a real row. The shape is identical to Part 1’s create_item, except the store is a database and the “secret” is a genuine password hash:

from fastapi import HTTPException
from sqlalchemy import select

@app.post("/register", response_model=UserPublic, status_code=201)
def register(payload: UserCreate, db: Db):
    if db.scalar(select(User).where(User.email == payload.email)):
        raise HTTPException(status_code=409, detail="Email already registered")  # conflict
    user = User(email=payload.email,
                hashed_password=hash_password(payload.password),   # HASH, never plaintext
                role="user")
    db.add(user); db.commit(); db.refresh(user)      # write, commit, reload the id
    return user                                       # UserPublic strips the hash

Three database calls matter here. db.add(user) stages the new row; db.commit() makes it real (without it, the write is invisible and thrown away when the session closes — the “it worked in my script but the table is empty” bug from the databases lesson); db.refresh(user) reloads the row so user.id (assigned by the database) is populated for the response. The duplicate-email check returns 409 Conflict, the correct code when a unique constraint would be violated — and the database’s unique=True on email is the real backstop if two requests race.

A CRUD-to-status-code map, now that a database is involved:

Operation Route DB action Success Failure
Create POST /register add + commit 201 409 duplicate, 422 bad body
Read one GET /users/{id} db.get(User, id) 200 404 missing
List GET /users select().limit() 200 — (never unbounded)
Update PATCH /users/{id} mutate + commit 200 404, 422
Delete DELETE /users/{id} db.delete + commit 204 404

Migrations with Alembic (briefly)

Base.metadata.create_all(engine) creates tables that don’t exist yet — fine for a demo or a first run, useless the moment the schema changes, because it never alters an existing table. Real projects use Alembic, SQLAlchemy’s migration tool, which versions your schema as a chain of scripts:

Command What it does
alembic init alembic Scaffold the migrations directory (once)
alembic revision --autogenerate -m "add users" Diff models vs DB, write a migration script
alembic upgrade head Apply all pending migrations
alembic downgrade -1 Roll back the last migration
alembic current / history Show where the DB is / the full chain

The workflow: change your models, --autogenerate a revision, review the generated script (autogenerate is good, not perfect — it misses some column-type changes), commit it, and upgrade head in every environment including production as a deploy step. ⚠️ Never create_all and Alembic on the same database — pick one; mixing them confuses Alembic about what state the schema is in.

Connection pooling (briefly)

Opening a database connection is expensive — a TCP handshake plus authentication, milliseconds each on Postgres. A server can’t afford that per request, so SQLAlchemy keeps a pool of open connections and lends them out. The knobs you’ll actually set:

create_engine(...) arg Default Meaning
pool_size 5 Connections kept open in the pool
max_overflow 10 Extra connections allowed under burst
pool_timeout 30 Seconds to wait for a free connection before raising
pool_recycle -1 (off) Recycle a connection after N seconds (dodges server-side timeouts)
pool_pre_ping False ✅ Test a connection is alive before using it (recommended in prod)

The one that saves you a 3am page is pool_pre_ping=True: without it, a connection the database silently dropped (idle timeout, failover) blows up the next request with a stale-connection error; with it, SQLAlchemy quietly checks and replaces dead connections. The get_db dependency’s finally: db.close() is what keeps the pool healthy — it returns each connection so the pool never starves.


Authentication vs authorization

Two words that sound alike, mean different things, and map to two different HTTP status codes. Getting them straight is the backbone of the rest of the lesson.

Authentication Authorization
Question Who are you? What may you do?
Happens First — at login and on every request After authn — per action
Mechanism Password → token; verify token Roles, scopes, ownership checks
Failure code 401 Unauthorized 403 Forbidden
“no token” 401
“valid token, wrong role” 403
FastAPI piece get_current_user dependency require_role / scope check

The distinction is not pedantic — returning the wrong code leaks information or confuses clients. A 401 tells a client “authenticate and retry”; a 403 tells it “don’t bother, you’ll never be allowed.” Send 401 when you meant 403 and a legitimate client retries forever; send 403 when you meant 401 and you’ve told an anonymous stranger that the resource exists. In the lab you’ll see all three cases fire: no token → 401, tampered token → 401, valid token but role=user on an admin route → 403.


Passwords: hash right, or not at all

This section is the most important one in the lesson, and the rule is absolute: never store a password. Store a hash — a one-way fingerprint you can verify against but can’t reverse — and make it salted and slow. Get this wrong and a database leak becomes a catastrophe for every user who reused that password elsewhere. Get it right and it’s one function call.

Why not just store the password? Because databases leak — through SQL injection, a stolen backup, an insider, a misconfigured bucket. When (not if) yours does, plaintext passwords are game over. Why not a fast hash like MD5 or SHA-256? Because “one-way” isn’t enough when the function is fast: an attacker with the hashes runs billions of guesses per second on a GPU against a dictionary and cracks the weak ones in minutes. The defence is a hash that is deliberately slow and salted (so identical passwords get different hashes and precomputed “rainbow tables” are useless).

passlib wraps the right algorithms behind one clean interface. Here it is hashing and verifying, run for real:

from passlib.context import CryptContext
pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")

password = "correct horse battery staple"
h1 = pwd.hash(password)
h2 = pwd.hash(password)                          # SAME password, hashed twice
print("hash #1 :", h1)
print("hash #2 :", h2)
print("equal?  :", h1 == h2, "(salted, so no)")
print("verify correct:", pwd.verify(password, h1))
print("verify wrong  :", pwd.verify("Correct Horse Battery Staple", h1))
hash #1 : $2b$12$jefLN/dLDCK3CiGIdJd6dOBn8M7MdSSPxsCgZv3dp8APvpoc75mOa
hash #2 : $2b$12$rGRpVUVI3Zt5.lwPDnQ/HujqAXiK6kx3livgVWmV/eg7D2UqQSVaa
equal?  : False (salted, so no)
verify correct: True
verify wrong  : False

Read that carefully. The same password produced two different hashes — because bcrypt generates a random salt each time and stores it inside the hash string. That’s why you can’t compare hashes directly; you call verify, which extracts the salt from the stored hash and re-hashes the candidate with it. And verify returns True only for the exact password — one capital letter off and it’s False. You never decrypt anything, because there’s nothing to decrypt; hashing is one-way by design.

The $2b$12$... string isn’t random noise — it’s self-describing:

Segment Value Meaning
$2b$ 2b The bcrypt algorithm identifier
$12$ 12 Cost factor — 2¹² = 4096 rounds. Higher = slower = safer
next 22 chars jefLN/... The random salt (base64)
final 31 chars Bn8M7M... The actual hash digest

Because the salt and cost are in the string, the hash is fully portable — verify needs nothing but the stored hash and the candidate password. The total is always 60 characters.

Why “slow” is the whole point

This is the part people skip, so let’s measure it. bcrypt at cost 12 versus a plain SHA-256, timed on the same machine:

import hashlib, time
from passlib.context import CryptContext
pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
password = "correct horse battery staple"

t = time.perf_counter()
for _ in range(20): pwd.hash(password)
bcrypt_ms = (time.perf_counter() - t) / 20 * 1000

t = time.perf_counter()
for _ in range(20_000): hashlib.sha256(password.encode()).hexdigest()
sha_ms = (time.perf_counter() - t) / 20_000 * 1000

print("bcrypt(cost=12): %.1f ms per hash" % bcrypt_ms)
print("sha256        : %.6f ms per hash" % sha_ms)
print("ratio         : sha256 is ~%d,000x faster" % round(bcrypt_ms / sha_ms / 1000))
bcrypt(cost=12): 226.4 ms per hash
sha256        : 0.000380 ms per hash
ratio         : sha256 is ~596,000x faster

That factor of ~600,000 is the security. A login hashes one password once, so 226 ms is invisible to your user. But an attacker who steals your database has to hash every guess: with SHA-256 they try millions per second per GPU and crack weak passwords fast; with bcrypt, every guess costs 226 ms, and a serious wordlist attack becomes years instead of minutes. “Fast hash” is a synonym for “wrong hash” when the input is a password. The slowness you’d never want in a checksum is precisely the feature you want here.

Algorithm Speed Salted Password use?
argon2 Slow, tunable (memory-hard) ✅ Best modern choice (OWASP #1)
bcrypt Slow, tunable (cost factor) ✅ Excellent, ubiquitous
scrypt Slow, memory-hard ✅ Good
PBKDF2 Slow, tunable ✅ OK (FIPS contexts)
MD5 Blazing fast Broken — never
SHA-256 / SHA-3 Blazing fast ❌ Wrong tool — never for passwords
plaintext ❌❌ Career-ending

To use argon2 instead of bcrypt, you install passlib[argon2] and change one line: CryptContext(schemes=["argon2"], deprecated="auto"). The deprecated="auto" flag is a quiet superpower — list several schemes and passlib verifies against any of them but re-hashes to the first (newest) one on successful login, so you can migrate a whole user base from bcrypt to argon2 transparently as people log in.

⚠️ bcrypt silently truncates at 72 bytes. It only ever looks at the first 72 bytes of the password, so "a"*100 and "a"*72 hash identically — which is why UserCreate caps password at max_length=72. For passwords long enough to hit that (passphrases, or if you pre-hash), reach for argon2, which has no such limit. And the two iron rules that have no exceptions: never log a password (not in debug logs, not in an exception, not in a request trace — the logging lesson shows how a careless logger.info(f"login {payload}") ends up with plaintext in your log files), and never return the hash (that’s what UserPublic is for).


Sessions vs tokens: two ways to stay logged in

Once a user has proven who they are, the server needs to remember it across their next requests, because HTTP itself is stateless. There are two dominant approaches, and FastAPI’s OAuth2 flow uses the second.

Session (cookie) Token (JWT)
Server stores A session record per login Nothing (just the signing key)
Client sends Opaque session ID (cookie) Signed JWT (Authorization header)
Who am I? Look up the session store Verify the signature, read the payload
Scales across servers Needs shared session store (Redis) ✅ Any server with the key can verify
Revoke instantly ✅ Delete the session record ⚠️ Hard — token valid until it expires
Mobile / API clients Awkward (cookies) ✅ Natural (a header)
Vulnerable to CSRF (it’s a cookie) XSS if stored badly; no built-in revoke
Best for Server-rendered web apps ✅ APIs, microservices, mobile

Neither is “better” — they trade instant revocation against statelessness. Sessions let you kill a login on the spot (delete the record) but need a shared store to scale horizontally. Tokens scale to any number of servers with zero shared state but can’t be un-issued — a stolen token works until it expires, which is exactly why token expiry is short (minutes to an hour) and why serious systems add a refresh token and sometimes a revocation list for the rare “log this user out everywhere now.” For a JSON API, mobile backend, or microservice — the FastAPI sweet spot — tokens win, and that’s what we build.


JWT: structure, signing, and what NOT to put in it

A JSON Web Token is three base64url-encoded parts joined by dots: header.payload.signature. Decode the first two and they’re just JSON; the third is a cryptographic signature over the first two. Let’s build one and take it apart, for real:

from datetime import datetime, timedelta, timezone
from jose import jwt
import base64, json

SECRET = "a-long-random-secret-from-the-environment"
token = jwt.encode(
    {"sub": "vinod@kloudvin.com", "role": "user",
     "exp": datetime.now(timezone.utc) + timedelta(minutes=30)},
    SECRET, algorithm="HS256")

h, p, s = token.split(".")
def b64url(seg): return base64.urlsafe_b64decode(seg + "=" * (-len(seg) % 4))
print("header   :", json.loads(b64url(h)))
print("payload  :", json.loads(b64url(p)))
print("signature:", s[:24], "... (%d chars)" % len(s))
header   : {'alg': 'HS256', 'typ': 'JWT'}
payload  : {'sub': 'vinod@kloudvin.com', 'role': 'user', 'exp': 1784288818}
signature: k-xOg-S64fJXVBz-GD9N-Tjs ... (43 chars)

(Your exp and signature will differ every run — exp is a timestamp and the signature depends on both the payload and the secret key.) The three parts, and what each is for:

Part Contains Example Secret?
Header Algorithm + type {"alg": "HS256", "typ": "JWT"} ❌ public
Payload Claims (the data) {"sub": ..., "role": ..., "exp": ...} public — readable by anyone
Signature HMAC of header+payload with the key k-xOg-S64f... Proves integrity

The payload holds claims. Some names are standardised (“registered claims”) and the libraries treat them specially:

Claim Name Meaning Enforced by the library?
sub Subject Who the token is about (user id/email) No (you read it)
exp Expiration Unix time after which it’s invalid ✅ Yes — decode rejects if past
iat Issued At When it was minted No
nbf Not Before Valid only after this time ✅ Yes
iss / aud Issuer / Audience Who made it / who it’s for ✅ If you pass issuer/audience
role, email, … Custom Whatever you add No

The one thing you must internalise: signed ≠ encrypted

The signature guarantees the token was minted by someone with the secret key and hasn’t been altered — it does not hide anything. The payload is base64, not ciphertext. Anyone who has the token can read every claim without the key. Here’s an attacker doing exactly that to a token that foolishly embedded a secret:

bad = jwt.encode({"sub": "vinod@kloudvin.com",
                  "credit_card": "4111-1111-1111-1111",     # ⚠️ NEVER
                  "exp": datetime.now(timezone.utc) + timedelta(minutes=5)},
                 "server-secret", algorithm="HS256")
_, payload_b64, _ = bad.split(".")
seen = json.loads(base64.urlsafe_b64decode(payload_b64 + "=" * (-len(payload_b64) % 4)))
print("attacker reads (NO secret needed):", seen)
attacker reads (NO secret needed): {'sub': 'vinod@kloudvin.com', 'credit_card': '4111-1111-1111-1111', 'exp': 1784286320}

No key, no cracking — a split(".") and a base64 decode, and the card number is right there. So: put identifiers in a JWT, never secrets. A user id, a role, an email — fine. A password, a card number, an API key, anything you’d hate to see in a log — never. If you genuinely need to hide contents, that’s JWE (encrypted tokens), a different tool; a plain JWT is a signed postcard, not a sealed envelope.

What the signature does protect: tampering and expiry

Change one byte of the payload and the signature no longer matches, so verification fails. Let an attacker with no key try to escalate their role, and try an expired token, and a token signed with a guessed key — all rejected, with real error types:

from jose import jwt, JWTError, ExpiredSignatureError

# expired: exp in the past
expired = jwt.encode({"sub": "vinod@kloudvin.com",
                      "exp": datetime.now(timezone.utc) - timedelta(seconds=1)},
                     SECRET, algorithm="HS256")
try: jwt.decode(expired, SECRET, algorithms=["HS256"])
except ExpiredSignatureError as e: print("expired  ->", type(e).__name__, "|", e)

# wrong key: a forged token
try: jwt.decode(token, "attacker-guessed-secret", algorithms=["HS256"])
except JWTError as e: print("wrong key->", type(e).__name__, "|", e)

# tampered: attacker flips role to admin but can't re-sign
payload = json.loads(b64url(p)); payload["role"] = "admin"
p2 = base64.urlsafe_b64encode(json.dumps(payload).encode()).rstrip(b"=").decode()
try: jwt.decode(f"{h}.{p2}.{s}", SECRET, algorithms=["HS256"])
except JWTError as e: print("tampered ->", type(e).__name__, "|", e)

# no expiry: valid forever
forever = jwt.encode({"sub": "vinod@kloudvin.com"}, SECRET, algorithm="HS256")
print("no-exp   ->", jwt.decode(forever, SECRET, algorithms=["HS256"]), "<- never dies")
expired  -> ExpiredSignatureError | Signature has expired.
wrong key-> JWTError | Signature verification failed.
tampered -> JWTError | Signature verification failed.
no-exp   -> {'sub': 'vinod@kloudvin.com'} <- never dies

Three protections and one warning, all demonstrated. jwt.decode checks the signature (so a tampered payload or a wrong key is rejected — the attacker can’t forge role: admin without the secret) and the expiry (so an old token is rejected automatically). But the last line is the trap: a token with no exp is valid forever — a single leaked one is a permanent backdoor. Always set exp, and keep it short. Which brings us to the mistakes worth stating outright:

❌ JWT mistake Why it’s bad ✅ Do instead
Secret/PII in the payload It’s public — anyone reads it Only identifiers (sub, role)
No exp claim Leaked token never expires Short expiry (15–30 min) + refresh token
Weak/guessable signing key Attacker forges valid tokens ≥32 random bytes from a CSPRNG
Signing key in the source code Leaks with the repo → forge anything From env / secret manager
Accepting alg: none Classic bypass — unsigned “valid” token Pin algorithms=["HS256"] on decode
Long-lived token as the only auth Can’t revoke a stolen one Short access + revocable refresh

That algorithms=["HS256"] on every decode is not optional — passing the allowed algorithms explicitly is what blocks the infamous alg: none and algorithm-confusion attacks. python-jose requires it, which is the library protecting you.


The OAuth2 password flow in FastAPI

Now assemble the pieces into FastAPI’s built-in OAuth2 password flow: a /token endpoint that takes a username and password and returns a JWT, and a get_current_user dependency that protects any route by requiring a valid token. FastAPI ships the plumbing.

Two helpers first — signing a token, and the security scheme that reads the header:

import os
from datetime import datetime, timedelta, timezone
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError

SECRET_KEY = os.environ["SECRET_KEY"]              # from the ENV, never hardcoded
ALGORITHM, EXPIRE_MIN = "HS256", 30

def create_access_token(sub: str, role: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=EXPIRE_MIN)
    return jwt.encode({"sub": sub, "role": role, "exp": expire},
                      SECRET_KEY, algorithm=ALGORITHM)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")   # extracts "Authorization: Bearer <jwt>"

OAuth2PasswordBearer(tokenUrl="token") is a dependency that pulls the token out of the Authorization: Bearer ... header (and, as a bonus, powers the Authorize button in /docs). Its tokenUrl just tells the docs where login lives. Now the two endpoints and the guard:

from typing import Annotated
from fastapi import Depends, HTTPException, status
from sqlalchemy import select

@app.post("/token", response_model=Token)
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()], db: Db):
    user = db.scalar(select(User).where(User.email == form.username))
    if user is None or not verify_password(form.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Incorrect email or password",
                            headers={"WWW-Authenticate": "Bearer"})
    return Token(access_token=create_access_token(sub=user.email, role=user.role))

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: Db) -> User:
    creds_exc = HTTPException(status_code=401, detail="Could not validate credentials",
                             headers={"WWW-Authenticate": "Bearer"})
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])   # verify sig + exp
        email = payload.get("sub")
        if email is None:
            raise creds_exc
    except JWTError:                                # bad signature, expired, malformed
        raise creds_exc
    user = db.scalar(select(User).where(User.email == email))
    if user is None:
        raise creds_exc
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

@app.get("/me", response_model=UserPublic)
def me(user: CurrentUser):                          # protected: just add the dependency
    return user

Trace the design. OAuth2PasswordRequestForm is a dependency that reads the standard username and password form fields (this is what needs python-multipart). /token looks the user up, calls verify_password against the stored hash, and — only on a match — signs and returns a JWT. get_current_user is the mirror: it receives the token via oauth2_scheme, jwt.decodes it (signature and expiry checked in one call), loads the user, and raises 401 on anything wrong. Any route becomes protected by adding one parameter, user: CurrentUser — the auth logic is written once and reused, and it even carries its own database session because get_current_user itself Depends(get_db). That’s the whole flow, and here it is as one picture:

Diagram of the login-to-protected-route flow across five zones, left to right. The client POSTs username and password to /token; the login zone runs verify_password against the bcrypt hash; the database-and-sign zone reads the users table via the get_db dependency and signs a short-lived JWT with the secret key; the client sends that JWT back as an Authorization Bearer header to /me; the access zone runs get_current_user, which verifies the signature and expiry through Depends; and the protected zone returns a 200 with UserPublic (the password hash stripped) or a 401/403 on a bad token or wrong role. Red badges mark the two mistakes: storing a plaintext or fast-hashed password at the verify step, and putting a secret or no expiry into the signed-not-encrypted JWT.

Read it left to right and the two round-trips share one path: login (client → /token → verify the bcrypt hash → sign a JWT) and access (client → /me with the Bearer token → get_current_user verifies signature and expiry → protected data). The badges mark the two places a mistake would be fatal — storing a plaintext password at zone 1, and putting a secret (or no expiry) in the signed-not-encrypted JWT at zone 2 — and the green badge is Part 1’s response_model stripping the hash on the way out.


Authorization: protecting routes by role

Authentication got us who. Authorization is what may they do, and the simplest model is roles. Our User has a role column, the JWT carries it, and a tiny dependency factory turns “must be an admin” into one line on any route:

def require_role(role: str):
    def checker(user: CurrentUser) -> User:
        if user.role != role:
            raise HTTPException(status_code=403,           # 403, NOT 401 — they ARE known
                                detail=f"Requires role '{role}'")
        return user
    return checker

@app.get("/admin", response_model=UserPublic)
def admin_only(user: Annotated[User, Depends(require_role("admin"))]):
    return user

require_role("admin") returns a dependency that first runs get_current_user (so the caller is authenticated) and then checks the role, raising 403 if it’s wrong — the caller is known, just not permitted. This is the authn/authz split made concrete: get_current_user handles who (401 on failure), require_role handles what (403 on failure). For finer-grained needs — “can edit this specific document” — you check ownership in the handler (does document.owner_id == user.id?); for APIs with many permissions you use scopes (FastAPI’s SecurityScopes), where the token carries a list of granted scopes and each route declares which it needs.

Authorization model Granularity How When
Roles Coarse (admin/user) user.role check Most apps start here
Scopes Per-permission Token lists scopes; route requires one OAuth2 APIs, third-party access
Ownership Per-object obj.owner_id == user.id “Your own resources only”
Policy / ABAC Arbitrary rules External engine (OPA, Casbin) Complex enterprise rules

The security essentials

Auth is the biggest piece, but a production API has a checklist of other defences. Most are one-liners or config; skipping them is how APIs get owned. Here’s the working set, several of which tie straight back to earlier lessons:

Concern The risk The defence in FastAPI
Input validation Malformed/malicious bodies ✅ pydantic — already done (Part 1’s 422)
SQL injection f"...{user_input}..." in SQL ✅ ORM / parameterised queries — the databases lesson
Secrets in code Key leaks with the repo Env vars / secret manager, never hardcoded
CORS Malicious sites calling your API CORSMiddleware with named origins
HTTPS/TLS Tokens/passwords sniffed on the wire Terminate TLS at a reverse proxy
Rate limiting Brute-force, scraping, DoS slowapi / proxy / gateway limits
Security headers Clickjacking, MIME sniffing Middleware sets X-Frame-Options, HSTS, …
Dependency CVEs Known holes in your packages pip-audit, Dependabot

Two deserve a second look because they connect to what you’ve already built. SQL injection is fully handled the moment you use the ORM or parameterised queries — as the databases lesson showed, the SQL text and the values travel to the database separately, so a value like "' OR '1'='1" is compared as a literal string, never parsed as SQL. The vulnerability is born only if you bypass that and f-string user input into a query. Since our select(User).where(User.email == form.username) is pure ORM, form.username can never be anything but data. And input validation is pydantic doing its job from Part 1 — every request body is validated before your handler runs; an invalid email is a 422, not a surprise:

POST /register {"email": "not-an-email", ...}  ->  422
{'type': 'value_error', 'msg': 'value is not a valid email address...'}

CORS, HTTPS, and rate limiting are the three most-forgotten. CORS is Part 1’s browser gotcha (add_middleware(CORSMiddleware, allow_origins=[...])). HTTPS you almost never terminate in Python — a reverse proxy (nginx, Traefik, Caddy) or a cloud load balancer holds the certificate and speaks TLS to the world, forwarding plain HTTP to uvicorn on localhost; without it, every password and token crosses the network in the clear. Rate limiting caps how fast a client can hit /token, turning a brute-force attack from “billions of guesses” into “a handful per minute” — do it at the proxy/gateway, or in-app with slowapi.

For a mental checklist, the OWASP API Security Top 10 names the categories that actually get APIs breached:

# OWASP API risk One-line defence
API1 Broken Object-Level Auth (BOLA) Check ownership, not just login — obj.owner_id == user.id
API2 Broken Authentication Strong hashing + short-lived tokens + exp
API3 Broken Object Property-Level Auth response_model out, strict schema in
API4 Unrestricted Resource Consumption Rate limits + pagination + body-size caps
API5 Broken Function-Level Auth require_role / scopes on every privileged route
API7 Server-Side Request Forgery Validate/allow-list any URL you fetch
API8 Security Misconfiguration No debug in prod, security headers, least privilege
API9 Improper Inventory Management Version your API, retire old endpoints

Notice how many you’ve already addressed: API2 (bcrypt + JWT expiry), API3 (response_model + pydantic), API5 (require_role). BOLA (API1) is the sneakiest — a logged-in user changing /orders/123 to /orders/124 and seeing someone else’s order — and no framework catches it for you; you must check ownership in the handler.


Taking it to production

Part 1 ran the dev server and flagged that it doesn’t belong in production. Here’s what does. The through-line: your Python process does one job — run the app — and everything else (TLS, static files, load balancing, restarts) is somebody else’s.

Run it with an ASGI server and multiple workers. In dev, uvicorn app:app --reload. In production, run several worker processes so you use every CPU core, managed by gunicorn (or uvicorn’s own --workers):

# production: 4 worker processes, no reload
gunicorn app:app -k uvicorn.workers.UvicornWorker -w 4 --bind 0.0.0.0:8000
Concern Development Production
Command uvicorn app:app --reload gunicorn -k uvicorn.workers.UvicornWorker -w N
Reload on edit --reload ❌ Never (slow, single-process)
Workers 1 ≈ 2 × cores (I/O-bound: more)
TLS none at the reverse proxy
Secrets .env file secret manager / orchestrator
Restarts manual systemd / container orchestrator

Put it behind a reverse proxy. nginx, Traefik, or Caddy sits in front and does what Python shouldn’t: terminate HTTPS, serve static files, load-balance across workers, enforce rate limits and body-size caps, and add security headers. uvicorn listens only on localhost; the proxy faces the internet. (One FastAPI detail: behind a proxy, pass --proxy-headers and set --forwarded-allow-ips so the app sees the real client IP and scheme from X-Forwarded-*.)

Configure from the environment, not code, with pydantic-settings — the same pydantic validation, applied to config:

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    secret_key: str                              # required — startup fails if unset
    access_token_expire_minutes: int = 30
    database_url: str = "sqlite:///./web2.db"
    model_config = SettingsConfigDict(env_file=".env")

settings = Settings()                            # reads env vars, coerces types, validates
$ SECRET_KEY=... ACCESS_TOKEN_EXPIRE_MINUTES=15 python -c "from config import settings; print(settings.access_token_expire_minutes)"
15

secret_key has no default, so the app refuses to start without it — far better than silently running on a dev key. Generate a real one with python -c "import secrets; print(secrets.token_hex(32))" (32 random bytes → 64 hex chars) and inject it via the environment. ⚠️ The .env file is for local dev only and must be in .gitignore; in production the secret comes from the orchestrator or a secret manager (AWS Secrets Manager, Vault, Azure Key Vault, Doppler).

Add a health check and structured logs. A trivial @app.get("/health") returning {"status": "ok"} lets your load balancer and orchestrator know the process is alive (and, if you check the DB in it, ready). Logs should be structured — the logging lesson’s logging configured for JSON output, with a request id, so you can search them; ⚠️ and never log tokens, passwords, or Authorization headers.

Dockerize it so the same image runs everywhere:

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["gunicorn", "app:app", "-k", "uvicorn.workers.UvicornWorker", \
     "-w", "4", "--bind", "0.0.0.0:8000"]

A production readiness checklist to run down before you ship:

✅ Ship-ready ❌ Not yet
Secrets from env / secret manager Signing key hardcoded or committed
HTTPS terminated at the proxy Plain HTTP to the internet
gunicorn + uvicorn workers --reload / bare dev server
Real DB (Postgres) + connection pool In-memory dict (dies per worker)
Alembic migrations as a deploy step create_all in prod
CORS locked to real origins allow_origins=["*"] with credentials
Rate limit on /token Unlimited login attempts
/health + structured logging No observability
bcrypt/argon2, short-lived JWTs Fast hashes, no exp

Managed hosting removes most of this: platforms like Railway, Render, Fly.io, Google Cloud Run, or AWS App Runner take your Docker image (or repo), give you HTTPS, scaling, health checks, and secret management, and run the container. For a small API the honest advice is to start there — you get the whole checklist’s infrastructure without hand-rolling nginx and systemd, and you can always move to raw VMs when scale or cost demands it.


Hands-on lab

Build the complete auth API — SQLAlchemy users with bcrypt-hashed passwords, a JWT login, and a protected route — then run the round-trip against a real server and watch every security property hold. This is the whole lesson executed end to end.

⚠️ Everything is local: a SQLite file, a localhost server, no external services, no cost.

Step 1 — Set up

mkdir auth-api && cd auth-api
python3 -m venv .venv
source .venv/bin/activate                 # Windows: .venv\Scripts\activate
python -m pip install fastapi uvicorn pydantic sqlalchemy \
    "passlib[bcrypt]" "python-jose[cryptography]" httpx python-multipart
python -m pip install "bcrypt==4.0.1"      # ⚠️ pin: newer bcrypt breaks passlib 1.7.4
export SECRET_KEY=$(python -c "import secrets; print(secrets.token_hex(32))")

What just happened: an isolated environment, the bcrypt pin that avoids the passlib crash, and a real 32-byte signing key in the environment (not the code).

Step 2 — The app (app.py) — this is the file every snippet above came from, assembled:

# app.py
from __future__ import annotations
import os
from datetime import datetime, timedelta, timezone
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, EmailStr, Field
from sqlalchemy import String, create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, sessionmaker

SECRET_KEY = os.environ.get("SECRET_KEY", "dev-only-change-me")   # ⚠️ demo default
ALGORITHM, ACCESS_TOKEN_EXPIRE_MINUTES = "HS256", 30

# ---- database ----
class Base(DeclarativeBase): pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String, unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String)
    role: Mapped[str] = mapped_column(String, default="user")

engine = create_engine("sqlite:///./web2.db", connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base.metadata.create_all(engine)

def get_db():
    db = SessionLocal()
    try: yield db
    finally: db.close()
Db = Annotated[Session, Depends(get_db)]

# ---- schemas ----
class UserCreate(BaseModel):
    email: EmailStr
    password: str = Field(min_length=8, max_length=72)

class UserPublic(BaseModel):
    id: int; email: EmailStr; role: str
    model_config = {"from_attributes": True}

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

# ---- passwords + JWT ----
pwd = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(p: str) -> str: return pwd.hash(p)
def verify_password(p: str, h: str) -> bool: return pwd.verify(p, h)

def create_access_token(sub: str, role: str) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    return jwt.encode({"sub": sub, "role": role, "exp": expire}, SECRET_KEY, algorithm=ALGORITHM)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: Db) -> User:
    exc = HTTPException(status.HTTP_401_UNAUTHORIZED, "Could not validate credentials",
                        {"WWW-Authenticate": "Bearer"})
    try:
        email = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]).get("sub")
        if email is None: raise exc
    except JWTError:
        raise exc
    user = db.scalar(select(User).where(User.email == email))
    if user is None: raise exc
    return user
CurrentUser = Annotated[User, Depends(get_current_user)]

def require_role(role: str):
    def checker(user: CurrentUser) -> User:
        if user.role != role:
            raise HTTPException(403, f"Requires role '{role}'")
        return user
    return checker

# ---- app ----
app = FastAPI(title="Auth API", version="1.0.0")

@app.post("/register", response_model=UserPublic, status_code=201)
def register(payload: UserCreate, db: Db):
    if db.scalar(select(User).where(User.email == payload.email)):
        raise HTTPException(409, "Email already registered")
    user = User(email=payload.email, hashed_password=hash_password(payload.password), role="user")
    db.add(user); db.commit(); db.refresh(user)
    return user

@app.post("/token", response_model=Token)
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()], db: Db):
    user = db.scalar(select(User).where(User.email == form.username))
    if user is None or not verify_password(form.password, user.hashed_password):
        raise HTTPException(401, "Incorrect email or password", {"WWW-Authenticate": "Bearer"})
    return Token(access_token=create_access_token(user.email, user.role))

@app.get("/me", response_model=UserPublic)
def me(user: CurrentUser): return user

@app.get("/admin", response_model=UserPublic)
def admin_only(user: Annotated[User, Depends(require_role("admin"))]): return user

What just happened: the ORM User (with hashed_password), three schemas (in/out/token), password hashing, JWT sign+verify, the get_db session dependency, get_current_user, a require_role authorization gate, and four routes — register, login, protected /me, admin-only /admin.

Step 3 — Run it

uvicorn app:app --port 8100

Open **http://127.0.0.1:8100/docs**: the Swagger UI has an Authorize button (from OAuth2PasswordBearer) — you can log in there and call /me in the browser. The OpenAPI security scheme is generated for free:

securitySchemes -> ['OAuth2PasswordBearer']
OAuth2 flow     -> {'type': 'oauth2', 'flows': {'password': {'scopes': {}, 'tokenUrl': 'token'}}}

Step 4 — The full round-trip (drive.py, in a second terminal, same venv) — register, prove the stored hash isn’t the password, log in for a JWT, and exercise every auth path:

# drive.py
import base64, json, httpx
from sqlalchemy import create_engine, text

B = "http://127.0.0.1:8100"
EMAIL, PW = "vinod@kloudvin.com", "s3cret-pa55phrase"
def b64url(s): return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))

with httpx.Client(base_url=B, timeout=5) as x:
    print("REGISTER        ->", (r := x.post("/register", json={"email": EMAIL, "password": PW})).status_code, r.json())

    with create_engine("sqlite:///./web2.db").connect() as c:   # peek at the stored row
        row = c.execute(text("SELECT email, hashed_password FROM users")).one()
    print("DB hashed_password ->", row.hashed_password)
    print("password in hash?  ->", PW in row.hashed_password)

    token = (r := x.post("/token", data={"username": EMAIL, "password": PW})).json()["access_token"]
    print("LOGIN           ->", r.status_code)
    h, p, s = token.split(".")
    print("JWT header      ->", json.loads(b64url(h)))
    print("JWT payload     ->", json.loads(b64url(p)))
    print("ME (token)      ->", (r := x.get("/me", headers={"Authorization": f"Bearer {token}"})).status_code, r.json())
    print("ME (no token)   ->", (r := x.get("/me")).status_code, r.json())
    print("LOGIN (bad pw)  ->", (r := x.post("/token", data={"username": EMAIL, "password": "WRONG"})).status_code, r.json())
    print("ME (tampered)   ->", (r := x.get("/me", headers={"Authorization": f"Bearer {h}.{p}.{s[:-2]}XY"})).status_code, r.json())
    print("ADMIN (role=user)->", (r := x.get("/admin", headers={"Authorization": f"Bearer {token}"})).status_code, r.json())
REGISTER        -> 201 {'id': 1, 'email': 'vinod@kloudvin.com', 'role': 'user'}
DB hashed_password -> $2b$12$Na0kKGdsagg18KpiOPb3Veit.TpTBtmkvzwn6Zel8CUT6yBh3PwMe
password in hash?  -> False
LOGIN           -> 200
JWT header      -> {'alg': 'HS256', 'typ': 'JWT'}
JWT payload     -> {'sub': 'vinod@kloudvin.com', 'role': 'user', 'exp': 1784287732}
ME (token)      -> 200 {'id': 1, 'email': 'vinod@kloudvin.com', 'role': 'user'}
ME (no token)   -> 401 {'detail': 'Not authenticated'}
LOGIN (bad pw)  -> 401 {'detail': 'Incorrect email or password'}
ME (tampered)   -> 401 {'detail': 'Could not validate credentials'}
ADMIN (role=user)-> 403 {'detail': "Requires role 'admin'"}

What just happened — read every line, because each is a security property proven over a real socket:

Step 5 — Offline tests (test_app.py) — TestClient runs the app in-process with a throwaway test database via dependency_overrides (Part 1’s DI test seam):

# test_app.py
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app import app, Base, get_db

test_engine = create_engine("sqlite:///./test.db", connect_args={"check_same_thread": False})
TestSession = sessionmaker(bind=test_engine)

def override_get_db():
    db = TestSession()
    try: yield db
    finally: db.close()

@pytest.fixture(autouse=True)
def fresh_db():
    Base.metadata.drop_all(test_engine); Base.metadata.create_all(test_engine)
    app.dependency_overrides[get_db] = override_get_db     # swap in the test DB
    yield
    app.dependency_overrides.clear()

client = TestClient(app)
REG = {"email": "a@b.com", "password": "password123"}

def test_register_emits_public_shape_no_password():
    r = client.post("/register", json=REG)
    assert r.status_code == 201 and set(r.json()) == {"id", "email", "role"}

def test_login_then_me_round_trip():
    client.post("/register", json=REG)
    tok = client.post("/token", data={"username": "a@b.com", "password": "password123"}).json()["access_token"]
    assert client.get("/me", headers={"Authorization": f"Bearer {tok}"}).status_code == 200

def test_me_without_token_is_401():
    assert client.get("/me").status_code == 401

def test_wrong_password_is_401():
    client.post("/register", json=REG)
    assert client.post("/token", data={"username": "a@b.com", "password": "WRONG"}).status_code == 401

def test_duplicate_email_is_409():
    client.post("/register", json=REG)
    assert client.post("/register", json=REG).status_code == 409

def test_short_password_is_422():
    assert client.post("/register", json={"email": "a@b.com", "password": "short"}).status_code == 422
python -m pytest test_app.py -q
......                                                                     [100%]
6 passed, 2 warnings in 1.74s

What just happened: six security behaviours — the public-shape response, the login→/me round trip, the 401 without a token, the 401 on a wrong password, the 409 on a duplicate, and the 422 on a too-short password — verified in 1.74 seconds with no server running. The two warnings are harmless: a StarletteDeprecationWarning about TestClient’s httpx (as in Part 1), and a DeprecationWarning that passlib imports the stdlib crypt module (removed in Python 3.13 — a sign passlib is aging; see the troubleshooting note).


Common mistakes and troubleshooting

Symptom / mistake Cause Fix
Passwords stored as plaintext / MD5 / SHA “One-way is enough” — it isn’t; fast hashes crack in minutes bcrypt/argon2 via passlib; slow + salted is the point
ValueError: password cannot be longer than 72 bytes at startup bcrypt 5.x breaks passlib 1.7.4’s backend self-test Pin bcrypt==4.0.1 (or use argon2, or pwdlib)
A password or hash appears in a log logger.info(f"login {payload}") / logged the row Never log credentials; log a user id, not the body
Secret / card number readable in a JWT JWT is signed, not encrypted — payload is public base64 Only put identifiers in the token; use JWE to hide data
Token never expires; a leaked one works forever No exp claim set Always set exp; keep it short (15–30 min) + refresh
Signing key hardcoded or committed to git Anyone with the repo can forge any token os.environ["SECRET_KEY"]; .env in .gitignore
Signing key too weak ("secret", "dev") Brute-forced / guessed → forged tokens secrets.token_hex(32) — ≥32 random bytes
401 where you expected 403 (or vice-versa) Confusing authn (who) with authz (what) 401 = not authenticated; 403 = authenticated, not allowed
TimeoutError: QueuePool limit ... reached under load DB session opened per request but never closed Depends(get_db) with try/yield/finally: db.close()
Password hash / internal field ships in the response Returned the raw ORM User with no response_model response_model=UserPublic; keep secrets out of the out-schema
Server slow under load, no error Sync DB call inside async def (Part 1’s trap) Plain def handler (threadpool), or an async driver
Browser can’t call the API; curl works Missing/incorrect CORS CORSMiddleware with your front-end origin
SQL injection possible Bypassed the ORM with an f-string query Use the ORM / parameterised queries — never f-string input
Passwords/tokens sniffable on the network Serving plain HTTP in production Terminate HTTPS/TLS at a reverse proxy
Token in the URL (?token=...) or in logs Query strings land in logs, history, referrers Send the token in the Authorization header only
Form data requires "python-multipart" at startup OAuth2PasswordRequestForm needs it pip install python-multipart
ImportError/AttributeError on EmailStr email-validator not installed pip install "pydantic[email]"

Three are worth more than a row.

Storing a password the wrong way is the mistake that ruins users, not just your afternoon. Every other bug here is contained — a 500, a leaked field, a slow endpoint. This one detonates when your database leaks, which is a when: plaintext means every account is instantly compromised, and because people reuse passwords, so are their accounts on other sites. A fast hash (MD5, SHA-256) is barely better — the attacker just runs a GPU dictionary attack and cracks everything weak. There is exactly one acceptable answer: a slow, salted, adaptive hash (bcrypt or argon2) behind passlib, one pwd.hash() on the way in and one pwd.verify() on the way out. It is less code than doing it wrong. There is no scenario, no deadline, no “we’ll fix it later” that justifies plaintext.

“Signed, not encrypted” is the JWT misconception that leaks data. The word “token” makes people imagine something opaque and secret, so they stash a session’s worth of sensitive data in the payload — internal ids, an email, once memorably a password. But a JWT payload is base64, publicly readable by anyone holding the token (we decoded one with no key at all). The signature only proves integrity and origin — that the server minted it and nobody altered it — not confidentiality. The rule that follows is simple: a JWT may carry identifiers you’d be fine printing on a billboard (a user id, a role), and nothing else. If you need the contents hidden, that’s a different, encrypted token (JWE), or keep the data server-side and put only a reference in the token.

The passlib/bcrypt version crash will bite you specifically. passlib 1.7.4 is from 2020 and unmaintained, and the modern bcrypt (5.x) removed the internal __about__ attribute passlib probes for and turned bcrypt’s silent 72-byte truncation into a hard ValueError — which fires inside passlib’s own backend self-test, so you get ValueError: password cannot be longer than 72 bytes before you’ve hashed anything of your own. The pragmatic fix, and what this lesson pins, is bcrypt==4.0.1, the last release passlib is happy with. The forward-looking fix is to move off passlib: hash with the bcrypt or argon2-cffi library directly, or use the maintained pwdlib (which FastAPI’s own docs have shifted toward). Know this, because you’ll pip install "passlib[bcrypt]", get bcrypt 5, and hit a wall the first time you hash — and the traceback points into passlib’s internals, not your code.


Cheat-sheet

Database (SQLAlchemy + FastAPI) What it does
create_engine(url, connect_args=...) The connection pool + dialect
sessionmaker(bind=engine) A factory for Session objects
def get_db(): db=...; try: yield db; finally: db.close() ✅ Per-request session dependency
db: Annotated[Session, Depends(get_db)] Inject the session into a handler
db.scalar(select(User).where(...)) Query one object (or None)
db.add(x); db.commit(); db.refresh(x) Insert, persist, reload the id
Base.metadata.create_all(engine) Create tables (demos; use Alembic for real)
pool_pre_ping=True ✅ Drop dead connections before use
Passwords + JWT What it does
CryptContext(schemes=["bcrypt"]) The hashing context
pwd.hash(plain) ✅ Salted, slow hash (store this)
pwd.verify(plain, hashed) Check a password → True/False
jwt.encode({...,"exp":...}, KEY, algorithm="HS256") Sign a token
jwt.decode(tok, KEY, algorithms=["HS256"]) ✅ Verify signature and expiry
OAuth2PasswordBearer(tokenUrl="token") Read the Bearer header + docs auth
OAuth2PasswordRequestForm Parse the /token login form
Depends(get_current_user) Protect a route (401 if invalid)
secrets.token_hex(32) Generate a strong signing key
Status code Return it when
201 Registered a new user
401 Not authenticated (no/bad/expired token, wrong password)
403 Authenticated but not allowed (wrong role)
409 Duplicate (email already registered)
422 Validation failed (bad email, short password)
Run it Command
Dev uvicorn app:app --reload
Prod gunicorn app:app -k uvicorn.workers.UvicornWorker -w 4
Signing key python -c "import secrets; print(secrets.token_hex(32))"
Migrations alembic revision --autogenerate -m msgalembic upgrade head

Interview and exam questions

Q: Why must the database session be a per-request dependency and not a module-level global? A: A single global Session shared across concurrent requests shares one transaction and connection: one request’s uncommitted writes become visible to another, a rollback in one wipes another’s work, and Session isn’t thread-safe — so you get intermittent data corruption under load. The correct pattern is a Depends(get_db) generator dependency that opens a Session per request, yields it, and closes it in a finally. That gives each request its own isolated transaction and guarantees the connection returns to the pool even on error. The two failure modes it prevents: a shared global (corruption) and a per-request session that’s never closed (a connection leak that exhausts the pool and raises QueuePool limit reached).

Q: Why hash passwords with bcrypt instead of SHA-256, and what does “salted” and “slow” buy you? A: SHA-256 is a fast hash — a GPU computes billions per second — so if your hash database leaks, an attacker cracks weak passwords in minutes via a dictionary attack. bcrypt is deliberately slow (a tunable cost factor; measured here at ~226 ms vs SHA-256’s ~0.0004 ms, roughly 600,000x slower), which is invisible for a single login but makes mass cracking infeasible. Salted means each hash includes a random salt, so identical passwords produce different hashes (verified: two hashes of the same password differ) and precomputed rainbow tables are useless. bcrypt stores the salt and cost inside the $2b$12$... string, so verify needs only the stored hash. argon2 is the modern first choice (memory-hard); the cardinal sins are plaintext and any fast hash.

Q: A JWT is “signed but not encrypted.” What does that mean for what you put in it? A: The signature proves integrity and origin — that a holder of the secret key minted the token and nobody altered it — but the payload is just base64-encoded JSON, readable by anyone who has the token without the key (demonstrated by decoding one with a plain split(".") + base64). So a JWT provides authentication, not confidentiality. You may put non-secret identifiers in it — a user id (sub), a role, an email — and you must never put secrets: passwords, card numbers, API keys. If you tamper with the payload (say, flip role to admin), the signature no longer verifies and jwt.decode rejects it, because you can’t re-sign without the key. To actually hide contents you need an encrypted token (JWE).

Q: What’s the difference between 401 and 403, and when does each fire in a token-based API? A: 401 Unauthorized means not authenticated — we don’t know who you are: no token, a malformed or tampered token, an expired token, or a wrong password at login. 403 Forbidden means authenticated but not authorized — we know exactly who you are, but you’re not allowed to do this: the right user with the wrong role, or accessing someone else’s object. In the lab, /me with no token and /me with a tampered token both return 401 (authentication failed), while /admin with a valid role=user token returns 403 (authenticated, but the role is insufficient). Returning the wrong code matters: 401 invites a retry-after-login, 403 says don’t bother; swapping them either loops legitimate clients or reveals a resource’s existence to strangers.

Q: Walk through the OAuth2 password flow end to end in FastAPI. A: (1) The client POSTs username and password as form data to /token. (2) The login handler looks the user up and calls verify_password(submitted, user.hashed_password) against the stored bcrypt hash — on mismatch, 401. (3) On success it signs a JWT with create_access_token, embedding sub, role, and a short exp, and returns {access_token, token_type: "bearer"}. (4) The client stores the token and sends it on later requests as Authorization: Bearer <jwt>. (5) Any protected route declares user: CurrentUser (i.e. Depends(get_current_user)); that dependency extracts the token via OAuth2PasswordBearer, jwt.decodes it (verifying signature and expiry), loads the user from the DB, and raises 401 on anything wrong. (6) The handler runs with the authenticated user, and response_model=UserPublic strips the hash on the way out. The auth logic lives once in the dependency and protects any route by adding one parameter.

Q: How do you keep a password hash from ever reaching the client? A: Keep the ORM model and the response schema separate. The SQLAlchemy User has hashed_password; the pydantic UserPublic (the response_model) has only id, email, role. Because the route declares response_model=UserPublic, the handler can return user (the whole row) and FastAPI serializes only the schema’s fields — the hash is stripped even though the handler returned it. from_attributes=True (v1’s orm_mode) lets pydantic build the schema by reading attributes off the ORM object. The anti-pattern is returning a raw ORM object or full dict with no response_model, which ships every column including the hash — the most common Python API leak, and a code-review red flag.

Q: Your app crashes with ValueError: password cannot be longer than 72 bytes the first time it hashes. Why? A: It’s the passlib 1.7.4 + modern-bcrypt incompatibility. passlib is unmaintained (2020), and bcrypt 5.x both removed the __about__ attribute passlib reads for its version and changed bcrypt’s old silent 72-byte truncation into a hard ValueError. passlib’s backend self-test hashes a 73-byte probe string, which now raises — so the error fires inside passlib before your own code hashes anything. Fixes: pin bcrypt==4.0.1 (the pragmatic one, and what avoids the crash immediately), or move off passlib entirely — hash with the bcrypt/argon2-cffi libraries directly, or use the maintained pwdlib. Separately, remember bcrypt genuinely ignores everything past 72 bytes, so cap password length or use argon2 for long passphrases.

Q: Sessions vs tokens — when would you choose each? A: Session-based auth stores a record server-side and hands the client an opaque session id in a cookie; the server looks it up per request. Token-based (JWT) auth stores nothing server-side — the signed token carries the identity, and any server with the key verifies it. Sessions give instant revocation (delete the record) but need a shared store (Redis) to scale across servers and are cookie-bound (CSRF-prone, awkward for mobile). Tokens are stateless — trivial to scale horizontally and natural for APIs/mobile via the Authorization header — but can’t be un-issued, so a stolen token works until it expires (hence short expiries + refresh tokens, and sometimes a denylist). Rule of thumb: server-rendered web app → sessions; JSON API, microservices, mobile backend → tokens.

Q: Name five production security measures beyond authentication, and tie each to a defence. A: (1) HTTPS/TLS terminated at a reverse proxy — otherwise tokens and passwords cross the wire in cleartext. (2) Secrets from the environment / a secret manager, never in code — a committed signing key lets anyone forge tokens; generate it with secrets.token_hex(32). (3) Rate limiting on /token (proxy or slowapi) — turns brute-force from billions of guesses into a few per minute. (4) CORS locked to real origins — stops malicious sites calling your API from a browser. (5) Input validation via pydantic and parameterised/ORM queries — closes the injection classes (422 on bad input; SQL text and values travel separately so injection can’t happen). Bonus: object-level authorization (BOLA / OWASP API1) — check ownership in the handler, because no framework stops a logged-in user from changing /orders/123 to /orders/124.

Q (coding): Add a require_role dependency so only admins can call /admin, returning 403 otherwise. A: A dependency factory that runs after get_current_user and checks the role:

def require_role(role: str):
    def checker(user: CurrentUser) -> User:           # CurrentUser = Depends(get_current_user)
        if user.role != role:
            raise HTTPException(403, f"Requires role '{role}'")   # 403, not 401
        return user
    return checker

@app.get("/admin", response_model=UserPublic)
def admin_only(user: Annotated[User, Depends(require_role("admin"))]):
    return user

require_role("admin") returns a dependency; because its inner checker depends on CurrentUser, FastAPI first authenticates (401 if the token’s bad) and then authorizes (403 if the role’s wrong). This is the authn/authz split in code: identity first, permission second, two different status codes.

Q (coding): Write the get_current_user dependency that turns a Bearer token into a user or a 401. A:

def get_current_user(token: Annotated[str, Depends(oauth2_scheme)], db: Db) -> User:
    exc = HTTPException(401, "Could not validate credentials", {"WWW-Authenticate": "Bearer"})
    try:
        email = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]).get("sub")
        if email is None:
            raise exc
    except JWTError:                 # bad signature, expired, malformed → all 401
        raise exc
    user = db.scalar(select(User).where(User.email == email))
    if user is None:
        raise exc
    return user

Points tested: reading the token via OAuth2PasswordBearer, jwt.decode verifying signature and expiry (with algorithms=[...] pinned to block alg: none), mapping every failure — bad signature, expiry, unknown user — to a single 401 with a WWW-Authenticate: Bearer header, and pulling the DB session through the same dependency injection. Any route protects itself by declaring user: Annotated[User, Depends(get_current_user)].


Key takeaways

pythonfastapisqlalchemyauthenticationjwtoauth2bcryptpasslibsecurityormpydanticproductiondockeradvanced
Need this built for real?

Vinod is a Senior Cloud Architect (22+ yrs) — available for Azure / AWS / GCP architecture, landing zones, and migrations.

Work with me

Comments