from fastapi import FastAPI, Depends

app = FastAPI()

# ---- Class-based dependency ----
class PaginationParams:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit

@app.get("/items")
def get_items(params: PaginationParams = Depends(PaginationParams)):
    # params is an instance of PaginationParams
    return {
        "skip": params.skip,
        "limit": params.limit,
        "items": ["a", "b", "c"]
    }

# GET /items?skip=20&limit=5  →  {"skip": 20, "limit": 5, "items": [...]}
# Notice: Depends(PaginationParams) is same as Depends(PaginationParams.__init__)

from fastapi import FastAPI, Depends, HTTPException

app = FastAPI()

# ---- A "factory" class that creates dependencies with config ----
class RoleChecker:
    def __init__(self, allowed_roles: list[str]):
        self.allowed_roles = allowed_roles

    def __call__(self, current_user_role: str = "viewer") -> bool:
        # __call__ makes the instance callable → FastAPI uses it as a dependency
        if current_user_role not in self.allowed_roles:
            raise HTTPException(
                status_code=403,
                detail=f"Role '{current_user_role}' not permitted. Need: {self.allowed_roles}"
            )
        return True

# ---- Create role-specific dependency instances ----
admin_only = RoleChecker(allowed_roles=["admin"])
editor_or_admin = RoleChecker(allowed_roles=["admin", "editor"])

@app.get("/admin-panel")
def admin_panel(_=Depends(admin_only)):
    return {"message": "Welcome, Admin!"}

@app.get("/editor-panel")
def editor_panel(_=Depends(editor_or_admin)):
    return {"message": "Welcome, Editor or Admin!"}

from fastapi import FastAPI, Depends

app = FastAPI()

class Logger:
    def __init__(self, prefix: str):
        self.prefix = prefix
        self.logs = []

    def __call__(self, message: str) -> "Logger":
        self.logs.append(f"[{self.prefix}] {message}")
        return self

# Create instance once (shared state)
request_logger = Logger(prefix="REQUEST")

@app.get("/log-test")
def log_test(logger: Logger = Depends(request_logger)):
    logger(message="Someone hit /log-test")
    return {"logs": logger.logs}

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

app = FastAPI()

# ---- Level 1: Lowest-level dependency (settings) ----
def get_settings() -> dict:
    return {"secret": "my-secret", "env": "production"}

# ---- Level 2: Middle dependency (DB) depends on settings ----
def get_db(settings: dict = Depends(get_settings)):
    # Uses settings to create DB connection
    return {"connection": f"db-connected-to-{settings['env']}"}

# ---- Level 3: Top-level dependency (auth) depends on DB ----
def get_current_user(
    authorization: Annotated[str, Header()],
    db: dict = Depends(get_db)
) -> dict:
    if authorization != "Bearer valid-token":
        raise HTTPException(status_code=401, detail="Unauthorized")
    # In real app: query db for user by token
    return {"user_id": 1, "name": "Alice", "db": db["connection"]}

# ---- Route: just asks for the top-level result ----
@app.get("/profile")
def profile(user: dict = Depends(get_current_user)):
    return {"hello": user["name"], "via": user["db"]}

# FastAPI automatically chains:
# get_settings → get_db → get_current_user → profile

from fastapi import FastAPI, Depends

app = FastAPI()
call_count = 0  # Track how many times get_db is called

# ---- Shared base dependency ----
def get_db() -> dict:
    global call_count
    call_count += 1
    return {"session": "db-session", "call_number": call_count}

# ---- Two dependencies that both need DB ----
def get_user(db: dict = Depends(get_db)) -> dict:
    return {"user": "Alice", "db_call": db["call_number"]}

def get_logger(db: dict = Depends(get_db)) -> str:
    return f"Logger using DB call #{db['call_number']}"

# ---- Route uses BOTH above dependencies ----
@app.get("/dashboard")
def dashboard(
    user: dict = Depends(get_user),
    logger: str = Depends(get_logger)
):
    return {
        "user": user,
        "logger": logger,
        "note": "get_db was called ONCE even though two deps needed it!"
    }
# Result: both user and logger show call_number = 1 (same cached DB call)

# Force fresh call each time — use_cache=False
@app.get("/no-cache")
def no_cache_route(
    user: dict = Depends(get_user),
    logger: str = Depends(get_logger, use_cache=False)
):
    # get_db will now be called TWICE — once for user, once for logger
    return {"user": user, "logger": logger}

from fastapi import FastAPI, Depends
from contextlib import asynccontextmanager

app = FastAPI()

# ---- Generator dependency (with cleanup) ----
def get_db_session():
    # Code before yield → setup
    db = {"session": "fake-db-session", "open": True}
    print("DB session opened")
    try:
        yield db          # this value is injected into the route
    finally:
        # Code after yield → cleanup (always runs, even on error)
        db["open"] = False
        print("DB session closed")

@app.get("/with-db")
def with_db(db=Depends(get_db_session)):
    return {"session_open": db["open"], "session": db["session"]}

# Execution order:
# 1. DB session opened
# 2. Route runs → response built
# 3. Response sent to client  
# 4. DB session closed  ← finally block runs

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

app = FastAPI()

# ===== MODELS =====
class UserCreate(BaseModel):
    name: str
    email: str
    age: int

# ===== REPOSITORY LAYER (data access) =====
class UserRepository:
    def __init__(self):
        # In real app: inject DB session here
        self._db: dict = {}
        self._next_id = 1

    def find_by_email(self, email: str):
        for user in self._db.values():
            if user["email"] == email:
                return user
        return None

    def create(self, data: dict) -> dict:
        user = {"id": self._next_id, **data}
        self._db[self._next_id] = user
        self._next_id += 1
        return user

    def get_all(self) -> list:
        return list(self._db.values())

# ===== SERVICE LAYER (business logic) =====
class UserService:
    def __init__(self, repo: UserRepository = Depends(UserRepository)):
        self.repo = repo

    def create_user(self, user_data: UserCreate) -> dict:
        # Business rule: user must be 18+
        if user_data.age < 18:
            raise HTTPException(
                status_code=400,
                detail="User must be at least 18 years old"
            )
        # Business rule: email must be unique
        existing = self.repo.find_by_email(user_data.email)
        if existing:
            raise HTTPException(
                status_code=409,
                detail=f"Email {user_data.email} already registered"
            )
        return self.repo.create(user_data.model_dump())

    def list_users(self) -> list:
        return self.repo.get_all()

# ===== ROUTES (controller layer) =====
@app.post("/users", status_code=201)
def create_user(
    user_data: UserCreate,
    service: UserService = Depends(UserService)
):
    # Route just calls service — no business logic here
    return service.create_user(user_data)

@app.get("/users")
def list_users(service: UserService = Depends(UserService)):
    return service.list_users()

from fastapi import FastAPI, Depends
from typing import Protocol

app = FastAPI()

# ---- Define a Repository Protocol (interface) ----
class IProductRepository(Protocol):
    def get_by_id(self, product_id: int) -> dict | None: ...
    def save(self, product: dict) -> dict: ...

# ---- Concrete implementation (in-memory, swap for SQLAlchemy in prod) ----
class InMemoryProductRepo:
    def __init__(self):
        self._store = {
            1: {"id": 1, "name": "Laptop", "price": 999.99},
            2: {"id": 2, "name": "Phone", "price": 499.99},
        }

    def get_by_id(self, product_id: int) -> dict | None:
        return self._store.get(product_id)

    def save(self, product: dict) -> dict:
        pid = max(self._store.keys()) + 1
        product["id"] = pid
        self._store[pid] = product
        return product

# ---- Dependency provider ----
def get_product_repo() -> InMemoryProductRepo:
    return InMemoryProductRepo()

# ---- Routes ----
@app.get("/products/{product_id}")
def get_product(
    product_id: int,
    repo: InMemoryProductRepo = Depends(get_product_repo)
):
    product = repo.get_by_id(product_id)
    if not product:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Product not found")
    return product

from fastapi import FastAPI, Depends, HTTPException
from contextlib import contextmanager
from pydantic import BaseModel

app = FastAPI()

# ---- Simulated DB ----
fake_db = {
    "orders": [],
    "inventory": {"laptop": 5},
}

# ---- Repository classes ----
class OrderRepo:
    def __init__(self, db): self.db = db
    def create(self, order): self.db["orders"].append(order)

class InventoryRepo:
    def __init__(self, db): self.db = db
    def deduct(self, item: str, qty: int):
        if self.db["inventory"].get(item, 0) < qty:
            raise ValueError(f"Not enough stock for {item}")
        self.db["inventory"][item] -= qty

# ---- Unit of Work ----
class UnitOfWork:
    def __init__(self):
        # All repos share the same db reference
        self.orders = OrderRepo(fake_db)
        self.inventory = InventoryRepo(fake_db)
        self._committed = False

    def commit(self):
        self._committed = True
        print("✅ Transaction committed")

    def rollback(self):
        print("❌ Transaction rolled back")
        # In real SQLAlchemy: session.rollback()

# ---- Dependency (generator with cleanup) ----
def get_uow():
    uow = UnitOfWork()
    try:
        yield uow
        uow.commit()  # only if no exception was raised
    except Exception:
        uow.rollback()
        raise

# ---- Request model ----
class OrderRequest(BaseModel):
    item: str
    quantity: int

# ---- Route ----
@app.post("/orders", status_code=201)
def create_order(
    req: OrderRequest,
    uow: UnitOfWork = Depends(get_uow)
):
    try:
        uow.inventory.deduct(req.item, req.quantity)
        uow.orders.create({"item": req.item, "qty": req.quantity})
        return {"status": "Order placed!", "item": req.item}
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

# GET /orders:
# POST /orders {"item": "laptop", "quantity": 2} → places order, deducts inventory
# POST /orders {"item": "laptop", "quantity": 99} → 400 error, rolled back

app/
├── main.py              ← FastAPI app, mounts routers
├── routers/
│   └── users.py         ← Route handlers (controller layer)
├── services/
│   └── user_service.py  ← Business logic (service layer)
├── repositories/
│   └── user_repo.py     ← DB access (repository layer)
├── models/
│   └── user.py          ← Pydantic models
└── database.py          ← DB session dependency

# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

DATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db() -> Session:
    db = SessionLocal()
    try:
        yield db          # inject into routes/services
        db.commit()       # commit on success
    except Exception:
        db.rollback()     # rollback on error
        raise
    finally:
        db.close()        # always close

# repositories/user_repo.py
from fastapi import Depends
from sqlalchemy.orm import Session
from database import get_db

class UserRepository:
    def __init__(self, db: Session = Depends(get_db)):
        self.db = db

    def get_by_id(self, user_id: int):
        # return self.db.query(User).filter(User.id == user_id).first()
        return {"id": user_id, "name": "Alice"}  # simplified

# services/user_service.py
from fastapi import Depends, HTTPException
from repositories.user_repo import UserRepository

class UserService:
    def __init__(self, repo: UserRepository = Depends(UserRepository)):
        self.repo = repo

    def get_user(self, user_id: int) -> dict:
        user = self.repo.get_by_id(user_id)
        if not user:
            raise HTTPException(status_code=404, detail="User not found")
        return user

# routers/users.py
from fastapi import APIRouter, Depends
from services.user_service import UserService

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/{user_id}")
def get_user(user_id: int, service: UserService = Depends(UserService)):
    # FastAPI auto-resolves: UserService → UserRepository → get_db → Session
    return service.get_user(user_id)

# main.py
from fastapi import FastAPI
from routers import users

app = FastAPI()
app.include_router(users.router)

# FastAPI DI chain for GET /users/1:
# get_db() → UserRepository(db) → UserService(repo) → route(service)

# tests/test_users.py
from fastapi.testclient import TestClient
from main import app
from database import get_db

def get_test_db():
    # Return a test/mock DB session instead
    yield {"test_session": True}

# Override the dependency
app.dependency_overrides[get_db] = get_test_db

client = TestClient(app)

def test_get_user():
    response = client.get("/users/1")
    assert response.status_code == 200
    assert response.json()["id"] == 1