from contextlib import asynccontextmanager
from fastapi import FastAPI

# Simulating a database connection pool
db_pool = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    # ── STARTUP ── (runs before first request)
    print("🚀 Starting up...")
    db_pool["conn"] = await create_db_connection()
    print("✅ Database connected!")

    yield  # ← app runs here, handling requests

    # ── SHUTDOWN ── (runs after last request)
    print("🛑 Shutting down...")
    await db_pool["conn"].close()
    print("✅ Database connection closed.")


# Pass lifespan to the app
app = FastAPI(lifespan=lifespan)

@app.get("/")
async def root():
    conn = db_pool["conn"]  # available everywhere!
    return {"status": "ready"}

async def create_db_connection():
    return {"host": "localhost"}

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

# GET   — retrieve a list
@app.get("/items")
async def list_items():
    return [{"id": 1, "name": "Widget"}]

# GET   — retrieve one
@app.get("/items/{item_id}")
async def get_item(item_id: int):
    return {"id": item_id, "name": "Widget"}

# POST  — create
@app.post("/items", status_code=201)
async def create_item(item: Item):
    return {"id": 42, **item.model_dump()}

# PUT   — replace
@app.put("/items/{item_id}")
async def replace_item(item_id: int, item: Item):
    return {"id": item_id, **item.model_dump()}

# PATCH — partial update
@app.patch("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"id": item_id, "updated": item.model_dump()}

# DELETE
@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    return None  # 204 No Content

@app.get(
    "/items/{item_id}",

    # Groups this endpoint in Swagger UI under "Items"
    tags=["items"],

    # Summary shown in the endpoint list
    summary="Get a single item by ID",

    # Longer description (supports Markdown)
    description="Returns the item with the given ID. Raises 404 if not found.",

    # HTTP status code for the "success" response (default: 200)
    status_code=200,

    # Unique name for reverse URL generation
    name="get_item",

    # Mark as deprecated in docs (shown with strikethrough)
    deprecated=False,

    # Hide from docs entirely
    include_in_schema=True,
)
async def get_item(item_id: int):
    return {"id": item_id}

# You can also use docstrings as the description:
@app.get("/users", tags=["users"])
async def list_users():
    """
    List all active users.

    Returns a paginated list of users sorted by creation date.
    """
    return []

# routers/users.py
from fastapi import APIRouter

# Create a router (mini-app)
router = APIRouter(
    prefix="/users",          # all routes start with /users
    tags=["users"],           # group in Swagger UI under "users"
    responses={404: {"description": "Not found"}},
)

@router.get("/")             # → GET /users/
async def list_users():
    return [{"id": 1, "name": "Alice"}]

@router.get("/{user_id}")    # → GET /users/{user_id}
async def get_user(user_id: int):
    return {"id": user_id, "name": "Alice"}

@router.post("/")            # → POST /users/
async def create_user():
    return {"id": 99}

# main.py
from fastapi import FastAPI
from routers import users, items, orders

app = FastAPI()

# Include each router — optionally override their prefix/tags
app.include_router(users.router)
app.include_router(items.router)
app.include_router(
    orders.router,
    prefix="/api/v1",   # additional prefix: /api/v1/orders/...
    tags=["orders"],
)

# Result:
# GET  /users/         ← from users router
# GET  /users/{id}     ← from users router
# POST /users/         ← from users router
# GET  /items/         ← from items router
# GET  /api/v1/orders/ ← from orders router (extra prefix)

from fastapi import FastAPI, Path

app = FastAPI()

# Basic: type annotation = automatic validation + conversion
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # FastAPI auto-converts "42" (string) to 42 (int)
    # GET /users/42   → user_id = 42  ✓
    # GET /users/abc  → 422 Validation Error ✗ (not an int!)
    return {"user_id": user_id}

# String path param
@app.get("/files/{filename}")
async def get_file(filename: str):
    return {"file": filename}

# Multiple path params
@app.get("/users/{user_id}/posts/{post_id}")
async def get_post(user_id: int, post_id: int):
    return {"user_id": user_id, "post_id": post_id}

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

app = FastAPI()

# Required query param (no default)
@app.get("/search")
async def search(q: str):
    # GET /search?q=python  → q="python"
    # GET /search           → 422 (q is required!)
    return {"query": q}

# Optional query param (has default)
@app.get("/items")
async def list_items(
    page: int = 1,
    limit: int = 10,
    active: Optional[bool] = None,   # truly optional
):
    # GET /items                        → page=1, limit=10, active=None
    # GET /items?page=2&limit=5         → page=2, limit=5
    # GET /items?active=true            → active=True (auto bool!)
    return {"page": page, "limit": limit, "active": active}

# List query param (repeated key)
@app.get("/filter")
async def filter_items(
    tags: Annotated[list[str], Query()] = []
):
    # GET /filter?tags=python&tags=web&tags=api
    # → tags=["python", "web", "api"]
    return {"tags": tags}

# Query with validation
@app.get("/users")
async def list_users(
    q: Annotated[Optional[str], Query(
        min_length=3,
        max_length=50,
        description="Search query string",
    )] = None
):
    return {"q": q}

from fastapi import FastAPI, Header
from typing import Annotated, Optional

app = FastAPI()

@app.get("/info")
async def get_info(
    # "User-Agent" header → user_agent param
    user_agent: Annotated[Optional[str], Header()] = None,

    # "X-Request-ID" header → x_request_id param
    x_request_id: Annotated[Optional[str], Header()] = None,
):
    return {"user_agent": user_agent, "request_id": x_request_id}

# Authorization header (most common use case)
@app.get("/secure")
async def secure_endpoint(
    authorization: Annotated[str, Header(
        description="Bearer token: 'Bearer your-token-here'"
    )]
):
    # Reads the "Authorization" header
    # Value: "Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."
    token = authorization.replace("Bearer ", "")
    return {"token_preview": token[:10] + "..."}

# List header (same header sent multiple times)
@app.get("/multi-header")
async def multi(
    x_token: Annotated[Optional[list[str]], Header()] = None
):
    return {"x_tokens": x_token}

from fastapi import FastAPI, Cookie, Response
from typing import Annotated, Optional

app = FastAPI()

# Read a cookie
@app.get("/me")
async def get_current_user(
    session_id: Annotated[Optional[str], Cookie()] = None
):
    # Reads the "session_id" cookie from the request
    if not session_id:
        return {"user": "anonymous"}
    return {"session_id": session_id}

# Set a cookie in the response
@app.post("/login")
async def login(response: Response):
    # Inject Response to set cookies
    response.set_cookie(
        key="session_id",
        value="abc123xyz",
        max_age=3600,        # expire in 1 hour
        httponly=True,       # JS can't access it
        secure=True,         # HTTPS only
        samesite="lax",      # CSRF protection
    )
    return {"message": "Logged in!"}

# Delete a cookie (logout)
@app.post("/logout")
async def logout(response: Response):
    response.delete_cookie("session_id")
    return {"message": "Logged out!"}

from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class CreateUserRequest(BaseModel):
    username: str
    email: str
    age: Optional[int] = None

@app.post("/users")
async def create_user(user: CreateUserRequest):
    # FastAPI automatically:
    # 1. Reads request body as JSON
    # 2. Validates against CreateUserRequest
    # 3. Returns 422 if validation fails
    # 4. Passes typed Python object to your function

    print(user.username)   # "alice"  ← typed!
    print(user.email)      # "alice@example.com"
    print(user.age)        # None or int
    return {"created": user.model_dump()}

# Mixing path + query + body:
# All 3 can coexist in one endpoint!
@app.put("/users/{user_id}")
async def update_user(
    user_id: int,              # ← path param
    notify: bool = False,     # ← query param
    user: CreateUserRequest = ...,  # ← body
):
    return {
        "id": user_id,
        "notify": notify,
        "data": user.model_dump()
    }

pip install python-multipart

from fastapi import FastAPI, Form
from typing import Annotated

app = FastAPI()

# URL-encoded form (typical HTML form POST)
@app.post("/login")
async def login(
    username: Annotated[str, Form()],
    password: Annotated[str, Form()],
):
    # Content-Type: application/x-www-form-urlencoded
    # Body: username=alice&password=secret
    return {"username": username, "logged_in": True}

# Form with validation
@app.post("/register")
async def register(
    username: Annotated[str, Form(min_length=3, max_length=20)],
    email: Annotated[str, Form()],
    age: Annotated[int, Form(ge=18)],  # must be 18+
):
    return {"registered": username}

# ⚠️ You cannot mix Form() and JSON body in the same endpoint
# They use different Content-Types

from fastapi import FastAPI, File, UploadFile, Form
from typing import Annotated

app = FastAPI()

# Single file upload
@app.post("/upload")
async def upload_file(file: UploadFile):
    print(file.filename)      # "photo.jpg"
    print(file.content_type)  # "image/jpeg"
    print(file.size)          # file size in bytes

    # Read entire file into memory
    contents = await file.read()

    # Or stream it in chunks (better for large files!)
    while chunk := await file.read(1024):  # 1KB chunks
        process_chunk(chunk)

    return {"filename": file.filename, "size": file.size}

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

# File + form fields together (use multipart/form-data)
@app.post("/upload-with-meta")
async def upload_with_meta(
    file: UploadFile,
    description: Annotated[str, Form()],
    public: Annotated[bool, Form()] = False,
):
    return {
        "filename": file.filename,
        "description": description,
        "public": public,
    }