FORGE-X: Upload source (2c211a42-api.zip)

2c211a42/Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 8000
+
+HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
+    CMD python -c "import httpx; httpx.get('http://localhost:8000/health').raise_for_status()"
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

2c211a42/README.md

@@ -0,0 +1,89 @@
+# Cloud Infrastructure (AWS/GCP/Azure) Scheduling and Availability API
+
+> Complex scheduling logic, conflict detection, and multi-timezone support. Calendly-level scheduling in your own app.
+
+## Features
+
+- Full REST API
+
+## Quick Start
+
+```bash
+# 1. Install dependencies
+pip install -r requirements.txt
+
+# 2. Configure environment
+cp .env.example .env
+# Edit .env with your settings
+
+# 3. Run locally
+uvicorn main:app --reload --port 8000
+
+# 4. View interactive docs
+open http://localhost:8000/docs
+```
+
+## Docker Deployment
+
+```bash
+# Build and run
+docker compose up -d
+
+# Check health
+curl http://localhost:8000/health
+```
+
+## Authentication
+
+Get a token first:
+```bash
+curl -X POST "http://localhost:8000/auth/token?username=admin&password=admin123"
+```
+
+Use the token in subsequent requests:
+```bash
+curl -H "Authorization: Bearer YOUR_TOKEN" http://localhost:8000/items
+```
+
+## API Endpoints
+
+| Method | Path | Description |
+|--------|------|-------------|
+| GET | `/health` | System health |
+| POST | `/auth/token` | Get JWT token |
+| GET | `/items` | List all items |
+| POST | `/items` | Create item |
+| GET | `/items/{id}` | Get item |
+| PATCH | `/items/{id}` | Update item |
+| DELETE | `/items/{id}` | Delete item |
+| GET | `/stats` | API statistics |
+
+Full interactive docs: `http://localhost:8000/docs`
+
+## Rate Limits
+
+| Endpoint | Limit |
+|----------|-------|
+| `/auth/token` | 10/minute |
+| `GET /items` | 60/minute |
+| `POST /items` | 30/minute |
+| `DELETE /items` | 20/minute |
+
+## Running Tests
+
+```bash
+pip install pytest httpx
+pytest tests/ -v
+```
+
+## Production Notes
+
+- Change `SECRET_KEY` in `.env` before deploying
+- Replace in-memory `_db` with a real database
+- Add proper user management to `auth.py`
+- Configure `ALLOWED_ORIGINS` for CORS
+- Use Nginx/Traefik as reverse proxy
+
+## License
+
+MIT

2c211a42/auth.py

@@ -0,0 +1,53 @@
+"""
+Simple JWT auth — swap for OAuth2/API keys in production.
+"""
+import os
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from jose import JWTError, jwt
+from passlib.context import CryptContext
+
+SECRET_KEY = os.getenv("SECRET_KEY", "changeme-in-production-please")
+ALGORITHM = os.getenv("ALGORITHM", "HS256")
+EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "30"))
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+security = HTTPBearer()
+
+# Demo users — replace with real DB lookup
+USERS: dict[str, str] = {
+    "admin": pwd_context.hash("admin123"),
+    "demo": pwd_context.hash("demo123"),
+}
+
+
+def verify_password(username: str, password: str) -> bool:
+    hashed = USERS.get(username)
+    if not hashed:
+        return False
+    return pwd_context.verify(password, hashed)
+
+
+def create_access_token(data: dict) -> str:
+    to_encode = data.copy()
+    expire = datetime.now(timezone.utc) + timedelta(minutes=EXPIRE_MINUTES)
+    to_encode["exp"] = expire
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+
+def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str:
+    token = credentials.credentials
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: Optional[str] = payload.get("sub")
+        if not username:
+            raise HTTPException(status_code=401, detail="Invalid token payload")
+        return username
+    except JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid or expired token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )

2c211a42/docker-compose.yml

@@ -0,0 +1,10 @@
+version: '3.8'
+services:
+  api:
+    build: .
+    ports:
+      - "8000:8000"
+    env_file: .env
+    restart: unless-stopped
+    volumes:
+      - ./data:/app/data

2c211a42/main.py

@@ -0,0 +1,184 @@
+"""
+Cloud Infrastructure (AWS/GCP/Azure) Scheduling and Availability API — REST API Service
+Complex scheduling logic, conflict detection, and multi-timezone support. Calendly-level scheduling in your own app.
+
+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="Cloud Infrastructure (AWS/GCP/Azure) Scheduling and Availability API",
+    description="Complex scheduling logic, conflict detection, and multi-timezone support. Calendly-level scheduling in your own app.",
+    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": "Cloud Infrastructure (AWS/GCP/Azure) Scheduling and Availability 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": "Cloud Infrastructure (AWS/GCP/Azure) Scheduling and Availability API",
+        "niche": "cloud_infrastructure",
+    }

2c211a42/models.py

@@ -0,0 +1,47 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+from datetime import datetime
+
+
+class ItemCreate(BaseModel):
+    name: str = Field(..., min_length=1, max_length=200, description="Item name")
+    description: Optional[str] = Field(None, max_length=2000)
+    category: Optional[str] = Field(None, max_length=100)
+    tags: list[str] = Field(default_factory=list)
+    metadata: dict = Field(default_factory=dict)
+
+    model_config = {"json_schema_extra": {"example": {
+        "name": "Sample Item",
+        "description": "A sample cloud_infrastructure item",
+        "category": "default",
+        "tags": ["cloud_infrastructure"],
+        "metadata": {},
+    }}}
+
+
+class ItemUpdate(BaseModel):
+    name: Optional[str] = Field(None, min_length=1, max_length=200)
+    description: Optional[str] = Field(None, max_length=2000)
+    category: Optional[str] = None
+    tags: Optional[list[str]] = None
+    metadata: Optional[dict] = None
+
+
+class ItemResponse(BaseModel):
+    id: str
+    name: str
+    description: Optional[str]
+    category: Optional[str]
+    tags: list[str]
+    metadata: dict
+    created_by: str
+    created_at: str
+    updated_at: str
+
+
+class HealthResponse(BaseModel):
+    status: str
+    service: str
+    version: str
+    timestamp: str
+    items_count: int

2c211a42/requirements.txt

@@ -0,0 +1,9 @@
+fastapi>=0.110.0
+uvicorn[standard]>=0.27.0
+pydantic>=2.0.0
+python-jose[cryptography]>=3.3.0
+passlib[bcrypt]>=1.7.4
+python-multipart>=0.0.9
+slowapi>=0.1.9
+python-dotenv>=1.0.0
+httpx>=0.27.0

2c211a42/tests/test_main.py

@@ -0,0 +1,96 @@
+"""
+Tests for Cloud Infrastructure (AWS/GCP/Azure) Scheduling and Availability API API.
+"""
+import pytest
+from fastapi.testclient import TestClient
+from main import app
+
+client = TestClient(app)
+
+
+def test_health():
+    r = client.get("/health")
+    assert r.status_code == 200
+    data = r.json()
+    assert data["status"] == "healthy"
+    assert "version" in data
+
+
+def test_unauthorized():
+    r = client.get("/items")
+    assert r.status_code == 403
+
+
+def test_login():
+    r = client.post("/auth/token", params={"username": "admin", "password": "admin123"})
+    assert r.status_code == 200
+    assert "access_token" in r.json()
+
+
+def _auth_headers():
+    r = client.post("/auth/token", params={"username": "admin", "password": "admin123"})
+    token = r.json()["access_token"]
+    return {"Authorization": f"Bearer {token}"}
+
+
+def test_create_item():
+    headers = _auth_headers()
+    payload = {"name": "Test Item", "description": "A test", "tags": ["test"]}
+    r = client.post("/items", json=payload, headers=headers)
+    assert r.status_code == 201
+    data = r.json()
+    assert data["name"] == "Test Item"
+    assert "id" in data
+
+
+def test_list_items():
+    headers = _auth_headers()
+    r = client.get("/items", headers=headers)
+    assert r.status_code == 200
+    assert isinstance(r.json(), list)
+
+
+def test_get_item():
+    headers = _auth_headers()
+    # Create
+    payload = {"name": "Get Test", "tags": []}
+    r = client.post("/items", json=payload, headers=headers)
+    item_id = r.json()["id"]
+    # Get
+    r = client.get(f"/items/{item_id}", headers=headers)
+    assert r.status_code == 200
+    assert r.json()["id"] == item_id
+
+
+def test_update_item():
+    headers = _auth_headers()
+    payload = {"name": "Update Test", "tags": []}
+    r = client.post("/items", json=payload, headers=headers)
+    item_id = r.json()["id"]
+    r = client.patch(f"/items/{item_id}", json={"name": "Updated"}, headers=headers)
+    assert r.status_code == 200
+    assert r.json()["name"] == "Updated"
+
+
+def test_delete_item():
+    headers = _auth_headers()
+    payload = {"name": "Delete Test", "tags": []}
+    r = client.post("/items", json=payload, headers=headers)
+    item_id = r.json()["id"]
+    r = client.delete(f"/items/{item_id}", headers=headers)
+    assert r.status_code == 204
+    r = client.get(f"/items/{item_id}", headers=headers)
+    assert r.status_code == 404
+
+
+def test_stats():
+    headers = _auth_headers()
+    r = client.get("/stats", headers=headers)
+    assert r.status_code == 200
+    assert "total_items" in r.json()
+
+
+def test_openapi_docs():
+    r = client.get("/openapi.json")
+    assert r.status_code == 200
+    assert "openapi" in r.json()

README.md

@@ -1,15 +1,21 @@
 ---
 title: "Cloud Infrastructure (AWS/GCP/Azure) Scheduling and Availability API"
-emoji: "🤖"
+emoji: 🤖
 colorFrom: blue
-colorTo: indigo
+colorTo: gray
 sdk: gradio
-sdk_version: "4.44.0"
+sdk_version: 4.44.0
 app_file: app.py
 pinned: false
 license: mit
+tags:
+  - cloud
+  - aws
+  - infrastructure
+  - devops
+  - gcp
+  - azure
 ---
-
 # Cloud Infrastructure (AWS/GCP/Azure) Scheduling and Availability API
 
 Stop overpaying for idle cloud resources across AWS, GCP, and Azure—this API automates scheduling so you only pay for what you actually use.
@@ -61,11 +67,28 @@ Absolutely – no contracts. You keep the API key and access to the dashboard ev
 
 **Get your API key now and start saving hundreds on cloud costs with one simple integration.**
 
-## Pricing
+## 🚀 Usage
+
+1. Click **Use in Spaces** above to run the demo directly
+2. Or clone the repository and run locally:
+
+```bash
+git clone https://huggingface.co/spaces/WealthFromAI/cloud-infrastructure-awsgcpazure-scheduling-and-availability-api-2c21
+cd cloud-infrastructure-awsgcpazure-scheduling-and-availability-api-2c21
+pip install -r requirements.txt
+python app.py
+```
+
+## 💰 Pricing
 
 - **Demo**: Free on Hugging Face Spaces
 - **Full Source Code**: $42.23
+- Available on [Gumroad](https://gumroad.com) and [Whop](https://whop.com)
+
+## 📄 License
+
+MIT License — free to use, modify, and distribute.
 
 ---
 
-*Built with [FORGE-X](https://github.com/WealthFromAI)*
+*Built with [FORGE-X](https://github.com/WealthFromAI) — automated digital product engine*