FORGE-X: Upload source (3bc83f83-api.zip)

3bc83f83/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"]

3bc83f83/README.md

@@ -0,0 +1,89 @@
+# Logistics Fleet Tracking and Driver Performance Management API
+
+> A REST API for transportation companies, last-mile delivery platforms, and fleet management software. Tracks real-time vehicle locations, calculates driver efficiency scores using fuel consumption, on-time delivery rate, and idle time metrics. Generates automated driver scorecards, identifies route optimization opportunities, predicts vehicle maintenance windows using mileage and engine data, and triggers compliance alerts for HOS regulation violations.
+
+## 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

3bc83f83/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"},
+        )

3bc83f83/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

3bc83f83/main.py

@@ -0,0 +1,184 @@
+"""
+Logistics Fleet Tracking and Driver Performance Management API — REST API Service
+A REST API for transportation companies, last-mile delivery platforms, and fleet management software. Tracks real-time vehicle locations, calculates driver efficiency scores using fuel consumption, on-time delivery rate, and idle time metrics. Generates automated driver scorecards, identifies route optimization opportunities, predicts vehicle maintenance windows using mileage and engine data, and triggers compliance alerts for HOS regulation violations.
+
+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="Logistics Fleet Tracking and Driver Performance Management API",
+    description="A REST API for transportation companies, last-mile delivery platforms, and fleet management software. Tracks real-time vehicle locations, calculates driver efficiency scores using fuel consumption, on-time delivery rate, and idle time metrics. Generates automated driver scorecards, identifies route optimization opportunities, predicts vehicle maintenance windows using mileage and engine data, and triggers compliance alerts for HOS regulation violations.",
+    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": "Logistics Fleet Tracking and Driver Performance Management 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": "Logistics Fleet Tracking and Driver Performance Management API",
+        "niche": "logistics",
+    }

3bc83f83/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 logistics item",
+        "category": "default",
+        "tags": ["logistics"],
+        "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

3bc83f83/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

3bc83f83/tests/test_main.py

@@ -0,0 +1,96 @@
+"""
+Tests for Logistics Fleet Tracking and Driver Performance Management 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,13 +1,98 @@
 ---
-title: Logistics Fleet Tracking And Driver Performance Management Api 3bc8
-emoji: 😻
+title: "Logistics Fleet Tracking and Driver Performance Management API"
+emoji: 🚚
 colorFrom: blue
-colorTo: pink
+colorTo: gray
 sdk: gradio
-sdk_version: 6.14.0
-python_version: '3.13'
+sdk_version: 4.44.0
 app_file: app.py
 pinned: false
+license: mit
+tags:
+  - logistics
+  - supply-chain
+  - shipping
+  - inventory
+  - maritime
+  - fleet
+  - management
+  - voyage
+  - tracking
+  - driver
 ---
+# Logistics Fleet Tracking and Driver Performance Management API
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+Are inefficient routes and underperforming drivers eating into your logistics business's profits? The Logistics Fleet Tracking and Driver Performance Management API helps you optimize your fleet operations.
+
+This powerful API solves the puzzle of fleet management by providing real-time tracking, automated route optimization, and in-depth driver performance analysis, ensuring you stay ahead of the competition. With our API, you can reduce fuel costs, lower emissions, and enhance customer satisfaction. By integrating our solution, you'll be able to make data-driven decisions to improve your logistics operations.
+
+## What's Included
+
+- Real-time vehicle tracking
+- Automated route optimization
+- Driver performance scoring
+- Fuel consumption monitoring
+- Integration with existing logistics systems
+
+## Who Is This For
+
+- Logistics companies seeking to reduce operational costs
+- Fleet managers wanting to improve driver performance
+- Transportation providers aiming to enhance customer experience
+- Delivery services looking to optimize routes
+
+## How It Works
+
+To get started, simply integrate our API into your existing logistics system, and begin tracking your fleet's performance in real-time. Our comprehensive documentation and dedicated support team ensure a seamless onboarding process. With our API, you can easily monitor and manage your fleet from a single dashboard.
+
+## Frequently Asked Questions
+
+**What programming languages does the API support?**
+Our API supports a wide range of programming languages, including Java, Python, and C#.
+
+**How do I integrate the API with my existing logistics system?**
+Our comprehensive documentation and dedicated support team will guide you through the integration process.
+
+**Can I customize the API to fit my specific business needs?**
+Yes, our API is highly customizable to meet the unique requirements of your logistics business.
+
+**What is the typical ROI for businesses using the Logistics Fleet Tracking and Driver Performance Management API?**
+Our customers have reported an average ROI of 20% through reduced fuel costs and improved operational efficiency.
+
+**Is the API scalable to support large fleets?**
+Yes, our API is designed to support fleets of all sizes, from small to large, and can scale to meet your growing business needs.
+
+## What You Get
+
+- Instant digital download
+- Complete REST API with full documentation
+- Free updates for life — pay once, own forever
+- Setup guide and usage instructions
+
+**Get instant access to the Logistics Fleet Tracking and Driver Performance Management API for just $59.99 and start optimizing your fleet operations today!**
+
+## 🚀 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/logistics-fleet-tracking-and-driver-performance-management-api-3bc8
+cd logistics-fleet-tracking-and-driver-performance-management-api-3bc8
+pip install -r requirements.txt
+python app.py
+```
+
+## 💰 Pricing
+
+- **Demo**: Free on Hugging Face Spaces
+- **Full Source Code**: $59.99
+- 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) — automated digital product engine*