Moved to FastAPI router example

2023/sqlalchemy/EXAMPLES.md → 2023/fastapi-router/EXAMPLES.md

@@ -1,5 +1,15 @@
+# Start the server
+
+```bash
+uvicorn main:app --reload
+```
+
 # Create an item
 
 curl -X POST -H "Content-Type: application/json" \
  -d '{"name": "Espresso machine", "description": "Silver dual boiler machine"}' \
  http://0.0.0.0:8000/items
+
+# Read an item
+
+curl -X GET http://0.0.0.0:8000/items/1

2023/fastapi-router/db/core.py

@@ -0,0 +1,30 @@
+from typing import Optional
+from sqlalchemy import create_engine, String
+from sqlalchemy.orm import sessionmaker, DeclarativeBase, Mapped, mapped_column
+
+DATABASE_URL = "sqlite:///test.db"
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class DBItem(Base):
+    __tablename__ = "items"
+
+    id: Mapped[int] = mapped_column(primary_key=True, index=True)
+    name: Mapped[str] = mapped_column(String(30))
+    description: Mapped[Optional[str]]
+
+
+engine = create_engine(DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+
+# Dependency to get the database session
+def get_db():
+    database = SessionLocal()
+    try:
+        yield database
+    finally:
+        database.close()

2023/fastapi-router/main.py

@@ -0,0 +1,18 @@
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from routers.items import router as items_router
+from db.core import Base, engine
+
+@asynccontextmanager
+async def lifespan(_: FastAPI):
+    Base.metadata.create_all(bind=engine)
+    yield
+
+app = FastAPI(lifespan=lifespan)
+
+app.include_router(items_router)
+
+
+@app.get("/")
+def read_root():
+    return "Server is running."

2023/sqlalchemy/main.py → 2023/fastapi-router/routers/items.py

@@ -1,10 +1,9 @@
-from contextlib import asynccontextmanager
 from typing import Optional
-from fastapi import FastAPI, HTTPException
+from fastapi import APIRouter, HTTPException
 from fastapi.params import Depends
-from sqlalchemy import create_engine, String
-from sqlalchemy.orm import sessionmaker, Session, DeclarativeBase, Mapped, mapped_column
+from sqlalchemy.orm import Session
 from pydantic import BaseModel
+from db.core import DBItem, get_db
 
 
 class Item(BaseModel):
@@ -23,61 +22,30 @@ class ItemUpdate(BaseModel):
     description: Optional[str]
 
 
-DATABASE_URL = "sqlite:///test.db"
+router = APIRouter(
+    prefix="/items",
+)
 
 
-class Base(DeclarativeBase):
-    pass
-
-
-class DBItem(Base):
-    __tablename__ = "items"
-
-    id: Mapped[int] = mapped_column(primary_key=True, index=True)
-    name: Mapped[str] = mapped_column(String(30))
-    description: Mapped[Optional[str]]
-
-
-engine = create_engine(DATABASE_URL)
-SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-
-
-# Dependency to get the database session
-def get_db():
-    database = SessionLocal()
-    try:
-        yield database
-    finally:
-        database.close()
-
-
-@asynccontextmanager
-async def lifespan(_: FastAPI):
-    Base.metadata.create_all(bind=engine)
-    yield
-
-
-app = FastAPI(lifespan=lifespan)
-
-
-@app.post("/items")
+@router.post("")
 def create_item(item: ItemCreate, db: Session = Depends(get_db)) -> Item:
+    print(item)
     db_item = DBItem(**item.model_dump())
     db.add(db_item)
     db.commit()
     db.refresh(db_item)
     return Item(**db_item.__dict__)
 
 
-@app.get("/items/{item_id}")
+@router.get("/{item_id}")
 def read_item(item_id: int, db: Session = Depends(get_db)) -> Item:
     db_item = db.query(DBItem).filter(DBItem.id == item_id).first()
     if db_item is None:
         raise HTTPException(status_code=404, detail="Item not found")
     return Item(**db_item.__dict__)
 
 
-@app.put("/items/{item_id}")
+@router.put("/{item_id}")
 def update_item(item_id: int, item: ItemUpdate, db: Session = Depends(get_db)) -> Item:
     db_item = db.query(DBItem).filter(DBItem.id == item_id).first()
     if db_item is None:
@@ -89,7 +57,7 @@ def update_item(item_id: int, item: ItemUpdate, db: Session = Depends(get_db)) -
     return Item(**db_item.__dict__)
 
 
-@app.delete("/items/{item_id}")
+@router.delete("/{item_id}")
 def delete_item(item_id: int, db: Session = Depends(get_db)) -> Item:
     db_item = db.query(DBItem).filter(DBItem.id == item_id).first()
     if db_item is None: