# Create and activate virtual environment
python -m venv venv
source venv/bin/activate       # Linux/macOS
venv\Scripts\activate          # Windows

# Install FastAPI with all standard dependencies
pip install "fastapi[standard]"

# Or install individually
pip install fastapi uvicorn[standard] pydantic

# Verify installation
python -c "import fastapi; print(fastapi.__version__)"

from fastapi import FastAPI

# Create the FastAPI application instance
app = FastAPI(
    title="My API",
    description="A simple FastAPI example",
    version="1.0.0"
)

@app.get("/")
def read_root():
    """Root endpoint — returns a greeting."""
    return {"message": "Hello, World!"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str | None = None):
    """Get an item by ID with an optional query parameter."""
    return {"item_id": item_id, "q": q}

# Development mode — auto-reload on file changes
uvicorn main:app --reload

# Custom host and port
uvicorn main:app --reload --host 0.0.0.0 --port 8080

# Production (no reload, multiple workers)
uvicorn main:app --workers 4 --host 0.0.0.0 --port 80

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: runs before the app starts serving
    print("Starting up... connecting to DB, loading models, etc.")
    yield
    # Shutdown: runs when the app is stopping
    print("Shutting down... closing connections.")

app = FastAPI(lifespan=lifespan)

@app.get("/")
def root():
    return {"status": "running"}

from fastapi import FastAPI, Path
from enum import Enum

app = FastAPI()

# Basic typed path parameter — FastAPI auto-validates int
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

# Using Path() for extra validation and metadata
@app.get("/products/{product_id}")
def get_product(
    product_id: int = Path(
        title="Product ID",
        description="The ID of the product to retrieve",
        ge=1,          # greater than or equal to 1
        le=9999        # less than or equal to 9999
    )
):
    return {"product_id": product_id}

# Enum-based path parameters (restricts allowed values)
class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet  = "resnet"
    lenet   = "lenet"

@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return {"model": model_name, "message": "Deep Learning FTW!"}
    elif model_name.value == "lenet":
        return {"model": model_name, "message": "LeCNN all the images"}
    return {"model": model_name, "message": "Have some residuals"}

# File path parameter — captures slashes too
@app.get("/files/{file_path:path}")
def read_file(file_path: str):
    # URL: /files/home/user/docs/report.pdf
    return {"file_path": file_path}

from fastapi import FastAPI
app = FastAPI()

# IMPORTANT: Fixed routes must come BEFORE dynamic ones
# Otherwise "/me" would be captured as {user_id}

@app.get("/users/me")         # ✅ Defined first — takes priority
def get_current_user():
    return {"user_id": "current_user"}

@app.get("/users/{user_id}")  # Fallback for other user IDs
def get_user(user_id: str):
    return {"user_id": user_id}

from fastapi import FastAPI, Query
from typing import Annotated

app = FastAPI()

# Optional query params (None default = optional)
# GET /items?skip=0&limit=10
@app.get("/items")
def get_items(skip: int = 0, limit: int = 10):
    fake_db = list(range(100))
    return fake_db[skip : skip + limit]

# Required query parameter (no default = required)
# GET /search?q=apple  ← q is mandatory
@app.get("/search")
def search(q: str):
    return {"query": q, "results": []}

# Optional with None
# GET /users   ← q is optional
@app.get("/users")
def list_users(q: str | None = None):
    if q:
        return {"users": [f"user matching '{q}'"]}
    return {"users": ["alice", "bob", "carol"]}

# Advanced validation with Query()
@app.get("/products")
def list_products(
    q: Annotated[
        str | None,
        Query(
            min_length=3,
            max_length=50,
            title="Search query",
            description="Query string for product search"
        )
    ] = None,
    page:  Annotated[int, Query(ge=1)] = 1,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
):
    return {"q": q, "page": page, "limit": limit}

# Multiple values for the same query key
# GET /tags?tags=python&tags=api&tags=web
@app.get("/tags")
def get_by_tags(tags: list[str] = Query(default=[])):
    return {"tags": tags}

from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr
from typing import Annotated
from decimal import Decimal

app = FastAPI()

# --- Basic Model ---
class Item(BaseModel):
    name: str
    description: str | None = None   # Optional field
    price: float
    tax: float | None = None

@app.post("/items")
def create_item(item: Item):
    item_dict = item.model_dump()
    if item.tax is not None:
        price_with_tax = item.price + item.tax
        item_dict["price_with_tax"] = price_with_tax
    return item_dict

# --- Model with Validation (Field) ---
class UserCreate(BaseModel):
    username: Annotated[str, Field(min_length=3, max_length=50)]
    email: EmailStr                   # requires: pip install pydantic[email]
    full_name: str | None = Field(default=None, title="Full Name")
    age: Annotated[int, Field(ge=0, le=120)]

    # Pydantic model config
    model_config = {
        "json_schema_extra": {
            "examples": [{
                "username": "johndoe",
                "email": "john@example.com",
                "full_name": "John Doe",
                "age": 30
            }]
        }
    }

@app.post("/users")
def create_user(user: UserCreate):
    return {"created": user.model_dump()}

# --- Nested Models ---
class Address(BaseModel):
    street: str
    city: str
    country: str = "US"

class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(ge=1)
    unit_price: Decimal

class Order(BaseModel):
    customer_name: str
    shipping_address: Address         # Nested model
    items: list[OrderItem]            # List of nested models
    notes: str | None = None

@app.post("/orders")
def create_order(order: Order):
    total = sum(i.quantity * i.unit_price for i in order.items)
    return {
        "order": order.model_dump(),
        "total": float(total)
    }

# --- Combining Path, Query, and Body ---
@app.put("/items/{item_id}")
def update_item(
    item_id: int,          # path parameter
    q: str | None = None,  # query parameter
    item: Item | None = None  # body (optional)
):
    result: dict = {"item_id": item_id}
    if q:
        result["q"] = q
    if item:
        result["item"] = item.model_dump()
    return result

from pydantic import BaseModel, field_validator, model_validator

class PasswordChange(BaseModel):
    current_password: str
    new_password: str
    confirm_password: str

    @field_validator("new_password")
    @classmethod
    def password_strength(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("Password must be at least 8 characters")
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain an uppercase letter")
        return v

    @model_validator(mode="after")
    def passwords_match(self) -> "PasswordChange":
        if self.new_password != self.confirm_password:
            raise ValueError("Passwords do not match")
        return self

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserIn(BaseModel):
    username: str
    password: str    # Never return this!
    email: str
    full_name: str | None = None

class UserOut(BaseModel):
    username: str    # Password excluded
    email: str
    full_name: str | None = None

# response_model filters the output — password never leaks!
@app.post("/users", response_model=UserOut)
def create_user(user: UserIn) -> UserIn:
    # Business logic here... save to DB etc.
    return user  # FastAPI strips 'password' automatically

# ── Response model with exclude / include ──
@app.get("/users/me", response_model=UserOut, response_model_exclude_unset=True)
def get_me():
    # Only fields that were actually set are returned
    return UserOut(username="alice", email="alice@example.com")

# ── Union response types ──
class Cat(BaseModel):
    name: str
    meows: bool

class Dog(BaseModel):
    name: str
    barks: bool

@app.get("/animals/{animal_id}", response_model=Cat | Dog)
def get_animal(animal_id: int):
    if animal_id == 1:
        return Cat(name="Whiskers", meows=True)
    return Dog(name="Rex", barks=True)

# ── List responses ──
@app.get("/users/all", response_model=list[UserOut])
def get_all_users():
    return [
        UserOut(username="alice", email="alice@example.com"),
        UserOut(username="bob",   email="bob@example.com"),
    ]

# ── Return type annotations (modern style) ──
@app.get("/items/{item_id}")
def get_item(item_id: int) -> dict:
    return {"item_id": item_id, "name": "Widget"}

from fastapi import FastAPI, HTTPException, status, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()
fake_db: dict[int, dict] = {1: {"name": "Foo"}, 2: {"name": "Bar"}}

# --- Using status_code on route ---
@app.post("/items", status_code=status.HTTP_201_CREATED)
def create_item(name: str):
    return {"name": name}

# --- Raising HTTPException ---
@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in fake_db:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Item {item_id} not found"
        )
    return fake_db[item_id]

# --- Custom Exception Class ---
class InsufficientFundsError(Exception):
    def __init__(self, amount: float):
        self.amount = amount

# --- Custom Exception Handler ---
@app.exception_handler(InsufficientFundsError)
async def insufficient_funds_handler(
    request: Request,
    exc: InsufficientFundsError
):
    return JSONResponse(
        status_code=402,
        content={
            "error": "insufficient_funds",
            "shortfall": exc.amount,
            "message": f"Need ${exc.amount:.2f} more funds"
        }
    )

@app.post("/transfer")
def transfer_funds(amount: float, balance: float):
    if amount > balance:
        raise InsufficientFundsError(amount - balance)
    return {"transferred": amount, "remaining": balance - amount}

# --- Override default 422 validation error ---
from fastapi.exceptions import RequestValidationError

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=400,  # Override 422 → 400
        content={"detail": exc.errors(), "body": str(exc.body)}
    )

from fastapi import FastAPI, Form, File, UploadFile, HTTPException
from typing import Annotated
import shutil, os

app = FastAPI()

# --- Form Data ---
@app.post("/login")
def login(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()]
):
    # Content-Type: application/x-www-form-urlencoded
    if username == "admin" and password == "secret":
        return {"access": "granted"}
    raise HTTPException(status_code=401, detail="Invalid credentials")

# --- Single File Upload ---
@app.post("/upload")
async def upload_file(file: UploadFile):
    # UploadFile gives you: filename, content_type, file (SpooledTemporaryFile)
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents)
    }

# --- Save file to disk ---
UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)

@app.post("/upload/save")
async def save_file(file: UploadFile):
    allowed_types = ["image/jpeg", "image/png", "application/pdf"]
    if file.content_type not in allowed_types:
        raise HTTPException(400, "File type not allowed")

    max_size = 5 * 1024 * 1024  # 5 MB
    contents = await file.read()
    if len(contents) > max_size:
        raise HTTPException(413, "File too large")

    dest = os.path.join(UPLOAD_DIR, file.filename)
    with open(dest, "wb") as f:
        f.write(contents)
    return {"saved_to": dest, "bytes": len(contents)}

# --- Multiple Files ---
@app.post("/upload/multiple")
async def upload_multiple(files: list[UploadFile]):
    results = []
    for file in files:
        content = await file.read()
        results.append({"name": file.filename, "size": len(content)})
    return {"files": results, "count": len(files)}

# --- Form + File together ---
@app.post("/profile")
async def update_profile(
    name:   Annotated[str, Form()],
    bio:    Annotated[str, Form()],
    avatar: UploadFile | None = File(default=None)
):
    return {
        "name": name,
        "bio": bio,
        "avatar": avatar.filename if avatar else None
    }

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

app = FastAPI()

# ── Simple function dependency ──
def common_params(skip: int = 0, limit: int = 100):
    return {"skip": skip, "limit": limit}

CommonDeps = Annotated[dict, Depends(common_params)]

@app.get("/users")
def list_users(commons: CommonDeps):
    return {"params": commons, "users": []}

@app.get("/items")
def list_items(commons: CommonDeps):
    return {"params": commons, "items": []}

# ── Class-based dependency ──
class PaginationParams:
    def __init__(self, page: int = 1, size: int = 20):
        if size > 100:
            raise HTTPException(400, "Max page size is 100")
        self.skip  = (page - 1) * size
        self.limit = size
        self.page  = page

Pagination = Annotated[PaginationParams, Depends()]

@app.get("/products")
def list_products(p: Pagination):
    return {"skip": p.skip, "limit": p.limit, "page": p.page}

# ── Dependency with yield (for DB sessions etc.) ──
def get_db():
    db = {"connection": "fake_db_session"}  # Replace with real DB session
    try:
        yield db
    finally:
        # Cleanup runs after the request (even on error)
        db.clear()

DB = Annotated[dict, Depends(get_db)]

@app.get("/db-items")
def db_items(db: DB):
    return {"db": db}

# ── Chained / nested dependencies ──
async def verify_token(x_token: Annotated[str, Header()]):
    if x_token != "supersecret":
        raise HTTPException(403, "Invalid X-Token header")
    return x_token

async def verify_key(x_key: Annotated[str, Header()]):
    if x_key != "fakekey":
        raise HTTPException(403, "Invalid X-Key header")
    return x_key

@app.get("/admin", dependencies=[Depends(verify_token), Depends(verify_key)])
def admin_route():
    return {"admin": "access granted"}

# ── Router-level dependencies ──
from fastapi import APIRouter

router = APIRouter(
    prefix="/api/v1",
    dependencies=[Depends(verify_token)]  # Applies to ALL routes in router
)

@router.get("/protected-resource")
def protected():
    return {"data": "sensitive"}

app.include_router(router)

pip install "python-jose[cryptography]" "passlib[bcrypt]" python-multipart

from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# ── Configuration ──
SECRET_KEY    = "your-256-bit-secret-keep-this-safe"  # Use env var in prod!
ALGORITHM     = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# ── Password hashing ──
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

# ── Fake user database ──
fake_users_db = {
    "alice": {
        "username": "alice",
        "email": "alice@example.com",
        "hashed_password": hash_password("secret123"),
        "disabled": False,
    }
}

# ── Pydantic Models ──
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None

class User(BaseModel):
    username: str
    email: str | None = None
    disabled: bool = False

# ── OAuth2 Scheme ──
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# ── JWT Helpers ──
def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=15))
    to_encode["exp"] = expire
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def get_user(db: dict, username: str) -> dict | None:
    return db.get(username)

# ── Current User Dependency ──
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    user_data = get_user(fake_users_db, username)
    if user_data is None:
        raise credentials_exception
    return User(**user_data)

async def get_active_user(
    current_user: Annotated[User, Depends(get_current_user)]
) -> User:
    if current_user.disabled:
        raise HTTPException(400, "Inactive user")
    return current_user

CurrentUser = Annotated[User, Depends(get_active_user)]

# ── FastAPI App ──
app = FastAPI()

@app.post("/token", response_model=Token)
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user_data = get_user(fake_users_db, form_data.username)
    if not user_data or not verify_password(form_data.password, user_data["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token = create_access_token(
        data={"sub": user_data["username"]},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return Token(access_token=access_token, token_type="bearer")

@app.get("/users/me", response_model=User)
async def read_users_me(current_user: CurrentUser):
    return current_user

@app.get("/users/me/items")
async def read_own_items(current_user: CurrentUser):
    return [{"item_id": "1", "owner": current_user.username}]

pip install sqlalchemy "databases[sqlite]" aiosqlite

from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
from contextlib import asynccontextmanager
from typing import Annotated

# ── Database Setup ──
DATABASE_URL = "sqlite:///./test.db"
engine       = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base         = declarative_base()

# ── SQLAlchemy Model (DB Table) ──
class UserModel(Base):
    __tablename__ = "users"
    id       = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email    = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

# ── Pydantic Schemas ──
class UserCreate(BaseModel):
    username: str
    email: str
    password: str

class UserRead(BaseModel):
    id: int
    username: str
    email: str
    is_active: bool

    model_config = {"from_attributes": True}  # Pydantic v2

# ── DB Dependency ──
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

DB = Annotated[Session, Depends(get_db)]

# ── CRUD Helpers ──
def crud_get_user(db: Session, user_id: int) -> UserModel | None:
    return db.query(UserModel).filter(UserModel.id == user_id).first()

def crud_get_by_email(db: Session, email: str) -> UserModel | None:
    return db.query(UserModel).filter(UserModel.email == email).first()

def crud_create_user(db: Session, user: UserCreate) -> UserModel:
    hashed = f"hashed_{user.password}"  # Use passlib in production!
    db_user = UserModel(
        username=user.username,
        email=user.email,
        hashed_password=hashed
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

# ── App + Lifespan ──
@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)  # Create tables on startup
    yield

app = FastAPI(lifespan=lifespan)

@app.post("/users", response_model=UserRead, status_code=201)
def create_user(user: UserCreate, db: DB):
    if crud_get_by_email(db, user.email):
        raise HTTPException(400, "Email already registered")
    return crud_create_user(db, user)

@app.get("/users/{user_id}", response_model=UserRead)
def get_user(user_id: int, db: DB):
    user = crud_get_user(db, user_id)
    if not user:
        raise HTTPException(404, "User not found")
    return user

@app.get("/users", response_model=list[UserRead])
def list_users(skip: int = 0, limit: int = 100, db: DB = None):
    return db.query(UserModel).offset(skip).limit(limit).all()

@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int, db: DB):
    user = crud_get_user(db, user_id)
    if not user:
        raise HTTPException(404, "User not found")
    db.delete(user)
    db.commit()

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import time, logging

app = FastAPI()
logger = logging.getLogger(__name__)

# ── Simple background task ──
def send_welcome_email(email: str, username: str):
    """This runs AFTER the response is sent."""
    time.sleep(2)  # Simulate slow email sending
    logger.info(f"Sent welcome email to {email} for user {username}")

@app.post("/users/register")
def register_user(
    username: str,
    email: str,
    background_tasks: BackgroundTasks
):
    # Schedule email — runs after response
    background_tasks.add_task(send_welcome_email, email, username)
    return {"message": "Registration successful! Welcome email on the way."}

# ── Multiple background tasks ──
def log_activity(user_id: int, action: str):
    logger.info(f"User {user_id} performed: {action}")

def update_analytics(endpoint: str):
    logger.info(f"Analytics updated for: {endpoint}")

@app.post("/items")
def create_item(
    item_name: str,
    user_id: int,
    background_tasks: BackgroundTasks
):
    # Response sent immediately
    background_tasks.add_task(log_activity, user_id, f"created item: {item_name}")
    background_tasks.add_task(update_analytics, "/items")
    return {"item": item_name, "status": "created"}

# ── Background task via dependency ──
class Notifier:
    async def send(self, message: str):
        # Async background work
        logger.info(f"Notification: {message}")

notifier = Notifier()

@app.delete("/items/{item_id}")
async def delete_item(
    item_id: int,
    background_tasks: BackgroundTasks
):
    background_tasks.add_task(notifier.send, f"Item {item_id} was deleted")
    return {"deleted": item_id}

import time, uuid
from fastapi import FastAPI, Request, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

# ── CORS — Cross-Origin Resource Sharing ──
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://myapp.com", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["Authorization", "Content-Type"],
    # allow_origins=["*"]  ← for public APIs (no credentials)
)

# ── GZip compression ──
app.add_middleware(GZipMiddleware, minimum_size=1000)

# ── Trusted Hosts (security) ──
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["myapp.com", "*.myapp.com", "localhost"]
)

# ── Custom Request Timing Middleware ──
class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response: Response = await call_next(request)
        duration = time.perf_counter() - start
        response.headers["X-Process-Time"] = f"{duration:.4f}s"
        return response

app.add_middleware(TimingMiddleware)

# ── Request ID Middleware ──
class RequestIDMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response

app.add_middleware(RequestIDMiddleware)

# ── Logging Middleware ──
import logging
logger = logging.getLogger(__name__)

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        logger.info(f"→ {request.method} {request.url.path}")
        response = await call_next(request)
        logger.info(f"← {response.status_code} {request.url.path}")
        return response

app.add_middleware(LoggingMiddleware)

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

pip install httpx pytest pytest-asyncio

from fastapi import FastAPI
from fastapi.testclient import TestClient
import pytest

# ── The app under test ──
app = FastAPI()

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

@app.post("/items")
def create_item(name: str, price: float):
    if price <= 0:
        from fastapi import HTTPException
        raise HTTPException(400, "Price must be positive")
    return {"name": name, "price": price}

# ── Synchronous Tests (TestClient) ──
client = TestClient(app)

def test_read_item():
    response = client.get("/items/42")
    assert response.status_code == 200
    assert response.json() == {"item_id": 42, "q": None}

def test_read_item_with_query():
    response = client.get("/items/42?q=hello")
    assert response.status_code == 200
    assert response.json()["q"] == "hello"

def test_read_item_invalid_type():
    response = client.get("/items/not-a-number")
    assert response.status_code == 422  # Validation error

def test_create_item():
    response = client.post("/items?name=Widget&price=9.99")
    assert response.status_code == 200
    assert response.json()["name"] == "Widget"

def test_create_item_invalid_price():
    response = client.post("/items?name=Freebie&price=-1")
    assert response.status_code == 400

# ── Using Fixtures (pytest) ──
@pytest.fixture
def test_client():
    with TestClient(app) as client:
        yield client

def test_with_fixture(test_client):
    response = test_client.get("/items/1")
    assert response.status_code == 200

# ── Async Tests ──
import pytest
from httpx import AsyncClient, ASGITransport

@pytest.mark.asyncio
async def test_async_get_item():
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test"
    ) as ac:
        response = await ac.get("/items/10")
    assert response.status_code == 200
    assert response.json()["item_id"] == 10

# ── Override Dependencies ──
from fastapi import Depends

def get_db():
    return {"conn": "real_db"}

@app.get("/db-data")
def db_endpoint(db=Depends(get_db)):
    return {"db": str(db)}

def override_get_db():
    return {"conn": "test_db"}  # Use in-memory or SQLite for tests

app.dependency_overrides[get_db] = override_get_db

def test_with_overridden_db():
    response = client.get("/db-data")
    assert response.json()["db"] == "{'conn': 'test_db'}"

# ── Run with: pytest test_main.py -v ──

FROM python:3.12-slim

WORKDIR /app

# Install dependencies first (layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy app code
COPY . .

# Expose port
EXPOSE 8000

# Run with Gunicorn + Uvicorn workers
CMD ["gunicorn", "main:app", \
     "--workers", "4", \
     "--worker-class", "uvicorn.workers.UvicornWorker", \
     "--bind", "0.0.0.0:8000"]

version: "3.9"
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
      - SECRET_KEY=${SECRET_KEY}
    depends_on:
      - db

  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

from pydantic_settings import BaseSettings  # pip install pydantic-settings
from functools import lru_cache

class Settings(BaseSettings):
    app_name:    str = "My FastAPI App"
    debug:       bool = False
    database_url: str
    secret_key:  str
    allowed_origins: list[str] = ["http://localhost:3000"]

    model_config = {"env_file": ".env"}  # Load from .env file

@lru_cache
def get_settings():
    return Settings()

# Usage in routes
from fastapi import Depends
from typing import Annotated

SettingsDep = Annotated[Settings, Depends(get_settings)]

# In any route:
# def my_route(settings: SettingsDep):
#     return {"app": settings.app_name}