FastAPI Tutorial — From Zero to Production with Diagrams and Code

The Story That Explains FastAPI 🛂FastAPI समझाने वाली कहानी 🛂一个讲清 FastAPI 的故事 🛂

FastAPI is a modern web framework for building APIs with Python, based on standard type hints. It sits on top of Starlette (the web parts) and Pydantic (the data parts). You get async support, automatic validation and interactive docs — all from the function you would write anyway. Current versions need Python 3.10 or newer.

FastAPI Python में APIs बनाने का आधुनिक web framework है, जो सामान्य type hints पर आधारित है। यह Starlette (web वाला हिस्सा) और Pydantic (data वाला हिस्सा) के ऊपर बना है। आपको async support, अपने-आप validation और interactive docs मिलते हैं — उसी function से जो आप वैसे भी लिखते। मौजूदा versions को Python 3.10 या नया चाहिए।

FastAPI 是一个基于标准类型提示、用 Python 构建 API 的现代 Web 框架。它建立在 Starlette（Web 部分）和 Pydantic（数据部分）之上。你会得到异步支持、自动校验和交互式文档 — 全部来自你本来就要写的那个函数。当前版本需要 Python 3.10 或更新版本。

Installation and Your First API 🚀Installation और आपका पहला API 🚀安装与第一个 API 🚀

You need Python 3.10+ and two things: fastapi (the framework) and an ASGI server such as uvicorn that runs it.

आपको Python 3.10+ और दो चीज़ें चाहिए: fastapi (framework) और uvicorn जैसा ASGI server जो उसे चलाता है।

你需要 Python 3.10+ 以及两样东西：fastapi（框架）和运行它的 ASGI 服务器，例如 uvicorn。

# Recommended: FastAPI + Uvicorn + the fastapi CLI + common extras
pip install "fastapi[standard]"

# Or a minimal install (no "fastapi dev" command)
pip install fastapi uvicorn

Hello World — The Shortest Possible APIHello World — सबसे छोटा APIHello World — 最短的 API

Below is main.py, then a real run: the server log, three requests with curl, and a check that /docs exists.

नीचे main.py है, फिर एक असली run: server log, curl से तीन requests, और /docs मौजूद होने की जाँच।

下面是 main.py，然后是一次真实运行：服务器日志、用 curl 发出的三个请求，以及对 /docs 是否存在的检查。

# file: main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI!"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}

Notice the third request: /items/abc never reached read_item. FastAPI answered with a 422 error that says which field failed and why. You can also start the server with fastapi dev main.py (included in fastapi[standard]).

तीसरी request देखिए: /items/abc कभी read_item तक नहीं पहुँची। FastAPI ने 422 error दिया जो बताता है कौन-सा field fail हुआ और क्यों। आप server fastapi dev main.py से भी चला सकते हैं (fastapi[standard] में शामिल)।

注意第三个请求：/items/abc 根本没有到达 read_item。FastAPI 直接返回 422 错误，说明哪个字段失败以及原因。你也可以用 fastapi dev main.py 启动服务器（包含在 fastapi[standard] 中）。

The Three Pillars of Every Endpoint 🏛️हर Endpoint के तीन खंभे 🏛️每个接口的三大支柱 🏛️

Most data comes into an endpoint from three places. Understanding them is a big part of understanding FastAPI.

Endpoint में ज़्यादातर data तीन जगहों से आता है। इन्हें समझना FastAPI समझने का बड़ा हिस्सा है।

进入接口的数据大多来自三个地方。理解它们，就理解了 FastAPI 的一大半。

🧭 Simulator: Where Does Each Argument Come From?🧭 Simulator: हर Argument कहाँ से आता है?🧭 模拟器：每个参数从哪里来？

Pick a function signature. The coloured tags come from FastAPI's own OpenAPI schema for that endpoint, and the request below was really sent.

एक function signature चुनिए। रंगीन tags उस endpoint के लिए FastAPI के अपने OpenAPI schema से आते हैं, और नीचे की request सच में भेजी गई।

选择一个函数签名。彩色标签来自 FastAPI 为该接口生成的 OpenAPI 模式，下面的请求也是真实发送的。

Type Hints — The Heart of FastAPI ❤️Type Hints — FastAPI का दिल ❤️类型提示 — FastAPI 的核心 ❤️

The Same Function — Step by Step More Typedवही Function — कदम-दर-कदम ज़्यादा Typed同一个函数 — 逐步加上类型

from typing import Annotated
from fastapi import FastAPI, Path, Query

app = FastAPI()

# Level 0 - no hints: item_id arrives as a plain str
@app.get("/v0/items/{item_id}")
def get_item_v0(item_id):
    return {"id": item_id, "type": type(item_id).__name__}

# Level 1 - a type hint: FastAPI now converts AND validates
@app.get("/v1/items/{item_id}")
def get_item_v1(item_id: int):
    return {"id": item_id, "type": type(item_id).__name__}

# Level 2 - constraints with Path and Query
@app.get("/v2/items/{item_id}")
def get_item_v2(
    item_id: Annotated[int, Path(gt=0, le=1000)],
    q: Annotated[str | None, Query(max_length=10)] = None,
):
    return {"id": item_id, "q": q}

# ---- try it (TestClient sends real HTTP requests to the app, no server needed) ----
from fastapi.testclient import TestClient
client = TestClient(app)
def show(r):
    u = r.request.url
    path = u.path + ("?" + u.query.decode() if u.query else "")
    body = r.json() if r.content else "(empty)"
    if r.status_code == 422:
        body = [".".join(map(str, e["loc"])) + ": " + e["msg"] for e in body["detail"]]
    print(f"{r.request.method:6} {path:30} -> {r.status_code} {body}")

show(client.get("/v0/items/42"))
show(client.get("/v1/items/42"))
show(client.get("/v1/items/abc"))
show(client.get("/v2/items/0"))
show(client.get("/v2/items/7?q=far-too-long-text"))

Level 0 handed you the text "42". Level 1 converted it to the number 42 and refused "abc". Level 2 also enforced limits — all before your function ran.

Level 0 ने आपको text "42" दिया। Level 1 ने उसे संख्या 42 में बदला और "abc" को ठुकराया। Level 2 ने सीमाएँ भी लागू कीं — सब आपका function चलने से पहले।

第 0 级给你的是文本 "42"。第 1 级把它转换成数字 42，并拒绝了 "abc"。第 2 级还强制了范围限制 — 全都发生在你的函数运行之前。

🧪 Simulator: Validation Playground🧪 Simulator: Validation Playground🧪 模拟器：校验游乐场

Pick a type and an input. Each result is FastAPI's real answer to GET /check?v=… — including a few surprises.

एक type और एक input चुनिए। हर result GET /check?v=… पर FastAPI का असली जवाब है — कुछ हैरानियों के साथ।

选择一种类型和一个输入。每个结果都是 FastAPI 对 GET /check?v=… 的真实回答 — 其中有几个出人意料。

Visual Diagram — The Request Lifecycle 🔄चित्र — Request का जीवन-चक्र 🔄图解 — 请求生命周期 🔄

Every HTTP request travels through the same pipeline. Knowing each stage helps you put middleware, dependencies and error handling in the right place.

हर HTTP request एक ही pipeline से गुज़रती है। हर stage जानने से आप middleware, dependencies और error handling सही जगह रख पाते हैं।

每个 HTTP 请求都经过同一条流水线。了解每个阶段，有助于你把中间件、依赖和错误处理放在正确的位置。

🔄 Simulator: Request Lifecycle🔄 Simulator: Request Lifecycle🔄 模拟器：请求生命周期

One small app with a middleware, an auth dependency and a response_model. Pick a request and watch how far it gets. The log on the right is what the app really recorded.

एक छोटा app जिसमें एक middleware, एक auth dependency और एक response_model है। एक request चुनिए और देखिए वह कहाँ तक पहुँचती है। दाईं ओर का log वही है जो app ने सच में दर्ज किया।

一个小应用，带一个中间件、一个认证依赖和一个 response_model。选择一个请求，看看它能走多远。右侧的日志是应用真实记录的内容。

HTTP Methods — CRUD in FastAPI 🛠️HTTP Methods — FastAPI में CRUD 🛠️HTTP 方法 — FastAPI 中的 CRUD 🛠️

A REST API maps HTTP methods to data operations: Create, Read, Update, Delete (CRUD). FastAPI has one decorator per method.

REST API HTTP methods को data operations से जोड़ता है: Create, Read, Update, Delete (CRUD)। FastAPI में हर method के लिए एक decorator है।

REST API 把 HTTP 方法映射到数据操作：创建、读取、更新、删除（CRUD）。FastAPI 为每种方法提供一个装饰器。

Complete CRUD API — In-Memory Item Storeपूरा CRUD API — In-Memory Item Store完整的 CRUD API — 内存商品库

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from uuid import UUID, uuid4

app = FastAPI(title="Item API", version="1.0.0")

# -- Models ---------------------------------------------
class ItemIn(BaseModel):
    name:  str   = Field(min_length=1, max_length=80)
    price: float = Field(gt=0)
    stock: int   = Field(default=0, ge=0)

class ItemOut(ItemIn):
    id: UUID

# -- Fake in-memory DB ----------------------------------
DB: dict[UUID, ItemOut] = {}

# -- CREATE ---------------------------------------------
@app.post("/items", response_model=ItemOut, status_code=status.HTTP_201_CREATED)
def create_item(payload: ItemIn):
    new = ItemOut(id=uuid4(), **payload.model_dump())
    DB[new.id] = new
    return new

# -- READ (list) ----------------------------------------
@app.get("/items", response_model=list[ItemOut])
def list_items(skip: int = 0, limit: int = 20):
    return list(DB.values())[skip : skip + limit]

# -- READ (one) -----------------------------------------
@app.get("/items/{item_id}", response_model=ItemOut)
def get_item(item_id: UUID):
    if item_id not in DB:
        raise HTTPException(404, "Item not found")
    return DB[item_id]

# -- UPDATE ---------------------------------------------
@app.put("/items/{item_id}", response_model=ItemOut)
def update_item(item_id: UUID, payload: ItemIn):
    if item_id not in DB:
        raise HTTPException(404, "Item not found")
    DB[item_id] = ItemOut(id=item_id, **payload.model_dump())
    return DB[item_id]

# -- DELETE ---------------------------------------------
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(item_id: UUID):
    if DB.pop(item_id, None) is None:
        raise HTTPException(404, "Item not found")

Trying it with real requestsअसली requests से आज़माना用真实请求试一试

Notice DELETE answered 204 with an empty body, and the bad POST listed both problems at once. The ids are random uuid4() values, so they change on every run.

देखिए DELETE ने ख़ाली body के साथ 204 दिया, और ग़लत POST ने दोनों समस्याएँ एक साथ बताईं। ids random uuid4() values हैं, इसलिए हर run पर बदलते हैं।

注意 DELETE 返回了空响应体的 204，而错误的 POST 一次列出了 两个 问题。id 是随机的 uuid4() 值，每次运行都会变。

🗃️ Simulator: CRUD Lab🗃️ Simulator: CRUD Lab🗃️ 模拟器：CRUD 实验室

Step through ten real requests to the Item API above. The table on the right is the in-memory DB after each request.

ऊपर के Item API पर दस असली requests step-by-step देखिए। दाईं ओर की table हर request के बाद in-memory DB है।

逐步查看发给上面 Item API 的十个真实请求。右侧表格是每个请求之后的内存 DB。

Path and Query Parameters — Deep Dive 🔬Path और Query Parameters — गहराई से 🔬路径参数与查询参数 — 深入 🔬

Path parameter validationPath parameter validation路径参数校验

from enum import Enum
from typing import Annotated
from fastapi import FastAPI, Path

app = FastAPI()

class Category(str, Enum):
    electronics = "electronics"
    books       = "books"
    clothing    = "clothing"

@app.get("/products/{category}/{product_id}")
def get_product(
    category:   Category,                                  # only 3 allowed values
    product_id: Annotated[int, Path(ge=1, le=999999,      # must be 1..999999
                                    title="Product ID",
                                    description="Unique product identifier")],
):
    return {"category": category, "product_id": product_id}

# ---- try it (TestClient sends real HTTP requests to the app, no server needed) ----
from fastapi.testclient import TestClient
client = TestClient(app)
def show(r):
    u = r.request.url
    path = u.path + ("?" + u.query.decode() if u.query else "")
    body = r.json() if r.content else "(empty)"
    if r.status_code == 422:
        body = [".".join(map(str, e["loc"])) + ": " + e["msg"] for e in body["detail"]]
    print(f"{r.request.method:6} {path:30} -> {r.status_code} {body}")

show(client.get("/products/books/42"))
show(client.get("/products/toys/42"))
show(client.get("/products/books/0"))

The Enum limited category to three words — and the error message even lists them. We switched the original Path(..., ge=1) to the Annotated form, to match the advice below.

Enum ने category को तीन शब्दों तक सीमित किया — और error message उन्हें गिनाता भी है। नीचे की सलाह से मेल खाने के लिए हमने मूल Path(..., ge=1) को Annotated रूप में बदला।

Enum 把 category 限定为三个词 — 错误信息甚至把它们都列了出来。我们把原来的 Path(..., ge=1) 改成了 Annotated 写法，与下面的建议保持一致。

Query parameter patternsQuery parameter patterns查询参数的常见写法

from typing import Annotated
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/search")
def search(
    # Required query param
    q:       Annotated[str,  Query(min_length=3, max_length=50)],
    # Optional with default
    limit:   Annotated[int,  Query(ge=1, le=100)] = 10,
    # Boolean flag
    exact:   Annotated[bool, Query()] = False,
    # List - accept ?tag=a&tag=b&tag=c
    tag:     Annotated[list[str] | None, Query()] = None,
    # Regex validation
    sort_by: Annotated[str, Query(pattern="^(name|price|date)$")] = "name",
):
    return {"q": q, "limit": limit, "exact": exact,
            "tag": tag or [], "sort_by": sort_by}

# ---- try it (TestClient sends real HTTP requests to the app, no server needed) ----
from fastapi.testclient import TestClient
client = TestClient(app)
def show(r):
    u = r.request.url
    path = u.path + ("?" + u.query.decode() if u.query else "")
    body = r.json() if r.content else "(empty)"
    if r.status_code == 422:
        body = [".".join(map(str, e["loc"])) + ": " + e["msg"] for e in body["detail"]]
    print(f"{r.request.method:6} {path:30} -> {r.status_code} {body}")

show(client.get("/search?q=laptop"))
show(client.get("/search?q=laptop&limit=5&exact=yes&tag=new&tag=sale&sort_by=price"))
show(client.get("/search?q=tv&limit=500&sort_by=rating"))
show(client.get("/search"))

exact=yes became True, and two tag= values became a list. The third request broke three rules at once, and FastAPI reported all three.

exact=yes True बना, और दो tag= values एक list बनीं। तीसरी request ने एक साथ तीन नियम तोड़े, और FastAPI ने तीनों बताए।

exact=yes 变成了 True，两个 tag= 值变成了列表。第三个请求同时违反了三条规则，FastAPI 把三条都报告了出来。

Request Body with Pydantic Models 📦Pydantic Models के साथ Request Body 📦使用 Pydantic 模型的请求体 📦

Pydantic is FastAPI's validation engine. Declare a class that inherits from BaseModel, and every incoming JSON body is parsed, checked and turned into a typed Python object.

Pydantic FastAPI का validation engine है। BaseModel से बनी class घोषित कीजिए, और हर आने वाली JSON body parse, जाँची और typed Python object में बदली जाती है।

Pydantic 是 FastAPI 的校验引擎。声明一个继承自 BaseModel 的类，每个传入的 JSON 请求体都会被解析、检查，并变成带类型的 Python 对象。

from datetime import date
from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr, HttpUrl, field_validator

app = FastAPI()

class Address(BaseModel):
    street:      str
    city:        str
    postal_code: str = Field(pattern=r"^\d{5,6}$")
    country:     str = "IN"

class UserRegistration(BaseModel):
    username:   str = Field(min_length=3, max_length=20)
    email:      EmailStr                 # needs: pip install "pydantic[email]"
    password:   str = Field(min_length=8)
    birth_date: date
    website:    HttpUrl | None = None
    address:    Address                  # nested model
    tags:       list[str] = []

    @field_validator("password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if not any(c.isupper() for c in v):
            raise ValueError("Password needs at least one uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password needs at least one digit")
        return v

@app.post("/register")
def register(user: UserRegistration):
    # user is fully validated by the time we get here
    return {"status": "created", "user": user.model_dump(exclude={"password"})}

# ---- try it (TestClient sends real HTTP requests to the app, no server needed) ----
from fastapi.testclient import TestClient
client = TestClient(app)
def show(r):
    u = r.request.url
    path = u.path + ("?" + u.query.decode() if u.query else "")
    body = r.json() if r.content else "(empty)"
    if r.status_code == 422:
        body = [".".join(map(str, e["loc"])) + ": " + e["msg"] for e in body["detail"]]
    print(f"{r.request.method:6} {path:30} -> {r.status_code} {body}")

good = {"username": "mohit", "email": "mohit@example.com", "password": "Secret123",
        "birth_date": "1990-07-18", "address": {"street": "1 Mall Rd", "city": "Baddi",
        "postal_code": "173205"}}
show(client.post("/register", json=good))
bad = {"username": "a", "email": "not-an-email", "password": "weak",
       "birth_date": "32-13-2020", "address": {"street": "x", "city": "y", "postal_code": "12"}}
show(client.post("/register", json=bad))
show(client.post("/register", json={**good, "password": "longenough"}))

What happens on invalid inputग़लत input पर क्या होता है输入无效时会发生什么

Response Models and Status Codes 🎯Response Models और Status Codes 🎯响应模型与状态码 🎯

A response_model gives three things at once: it checks your output, filters out fields that should not leave the server, and documents the response.

response_model एक साथ तीन चीज़ें देता है: आपका output जाँचता है, server से बाहर न जाने वाले fields छाँटता है, और response का दस्तावेज़ बनाता है।

response_model 同时带来三件事：检查你的输出、过滤掉不应离开服务器的字段，并为响应生成文档。

from fastapi import FastAPI, status
from pydantic import BaseModel, EmailStr

app = FastAPI()

# Two versions: one with the password, one without
class UserIn(BaseModel):
    username: str
    email:    EmailStr
    password: str                 # client sends this

class UserOut(BaseModel):
    username: str
    email:    EmailStr            # NO password field

@app.post("/users",
          response_model=UserOut,
          status_code=status.HTTP_201_CREATED,
          summary="Register a new user",
          tags=["users"])
def create_user(user: UserIn) -> UserIn:
    # Return the whole UserIn (with password) - FastAPI filters it
    # out because response_model=UserOut has no password field.
    return user

@app.post("/users-unsafe", status_code=201)
def create_user_unsafe(user: UserIn):
    return user                   # no response_model -> everything goes out

# ---- try it (TestClient sends real HTTP requests to the app, no server needed) ----
from fastapi.testclient import TestClient
client = TestClient(app)
def show(r):
    u = r.request.url
    path = u.path + ("?" + u.query.decode() if u.query else "")
    body = r.json() if r.content else "(empty)"
    if r.status_code == 422:
        body = [".".join(map(str, e["loc"])) + ": " + e["msg"] for e in body["detail"]]
    print(f"{r.request.method:6} {path:30} -> {r.status_code} {body}")

show(client.post("/users", json={"username": "ada", "email": "ada@x.io", "password": "S3cret!"}))
show(client.post("/users-unsafe", json={"username": "ada", "email": "ada@x.io", "password": "S3cret!"}))

Same input, two endpoints. With response_model=UserOut the password was removed; without it, the password went straight back to the client.

एक ही input, दो endpoints। response_model=UserOut के साथ password हट गया; उसके बिना password सीधे client को वापस चला गया।

同样的输入，两个接口。有 response_model=UserOut 时密码被去掉了；没有它，密码就直接返回给了客户端。

Dependency Injection — FastAPI's Secret Weapon 🧩Dependency Injection — FastAPI का गुप्त हथियार 🧩依赖注入 — FastAPI 的秘密武器 🧩

from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, Header

app = FastAPI()
CALLS = {"get_current_user": 0}

# -- A simple dependency -------------------------------
def get_pagination(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": min(limit, 100)}

# -- A dependency with a sub-dependency ----------------
def verify_api_key(x_api_key: Annotated[str, Header()]):     # reads header X-Api-Key
    if x_api_key != "secret-key-123":
        raise HTTPException(401, "Invalid API key")
    return x_api_key

def get_current_user(api_key: Annotated[str, Depends(verify_api_key)]):
    CALLS["get_current_user"] += 1                # count how often it really runs
    return {"id": "u_42", "name": "Alice"}

def get_audit(user: Annotated[dict, Depends(get_current_user)]):
    return f"audit for {user['name']}"            # ALSO needs the current user

# -- Yield-based dependency (setup + cleanup) ----------
def get_db():
    db = {"conn": "fake-db-session"}              # real app: SessionLocal()
    print("  db: opened")
    try:
        yield db                                  # handed to the endpoint
    finally:
        print("  db: closed")                     # always runs, even on error

# -- Consuming all of them ------------------------------
@app.get("/orders")
def list_orders(
    pagination: Annotated[dict, Depends(get_pagination)],
    user:       Annotated[dict, Depends(get_current_user)],
    audit:      Annotated[str,  Depends(get_audit)],
    db:         Annotated[dict, Depends(get_db)],
):
    return {"user": user["name"], "page": pagination, "audit": audit}

# ---- try it (TestClient sends real HTTP requests to the app, no server needed) ----
from fastapi.testclient import TestClient
client = TestClient(app)
def show(r):
    u = r.request.url
    path = u.path + ("?" + u.query.decode() if u.query else "")
    body = r.json() if r.content else "(empty)"
    if r.status_code == 422:
        body = [".".join(map(str, e["loc"])) + ": " + e["msg"] for e in body["detail"]]
    print(f"{r.request.method:6} {path:30} -> {r.status_code} {body}")

show(client.get("/orders?limit=500", headers={"X-Api-Key": "secret-key-123"}))
print("get_current_user ran", CALLS["get_current_user"], "time(s) for that request")
show(client.get("/orders", headers={"X-Api-Key": "wrong"}))
show(client.get("/orders"))

Authentication with JWT (OAuth2) 🔐JWT (OAuth2) के साथ Authentication 🔐使用 JWT（OAuth2）认证 🔐

FastAPI includes OAuth2 helpers. Combined with PyJWT (to create and check tokens) and pwdlib (to hash passwords with Argon2), you get a solid login flow in about 60 lines. This is the setup the FastAPI documentation uses today.

FastAPI में OAuth2 helpers शामिल हैं। PyJWT (tokens बनाने और जाँचने के लिए) और pwdlib (Argon2 से passwords hash करने के लिए) के साथ लगभग 60 lines में मज़बूत login flow मिलता है। FastAPI documentation आज यही setup लेता है।

FastAPI 自带 OAuth2 辅助工具。配合 PyJWT（创建和校验令牌）和 pwdlib（用 Argon2 哈希密码），大约 60 行就能写出可靠的登录流程。这正是 FastAPI 文档如今采用的方案。

import os
from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt                                   # pip install pyjwt
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash              # pip install "pwdlib[argon2]"
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

SECRET_KEY = os.environ.get("SECRET_KEY", "dev-only-change-me-" + "x" * 32)
ALGORITHM  = "HS256"
TOKEN_TTL  = timedelta(minutes=30)

password_hash = PasswordHash.recommended()   # Argon2
oauth2 = OAuth2PasswordBearer(tokenUrl="/token")
app = FastAPI()

# Fake user DB - real app: query PostgreSQL etc.
USERS = {"alice": {"username": "alice", "hashed_pw": password_hash.hash("wonderland")}}

def authenticate(username: str, password: str):
    user = USERS.get(username)
    if not user or not password_hash.verify(password, user["hashed_pw"]):
        return None
    return user

def create_token(data: dict, ttl: timedelta = TOKEN_TTL) -> str:
    payload = data.copy()
    payload["exp"] = datetime.now(timezone.utc) + ttl
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token")
def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = authenticate(form.username, form.password)
    if not user:
        raise HTTPException(401, "Wrong username or password")
    return {"access_token": create_token({"sub": user["username"]}), "token_type": "bearer"}

def get_current_user(token: Annotated[str, Depends(oauth2)]):
    creds_exc = HTTPException(status.HTTP_401_UNAUTHORIZED, "Could not validate credentials",
                              headers={"WWW-Authenticate": "Bearer"})
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
    except InvalidTokenError:
        raise creds_exc
    if username is None or username not in USERS:
        raise creds_exc
    return USERS[username]

@app.get("/me")
def read_me(user: Annotated[dict, Depends(get_current_user)]):
    return {"username": user["username"]}

# ---- try it (TestClient sends real HTTP requests to the app, no server needed) ----
from fastapi.testclient import TestClient
client = TestClient(app)
def show(r):
    u = r.request.url
    path = u.path + ("?" + u.query.decode() if u.query else "")
    body = r.json() if r.content else "(empty)"
    if r.status_code == 422:
        body = [".".join(map(str, e["loc"])) + ": " + e["msg"] for e in body["detail"]]
    print(f"{r.request.method:6} {path:30} -> {r.status_code} {body}")

show(client.post("/token", data={"username": "alice", "password": "nope"}))
r = client.post("/token", data={"username": "alice", "password": "wonderland"})
token = r.json()["access_token"]
print("token:", token[:36] + "...", "| parts:", token.count(".") + 1)
show(client.get("/me", headers={"Authorization": f"Bearer {token}"}))
show(client.get("/me"))
old = create_token({"sub": "alice"}, ttl=timedelta(seconds=-1))   # already expired
show(client.get("/me", headers={"Authorization": f"Bearer {old}"}))
show(client.get("/me", headers={"Authorization": f"Bearer {token[:-3]}abc"}))   # tampered
print("stored hash:", USERS["alice"]["hashed_pw"][:30] + "...")

The run shows the whole story: a wrong password gets 401, the right one gets a three-part token, /me works with it, and three bad cases fail — no token, an expired token, and a token whose last characters were changed.

Run पूरी कहानी दिखाता है: ग़लत password पर 401, सही पर तीन हिस्सों वाला token, उसके साथ /me चलता है, और तीन ग़लत मामले fail होते हैं — कोई token नहीं, expired token, और ऐसा token जिसके आख़िरी अक्षर बदले गए।

这次运行展示了完整过程：密码错误得到 401，密码正确得到三段式令牌，用它访问 /me 成功，三种错误情况都失败了 — 没有令牌、令牌过期、令牌末尾字符被篡改。

# The original setup: passlib + bcrypt (bcrypt 5.x installed)
from passlib.context import CryptContext

pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
print(pwd_ctx.hash("wonderland"))

Database Integration with SQLAlchemy 🗄️SQLAlchemy के साथ Database Integration 🗄️与 SQLAlchemy 集成数据库 🗄️

# file: database.py
from sqlalchemy import create_engine, Integer, String, Float
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker

DATABASE_URL = "sqlite:///./app.db"

engine       = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autoflush=False, bind=engine)

class Base(DeclarativeBase):
    pass

class Item(Base):
    __tablename__ = "items"
    id:    Mapped[int]   = mapped_column(Integer, primary_key=True, index=True)
    name:  Mapped[str]   = mapped_column(String, nullable=False)
    price: Mapped[float] = mapped_column(Float, nullable=False)
    stock: Mapped[int]   = mapped_column(Integer, default=0)

Base.metadata.create_all(bind=engine)

# file: main.py
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from pydantic import BaseModel, ConfigDict

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:     yield db
    finally: db.close()

DB = Annotated[Session, Depends(get_db)]

class ItemIn(BaseModel):
    name:  str
    price: float
    stock: int = 0

class ItemOut(ItemIn):
    model_config = ConfigDict(from_attributes=True)   # read ORM objects directly
    id: int

@app.post("/items", response_model=ItemOut, status_code=201)
def create(payload: ItemIn, db: DB):
    item = Item(**payload.model_dump())
    db.add(item)
    db.commit()
    db.refresh(item)          # load the new id from the database
    return item

@app.get("/items/{item_id}", response_model=ItemOut)
def read(item_id: int, db: DB):
    item = db.get(Item, item_id)
    if not item:
        raise HTTPException(404, "Item not found")
    return item

# ---- try it (TestClient sends real HTTP requests to the app, no server needed) ----
from fastapi.testclient import TestClient
client = TestClient(app)
def show(r):
    u = r.request.url
    path = u.path + ("?" + u.query.decode() if u.query else "")
    body = r.json() if r.content else "(empty)"
    if r.status_code == 422:
        body = [".".join(map(str, e["loc"])) + ": " + e["msg"] for e in body["detail"]]
    print(f"{r.request.method:6} {path:30} -> {r.status_code} {body}")

show(client.post("/items", json={"name": "Pen", "price": 1.5, "stock": 40}))
show(client.get("/items/1"))
show(client.get("/items/99"))

Middleware and CORS 🧅Middleware और CORS 🧅中间件与 CORS 🧅

Middleware wraps every request — useful for logging, timing, CORS and compression. Think of an onion: the request goes in through each layer, and the response comes back out through the same layers in reverse.

Middleware हर request को लपेटता है — logging, timing, CORS और compression के लिए उपयोगी। प्याज़ जैसा सोचिए: request हर परत से अंदर जाती है, और response उन्हीं परतों से उल्टे क्रम में बाहर आता है।

中间件包裹着每个请求 — 适合做日志、计时、CORS 和压缩。可以想象成一个洋葱：请求穿过每一层进入，响应再按相反顺序穿过同样的层返回。

import time
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()
LOG = []

# -- CORS - allow these browser origins to call your API -
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# -- GZip - compress large responses -------------------
app.add_middleware(GZipMiddleware, minimum_size=1000)

# -- Custom middlewares (registered in this order) -----
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
    LOG.append("timing: in")
    start    = time.perf_counter()
    response = await call_next(request)
    response.headers["X-Process-Time"] = f"{time.perf_counter() - start:.4f}"
    LOG.append("timing: out")
    return response

@app.middleware("http")
async def log_requests(request: Request, call_next):
    LOG.append("logger: in")
    response = await call_next(request)
    LOG.append("logger: out")
    return response

@app.get("/data")
def data():
    LOG.append("endpoint")
    return {"numbers": list(range(400))}       # a large-ish JSON body

from fastapi.testclient import TestClient
client = TestClient(app)

r = client.get("/data", headers={"Accept-Encoding": "gzip", "Origin": "http://localhost:3000"})
print("order       :", " -> ".join(LOG))
print("status      :", r.status_code)
print("gzip        :", r.headers.get("content-encoding"), "| bytes on the wire:", r.headers.get("content-length"),
      "| JSON size:", len(r.content))
print("CORS header :", r.headers.get("access-control-allow-origin"))
print("timing hdr  :", "X-Process-Time" in r.headers)
r = client.options("/data", headers={"Origin": "https://evil.example",
                                      "Access-Control-Request-Method": "GET"})
print("preflight from evil.example ->", r.status_code, r.text)

The run also shows the other middlewares working: the 1,503-byte JSON travelled as 748 gzip bytes, the allowed origin got its CORS header, and a preflight request from an origin not on the list was refused.

Run बाक़ी middlewares को भी काम करते दिखाता है: 1,503-byte JSON 748 gzip bytes बनकर गया, अनुमति वाले origin को CORS header मिला, और सूची में न होने वाले origin की preflight request ठुकरा दी गई।

这次运行也展示了其他中间件的作用：1,503 字节的 JSON 以 748 字节的 gzip 形式传输，被允许的来源拿到了 CORS 响应头，而来自名单外来源的预检请求被拒绝。

Sync vs Async — When Speed Matters ⚡Sync बनाम Async — जब Speed मायने रखे ⚡同步 vs 异步 — 速度要紧时 ⚡

import asyncio, time
import requests, httpx
from fastapi import FastAPI

app = FastAPI()

# -- SYNC - FastAPI runs it in a thread pool ------------
@app.get("/weather-sync/{city}")
def weather_sync(city: str):
    r = requests.get(f"https://api.example-weather.com/{city}", timeout=5)
    return r.json()

# -- ASYNC - awaits, so the event loop stays free --------
@app.get("/weather-async/{city}")
async def weather_async(city: str):
    async with httpx.AsyncClient(timeout=5) as client:
        r = await client.get(f"https://api.example-weather.com/{city}")
    return r.json()

# -- WRONG - a blocking call inside async def -----------
@app.get("/weather-broken/{city}")
async def weather_broken(city: str):
    r = requests.get(f"https://api.example-weather.com/{city}", timeout=5)  # blocks the loop!
    return r.json()

Those endpoints call an outside service, so here is the same idea measured for real: each endpoint waits 0.2 s, and 50 requests arrive at once.

वे endpoints बाहरी service बुलाते हैं, इसलिए यहाँ वही बात असल में मापी गई है: हर endpoint 0.2 s इंतज़ार करता है, और 50 requests एक साथ आती हैं।

这些接口会调用外部服务，所以下面用真实测量来说明同样的道理：每个接口等待 0.2 秒，50 个请求同时到达。

import asyncio, threading, time
import httpx, uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/sync")                    # def -> runs in the thread pool
def sync_endpoint():
    time.sleep(0.2)                  # pretend: a blocking I/O call (requests, sqlite3)
    return {"ok": True}

@app.get("/async")                   # async def + await -> runs on the event loop
async def async_endpoint():
    await asyncio.sleep(0.2)         # pretend: an awaited I/O call (httpx, asyncpg)
    return {"ok": True}

@app.get("/broken")                  # async def + BLOCKING call
async def broken_endpoint():
    time.sleep(0.2)                  # blocks the whole event loop!
    return {"ok": True}

server = uvicorn.Server(uvicorn.Config(app, port=8012, log_level="warning"))
threading.Thread(target=server.run, daemon=True).start()
time.sleep(1.5)

async def hammer(path, n=50):
    async with httpx.AsyncClient(base_url="http://127.0.0.1:8012", timeout=60) as c:
        t0 = time.perf_counter()
        await asyncio.gather(*(c.get(path) for _ in range(n)))
        return time.perf_counter() - t0

for path in ["/sync", "/async", "/broken"]:
    print(f"50 requests at once to {path:8s}: {asyncio.run(hammer(path)):5.2f} s")
server.should_exit = True

⏱️ Simulator: Concurrency Lab⏱️ Simulator: Concurrency Lab⏱️ 模拟器：并发实验室

Each dot is one request finishing, placed at its real finish time. Each endpoint waits 0.2 s. Change how many requests arrive at once.

हर बिंदु एक पूरी हुई request है, अपने असली ख़त्म होने के समय पर। हर endpoint 0.2 s इंतज़ार करता है। बदलिए कि एक साथ कितनी requests आती हैं।

每个圆点代表一个请求完成，位于它真实完成的时间点。每个接口等待 0.2 秒。改变同时到达的请求数试试。

Testing FastAPI Applications 🧪FastAPI Applications की Testing 🧪测试 FastAPI 应用 🧪

FastAPI includes TestClient, built on httpx. It sends real requests to your app in memory — no server needed. Pair it with pytest for a fast test loop. Below, main.py is the CRUD app from Section 06.

FastAPI में httpx पर बना TestClient शामिल है। यह memory में आपके app को असली requests भेजता है — server की ज़रूरत नहीं। तेज़ test loop के लिए इसे pytest के साथ लीजिए। नीचे main.py Section 06 का CRUD app है।

FastAPI 自带基于 httpx 的 TestClient。它在内存中向你的应用发送真实请求 — 不需要服务器。配合 pytest，测试循环又快又稳。下面的 main.py 就是第 06 节的 CRUD 应用。

# file: test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_item():
    payload = {"name": "Notebook", "price": 2.99, "stock": 100}
    r = client.post("/items", json=payload)
    assert r.status_code == 201
    data = r.json()
    assert data["name"]  == "Notebook"
    assert data["price"] == 2.99
    assert "id" in data

def test_create_item_validation_error():
    payload = {"name": "", "price": -1}      # invalid
    r = client.post("/items", json=payload)
    assert r.status_code == 422
    errors = r.json()["detail"]
    assert len(errors) >= 2                  # name + price both wrong

def test_get_missing_item_returns_404():
    r = client.get("/items/00000000-0000-0000-0000-000000000000")
    assert r.status_code == 404
    assert r.json()["detail"] == "Item not found"

FastAPI vs Flask vs Django REST ⚖️FastAPI बनाम Flask बनाम Django REST ⚖️FastAPI vs Flask vs Django REST ⚖️

Deployment — From Laptop to Production 🚢Deployment — Laptop से Production तक 🚢部署 — 从笔记本到生产环境 🚢

Production server: Gunicorn + Uvicorn workersProduction server: Gunicorn + Uvicorn workers生产服务器：Gunicorn + Uvicorn worker

# A tiny app to deploy
from fastapi import FastAPI
app = FastAPI()

@app.get("/health")
def health():
    return {"status": "ok"}

DockerfileDockerfileDockerfile

FROM python:3.12-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["fastapi", "run", "main.py", "--port", "8000", "--workers", "4"]

When to Use FastAPI 🧭FastAPI कब लें 🧭何时使用 FastAPI 🧭

Bug Doctor, Golden Rules and Quiz 🏆Bug Doctor, सुनहरे नियम और Quiz 🏆Bug 医生、黄金法则与小测验 🏆

🩺 Simulator: Bug Doctor🩺 Simulator: Bug Doctor🩺 模拟器：Bug 医生

Six real FastAPI mistakes. Buggy code on the left, fixed code on the right, changed lines marked. Every output is from a real run.

छह असली FastAPI ग़लतियाँ। बाईं ओर buggy code, दाईं ओर ठीक किया हुआ, बदली lines चिह्नित। हर output असली run से है।

六个真实的 FastAPI 错误。左边是有 bug 的代码，右边是修正后的代码，改动的行已标出。每个输出都来自真实运行。

Quick Quiz — 8 Questionsछोटा Quiz — 8 सवाल小测验 — 8 道题