Generating images on demand through an API unlocks workflows that manual design tools cannot match. Whether you need product mockups, blog illustrations, or social media assets at scale, an automated pipeline saves hours of repetitive work.

This tutorial walks you through building a visual content pipeline that accepts text prompts, generates images via two leading APIs, and stores results for downstream use. By the end, you will have a working Python script that can plug into any content management system or marketing automation tool.

Prerequisites

Before starting, make sure you have:

Step 1: Set Up Your Project

Create a new directory and install the required dependencies.

mkdir image-pipeline && cd image-pipeline
python -m venv venv
source venv/bin/activate
pip install openai requests pillow python-dotenv

Create a .env file to store your API keys securely:

OPENAI_API_KEY=sk-your-openai-key-here
STABILITY_API_KEY=sk-your-stability-key-here
OUTPUT_DIR=./generated_images

Step 2: Build the Base Pipeline Class

Start with a base class that handles common operations like saving images and managing output directories.

# pipeline.py
import os
from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv
 
load_dotenv()
 
class ImagePipeline:
    def __init__(self):
        self.output_dir = Path(os.getenv("OUTPUT_DIR", "./generated_images"))
        self.output_dir.mkdir(parents=True, exist_ok=True)
 
    def save_image(self, image_bytes: bytes, prefix: str, prompt: str) -> Path:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        slug = prompt[:30].replace(" ", "_").lower()
        filename = f"{prefix}_{slug}_{timestamp}.png"
        filepath = self.output_dir / filename
        filepath.write_bytes(image_bytes)
        return filepath

Step 3: Integrate the OpenAI DALL-E API

Add a method to generate images using DALL-E 3. The API accepts a text prompt and returns image data directly.

# openai_generator.py
import base64
from openai import OpenAI
from pipeline import ImagePipeline
 
class DallEGenerator(ImagePipeline):
    def __init__(self):
        super().__init__()
        self.client = OpenAI()
 
    def generate(self, prompt: str, size: str = "1024x1024",
                 quality: str = "standard") -> dict:
        response = self.client.images.generate(
            model="dall-e-3",
            prompt=prompt,
            size=size,
            quality=quality,
            response_format="b64_json",
            n=1
        )
 
        image_data = base64.b64decode(response.data[0].b64_json)
        filepath = self.save_image(image_data, "dalle", prompt)
        revised_prompt = response.data[0].revised_prompt
 
        return {
            "filepath": str(filepath),
            "revised_prompt": revised_prompt,
            "model": "dall-e-3",
            "size": size
        }

Key parameters to understand:

  • size: DALL-E 3 supports 1024x1024, 1792x1024, and 1024x1792
  • quality: Use hd for more detailed images (costs 2x more)
  • response_format: b64_json returns raw bytes; url returns a temporary link

Step 4: Integrate the Stability AI API

Stability AI offers fine-grained control over the generation process including style presets and negative prompts.

# stability_generator.py
import requests
import os
from pipeline import ImagePipeline
 
class StabilityGenerator(ImagePipeline):
    def __init__(self):
        super().__init__()
        self.api_key = os.getenv("STABILITY_API_KEY")
        self.base_url = "https://api.stability.ai/v2beta"
 
    def generate(self, prompt: str, negative_prompt: str = "",
                 aspect_ratio: str = "1:1",
                 style_preset: str = None) -> dict:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Accept": "image/*"
        }
 
        payload = {
            "prompt": prompt,
            "negative_prompt": negative_prompt,
            "aspect_ratio": aspect_ratio,
            "output_format": "png"
        }
 
        if style_preset:
            payload["style_preset"] = style_preset
 
        response = requests.post(
            f"{self.base_url}/stable-image/generate/core",
            headers=headers,
            files={"none": ""},
            data=payload
        )
 
        if response.status_code != 200:
            raise Exception(f"Stability API error: {response.text}")
 
        filepath = self.save_image(response.content, "stability", prompt)
 
        return {
            "filepath": str(filepath),
            "model": "stable-diffusion",
            "aspect_ratio": aspect_ratio
        }

Step 5: Build the Orchestrator

The orchestrator ties both generators together, letting you run prompts through multiple models and compare results.

# orchestrator.py
import json
from datetime import datetime
from dalle_generator import DallEGenerator
from stability_generator import StabilityGenerator
 
class ContentOrchestrator:
    def __init__(self):
        self.dalle = DallEGenerator()
        self.stability = StabilityGenerator()
 
    def generate_batch(self, prompts: list[str],
                       providers: list[str] = None) -> list[dict]:
        if providers is None:
            providers = ["dalle", "stability"]
 
        results = []
 
        for prompt in prompts:
            result = {"prompt": prompt, "outputs": [], "timestamp": datetime.now().isoformat()}
 
            if "dalle" in providers:
                try:
                    dalle_result = self.dalle.generate(prompt)
                    result["outputs"].append(dalle_result)
                except Exception as e:
                    result["outputs"].append({"error": str(e), "model": "dall-e-3"})
 
            if "stability" in providers:
                try:
                    stability_result = self.stability.generate(prompt)
                    result["outputs"].append(stability_result)
                except Exception as e:
                    result["outputs"].append({"error": str(e), "model": "stable-diffusion"})
 
            results.append(result)
 
        # Save manifest for tracking
        manifest_path = self.dalle.output_dir / "manifest.json"
        with open(manifest_path, "w") as f:
            json.dump(results, f, indent=2)
 
        return results
 
# Usage
if __name__ == "__main__":
    orchestrator = ContentOrchestrator()
    prompts = [
        "A minimalist blog header showing abstract data visualization",
        "Professional product photo of a wireless headphone on marble surface",
        "Isometric illustration of a modern home office setup"
    ]
    results = orchestrator.generate_batch(prompts)
    print(f"Generated {len(results)} image sets")

Step 6: Add Cost Tracking

API calls cost money. Adding a simple cost tracker prevents unexpected bills.

# Add to orchestrator.py
COST_MAP = {
    "dall-e-3": {"standard": 0.040, "hd": 0.080},
    "stable-diffusion": {"core": 0.03}
}
 
def estimate_batch_cost(prompts: list, providers: list, quality: str = "standard") -> float:
    total = 0.0
    for _ in prompts:
        if "dalle" in providers:
            total += COST_MAP["dall-e-3"][quality]
        if "stability" in providers:
            total += COST_MAP["stable-diffusion"]["core"]
    return total

Practical Tips

  • Prompt engineering matters: Be specific about style, lighting, and composition. "A product photo with soft studio lighting on a white background" outperforms "a photo of a product" every time.
  • Use negative prompts with Stability AI: Excluding unwanted elements (blur, text, watermark) significantly improves output quality.
  • Cache results: Store generated images with their prompts in a database. Regenerating the same image wastes money.
  • Rate limiting: OpenAI allows 7 images per minute on the standard tier. Add time.sleep() calls or use a queue for batch processing.
  • Validate before saving: Check image dimensions and file sizes before writing to storage. Corrupted responses happen occasionally.
  • Version your prompts: Keep a log of prompt text alongside generated images. This makes it easy to refine and reproduce good results.

Cost Comparison

What to Build Next

Once your pipeline works, consider these extensions:

  • Connect to a CMS webhook so images generate automatically when new posts are created
  • Add an image upscaling step using Stability AI's upscale endpoint
  • Build a simple web UI with FastAPI to let non-technical team members submit prompts
  • Implement A/B testing by generating multiple variants and tracking engagement