diff --git a/docs/NEXTJS_INTEGRATION.md b/docs/NEXTJS_INTEGRATION.md new file mode 100644 index 0000000..86d6583 --- /dev/null +++ b/docs/NEXTJS_INTEGRATION.md @@ -0,0 +1,511 @@ +# Next.js Integration Guide for aiqso.io + +This guide explains how to integrate the QR Builder API with your Next.js website at aiqso.io. + +## Architecture Overview + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ aiqso.io (Next.js) │ +├─────────────────────────────────────────────────────────────────┤ +│ Frontend (React) │ Backend (API Routes) │ +│ - QR Builder UI │ - /api/qr-builder/validate-key │ +│ - User Portal │ - /api/qr-builder/create-key │ +│ - Stripe Checkout │ - Odoo Integration │ +└─────────────────────────────────────────────────────────────────┘ + │ + │ API Calls + ▼ +┌─────────────────────────────────────────────────────────────────┐ +│ QR Builder API (FastAPI) │ +├─────────────────────────────────────────────────────────────────┤ +│ /qr, /qr/logo, /qr/artistic │ /webhooks/update-tier │ +│ /qr/text, /qr/qart, /embed │ /usage/logs, /usage/stats │ +└─────────────────────────────────────────────────────────────────┘ +``` + +## Environment Variables + +Set these in your QR Builder API deployment: + +```bash +# Required for production +QR_BUILDER_AUTH_ENABLED=true +QR_BUILDER_BACKEND_SECRET=your-secure-secret-here +QR_BUILDER_BACKEND_URL=https://api.aiqso.io +QR_BUILDER_ALLOWED_ORIGINS=https://aiqso.io,https://www.aiqso.io + +# Optional +QR_BUILDER_HOST=0.0.0.0 +QR_BUILDER_PORT=8000 +``` + +## Step 1: Backend Endpoint for API Key Validation + +Create this endpoint in your Next.js API routes. The QR Builder API will call this to validate user API keys. + +```typescript +// pages/api/qr-builder/validate-key.ts (or app/api/qr-builder/validate-key/route.ts) + +import { NextRequest, NextResponse } from 'next/server'; + +const BACKEND_SECRET = process.env.QR_BUILDER_BACKEND_SECRET; + +export async function POST(request: NextRequest) { + // Verify the request is from QR Builder API + const authHeader = request.headers.get('authorization'); + if (authHeader !== `Bearer ${BACKEND_SECRET}`) { + return NextResponse.json({ valid: false, error: 'Unauthorized' }, { status: 401 }); + } + + const { api_key } = await request.json(); + + // Look up the API key in your database (Odoo or your user store) + // This is pseudocode - replace with your actual database logic + const user = await getUserByApiKey(api_key); + + if (!user) { + return NextResponse.json({ valid: false, error: 'Invalid API key' }); + } + + // Check if subscription is active + if (!user.subscriptionActive) { + return NextResponse.json({ valid: false, error: 'Subscription expired' }); + } + + return NextResponse.json({ + valid: true, + user_id: user.id, + tier: user.subscriptionTier, // 'free', 'pro', or 'business' + email: user.email, + }); +} + +// Helper function - implement based on your database +async function getUserByApiKey(apiKey: string) { + // Example with Odoo: + // const response = await fetch(`${ODOO_URL}/api/users/by-api-key`, { + // method: 'POST', + // headers: { 'Content-Type': 'application/json' }, + // body: JSON.stringify({ api_key: apiKey }), + // }); + // return response.json(); + + // Or with Prisma: + // return prisma.user.findUnique({ where: { apiKey } }); + + return null; // Replace with actual implementation +} +``` + +## Step 2: Generate API Keys for Users + +When a user signs up or upgrades, generate an API key for them: + +```typescript +// lib/qr-builder.ts + +import crypto from 'crypto'; + +export function generateApiKey(userId: string): string { + const random = crypto.randomBytes(16).toString('hex'); + return `qrb_${userId}_${random}`; +} + +// When user signs up or upgrades +async function onUserSubscribe(userId: string, tier: 'free' | 'pro' | 'business') { + const apiKey = generateApiKey(userId); + + // Store in your database + await saveUserApiKey(userId, apiKey, tier); + + return apiKey; +} +``` + +## Step 3: Frontend QR Builder Component + +```tsx +// components/QRBuilder.tsx + +'use client'; + +import { useState } from 'react'; + +const QR_BUILDER_API = process.env.NEXT_PUBLIC_QR_BUILDER_API || 'https://qr-api.aiqso.io'; + +interface QRBuilderProps { + apiKey: string; + userTier: 'free' | 'pro' | 'business'; +} + +export function QRBuilder({ apiKey, userTier }: QRBuilderProps) { + const [data, setData] = useState(''); + const [style, setStyle] = useState('basic'); + const [loading, setLoading] = useState(false); + const [qrImage, setQrImage] = useState(null); + const [error, setError] = useState(null); + + // Available styles based on tier + const availableStyles = userTier === 'free' + ? ['basic', 'text'] + : ['basic', 'text', 'logo', 'artistic', 'qart', 'embed']; + + const generateQR = async () => { + setLoading(true); + setError(null); + + try { + const formData = new FormData(); + formData.append('data', data); + formData.append('size', '500'); + + const response = await fetch(`${QR_BUILDER_API}/qr`, { + method: 'POST', + headers: { + 'X-API-Key': apiKey, + }, + body: formData, + }); + + if (!response.ok) { + const errorData = await response.json(); + throw new Error(errorData.detail || 'Failed to generate QR'); + } + + const blob = await response.blob(); + setQrImage(URL.createObjectURL(blob)); + } catch (err) { + setError(err instanceof Error ? err.message : 'Unknown error'); + } finally { + setLoading(false); + } + }; + + const generateLogoQR = async (logoFile: File) => { + if (userTier === 'free') { + setError('Logo QR requires Pro tier. Upgrade at /portal/upgrade'); + return; + } + + setLoading(true); + setError(null); + + try { + const formData = new FormData(); + formData.append('data', data); + formData.append('logo', logoFile); + formData.append('size', '500'); + + const response = await fetch(`${QR_BUILDER_API}/qr/logo`, { + method: 'POST', + headers: { + 'X-API-Key': apiKey, + }, + body: formData, + }); + + if (response.status === 403) { + setError('This feature requires a Pro subscription. Upgrade at /portal/upgrade'); + return; + } + + if (response.status === 429) { + setError('Rate limit exceeded. Please wait a moment and try again.'); + return; + } + + if (!response.ok) { + const errorData = await response.json(); + throw new Error(errorData.detail || 'Failed to generate QR'); + } + + const blob = await response.blob(); + setQrImage(URL.createObjectURL(blob)); + } catch (err) { + setError(err instanceof Error ? err.message : 'Unknown error'); + } finally { + setLoading(false); + } + }; + + return ( +
+
+ setData(e.target.value)} + placeholder="Enter URL or text" + /> + + + + +
+ + {error &&
{error}
} + + {qrImage && ( +
+ Generated QR Code + Download +
+ )} + + {userTier === 'free' && ( +
+ Want logo QR codes and more? Upgrade to Pro +
+ )} +
+ ); +} +``` + +## Step 4: Stripe Webhook for Tier Updates + +When a user's subscription changes, update their tier in the QR Builder API: + +```typescript +// pages/api/webhooks/stripe.ts + +import { NextRequest, NextResponse } from 'next/server'; +import Stripe from 'stripe'; + +const stripe = new Stripe(process.env.STRIPE_SECRET_KEY!); +const QR_BUILDER_API = process.env.QR_BUILDER_API_URL; +const QR_BUILDER_SECRET = process.env.QR_BUILDER_BACKEND_SECRET; + +export async function POST(request: NextRequest) { + const body = await request.text(); + const sig = request.headers.get('stripe-signature')!; + + let event: Stripe.Event; + + try { + event = stripe.webhooks.constructEvent(body, sig, process.env.STRIPE_WEBHOOK_SECRET!); + } catch (err) { + return NextResponse.json({ error: 'Webhook signature failed' }, { status: 400 }); + } + + switch (event.type) { + case 'customer.subscription.created': + case 'customer.subscription.updated': { + const subscription = event.data.object as Stripe.Subscription; + const tier = getTierFromPriceId(subscription.items.data[0].price.id); + const apiKey = await getApiKeyForCustomer(subscription.customer as string); + + if (apiKey) { + // Update tier in QR Builder API + await fetch(`${QR_BUILDER_API}/webhooks/update-tier`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Webhook-Secret': QR_BUILDER_SECRET!, + }, + body: JSON.stringify({ api_key: apiKey, tier }), + }); + } + break; + } + + case 'customer.subscription.deleted': { + const subscription = event.data.object as Stripe.Subscription; + const apiKey = await getApiKeyForCustomer(subscription.customer as string); + + if (apiKey) { + // Downgrade to free or invalidate + await fetch(`${QR_BUILDER_API}/webhooks/update-tier`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Webhook-Secret': QR_BUILDER_SECRET!, + }, + body: JSON.stringify({ api_key: apiKey, tier: 'free' }), + }); + } + break; + } + } + + return NextResponse.json({ received: true }); +} + +function getTierFromPriceId(priceId: string): string { + // Map your Stripe price IDs to tiers + const tierMap: Record = { + 'price_pro_monthly': 'pro', + 'price_pro_yearly': 'pro', + 'price_business_monthly': 'business', + 'price_business_yearly': 'business', + }; + return tierMap[priceId] || 'free'; +} + +async function getApiKeyForCustomer(customerId: string): Promise { + // Look up in your database + return null; // Replace with actual implementation +} +``` + +## Step 5: Odoo Integration for Usage Tracking + +Sync usage data from QR Builder to Odoo: + +```typescript +// lib/odoo-sync.ts + +const QR_BUILDER_API = process.env.QR_BUILDER_API_URL; +const QR_BUILDER_SECRET = process.env.QR_BUILDER_BACKEND_SECRET; + +let lastSyncTimestamp = 0; + +export async function syncUsageToOdoo() { + // Get usage logs since last sync + const response = await fetch( + `${QR_BUILDER_API}/usage/logs?since=${lastSyncTimestamp}`, + { + headers: { + 'X-Webhook-Secret': QR_BUILDER_SECRET!, + }, + } + ); + + const { logs, latest_timestamp } = await response.json(); + + if (logs.length === 0) return; + + // Send to Odoo + for (const log of logs) { + await sendToOdoo({ + model: 'qr.usage.log', + method: 'create', + args: [{ + user_id: log.user_id, + style: log.style, + success: log.success, + timestamp: new Date(log.timestamp * 1000).toISOString(), + metadata: JSON.stringify(log.metadata), + }], + }); + } + + lastSyncTimestamp = latest_timestamp; + + // Cleanup old logs in QR Builder (optional) + await fetch(`${QR_BUILDER_API}/usage/cleanup`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'X-Webhook-Secret': QR_BUILDER_SECRET!, + }, + body: JSON.stringify({ days: 7 }), // Keep 7 days in QR Builder + }); +} + +// Run this periodically (e.g., cron job every hour) +``` + +## Step 6: User Portal Page + +```tsx +// app/portal/qr-builder/page.tsx + +import { getServerSession } from 'next-auth'; +import { QRBuilder } from '@/components/QRBuilder'; +import { getUserSubscription, getUserApiKey } from '@/lib/user'; + +export default async function QRBuilderPage() { + const session = await getServerSession(); + + if (!session?.user) { + return
Please sign in to use QR Builder
; + } + + const subscription = await getUserSubscription(session.user.id); + const apiKey = await getUserApiKey(session.user.id); + + // If no API key, create one + if (!apiKey) { + // Redirect to setup or create automatically + } + + return ( +
+

QR Code Generator

+ +
+

Current Plan: {subscription.tier}

+

QR Codes Today: {subscription.usageToday} / {subscription.dailyLimit}

+ {subscription.tier === 'free' && ( + Upgrade for Logo QR + )} +
+ + +
+ ); +} +``` + +## API Reference + +### QR Generation Endpoints + +| Endpoint | Tier | Description | +|----------|------|-------------| +| `POST /qr` | Free | Basic QR code | +| `POST /qr/text` | Free | QR with text overlay | +| `POST /qr/logo` | Pro | QR with logo | +| `POST /qr/artistic` | Pro | Image blended into QR | +| `POST /qr/qart` | Pro | Halftone style | +| `POST /embed` | Pro | QR on background image | +| `POST /batch/embed` | Pro | Batch processing | + +### Backend Integration Endpoints + +| Endpoint | Auth | Description | +|----------|------|-------------| +| `POST /webhooks/update-tier` | X-Webhook-Secret | Update user tier | +| `POST /webhooks/invalidate-key` | X-Webhook-Secret | Invalidate API key | +| `GET /usage/logs` | X-Webhook-Secret | Get usage logs | +| `GET /usage/stats/{user_id}` | X-Webhook-Secret | Get user stats | + +### Response Headers + +The API includes helpful headers: + +- `X-RateLimit-Limit`: Requests allowed per minute +- `X-RateLimit-Remaining`: Requests remaining +- `X-Required-Tier`: Tier needed for blocked features + +## Pricing Recommendations + +Based on market research: + +| Tier | Price | Features | +|------|-------|----------| +| **Free** | $0 | Basic + Text QR, 10/day | +| **Pro** | $5/month | All styles, 500/day, batch (10) | +| **Business** | $15/month | All styles, 5000/day, batch (50) | + +Or per-QR pricing: +- Basic QR: Free +- Logo QR: $1 +- Artistic QR: $2 + +## Security Checklist + +- [ ] Set strong `QR_BUILDER_BACKEND_SECRET` +- [ ] Configure `QR_BUILDER_ALLOWED_ORIGINS` for your domains only +- [ ] Store API keys securely (hashed in database) +- [ ] Implement proper error handling for rate limits +- [ ] Set up monitoring for usage anomalies +- [ ] Use HTTPS for all communications diff --git a/qr_builder/__init__.py b/qr_builder/__init__.py index 3926a1b..1833a87 100644 --- a/qr_builder/__init__.py +++ b/qr_builder/__init__.py @@ -24,7 +24,16 @@ VALID_POSITIONS, ) -__version__ = "0.2.0" +from .auth import ( + UserTier, + TierLimits, + UserSession, + TIER_LIMITS, + get_all_tiers_info, + get_tier_info, +) + +__version__ = "0.3.0" __all__ = [ # Basic functions @@ -45,6 +54,13 @@ "QRConfig", "QRStyle", "ARTISTIC_PRESETS", + # Auth & Tiers + "UserTier", + "TierLimits", + "UserSession", + "TIER_LIMITS", + "get_all_tiers_info", + "get_tier_info", # Constants "MAX_DATA_LENGTH", "MAX_QR_SIZE", diff --git a/qr_builder/api.py b/qr_builder/api.py index 3df6bd9..d4e733b 100644 --- a/qr_builder/api.py +++ b/qr_builder/api.py @@ -11,12 +11,21 @@ - /qr/qart - Halftone/dithered style - /embed - QR placed on background - /batch/embed - Batch processing +- /webhooks/* - Backend integration endpoints +- /usage/* - Usage tracking for Odoo Entry points (after install): qr-builder-api # convenience wrapper defined in pyproject.toml Or manually: uvicorn qr_builder.api:app --reload + +Integration with aiqso.io: + Set environment variables: + - QR_BUILDER_AUTH_ENABLED=true + - QR_BUILDER_BACKEND_SECRET=your-secret + - QR_BUILDER_BACKEND_URL=https://api.aiqso.io + - QR_BUILDER_ALLOWED_ORIGINS=https://aiqso.io,https://www.aiqso.io """ from __future__ import annotations @@ -25,13 +34,14 @@ import logging import zipfile import tempfile +import time from pathlib import Path from typing import List, Optional from enum import Enum -from fastapi import FastAPI, UploadFile, File, Form, HTTPException, Query +from fastapi import FastAPI, UploadFile, File, Form, HTTPException, Query, Depends, Body from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import StreamingResponse +from fastapi.responses import StreamingResponse, JSONResponse from .core import ( generate_qr, @@ -44,6 +54,21 @@ ARTISTIC_PRESETS, ) +from .auth import ( + UserSession, + UserTier, + get_current_user, + require_auth, + check_rate_limit, + require_style, + verify_backend_webhook, + session_store, + get_tier_info, + get_all_tiers_info, + ALLOWED_ORIGINS, + AUTH_ENABLED, +) + logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) @@ -52,16 +77,33 @@ description=""" Generate and embed QR codes into images via HTTP. +## Authentication + +Include your API key in the `X-API-Key` header: +``` +X-API-Key: your_api_key_here +``` + +Get your API key at [aiqso.io/portal](https://aiqso.io/portal) + +## Tiers + +| Tier | Styles | Daily Limit | Batch | +|------|--------|-------------|-------| +| **Free** | basic, text | 10/day | No | +| **Pro** | All styles | 500/day | 10 images | +| **Business** | All styles | 5000/day | 50 images | + ## Available Styles -| Style | Endpoint | Description | -|-------|----------|-------------| -| **Basic** | `/qr` | Simple QR with custom colors | -| **Logo** | `/qr/logo` | Logo embedded in QR center | -| **Text** | `/qr/text` | Text/words in QR center | -| **Artistic** | `/qr/artistic` | Image IS the QR code (colorful) | -| **QArt** | `/qr/qart` | Halftone/dithered style | -| **Embed** | `/embed` | QR placed on background image | +| Style | Endpoint | Description | Tier | +|-------|----------|-------------|------| +| **Basic** | `/qr` | Simple QR with custom colors | Free | +| **Text** | `/qr/text` | Text/words in QR center | Free | +| **Logo** | `/qr/logo` | Logo embedded in QR center | Pro+ | +| **Artistic** | `/qr/artistic` | Image IS the QR code (colorful) | Pro+ | +| **QArt** | `/qr/qart` | Halftone/dithered style | Pro+ | +| **Embed** | `/embed` | QR placed on background image | Pro+ | ## Presets (Artistic mode) - `small` - Compact, high contrast (version 5) @@ -69,16 +111,17 @@ - `large` - High detail (version 15) - `hd` - Maximum detail (version 20) """, - version="0.2.0", + version="0.3.0", ) -# CORS middleware for web integration +# CORS middleware - configured for aiqso.io app.add_middleware( CORSMiddleware, - allow_origins=["*"], # Configure appropriately for production + allow_origins=ALLOWED_ORIGINS if AUTH_ENABLED else ["*"], allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], + allow_methods=["GET", "POST", "OPTIONS"], + allow_headers=["*", "X-API-Key", "X-Webhook-Secret"], + expose_headers=["X-RateLimit-Limit", "X-RateLimit-Remaining", "X-Required-Tier"], ) @@ -102,19 +145,26 @@ class PresetEnum(str, Enum): @app.get("/health", tags=["meta"]) async def health() -> dict: """Health check endpoint.""" - return {"status": "ok"} + return {"status": "ok", "auth_enabled": AUTH_ENABLED} @app.get("/styles", tags=["meta"]) -async def list_styles() -> dict: - """List all available QR code styles and presets.""" +async def list_styles(user: UserSession = Depends(get_current_user)) -> dict: + """List all available QR code styles and presets for current user.""" + user_styles = user.limits.allowed_styles + all_styles = [ + {"name": "basic", "description": "Simple QR with custom colors", "requires_image": False, "tier": "free"}, + {"name": "text", "description": "Text/words in QR center", "requires_image": False, "tier": "free"}, + {"name": "logo", "description": "Logo embedded in QR center", "requires_image": True, "tier": "pro"}, + {"name": "artistic", "description": "Image IS the QR code (colorful)", "requires_image": True, "tier": "pro"}, + {"name": "qart", "description": "Halftone/dithered style", "requires_image": True, "tier": "pro"}, + {"name": "embed", "description": "QR placed on background image", "requires_image": True, "tier": "pro"}, + ] + return { "styles": [ - {"name": "basic", "description": "Simple QR with custom colors", "requires_image": False}, - {"name": "logo", "description": "Logo embedded in QR center", "requires_image": True}, - {"name": "artistic", "description": "Image IS the QR code (colorful)", "requires_image": True}, - {"name": "qart", "description": "Halftone/dithered style", "requires_image": True}, - {"name": "embed", "description": "QR placed on background image", "requires_image": True}, + {**style, "available": style["name"] in user_styles} + for style in all_styles ], "artistic_presets": [ {"name": "small", "version": 5, "description": "Compact, high contrast"}, @@ -122,6 +172,38 @@ async def list_styles() -> dict: {"name": "large", "version": 15, "description": "High detail"}, {"name": "hd", "version": 20, "description": "Maximum detail"}, ], + "user_tier": user.tier.value, + "custom_colors": user.can_use_custom_colors(), + } + + +@app.get("/tiers", tags=["meta"]) +async def list_tiers() -> dict: + """List all available tiers and their features (for pricing page).""" + return {"tiers": get_all_tiers_info()} + + +@app.get("/me", tags=["meta"]) +async def get_current_user_info(user: UserSession = Depends(get_current_user)) -> dict: + """Get current user's tier, limits, and usage.""" + return { + "user_id": user.user_id, + "tier": user.tier.value, + "email": user.email, + "limits": { + "requests_per_minute": user.limits.requests_per_minute, + "requests_per_day": user.limits.requests_per_day, + "max_qr_size": user.limits.max_qr_size, + "batch_limit": user.limits.batch_limit, + }, + "usage": { + "requests_this_minute": user.requests_this_minute, + "requests_today": user.requests_today, + }, + "features": { + "allowed_styles": user.limits.allowed_styles, + "custom_colors": user.limits.custom_colors, + }, } @@ -135,8 +217,24 @@ async def create_qr( size: int = Form(500, description="Pixel size of the QR image."), fill_color: str = Form("black", description="QR foreground color."), back_color: str = Form("white", description="QR background color."), + user: UserSession = Depends(require_style("basic")), ): - """Generate a basic standalone QR code and return as PNG.""" + """Generate a basic standalone QR code and return as PNG. (Free tier)""" + # Check size limit for tier + if size > user.limits.max_qr_size: + raise HTTPException( + status_code=403, + detail=f"Size {size} exceeds your tier limit of {user.limits.max_qr_size}px. " + f"Upgrade at https://aiqso.io/portal", + ) + + # Check custom colors + if (fill_color.startswith("#") or back_color.startswith("#")) and not user.can_use_custom_colors(): + raise HTTPException( + status_code=403, + detail="Custom hex colors require Pro tier. Upgrade at https://aiqso.io/portal", + ) + try: img = generate_qr( data=data, @@ -146,8 +244,12 @@ async def create_qr( ) except Exception as exc: logger.exception("Failed to generate QR.") + session_store.log_usage(user.user_id, "basic", False, {"error": str(exc)}) raise HTTPException(status_code=400, detail=str(exc)) + # Log successful generation + session_store.log_usage(user.user_id, "basic", True, {"size": size}) + buf = io.BytesIO() img.save(buf, format="PNG") buf.seek(0) @@ -166,8 +268,16 @@ async def create_qr_with_logo( logo_scale: float = Form(0.25, description="Logo size as fraction of QR (0.1-0.4)."), fill_color: str = Form("black", description="QR foreground color."), back_color: str = Form("white", description="QR background color."), + user: UserSession = Depends(require_style("logo")), ): - """Generate a QR code with logo embedded in the center.""" + """Generate a QR code with logo embedded in the center. (Pro tier)""" + # Check size limit + if size > user.limits.max_qr_size: + raise HTTPException( + status_code=403, + detail=f"Size {size} exceeds your tier limit of {user.limits.max_qr_size}px", + ) + try: # Save uploaded file temporarily with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as tmp: @@ -194,11 +304,14 @@ async def create_qr_with_logo( out_path.unlink(missing_ok=True) except ValueError as ve: + session_store.log_usage(user.user_id, "logo", False, {"error": str(ve)}) raise HTTPException(status_code=400, detail=str(ve)) except Exception as exc: logger.exception("Failed to generate QR with logo.") + session_store.log_usage(user.user_id, "logo", False, {"error": str(exc)}) raise HTTPException(status_code=500, detail=str(exc)) + session_store.log_usage(user.user_id, "logo", True, {"size": size}) return StreamingResponse(io.BytesIO(result), media_type="image/png") @@ -216,8 +329,16 @@ async def create_qr_with_text_endpoint( back_color: str = Form("white", description="QR background color."), font_color: str = Form("black", description="Text color."), font_size: int = Form(None, description="Font size in pixels (auto if not set)."), + user: UserSession = Depends(require_style("text")), ): - """Generate a QR code with text/words embedded in the center.""" + """Generate a QR code with text/words embedded in the center. (Free tier)""" + # Check size limit + if size > user.limits.max_qr_size: + raise HTTPException( + status_code=403, + detail=f"Size {size} exceeds your tier limit of {user.limits.max_qr_size}px", + ) + try: with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as out: out_path = Path(out.name) @@ -239,11 +360,14 @@ async def create_qr_with_text_endpoint( out_path.unlink(missing_ok=True) except ValueError as ve: + session_store.log_usage(user.user_id, "text", False, {"error": str(ve)}) raise HTTPException(status_code=400, detail=str(ve)) except Exception as exc: logger.exception("Failed to generate QR with text.") + session_store.log_usage(user.user_id, "text", False, {"error": str(exc)}) raise HTTPException(status_code=500, detail=str(exc)) + session_store.log_usage(user.user_id, "text", True, {"size": size}) return StreamingResponse(io.BytesIO(result), media_type="image/png") @@ -260,9 +384,10 @@ async def create_artistic_qr( contrast: float = Form(1.0, description="Image contrast (try 1.2-1.5)."), brightness: float = Form(1.0, description="Image brightness (try 1.1-1.2)."), colorized: bool = Form(True, description="Keep colors (False for B&W)."), + user: UserSession = Depends(require_style("artistic")), ): """ - Generate an artistic QR code where the image IS the QR code. + Generate an artistic QR code where the image IS the QR code. (Pro tier) The image is blended into the QR pattern itself, creating a visually striking QR code that remains scannable. @@ -307,8 +432,10 @@ async def create_artistic_qr( except Exception as exc: logger.exception("Failed to generate artistic QR.") + session_store.log_usage(user.user_id, "artistic", False, {"error": str(exc)}) raise HTTPException(status_code=500, detail=str(exc)) + session_store.log_usage(user.user_id, "artistic", True, {"preset": preset.value if preset else "custom"}) return StreamingResponse(io.BytesIO(result), media_type="image/png") @@ -327,9 +454,10 @@ async def create_qart( color_r: int = Form(0, description="Red component (0-255)."), color_g: int = Form(0, description="Green component (0-255)."), color_b: int = Form(0, description="Blue component (0-255)."), + user: UserSession = Depends(require_style("qart")), ): """ - Generate a QArt-style halftone/dithered QR code. + Generate a QArt-style halftone/dithered QR code. (Pro tier) Creates a black & white (or single color) artistic QR using dithering techniques. Good for minimalist designs. @@ -366,8 +494,10 @@ async def create_qart( except Exception as exc: logger.exception("Failed to generate QArt.") + session_store.log_usage(user.user_id, "qart", False, {"error": str(exc)}) raise HTTPException(status_code=500, detail=str(exc)) + session_store.log_usage(user.user_id, "qart", True) return StreamingResponse(io.BytesIO(result), media_type="image/png") @@ -384,8 +514,9 @@ async def embed_qr( margin: int = Form(20, description="Margin from edge in pixels."), fill_color: str = Form("black", description="QR foreground color."), back_color: str = Form("white", description="QR background color."), + user: UserSession = Depends(require_style("embed")), ): - """Embed a QR into an uploaded background image and return the result as PNG.""" + """Embed a QR into an uploaded background image and return the result as PNG. (Pro tier)""" try: raw = await background.read() tmp_buf = io.BytesIO(raw) @@ -415,11 +546,14 @@ async def embed_qr( except ValueError as ve: logger.warning("Bad request for /embed: %s", ve) + session_store.log_usage(user.user_id, "embed", False, {"error": str(ve)}) raise HTTPException(status_code=400, detail=str(ve)) except Exception: logger.exception("Failed to embed QR.") + session_store.log_usage(user.user_id, "embed", False) raise HTTPException(status_code=500, detail="Internal server error") + session_store.log_usage(user.user_id, "embed", True) return StreamingResponse(out_buf, media_type="image/png") @@ -436,12 +570,27 @@ async def batch_embed_qr( margin: int = Form(20), fill_color: str = Form("black"), back_color: str = Form("white"), + user: UserSession = Depends(require_style("embed")), ): """ - Embed the same QR into multiple uploaded background images and return a ZIP. + Embed the same QR into multiple uploaded background images and return a ZIP. (Pro tier) Filenames inside the ZIP will be the original filename with `_qr` appended. """ + # Check batch limit + if len(backgrounds) > user.get_max_batch_size(): + raise HTTPException( + status_code=403, + detail=f"Batch size {len(backgrounds)} exceeds your tier limit of {user.get_max_batch_size()}. " + f"Upgrade at https://aiqso.io/portal", + ) + + if user.get_max_batch_size() == 0: + raise HTTPException( + status_code=403, + detail="Batch processing requires Pro or Business tier. Upgrade at https://aiqso.io/portal", + ) + try: zip_buf = io.BytesIO() with zipfile.ZipFile(zip_buf, "w", zipfile.ZIP_DEFLATED) as zf: @@ -483,11 +632,14 @@ async def batch_embed_qr( except ValueError as ve: logger.warning("Bad request for /batch/embed: %s", ve) + session_store.log_usage(user.user_id, "batch_embed", False, {"error": str(ve)}) raise HTTPException(status_code=400, detail=str(ve)) except Exception: logger.exception("Failed to batch embed QR.") + session_store.log_usage(user.user_id, "batch_embed", False) raise HTTPException(status_code=500, detail="Internal server error") + session_store.log_usage(user.user_id, "batch_embed", True, {"count": len(backgrounds)}) return StreamingResponse( zip_buf, media_type="application/zip", @@ -500,10 +652,24 @@ async def batch_artistic_qr( images: List[UploadFile] = File(..., description="Multiple images to transform."), data: str = Form(..., description="Text or URL to encode."), preset: Optional[PresetEnum] = Form(PresetEnum.large, description="Quality preset."), + user: UserSession = Depends(require_style("artistic")), ): """ - Generate artistic QR codes from multiple images and return a ZIP. + Generate artistic QR codes from multiple images and return a ZIP. (Pro tier) """ + # Check batch limit + if len(images) > user.get_max_batch_size(): + raise HTTPException( + status_code=403, + detail=f"Batch size {len(images)} exceeds your tier limit of {user.get_max_batch_size()}", + ) + + if user.get_max_batch_size() == 0: + raise HTTPException( + status_code=403, + detail="Batch processing requires Pro or Business tier. Upgrade at https://aiqso.io/portal", + ) + try: p = ARTISTIC_PRESETS[preset.value] version = p["version"] @@ -552,8 +718,10 @@ async def batch_artistic_qr( except Exception: logger.exception("Failed to batch generate artistic QR.") + session_store.log_usage(user.user_id, "batch_artistic", False) raise HTTPException(status_code=500, detail="Internal server error") + session_store.log_usage(user.user_id, "batch_artistic", True, {"count": len(images)}) return StreamingResponse( zip_buf, media_type="application/zip", @@ -561,6 +729,145 @@ async def batch_artistic_qr( ) +# ============================================================================= +# Webhook Endpoints (for aiqso.io backend integration) +# ============================================================================= + +@app.post("/webhooks/update-tier", tags=["webhooks"]) +async def webhook_update_tier( + api_key: str = Body(..., embed=True), + tier: str = Body(..., embed=True), + _: bool = Depends(verify_backend_webhook), +): + """ + Update a user's tier (called from your Odoo/Next.js backend). + + Headers required: + X-Webhook-Secret: your-backend-secret + + Body: + { + "api_key": "user_api_key_here", + "tier": "pro" // free, pro, or business + } + """ + try: + new_tier = UserTier(tier) + except ValueError: + raise HTTPException(status_code=400, detail=f"Invalid tier: {tier}") + + success = session_store.update_user_tier(api_key, new_tier) + + return { + "success": success, + "message": f"Tier updated to {tier}" if success else "User session not found (will apply on next request)", + } + + +@app.post("/webhooks/invalidate-key", tags=["webhooks"]) +async def webhook_invalidate_key( + api_key: str = Body(..., embed=True), + _: bool = Depends(verify_backend_webhook), +): + """ + Invalidate a user's API key (called when subscription is cancelled). + + This removes their session from the cache, forcing re-validation + on next request. + """ + if api_key in session_store._sessions: + del session_store._sessions[api_key] + return {"success": True, "message": "Session invalidated"} + + return {"success": True, "message": "Session not found (already invalidated)"} + + +# ============================================================================= +# Usage Tracking Endpoints (for Odoo integration) +# ============================================================================= + +@app.get("/usage/logs", tags=["usage"]) +async def get_usage_logs( + since: float = Query(0, description="Unix timestamp to get logs since"), + _: bool = Depends(verify_backend_webhook), +): + """ + Get usage logs since a timestamp (for Odoo sync). + + Your Odoo integration should call this periodically to sync usage data. + + Headers required: + X-Webhook-Secret: your-backend-secret + + Query params: + since: Unix timestamp (default 0 = all logs) + + Returns: + { + "logs": [ + { + "timestamp": 1234567890.123, + "user_id": "user_123", + "style": "logo", + "success": true, + "metadata": {"size": 500} + }, + ... + ], + "count": 42, + "latest_timestamp": 1234567890.123 + } + """ + logs = session_store.get_usage_since(since) + + return { + "logs": logs, + "count": len(logs), + "latest_timestamp": max((log["timestamp"] for log in logs), default=since), + } + + +@app.get("/usage/stats/{user_id}", tags=["usage"]) +async def get_user_stats( + user_id: str, + _: bool = Depends(verify_backend_webhook), +): + """ + Get usage statistics for a specific user. + + Headers required: + X-Webhook-Secret: your-backend-secret + + Returns: + { + "user_id": "user_123", + "total_requests": 42, + "successful": 40, + "by_style": { + "basic": 20, + "logo": 15, + "artistic": 5 + } + } + """ + stats = session_store.get_user_stats(user_id) + return {"user_id": user_id, **stats} + + +@app.post("/usage/cleanup", tags=["usage"]) +async def cleanup_old_logs( + days: int = Body(30, embed=True), + _: bool = Depends(verify_backend_webhook), +): + """ + Clean up logs older than N days. + + Call this periodically to prevent memory growth. + """ + removed = session_store.clear_old_logs(days) + return {"success": True, "removed_count": removed} + + def run() -> None: """Convenience entrypoint for `qr-builder-api` script.""" import os diff --git a/qr_builder/auth.py b/qr_builder/auth.py new file mode 100644 index 0000000..7b215df --- /dev/null +++ b/qr_builder/auth.py @@ -0,0 +1,546 @@ +""" +qr_builder.auth +--------------- + +Authentication, authorization, and rate limiting for QR Builder API. + +This module provides middleware and utilities for: +- API key authentication (validates tokens from your backend) +- Tier-based access control (free, pro, business) +- Rate limiting with configurable limits per tier +- Usage tracking for Odoo integration + +Integration with aiqso.io: +- Your Next.js frontend sends API keys from authenticated users +- Your Odoo backend manages user tiers and tracks usage +- This module validates requests and enforces limits +""" + +from __future__ import annotations + +import os +import time +import hashlib +import logging +from enum import Enum +from typing import Optional, Callable +from dataclasses import dataclass, field +from collections import defaultdict + +from fastapi import Request, HTTPException, Header, Depends +from fastapi.security import APIKeyHeader + +logger = logging.getLogger(__name__) + +# ============================================================================= +# Configuration (from environment variables) +# ============================================================================= + +# Backend secret for webhook authentication (set this in your deployment) +BACKEND_SECRET = os.getenv("QR_BUILDER_BACKEND_SECRET", "change-me-in-production") + +# Your backend URL for token validation (Odoo/Next.js backend) +BACKEND_VALIDATION_URL = os.getenv("QR_BUILDER_BACKEND_URL", "https://api.aiqso.io") + +# Enable/disable authentication (disable for local development) +AUTH_ENABLED = os.getenv("QR_BUILDER_AUTH_ENABLED", "true").lower() == "true" + +# Allowed origins for CORS +ALLOWED_ORIGINS = os.getenv( + "QR_BUILDER_ALLOWED_ORIGINS", + "https://aiqso.io,https://www.aiqso.io,https://api.aiqso.io" +).split(",") + + +# ============================================================================= +# Tier Definitions +# ============================================================================= + +class UserTier(str, Enum): + """User subscription tiers.""" + FREE = "free" + PRO = "pro" + BUSINESS = "business" + ADMIN = "admin" # Internal/backend use + + +@dataclass +class TierLimits: + """Rate limits and feature access per tier.""" + requests_per_minute: int + requests_per_day: int + max_qr_size: int + allowed_styles: list[str] + batch_limit: int # Max images in batch request + custom_colors: bool + priority: int # Higher = more priority in queue + + +# Tier configuration - adjust these values as needed +TIER_LIMITS: dict[UserTier, TierLimits] = { + UserTier.FREE: TierLimits( + requests_per_minute=5, + requests_per_day=10, + max_qr_size=500, + allowed_styles=["basic", "text"], # Free: basic + text only + batch_limit=0, # No batch for free + custom_colors=False, + priority=1, + ), + UserTier.PRO: TierLimits( + requests_per_minute=30, + requests_per_day=500, + max_qr_size=2000, + allowed_styles=["basic", "text", "logo", "artistic", "qart", "embed"], + batch_limit=10, + custom_colors=True, + priority=5, + ), + UserTier.BUSINESS: TierLimits( + requests_per_minute=100, + requests_per_day=5000, + max_qr_size=4000, + allowed_styles=["basic", "text", "logo", "artistic", "qart", "embed"], + batch_limit=50, + custom_colors=True, + priority=10, + ), + UserTier.ADMIN: TierLimits( + requests_per_minute=1000, + requests_per_day=100000, + max_qr_size=4000, + allowed_styles=["basic", "text", "logo", "artistic", "qart", "embed"], + batch_limit=100, + custom_colors=True, + priority=100, + ), +} + + +# ============================================================================= +# User Session Data +# ============================================================================= + +@dataclass +class UserSession: + """Authenticated user session data.""" + user_id: str + tier: UserTier + api_key: str + email: Optional[str] = None + + # Rate limiting tracking + requests_this_minute: int = 0 + requests_today: int = 0 + minute_reset_time: float = field(default_factory=time.time) + day_reset_time: float = field(default_factory=time.time) + + @property + def limits(self) -> TierLimits: + return TIER_LIMITS[self.tier] + + def check_rate_limit(self) -> tuple[bool, str]: + """Check if user is within rate limits. Returns (allowed, reason).""" + now = time.time() + + # Reset minute counter if minute has passed + if now - self.minute_reset_time > 60: + self.requests_this_minute = 0 + self.minute_reset_time = now + + # Reset daily counter if day has passed + if now - self.day_reset_time > 86400: + self.requests_today = 0 + self.day_reset_time = now + + # Check limits + if self.requests_this_minute >= self.limits.requests_per_minute: + return False, f"Rate limit exceeded: {self.limits.requests_per_minute}/minute" + + if self.requests_today >= self.limits.requests_per_day: + return False, f"Daily limit exceeded: {self.limits.requests_per_day}/day" + + return True, "OK" + + def record_request(self) -> None: + """Record a request for rate limiting.""" + self.requests_this_minute += 1 + self.requests_today += 1 + + def can_access_style(self, style: str) -> bool: + """Check if user's tier allows access to a style.""" + return style in self.limits.allowed_styles + + def can_use_custom_colors(self) -> bool: + """Check if user can use custom hex colors.""" + return self.limits.custom_colors + + def get_max_batch_size(self) -> int: + """Get maximum batch size for user's tier.""" + return self.limits.batch_limit + + +# ============================================================================= +# In-Memory Session Store (for rate limiting) +# ============================================================================= + +class SessionStore: + """ + In-memory session store for rate limiting. + + In production, consider using Redis for distributed rate limiting + across multiple instances. + """ + + def __init__(self): + self._sessions: dict[str, UserSession] = {} + self._usage_log: list[dict] = [] # For Odoo sync + + def get_or_create_session( + self, + user_id: str, + tier: UserTier, + api_key: str, + email: Optional[str] = None, + ) -> UserSession: + """Get existing session or create new one.""" + if api_key not in self._sessions: + self._sessions[api_key] = UserSession( + user_id=user_id, + tier=tier, + api_key=api_key, + email=email, + ) + else: + # Update tier in case it changed + self._sessions[api_key].tier = tier + + return self._sessions[api_key] + + def update_user_tier(self, api_key: str, new_tier: UserTier) -> bool: + """Update a user's tier (called via webhook from your backend).""" + if api_key in self._sessions: + self._sessions[api_key].tier = new_tier + logger.info(f"Updated tier for user to {new_tier}") + return True + return False + + def log_usage( + self, + user_id: str, + style: str, + success: bool, + metadata: Optional[dict] = None, + ) -> None: + """Log usage for Odoo sync.""" + self._usage_log.append({ + "timestamp": time.time(), + "user_id": user_id, + "style": style, + "success": success, + "metadata": metadata or {}, + }) + + def get_usage_since(self, timestamp: float) -> list[dict]: + """Get usage logs since timestamp (for Odoo sync).""" + return [log for log in self._usage_log if log["timestamp"] > timestamp] + + def get_user_stats(self, user_id: str) -> dict: + """Get usage statistics for a user.""" + user_logs = [log for log in self._usage_log if log["user_id"] == user_id] + return { + "total_requests": len(user_logs), + "successful": sum(1 for log in user_logs if log["success"]), + "by_style": self._count_by_style(user_logs), + } + + def _count_by_style(self, logs: list[dict]) -> dict[str, int]: + counts: dict[str, int] = defaultdict(int) + for log in logs: + counts[log["style"]] += 1 + return dict(counts) + + def clear_old_logs(self, days: int = 30) -> int: + """Clear logs older than N days. Returns count of removed logs.""" + cutoff = time.time() - (days * 86400) + old_count = len(self._usage_log) + self._usage_log = [log for log in self._usage_log if log["timestamp"] > cutoff] + return old_count - len(self._usage_log) + + +# Global session store +session_store = SessionStore() + + +# ============================================================================= +# API Key Authentication +# ============================================================================= + +api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) + + +async def validate_api_key_with_backend(api_key: str) -> Optional[dict]: + """ + Validate API key with your backend (Odoo/Next.js). + + Your backend should return: + { + "valid": true, + "user_id": "user_123", + "tier": "pro", + "email": "user@example.com" + } + + Or for invalid keys: + { + "valid": false, + "error": "Invalid or expired API key" + } + """ + # For development/testing without backend + if not AUTH_ENABLED: + return { + "valid": True, + "user_id": "dev_user", + "tier": "business", + "email": "dev@aiqso.io", + } + + # Check for internal/admin keys (for your backend services) + if api_key.startswith("qrb_admin_"): + expected_hash = hashlib.sha256( + f"{BACKEND_SECRET}:{api_key}".encode() + ).hexdigest()[:16] + if api_key.endswith(expected_hash): + return { + "valid": True, + "user_id": "admin", + "tier": "admin", + "email": "admin@aiqso.io", + } + + # Validate with your backend + try: + import httpx + async with httpx.AsyncClient() as client: + response = await client.post( + f"{BACKEND_VALIDATION_URL}/api/qr-builder/validate-key", + json={"api_key": api_key}, + headers={"Authorization": f"Bearer {BACKEND_SECRET}"}, + timeout=5.0, + ) + if response.status_code == 200: + return response.json() + except Exception as e: + logger.warning(f"Backend validation failed: {e}") + + return None + + +async def get_current_user( + request: Request, + api_key: Optional[str] = Depends(api_key_header), +) -> UserSession: + """ + Dependency to get current authenticated user. + + Usage in endpoints: + @app.post("/qr/logo") + async def create_logo_qr(user: UserSession = Depends(get_current_user)): + if not user.can_access_style("logo"): + raise HTTPException(403, "Upgrade to Pro for logo QR codes") + """ + # Allow unauthenticated access for free tier (with limits) + if not api_key: + if not AUTH_ENABLED: + # Dev mode - return business tier + return session_store.get_or_create_session( + user_id="anonymous", + tier=UserTier.BUSINESS, + api_key="dev_anonymous", + ) + + # Production - anonymous users get free tier + client_ip = request.client.host if request.client else "unknown" + anonymous_key = f"anon_{hashlib.md5(client_ip.encode()).hexdigest()[:8]}" + return session_store.get_or_create_session( + user_id=f"anonymous_{client_ip}", + tier=UserTier.FREE, + api_key=anonymous_key, + ) + + # Validate API key with backend + validation = await validate_api_key_with_backend(api_key) + + if not validation or not validation.get("valid"): + raise HTTPException( + status_code=401, + detail="Invalid API key. Get your key at https://aiqso.io/portal", + headers={"WWW-Authenticate": "ApiKey"}, + ) + + # Get or create session + tier = UserTier(validation.get("tier", "free")) + return session_store.get_or_create_session( + user_id=validation["user_id"], + tier=tier, + api_key=api_key, + email=validation.get("email"), + ) + + +async def require_auth( + user: UserSession = Depends(get_current_user), +) -> UserSession: + """Dependency that requires authentication (no anonymous access).""" + if user.user_id.startswith("anonymous"): + raise HTTPException( + status_code=401, + detail="Authentication required. Sign up at https://aiqso.io/portal", + ) + return user + + +# ============================================================================= +# Rate Limiting Middleware +# ============================================================================= + +async def check_rate_limit(user: UserSession = Depends(get_current_user)) -> UserSession: + """ + Dependency to check rate limits. + + Usage: + @app.post("/qr") + async def create_qr(user: UserSession = Depends(check_rate_limit)): + ... + """ + allowed, reason = user.check_rate_limit() + if not allowed: + raise HTTPException( + status_code=429, + detail=f"Rate limit exceeded: {reason}. Upgrade at https://aiqso.io/portal", + headers={ + "Retry-After": "60", + "X-RateLimit-Limit": str(user.limits.requests_per_minute), + "X-RateLimit-Remaining": "0", + }, + ) + + user.record_request() + return user + + +# ============================================================================= +# Style Access Control +# ============================================================================= + +def require_style(style: str) -> Callable: + """ + Dependency factory to check style access. + + Usage: + @app.post("/qr/logo") + async def create_logo_qr( + user: UserSession = Depends(require_style("logo")) + ): + ... + """ + async def check_style_access( + user: UserSession = Depends(check_rate_limit), + ) -> UserSession: + if not user.can_access_style(style): + raise HTTPException( + status_code=403, + detail=f"'{style}' style requires Pro or Business tier. " + f"Upgrade at https://aiqso.io/portal", + headers={"X-Required-Tier": "pro"}, + ) + return user + + return check_style_access + + +def require_custom_colors() -> Callable: + """Dependency to check if custom colors are allowed.""" + async def check_custom_colors( + user: UserSession = Depends(check_rate_limit), + fill_color: str = "", + back_color: str = "", + ) -> UserSession: + # Check if using hex colors + is_custom = ( + fill_color.startswith("#") or + back_color.startswith("#") + ) + if is_custom and not user.can_use_custom_colors(): + raise HTTPException( + status_code=403, + detail="Custom hex colors require Pro or Business tier. " + "Upgrade at https://aiqso.io/portal", + ) + return user + + return check_custom_colors + + +# ============================================================================= +# Webhook Authentication (for your backend) +# ============================================================================= + +async def verify_backend_webhook( + x_webhook_secret: str = Header(..., alias="X-Webhook-Secret"), +) -> bool: + """ + Verify webhook requests from your backend. + + Your backend should include the secret in the X-Webhook-Secret header. + """ + if x_webhook_secret != BACKEND_SECRET: + raise HTTPException( + status_code=401, + detail="Invalid webhook secret", + ) + return True + + +# ============================================================================= +# Helper Functions +# ============================================================================= + +def get_tier_info(tier: UserTier) -> dict: + """Get tier information for display.""" + limits = TIER_LIMITS[tier] + return { + "tier": tier.value, + "limits": { + "requests_per_minute": limits.requests_per_minute, + "requests_per_day": limits.requests_per_day, + "max_qr_size": limits.max_qr_size, + "batch_limit": limits.batch_limit, + }, + "features": { + "allowed_styles": limits.allowed_styles, + "custom_colors": limits.custom_colors, + }, + } + + +def get_all_tiers_info() -> list[dict]: + """Get information about all tiers for pricing page.""" + return [ + { + "tier": tier.value, + "limits": { + "requests_per_minute": limits.requests_per_minute, + "requests_per_day": limits.requests_per_day, + "max_qr_size": limits.max_qr_size, + "batch_limit": limits.batch_limit, + }, + "features": { + "allowed_styles": limits.allowed_styles, + "custom_colors": limits.custom_colors, + }, + } + for tier, limits in TIER_LIMITS.items() + if tier != UserTier.ADMIN # Don't expose admin tier + ] diff --git a/tests/test_api.py b/tests/test_api.py index cfaf6d8..e6aefbb 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,10 +1,14 @@ """Tests for qr_builder.api module.""" +import os import pytest -from fastapi.testclient import TestClient from PIL import Image import io +# Disable authentication for tests +os.environ["QR_BUILDER_AUTH_ENABLED"] = "false" + +from fastapi.testclient import TestClient from qr_builder.api import app @@ -20,7 +24,9 @@ class TestHealthEndpoint: def test_health_returns_ok(self, client): response = client.get("/health") assert response.status_code == 200 - assert response.json() == {"status": "ok"} + data = response.json() + assert data["status"] == "ok" + assert "auth_enabled" in data class TestQREndpoint: