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:
- The database session is per-request, and it arrives through
Depends. Not a global you share across requests (that corrupts transactions and isn’t thread-safe), not one you open and forget to close (that leaks connections until the pool times out). One session per request, yielded in and cleaned up after — the exactDependspattern Part 1 pointed at, now holding a real SQLAlchemySession. - A password is never stored. Ever. You store a slow, salted hash, and when someone logs in you hash what they typed and compare. If your database is stolen — and assume it will be — the attacker gets hashes they can’t feasibly reverse, not a spreadsheet of everyone’s password. Getting this wrong is the single most damaging mistake in this lesson, and it’s one function call to get it right.
- A token proves identity without proving secrecy. A JWT is signed, so the server can trust it came from the server and wasn’t altered — but it is not encrypted, so anyone can read what’s inside. Put a secret in it and you’ve published the secret. Give it no expiry and a leaked token is valid forever.
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 (authn) — who are you? Proving identity. Logging in with a password, presenting a valid token. Failure is 401 Unauthorized (a misnomer — it really means unauthenticated).
- Authorization (authz) — what are you allowed to do? Checking permissions. This user is an admin; that user owns this document. Failure is 403 Forbidden.
| 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-based (stateful). On login the server creates a session record (in memory, Redis, or a database) and hands the client a random, opaque session ID in a cookie. Every later request sends the cookie; the server looks the ID up to find who you are. The state lives on the server; the cookie is just a claim check.
- Token-based (stateless). On login the server signs a token (a JWT) containing the user’s identity and hands it back. The client sends it on every request in the
Authorizationheader — exactly the Bearer-token header you sent as a caller in the requests lesson, now arriving at your server. The server verifies the signature — no lookup, no server-side session store. The state lives in the token.
| 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:
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:
- REGISTER → 201 returns
{id, email, role}— no password, no hash.response_model=UserPublicdid its job. - The stored value is a bcrypt hash (
$2b$12$...), andpassword in hash? -> False— the plaintexts3cret-pa55phraseis nowhere in the database. If this DB leaked, the attacker gets a hash, not the password. - LOGIN → 200 returns a real JWT, whose header and payload are readable (
sub,role,exp) — signed, not secret. - ME (token) → 200 — the valid token authenticates; ME (no token) → 401
Not authenticated(the OAuth2 scheme rejects the missing header). - LOGIN (bad pw) → 401 —
verify_passwordfailed against the hash; the wrong password never gets a token. - ME (tampered) → 401
Could not validate credentials— flipping two characters of the signature broke verification. The token can’t be forged without the key. - ADMIN (role=user) → 403 — the token is valid (authenticated) but the role is wrong (not authorized). 401 vs 403, live.
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 msg → alembic 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
- The database session is a per-request
Depends(get_db)that yields and closes. Never a global (shared transactions corrupt data, and it’s not thread-safe) and never left open (leaked connections exhaust the pool). One session per request, cleaned up infinally— the exactDependsseam Part 1 promised, now holding a real SQLAlchemySession. Use plaindefhandlers with a sync driver to avoid the event-loop trap. - The ORM model is not the API schema — and that gap is a security boundary.
Userstoreshashed_password;UserCreate/UserPublicnever expose it. Withresponse_model=UserPublicandfrom_attributes=True, a handler canreturn userand the hash is stripped. Returning a raw ORM object with noresponse_modelis the most common way a Python API leaks secrets. - Never store a password — store a slow, salted hash, and it’s one line.
pwd.hash()in,pwd.verify()out, via bcrypt or argon2. Fast hashes (MD5/SHA) are the wrong tool — measured ~600,000x faster, i.e. crackable — and plaintext is indefensible. bcrypt caps at 72 bytes; ⚠️ pinbcrypt==4.0.1for passlib 1.7.4. Never log or return a password or its hash. - A JWT is signed, not encrypted: verify it, but never hide secrets in it.
jwt.decodechecks signature and expiry (pinalgorithms=[...]), so tampered and expired tokens are rejected — but anyone can read the payload without the key. Put only identifiers (sub,role) in it, always set a shortexp, and keep the signing key strong and in the environment. Noexpmeans a leaked token is a permanent backdoor. - Authentication is who; authorization is what — 401 vs 403.
get_current_userauthenticates (401 on a missing/bad/expired token);require_role/ownership authorizes (403 when the known user isn’t allowed). The lab proves all three live: no token → 401, tampered token → 401, valid token wrong role → 403. Confusing the codes leaks information or loops clients. - The OAuth2 password flow is
/token→ JWT →Depends(get_current_user)on every protected route. Login verifies the hash and signs a token; each request re-presents it asAuthorization: Bearer; the dependency verifies and injects the user. Written once, it protects any route by adding one parameter — and powers the/docsAuthorize button for free. - Production is a checklist, and Python’s job is small. HTTPS at a reverse proxy, secrets from the environment (
secrets.token_hex(32)),gunicorn+ uvicorn workers (never--reload), a real database with Alembic migrations, CORS locked down, rate-limited login, a/healthroute, structured logs that never print tokens, and a Dockerfile. Managed hosting (Railway, Render, Fly, Cloud Run) hands you most of it — start there, and you’ve shipped a secure API.