Skip to main content

Code Examples

This page contains complete, end-to-end examples for using the GENERIO Flows API.

All examples follow the same lifecycle:

create → poll → download

Setup​

import requests
import time
import base64
from pathlib import Path

# Get your API key at: https://account.generio.ai → API Keys tab
API_KEY = "YOUR_API_KEY_HERE" 
BASE_URL = "https://flows.generio.ai"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

OUT_DIR = Path("out")
OUT_DIR.mkdir(exist_ok=True)

Helper Functions​

POLL_INTERVAL = 2
MAX_WAIT_SECONDS = 360 # 6 Minutes

def wait_for_flow(flow_id: str) -> dict | None:
    """
    Poll the flow status until completion, failure, abort, or timeout.
    
    Args:
        flow_id: The ID of the flow to monitor
        
    Returns:
        The final status dict if completed/failed/aborted, None on timeout or error
    """
    start = time.time()

    while True:
        try:
            response = requests.get(
                f"{BASE_URL}/flows/{flow_id}",
                headers=HEADERS
            )
            
            if response.status_code != 200:
                print(f"⚠️ Error checking status: {response.status_code}")
                return None
            
            status = response.json()
            state = status['state']
            progress = status['progress']
            
            # Progress is 0 (processing) or 1 (finished)
            progress_text = "Processing" if progress == 0 else "Complete"
            print(f"→ {progress_text} | {state}")

            if state in ["completed", "failed", "aborted"]:
                return status

            if time.time() - start >= MAX_WAIT_SECONDS:
                print("⚠️ Timeout reached")
                return None

            time.sleep(POLL_INTERVAL)
            
        except requests.exceptions.RequestException as e:
            print(f"⚠️ Network error: {e}")
            return None
        except KeyError as e:
            print(f"⚠️ Unexpected response format: {e}")
            return None


def download_outputs(flow_id: str, prefix: str = ""):
    outputs = requests.get(
        f"{BASE_URL}/flows/{flow_id}/outputs",
        headers=HEADERS
    ).json().get("outputs", [])

    for i, o in enumerate(outputs, 1):
        asset_id = o["asset_id"]
        r = requests.get(
            f"{BASE_URL}/flows/{flow_id}/outputs/{asset_id}?include_data=true",
            headers=HEADERS
        )

        # Parse JSON and decode base64
        response_json = r.json()
        data_uri = response_json["data"]
        base64_data = data_uri.split(",", 1)[1]
        binary_data = base64.b64decode(base64_data)

        filename = f"model_{prefix}_{asset_id}.glb"
        with open(filename, "wb") as f:
            f.write(binary_data)

        print(f"✓ Model saved: {filename}")


def image_to_data_url(path: Path) -> str:
    b64 = base64.b64encode(path.read_bytes()).decode()
    return f"data:image/png;base64,{b64}"


def glb_to_data_url(path: Path) -> str:
    b64 = base64.b64encode(path.read_bytes()).decode()
    return f"data:model/gltf-binary;base64,{b64}"

Example 1: Text → 3D​

payload = {
    "template": "model_generate_fromprompt",
    "parameters": {
        "quality": "high",
        "texture": True
    },
    "inputs": [
        {
            "data": "A modern office chair with armrests"
        }
    ]
}

r = requests.post(f"{BASE_URL}/flows", headers=HEADERS, json=payload)
flow_id = r.json()["flow_id"]

status = wait_for_flow(flow_id)
if status and status["state"] == "completed":
    download_outputs(flow_id, "fromprompt")

Example 2: Image → 3D​

payload = {
    "template": "model_generate_fromimage",
    "parameters": {
        "quality": "high",
        "texture": True
    },
    "inputs": [
        {
            "data": image_to_data_url(Path("input.png"))
        }
    ]
}

r = requests.post(f"{BASE_URL}/flows", headers=HEADERS, json=payload)
flow_id = r.json()["flow_id"]

status = wait_for_flow(flow_id)
if status and status["state"] == "completed":
    download_outputs(flow_id, "fromimage")

Example 3: Optimize a 3D Model​

payload = {
    "template": "model_optimize",
    "parameters": {
        "polygon_count": 50000
    },
    "inputs": [
        {
            "data": glb_to_data_url(Path("model.glb"))
        }
    ]
}

r = requests.post(f"{BASE_URL}/flows", headers=HEADERS, json=payload)
flow_id = r.json()["flow_id"]

status = wait_for_flow(flow_id)
if status and status["state"] == "completed":
    download_outputs(flow_id, "optimized")