cc89b44f/main.py

"""

QR Code Generator with Analytics API — REST API Service





Features:



"""
import os
from contextlib import asynccontextmanager
from typing import Optional, List
from datetime import datetime, timezone

from fastapi import FastAPI, Depends, HTTPException, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from dotenv import load_dotenv

from models import ItemCreate, ItemUpdate, ItemResponse, HealthResponse
from auth import get_current_user, create_access_token, verify_password

load_dotenv()

limiter = Limiter(key_func=get_remote_address)
_db: dict = {}  # In-memory store — swap for real DB in production


@asynccontextmanager
async def lifespan(app: FastAPI):
    print(f"{app.title} starting up...")
    yield
    print(f"{app.title} shutting down...")


app = FastAPI(
    title="QR Code Generator with Analytics API",
    description="",
    version="1.0.0",
    lifespan=lifespan,
    docs_url="/docs",
    redoc_url="/redoc",
    openapi_url="/openapi.json",
)

app.state.limiter = limiter

app.add_middleware(
    CORSMiddleware,
    allow_origins=os.getenv("ALLOWED_ORIGINS", "*").split(","),
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded):
    return JSONResponse(
        status_code=status.HTTP_429_TOO_MANY_REQUESTS,
        content={"error": "Rate limit exceeded", "detail": str(exc)},
    )


@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"error": "Internal server error", "detail": str(exc)},
    )


# ── Health ────────────────────────────────────────────────────────────────────

@app.get("/health", response_model=HealthResponse, tags=["System"])
async def health():
    return {
        "status": "healthy",
        "service": "QR Code Generator with Analytics API",
        "version": "1.0.0",
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "items_count": len(_db),
    }


# ── Auth ──────────────────────────────────────────────────────────────────────

@app.post("/auth/token", tags=["Auth"], summary="Get API token")
@limiter.limit("10/minute")
async def login(request: Request, username: str, password: str):
    if not verify_password(username, password):
        raise HTTPException(status_code=401, detail="Invalid credentials")
    token = create_access_token({"sub": username})
    return {"access_token": token, "token_type": "bearer"}


# ── CRUD ──────────────────────────────────────────────────────────────────────

@app.get("/items", response_model=List[ItemResponse], tags=["Items"])
@limiter.limit("60/minute")
async def list_items(

    request: Request,

    skip: int = 0,

    limit: int = 50,

    search: Optional[str] = None,

    current_user: str = Depends(get_current_user),

):
    items = list(_db.values())
    if search:
        items = [i for i in items if search.lower() in i.get("name", "").lower()]
    return items[skip : skip + limit]


@app.post("/items", response_model=ItemResponse, status_code=201, tags=["Items"])
@limiter.limit("30/minute")
async def create_item(

    request: Request,

    item: ItemCreate,

    current_user: str = Depends(get_current_user),

):
    item_id = f"item_{len(_db) + 1:06d}"
    record = {
        "id": item_id,
        **item.model_dump(),
        "created_by": current_user,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "updated_at": datetime.now(timezone.utc).isoformat(),
    }
    _db[item_id] = record
    return record


@app.get("/items/{item_id}", response_model=ItemResponse, tags=["Items"])
@limiter.limit("120/minute")
async def get_item(

    request: Request,

    item_id: str,

    current_user: str = Depends(get_current_user),

):
    if item_id not in _db:
        raise HTTPException(status_code=404, detail=f"Item {item_id} not found")
    return _db[item_id]


@app.patch("/items/{item_id}", response_model=ItemResponse, tags=["Items"])
@limiter.limit("30/minute")
async def update_item(

    request: Request,

    item_id: str,

    item: ItemUpdate,

    current_user: str = Depends(get_current_user),

):
    if item_id not in _db:
        raise HTTPException(status_code=404, detail=f"Item {item_id} not found")
    record = _db[item_id]
    updates = item.model_dump(exclude_unset=True)
    record.update({**updates, "updated_at": datetime.now(timezone.utc).isoformat()})
    _db[item_id] = record
    return record


@app.delete("/items/{item_id}", status_code=204, tags=["Items"])
@limiter.limit("20/minute")
async def delete_item(

    request: Request,

    item_id: str,

    current_user: str = Depends(get_current_user),

):
    if item_id not in _db:
        raise HTTPException(status_code=404, detail=f"Item {item_id} not found")
    del _db[item_id]


# ── Stats ─────────────────────────────────────────────────────────────────────

@app.get("/stats", tags=["System"])
@limiter.limit("30/minute")
async def stats(request: Request, current_user: str = Depends(get_current_user)):
    return {
        "total_items": len(_db),
        "service": "QR Code Generator with Analytics API",
        "niche": "digital_marketing",
    }