Mini Project: Online Store API
35 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)You now have routes, validation, a database, background tasks and caches. This project joins them in one small API: a shop with products and orders. The rule that matters is simple: never sell stock you do not have.
Plan before you code
Order line is one row of an order: a product, a quantity, and the unit price at the time of the sale. Saving the price on the line keeps old orders correct when the product price changes later.
Write down the data and the routes first. A product has a name, a price and a stock count. An order has a customer and lines, and each line says which product and how many.
- Tables:
products,orders,order_lines. POST /productsandGET /productsmanage the catalogue.POST /orderschecks stock, reduces it and saves the order, or fails as a whole.GET /orders/{id}shows a saved order with its lines.
Models, schemas and the session dependency
Session dependency is a function with yield that FastAPI calls for each request: it opens a database session, hands it to the route, and closes it after the response, even after an error.
SQLAlchemy classes describe the tables. Pydantic classes describe what the API accepts and returns. Keep them apart. A dependency with yield opens one session per request and closes it afterwards.
def get_session(): with SessionLocal() as session: yield session
@app.get("/products", response_model=list[ProductOut])def list_products(session: Session = Depends(get_session)): query = select(Product).order_by(Product.id).limit(50) return session.scalars(query).all()Example 1 · Models, schemas and product routes
from fastapi import Depends, FastAPI, HTTPExceptionfrom fastapi.testclient import TestClientfrom pydantic import BaseModel, ConfigDict, Fieldfrom sqlalchemy import ForeignKey, create_engine, select, updatefrom sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship, sessionmakerfrom sqlalchemy.pool import StaticPool
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)SessionLocal = sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase): pass
class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] price_cents: Mapped[int] stock: Mapped[int]
class Order(Base): __tablename__ = "orders" id: Mapped[int] = mapped_column(primary_key=True) customer: Mapped[str] total_cents: Mapped[int] = mapped_column(default=0) lines: Mapped[list["OrderLine"]] = relationship(cascade="all, delete-orphan")
class OrderLine(Base): __tablename__ = "order_lines" id: Mapped[int] = mapped_column(primary_key=True) order_id: Mapped[int] = mapped_column(ForeignKey("orders.id")) product_id: Mapped[int] = mapped_column(ForeignKey("products.id")) quantity: Mapped[int] unit_price_cents: Mapped[int]
Base.metadata.create_all(engine)
class ProductIn(BaseModel): name: str = Field(min_length=1) price_cents: int = Field(gt=0) stock: int = Field(ge=0)
class ProductOut(ProductIn): model_config = ConfigDict(from_attributes=True) id: int
class LineIn(BaseModel): product_id: int quantity: int = Field(gt=0)
class OrderIn(BaseModel): customer: str = Field(min_length=1) lines: list[LineIn] = Field(min_length=1)
class LineOut(LineIn): model_config = ConfigDict(from_attributes=True) unit_price_cents: int
class OrderOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int customer: str total_cents: int lines: list[LineOut]
def get_session(): with SessionLocal() as session: yield session
app = FastAPI()
@app.post("/products", response_model=ProductOut, status_code=201)def add_product(body: ProductIn, session: Session = Depends(get_session)): product = Product(**body.model_dump()) session.add(product) session.commit() return product
@app.get("/products", response_model=list[ProductOut])def list_products(skip: int = 0, limit: int = 50, session: Session = Depends(get_session)): query = select(Product).order_by(Product.id).offset(skip).limit(limit) return session.scalars(query).all()
client = TestClient(app)
res = client.post("/products", json={"name": "Pen", "price_cents": 250, "stock": 10})print(res.status_code, res.json())client.post("/products", json={"name": "Notebook", "price_cents": 1200, "stock": 3})print([p["name"] for p in client.get("/products").json()])bad = client.post("/products", json={"name": "", "price_cents": -5, "stock": 1})print(bad.status_code, sorted(e["loc"][-1] for e in bad.json()["detail"]))Output
201 {'name': 'Pen', 'price_cents': 250, 'stock': 10, 'id': 1}
['Pen', 'Notebook']
422 ['name', 'price_cents']What this code does
ProductOuthasfrom_attributes=True, so FastAPI can turn the SQLAlchemy object into JSON.- The id is made by the database. Validation errors name every field that broke a rule.
StaticPoolmakes all requests share one in-memory database, which is what a TestClient needs.
The route never creates or closes a session itself. Money is stored as whole cents (price_cents), because decimal fractions of floats cause rounding errors.
Orders and the stock check
Atomic update is a single database statement that checks and changes data in one step, so no other request can slip in between the check and the change.
Reading the stock and then writing it back is risky: two requests can both read 1 and both sell it. Instead let the database check and reduce in one statement, and look at how many rows it changed.
taken = session.execute( update(Product) .where(Product.id == line.product_id, Product.stock >= line.quantity) .values(stock=Product.stock - line.quantity))if taken.rowcount == 0: session.rollback() raise HTTPException(status_code=409, detail="Not enough stock")Example 2 · The full API: a walk-through
from fastapi import Depends, FastAPI, HTTPExceptionfrom fastapi.testclient import TestClientfrom pydantic import BaseModel, ConfigDict, Fieldfrom sqlalchemy import ForeignKey, create_engine, select, updatefrom sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship, sessionmakerfrom sqlalchemy.pool import StaticPool
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)SessionLocal = sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase): pass
class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] price_cents: Mapped[int] stock: Mapped[int]
class Order(Base): __tablename__ = "orders" id: Mapped[int] = mapped_column(primary_key=True) customer: Mapped[str] total_cents: Mapped[int] = mapped_column(default=0) lines: Mapped[list["OrderLine"]] = relationship(cascade="all, delete-orphan")
class OrderLine(Base): __tablename__ = "order_lines" id: Mapped[int] = mapped_column(primary_key=True) order_id: Mapped[int] = mapped_column(ForeignKey("orders.id")) product_id: Mapped[int] = mapped_column(ForeignKey("products.id")) quantity: Mapped[int] unit_price_cents: Mapped[int]
Base.metadata.create_all(engine)
class ProductIn(BaseModel): name: str = Field(min_length=1) price_cents: int = Field(gt=0) stock: int = Field(ge=0)
class ProductOut(ProductIn): model_config = ConfigDict(from_attributes=True) id: int
class LineIn(BaseModel): product_id: int quantity: int = Field(gt=0)
class OrderIn(BaseModel): customer: str = Field(min_length=1) lines: list[LineIn] = Field(min_length=1)
class LineOut(LineIn): model_config = ConfigDict(from_attributes=True) unit_price_cents: int
class OrderOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int customer: str total_cents: int lines: list[LineOut]
def get_session(): with SessionLocal() as session: yield session
app = FastAPI()
@app.post("/products", response_model=ProductOut, status_code=201)def add_product(body: ProductIn, session: Session = Depends(get_session)): product = Product(**body.model_dump()) session.add(product) session.commit() return product
@app.get("/products", response_model=list[ProductOut])def list_products(skip: int = 0, limit: int = 50, session: Session = Depends(get_session)): query = select(Product).order_by(Product.id).offset(skip).limit(limit) return session.scalars(query).all()
@app.post("/orders", response_model=OrderOut, status_code=201)def place_order(body: OrderIn, session: Session = Depends(get_session)): order = Order(customer=body.customer, total_cents=0) for line in body.lines: product = session.get(Product, line.product_id) if product is None: session.rollback() raise HTTPException(status_code=404, detail=f"Product {line.product_id} not found") taken = session.execute( update(Product) .where(Product.id == line.product_id, Product.stock >= line.quantity) .values(stock=Product.stock - line.quantity) ) if taken.rowcount == 0: session.rollback() raise HTTPException(status_code=409, detail=f"Not enough stock for {product.name}") order.lines.append(OrderLine(product_id=product.id, quantity=line.quantity, unit_price_cents=product.price_cents)) order.total_cents += product.price_cents * line.quantity session.add(order) session.commit() return order
@app.get("/orders/{order_id}", response_model=OrderOut)def get_order(order_id: int, session: Session = Depends(get_session)): order = session.get(Order, order_id) if order is None: raise HTTPException(status_code=404, detail="Order not found") return order
client = TestClient(app)
def stock(product_id): return next(p["stock"] for p in client.get("/products").json() if p["id"] == product_id)
client.post("/products", json={"name": "Pen", "price_cents": 250, "stock": 10})client.post("/products", json={"name": "Notebook", "price_cents": 1200, "stock": 3})
res = client.post("/orders", json={"customer": "Asha", "lines": [{"product_id": 1, "quantity": 4}, {"product_id": 2, "quantity": 1}]})print(res.status_code, res.json())print("stock now:", stock(1), stock(2))
res = client.post("/orders", json={"customer": "Ravi", "lines": [{"product_id": 2, "quantity": 5}]})print(res.status_code, res.json())print(client.post("/orders", json={"customer": "Ravi", "lines": [{"product_id": 99, "quantity": 1}]}).json())print(client.get("/orders/1").json()["total_cents"], client.get("/orders/7").status_code)Output
201 {'id': 1, 'customer': 'Asha', 'total_cents': 2200, 'lines': [{'product_id': 1, 'quantity': 4, 'unit_price_cents': 250}, {'product_id': 2, 'quantity': 1, 'unit_price_cents': 1200}]}
stock now: 6 2
409 {'detail': 'Not enough stock for Notebook'}
{'detail': 'Product 99 not found'}
2200 404What this code does
- The first order takes 4 pens and 1 notebook. The total is 4 x 250 + 1200 = 2200 cents. The line stores the price at the time.
- The second order wants 5 notebooks but only 2 are left:
409, and the stock is unchanged. - An unknown product gives
404. A missing order gives404.
If the stock is too low, the WHERE matches no row, rowcount is 0 and we answer 409 Conflict. Nothing was reduced.
All or nothing, and testing the whole API
Transaction is a group of database changes that are saved together with commit or undone together with rollback, so the data never stays half changed.
An order with two lines must not take stock for the first line and then fail on the second. All changes live in one transaction, so rollback() undoes them together. TestClient lets you walk through the API in memory.
client = TestClient(app)res = client.post("/orders", json={ "customer": "Asha", "lines": [{"product_id": 1, "quantity": 4}],})print(res.status_code)Example 3 · A failed order changes nothing
from fastapi import Depends, FastAPI, HTTPExceptionfrom fastapi.testclient import TestClientfrom pydantic import BaseModel, ConfigDict, Fieldfrom sqlalchemy import ForeignKey, create_engine, select, updatefrom sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship, sessionmakerfrom sqlalchemy.pool import StaticPool
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)SessionLocal = sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase): pass
class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] price_cents: Mapped[int] stock: Mapped[int]
class Order(Base): __tablename__ = "orders" id: Mapped[int] = mapped_column(primary_key=True) customer: Mapped[str] total_cents: Mapped[int] = mapped_column(default=0) lines: Mapped[list["OrderLine"]] = relationship(cascade="all, delete-orphan")
class OrderLine(Base): __tablename__ = "order_lines" id: Mapped[int] = mapped_column(primary_key=True) order_id: Mapped[int] = mapped_column(ForeignKey("orders.id")) product_id: Mapped[int] = mapped_column(ForeignKey("products.id")) quantity: Mapped[int] unit_price_cents: Mapped[int]
Base.metadata.create_all(engine)
class ProductIn(BaseModel): name: str = Field(min_length=1) price_cents: int = Field(gt=0) stock: int = Field(ge=0)
class ProductOut(ProductIn): model_config = ConfigDict(from_attributes=True) id: int
class LineIn(BaseModel): product_id: int quantity: int = Field(gt=0)
class OrderIn(BaseModel): customer: str = Field(min_length=1) lines: list[LineIn] = Field(min_length=1)
class LineOut(LineIn): model_config = ConfigDict(from_attributes=True) unit_price_cents: int
class OrderOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int customer: str total_cents: int lines: list[LineOut]
def get_session(): with SessionLocal() as session: yield session
app = FastAPI()
@app.post("/products", response_model=ProductOut, status_code=201)def add_product(body: ProductIn, session: Session = Depends(get_session)): product = Product(**body.model_dump()) session.add(product) session.commit() return product
@app.get("/products", response_model=list[ProductOut])def list_products(skip: int = 0, limit: int = 50, session: Session = Depends(get_session)): query = select(Product).order_by(Product.id).offset(skip).limit(limit) return session.scalars(query).all()
@app.post("/orders", response_model=OrderOut, status_code=201)def place_order(body: OrderIn, session: Session = Depends(get_session)): order = Order(customer=body.customer, total_cents=0) for line in body.lines: product = session.get(Product, line.product_id) if product is None: session.rollback() raise HTTPException(status_code=404, detail=f"Product {line.product_id} not found") taken = session.execute( update(Product) .where(Product.id == line.product_id, Product.stock >= line.quantity) .values(stock=Product.stock - line.quantity) ) if taken.rowcount == 0: session.rollback() raise HTTPException(status_code=409, detail=f"Not enough stock for {product.name}") order.lines.append(OrderLine(product_id=product.id, quantity=line.quantity, unit_price_cents=product.price_cents)) order.total_cents += product.price_cents * line.quantity session.add(order) session.commit() return order
@app.get("/orders/{order_id}", response_model=OrderOut)def get_order(order_id: int, session: Session = Depends(get_session)): order = session.get(Order, order_id) if order is None: raise HTTPException(status_code=404, detail="Order not found") return order
client = TestClient(app)
client.post("/products", json={"name": "Pen", "price_cents": 250, "stock": 10})client.post("/products", json={"name": "Notebook", "price_cents": 1200, "stock": 3})
res = client.post("/orders", json={"customer": "Ravi", "lines": [{"product_id": 1, "quantity": 5}, {"product_id": 2, "quantity": 9}]})print(res.status_code, res.json())print([p["stock"] for p in client.get("/products").json()])print(client.get("/orders/1").status_code)
bad = client.post("/orders", json={"customer": "", "lines": []})print(bad.status_code, sorted(e["loc"][-1] for e in bad.json()["detail"]))Output
409 {'detail': 'Not enough stock for Notebook'}
[10, 3]
404
422 ['customer', 'lines']What this code does
- The first line (5 pens) is fine, but the second (9 notebooks) fails. The pen stock is still 10, so the whole order was undone.
- No order was saved, so
GET /orders/1is404. - An empty customer and an empty line list are both refused with
422before the route starts.
Example 4 · Test isolation with dependency_overrides
from fastapi import Depends, FastAPI, HTTPExceptionfrom fastapi.testclient import TestClientfrom pydantic import BaseModel, ConfigDict, Fieldfrom sqlalchemy import ForeignKey, create_engine, select, updatefrom sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship, sessionmakerfrom sqlalchemy.pool import StaticPool
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)SessionLocal = sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase): pass
class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] price_cents: Mapped[int] stock: Mapped[int]
class Order(Base): __tablename__ = "orders" id: Mapped[int] = mapped_column(primary_key=True) customer: Mapped[str] total_cents: Mapped[int] = mapped_column(default=0) lines: Mapped[list["OrderLine"]] = relationship(cascade="all, delete-orphan")
class OrderLine(Base): __tablename__ = "order_lines" id: Mapped[int] = mapped_column(primary_key=True) order_id: Mapped[int] = mapped_column(ForeignKey("orders.id")) product_id: Mapped[int] = mapped_column(ForeignKey("products.id")) quantity: Mapped[int] unit_price_cents: Mapped[int]
Base.metadata.create_all(engine)
class ProductIn(BaseModel): name: str = Field(min_length=1) price_cents: int = Field(gt=0) stock: int = Field(ge=0)
class ProductOut(ProductIn): model_config = ConfigDict(from_attributes=True) id: int
class LineIn(BaseModel): product_id: int quantity: int = Field(gt=0)
class OrderIn(BaseModel): customer: str = Field(min_length=1) lines: list[LineIn] = Field(min_length=1)
class LineOut(LineIn): model_config = ConfigDict(from_attributes=True) unit_price_cents: int
class OrderOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int customer: str total_cents: int lines: list[LineOut]
def get_session(): with SessionLocal() as session: yield session
app = FastAPI()
@app.post("/products", response_model=ProductOut, status_code=201)def add_product(body: ProductIn, session: Session = Depends(get_session)): product = Product(**body.model_dump()) session.add(product) session.commit() return product
@app.get("/products", response_model=list[ProductOut])def list_products(skip: int = 0, limit: int = 50, session: Session = Depends(get_session)): query = select(Product).order_by(Product.id).offset(skip).limit(limit) return session.scalars(query).all()
@app.post("/orders", response_model=OrderOut, status_code=201)def place_order(body: OrderIn, session: Session = Depends(get_session)): order = Order(customer=body.customer, total_cents=0) for line in body.lines: product = session.get(Product, line.product_id) if product is None: session.rollback() raise HTTPException(status_code=404, detail=f"Product {line.product_id} not found") taken = session.execute( update(Product) .where(Product.id == line.product_id, Product.stock >= line.quantity) .values(stock=Product.stock - line.quantity) ) if taken.rowcount == 0: session.rollback() raise HTTPException(status_code=409, detail=f"Not enough stock for {product.name}") order.lines.append(OrderLine(product_id=product.id, quantity=line.quantity, unit_price_cents=product.price_cents)) order.total_cents += product.price_cents * line.quantity session.add(order) session.commit() return order
@app.get("/orders/{order_id}", response_model=OrderOut)def get_order(order_id: int, session: Session = Depends(get_session)): order = session.get(Order, order_id) if order is None: raise HTTPException(status_code=404, detail="Order not found") return order
client = TestClient(app)
test_engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)Base.metadata.create_all(test_engine)TestSession = sessionmaker(test_engine, expire_on_commit=False)
def override_session(): with TestSession() as session: yield session
app.dependency_overrides[get_session] = override_session
client.post("/products", json={"name": "Pen", "price_cents": 250, "stock": 10})print(len(client.get("/products").json()), "product in the test database")with SessionLocal() as session: print(len(session.scalars(select(Product)).all()), "products in the normal database")Output
1 product in the test database 0 products in the normal database
What this code does
dependency_overridesswapsget_sessionfor a version that uses a fresh test database.- The normal database stays empty. Every test file can build its own database this way.
Example 5 · Run it for real (sample, not run here)
Runs in: Your terminal
pip install fastapi uvicorn sqlalchemyuvicorn store:app --reloadcurl -X POST localhost:8000/products -H "Content-Type: application/json" -d '{"name": "Pen", "price_cents": 250, "stock": 10}'Sample output
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
{"name":"Pen","price_cents":250,"stock":10,"id":1}What this code does
- Save the code as
store.py.uvicornis the server that runs the app;--reloadrestarts it when the file changes. - Open
http://localhost:8000/docsto try every route in the browser.
Use 201 for a created order, 404 for an unknown product, 409 for not enough stock and 422 for a bad request body, which FastAPI sends by itself.
What to add next
This API is small on purpose. Each piece you learned can be added without changing its shape.
- Move the models and routes into routers and layers, and read the database URL from settings.
- Add an Alembic migration instead of
create_all, and use PostgreSQL withwith_for_update()for row locks. - Cache
GET /productsfor a short time and delete the key when a product changes. - Send the order confirmation as a background task, or add login with JWT in the security section.
Next step
Check your understanding
12 quick questions