Skip to content

API_EXAMPLES.md

Latest commit

 

History

History
243 lines (196 loc) · 5.84 KB

API_EXAMPLES.md

File metadata and controls

243 lines (196 loc) · 5.84 KB
Raw

API Integration Examples for Backend Developers

How to Use Data Layer Functions

This document shows backend developers how to integrate the data infrastructure functions into their FastAPI/Flask applications.

Installation

# Install the data infrastructure package
pip install -r requirements.txt

# Or if packaged:
pip install ./awdb-data-infrastructure

Users Operations

Create User

from db_layer.users import create_user

# Create a new user
user_id = create_user(
    user_type="mobile",  # "mobile" or "web"
    name="John Doe",
    email="john@example.com"
)
print(f"Created user with ID: {user_id}")

Get User by Email

from db_layer.users import get_user_by_email

# Retrieve user information
user = get_user_by_email("john@example.com")
if user:
    print(f"Found user: {user['name']}")
else:
    print("User not found")

Scans Operations

Insert Scan Result

from db_layer.scans import insert_scan

# Store a clothing scan result
scan_id = insert_scan(
    user_id="507f1f77bcf86cd799439012",
    image_url="https://storage.com/shirt_scan_123.jpg",
    predicted_class="Cotton",
    confidence=0.95
)
print(f"Stored scan with ID: {scan_id}")

Get User's Scans

from db_layer.scans import get_scans_by_user

# Retrieve all scans for a user
scans = get_scans_by_user("507f1f77bcf86cd799439012")
for scan in scans:
    print(f"Scan: {scan['predicted_class']} ({scan['confidence']:.2f})")

Get Specific Scan

from db_layer.scans import get_scan_by_id

# Retrieve a specific scan
scan = get_scan_by_id("507f1f77bcf86cd799439011")
if scan:
    print(f"Scan result: {scan['predicted_class']}")

Images Operations

Insert Image with Prediction

from db_layer.images import insert_image

# Store image data with ML prediction
image_id = insert_image(
    user_id="507f1f77bcf86cd799439012",
    image_url="https://storage.com/shirt_scan_123.jpg",
    predicted_class="Cotton",
    confidence=0.95
)
print(f"Stored image with ID: {image_id}")

Get User's Images

from db_layer.images import get_images_by_user

# Retrieve all images for a user
images = get_images_by_user("507f1f77bcf86cd799439012")
for image in images:
    print(f"Image: {image['predicted_class']} ({image['confidence']:.2f})")

Get Specific Image

from db_layer.images import get_image_by_id

# Retrieve a specific image
image = get_image_by_id("507f1f77bcf86cd799439013")
if image:
    print(f"Image result: {image['predicted_class']}")

Analytics Operations

Get Scan Counts by Class

from db_layer.analytics import get_scan_counts_by_class

# Get analytics on scan predictions
analytics = get_scan_counts_by_class()
for item in analytics:
    print(f"{item['_id']}: {item['count']} scans")

Example FastAPI Integration

from fastapi import FastAPI, Depends, HTTPException
from db_layer.users import create_user, get_user_by_email
from db_layer.scans import insert_scan, get_scans_by_user
from db_layer.images import insert_image, get_images_by_user

app = FastAPI()

# User endpoints
@app.post("/users")
def register_user(user_type: str, name: str, email: str):
    try:
        user_id = create_user(user_type, name, email)
        return {"status": "success", "user_id": str(user_id)}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/users/{email}")
def get_user(email: str):
    user = get_user_by_email(email)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return {"user": user}

# Scan endpoints
@app.post("/scans")
def add_scan(user_id: str, image_url: str, predicted_class: str, confidence: float):
    try:
        scan_id = insert_scan(user_id, image_url, predicted_class, confidence)
        return {"status": "success", "scan_id": str(scan_id)}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/scans/{user_id}")
def get_user_scans(user_id: str):
    scans = get_scans_by_user(user_id)
    return {"scans": scans}

# Image endpoints
@app.post("/images")
def add_image(user_id: str, image_url: str, predicted_class: str, confidence: float):
    try:
        image_id = insert_image(user_id, image_url, predicted_class, confidence)
        return {"status": "success", "image_id": str(image_id)}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/images/{user_id}")
def get_user_images(user_id: str):
    images = get_images_by_user(user_id)
    return {"images": images}

Data Models

User Document

{
    "_id": ObjectId("507f1f77bcf86cd799439012"),
    "user_type": "mobile",  # or "web"
    "name": "John Doe",
    "email": "john@example.com"
}

Scan Document

{
    "_id": ObjectId("507f1f77bcf86cd799439011"),
    "user_id": "507f1f77bcf86cd799439012",
    "image_url": "https://storage.com/shirt_scan_123.jpg",
    "timestamp": datetime.utcnow(),
    "predicted_class": "Cotton",
    "confidence": 0.95
}

Image Document

{
    "_id": ObjectId("507f1f77bcf86cd799439013"),
    "user_id": "507f1f77bcf86cd799439012",
    "image_url": "https://storage.com/shirt_scan_123.jpg",
    "timestamp": datetime.utcnow(),
    "predicted_class": "Cotton",
    "confidence": 0.95
}

Error Handling

All functions return MongoDB ObjectIds on success. Handle exceptions appropriately:

try:
    user_id = create_user("mobile", "John Doe", "john@example.com")
    print(f"Success: {user_id}")
except Exception as e:
    print(f"Error: {e}")
    # Handle the error (duplicate email, connection issues, etc.)

Environment Setup

Make sure to set up your environment variables:

# Set MongoDB connection
python set_env.py

# Or manually set:
export MONGO_URI="your_mongodb_connection_string"
export DB_NAME="your_database_name"