Skip to main content
View Categories

Code Examples

Code Examples & Recipes #

Practical code examples for common Nellie API use cases.

Table of Contents #

The simplest possible API call — Nellie generates everything automatically:

from nellie_api import Nellie

client = Nellie(api_key="nel_...")
book = client.books.create()  # All defaults

print(f"Started: {book.request_id}")

Custom Book Request #

Full control over generation parameters:

from nellie_api import Nellie

client = Nellie(api_key="nel_...")

book = client.books.create(
    prompt=""
    A noir detective story set in 1940s Los Angeles.
    
    Main character: Jack Marlowe, a cynical private eye
    Setting: Smoky bars, rain-slicked streets, glamorous mansions
    Tone: Hard-boiled, witty dialogue, atmospheric descriptions
    
    The story should involve a missing heiress and a mysterious 
    jade statue that everyone seems to be looking for.
    "",
    style="noir",
    type="novel",
    images=True,
    author="Raymond Handler",
    custom_tone="First-person narration, short punchy sentences",
    model="3.0",
    output_format="epub"
)

print(f"Request ID: {book.request_id}")
print(f"Status URL: {book.status_url}")

Using cURL #

curl -X POST https://api.nelliewriter.com/v1/book 
  -H "X-API-Key: nel_your_api_key" 
  -H "Content-Type: application/json" 
  -d '{
    "prompt": "A fantasy adventure about a young wizard",
    "style": "fantasy",
    "type": "novel",
    "images": true,
    "output_format": "pdf"
  }'

Polling Patterns #

Simple Polling Loop #

import time
from nellie_api import Nellie

client = Nellie(api_key="nel_...")

# Start generation
book = client.books.create(prompt="A sci-fi thriller")
print(f"Started: {book.request_id}")

# Poll until complete
while True:
    status = client.books.retrieve(book.request_id)
    print(f"[{status.progress:3d}%] {status.status}")
    
    if status.status == "completed":
        print(f"n✅ Done! Download: {status.result_url}")
        break
    elif status.status == "failed":
        print(f"n❌ Failed: {status.error}")
        break
    
    time.sleep(120)  # Wait 2 minutes

Using wait_for_completion() #

The SDK provides a built-in polling helper:

from nellie_api import Nellie

client = Nellie(api_key="nel_...")

book = client.books.create(prompt="A mystery novel")

# Block until done
result = client.books.wait_for_completion(
    book.request_id,
    poll_interval=120,
    timeout=7200,
    on_progress=lambda s: print(f"[{s.progress}%] {s.status}")
)

if result.is_successful():
    print(f"Download: {result.result_url}")
else:
    print(f"Error: {result.error}")

Async Polling (asyncio) #

import asyncio
import httpx

async def poll_book(request_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        while True:
            response = await client.get(
                f"https://api.nelliewriter.com/v1/status/{request_id}"
            )
            data = response.json()
            
            print(f"[{data['progress']}%] {data['status']}")
            
            if data['status'] in ('completed', 'failed'):
                return data
            
            await asyncio.sleep(120)

# Usage
async def main():
    # Start generation (sync is fine for one-off requests)
    from nellie_api import Nellie
    client = Nellie(api_key="nel_...")
    book = client.books.create(prompt="An adventure story")
    
    # Poll async
    result = await poll_book(book.request_id)
    print(result)

asyncio.run(main())

Webhook Servers #

Flask Webhook Handler #

from flask import Flask, request, jsonify
from nellie_api import Webhook, WebhookSignatureError
import os

app = Flask(__name__)
WEBHOOK_SECRET = os.environ["NELLIE_WEBHOOK_SECRET"]

@app.route("/webhooks/nellie", methods=["POST"])
def handle_webhook():
    # Verify signature
    try:
        event = Webhook.construct_event(
            request.data,
            request.headers.get("X-Nellie-Signature"),
            WEBHOOK_SECRET
        )
    except WebhookSignatureError:
        return jsonify({"error": "Invalid signature"}), 400
    
    # Handle the event
    if event.is_successful():
        print(f"✅ Book {event.request_id} completed!")
        print(f"   Download: {event.result_url}")
        print(f"   Credits: {event.credits_used}")
        
        # Trigger your business logic here
        # e.g., send email, update database, etc.
        
    else:
        print(f"❌ Book {event.request_id} failed: {event.error}")
    
    return jsonify({"received": True}), 200

if __name__ == "__main__":
    app.run(port=5000)

FastAPI Webhook Handler #

from fastapi import FastAPI, Request, HTTPException
from nellie_api import Webhook, WebhookSignatureError
import os

app = FastAPI()
WEBHOOK_SECRET = os.environ["NELLIE_WEBHOOK_SECRET"]

@app.post("/webhooks/nellie")
async def handle_webhook(request: Request):
    body = await request.body()
    sig_header = request.headers.get("x-nellie-signature")
    
    try:
        event = Webhook.construct_event(body, sig_header, WEBHOOK_SECRET)
    except WebhookSignatureError:
        raise HTTPException(status_code=400, detail="Invalid signature")
    
    if event.is_successful():
        # Queue background task
        from .tasks import process_story
        process_story.delay(event.request_id, event.result_url)
    
    return {"received": True}

Express.js Webhook Handler #

const express = require('express');
const crypto = require('crypto');

const app = express();
const WEBHOOK_SECRET = process.env.NELLIE_WEBHOOK_SECRET;

function verifySignature(payload, sigHeader, secret) {
  const parts = Object.fromEntries(sigHeader.split(',').map(x => x.split('=')));
  const timestamp = parts.t;
  const signature = parts.v1;
  
  if (Math.abs(Date.now() / 1000 - parseInt(timestamp)) > 300) {
    return false;
  }
  
  const expected = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.${payload}`)
    .digest('hex');
  
  return crypto.timingSafeEqual(Buffer.from(expected), Buffer.from(signature));
}

app.post('/webhooks/nellie', express.raw({type: '*/*'}), (req, res) => {
  const payload = req.body.toString();
  const sigHeader = req.headers['x-nellie-signature'];
  
  if (!verifySignature(payload, sigHeader, WEBHOOK_SECRET)) {
    return res.status(400).json({ error: 'Invalid signature' });
  }
  
  const event = JSON.parse(payload);
  
  if (event.status === 'completed') {
    console.log(`Book ready: ${event.resultUrl}`);
  } else {
    console.log(`Book failed: ${event.error}`);
  }
  
  res.json({ received: true });
});

app.listen(3000);

Batch Generation #

Sequential Batch #

Process multiple stories one at a time:

from nellie_api import Nellie
import time

client = Nellie(api_key="nel_...")

prompts = [
    "A mystery in a small town",
    "A sci-fi adventure on Mars",
    "A romance in Paris",
]

results = []

for i, prompt in enumerate(prompts):
    print(f"n[{i+1}/{len(prompts)}] Starting: {prompt[:30]}...")
    
    book = client.books.create(prompt=prompt)
    result = client.books.wait_for_completion(
        book.request_id,
        on_progress=lambda s: print(f"  [{s.progress}%]", end="r")
    )
    
    results.append({
        "prompt": prompt,
        "status": result.status,
        "url": result.result_url if result.is_successful() else None
    })
    
    print(f"  Done: {result.status}")
    
    # Respect rate limits between jobs
    if i < len(prompts) - 1:
        time.sleep(10)

# Summary
print("n=== Results ===")
for r in results:
    status = "✅" if r["url"] else "❌"
    print(f"{status} {r['prompt'][:30]}...")

Parallel Start with Webhook #

Start multiple jobs and receive results via webhook:

from nellie_api import Nellie
import time

client = Nellie(api_key="nel_...")
WEBHOOK_URL = "https://myapp.com/webhooks/nellie"

prompts = [
    "A mystery novel",
    "A sci-fi thriller",
    "A fantasy epic",
]

# Start all jobs (respecting rate limits)
job_ids = []
for prompt in prompts:
    book = client.books.create(
        prompt=prompt,
        webhook_url=WEBHOOK_URL
    )
    job_ids.append(book.request_id)
    print(f"Started: {book.request_id}")
    time.sleep(7)  # Respect 6-second rate limit

print(f"nStarted {len(job_ids)} jobs. Results will arrive via webhook.")

Error Handling #

Comprehensive Error Handling #

from nellie_api import (
    Nellie,
    NellieError,
    AuthenticationError,
    RateLimitError,
    APIError
)
import time

def generate_with_retries(prompt: str, max_retries: int = 3) -> str | None:
    ""Generate a book with retry logic.""
    client = Nellie()
    
    for attempt in range(max_retries):
        try:
            # Start generation
            book = client.books.create(prompt=prompt)
            print(f"Started: {book.request_id}")
            
            # Wait for completion
            result = client.books.wait_for_completion(
                book.request_id,
                timeout=7200
            )
            
            if result.is_successful():
                return result.result_url
            else:
                print(f"Generation failed: {result.error}")
                return None
                
        except AuthenticationError:
            print("❌ Invalid API key")
            return None  # Don't retry auth errors
            
        except RateLimitError:
            wait_time = 60 * (attempt + 1)
            print(f"⏳ Rate limited. Waiting {wait_time}s...")
            time.sleep(wait_time)
            continue
            
        except TimeoutError:
            print("⏰ Generation timed out")
            return None
            
        except APIError as e:
            print(f"❌ API Error ({e.status_code}): {e}")
            if e.status_code >= 500:
                # Retry server errors
                time.sleep(30)
                continue
            return None
            
        except NellieError as e:
            print(f"❌ Unexpected error: {e}")
            return None
    
    print("❌ Max retries exceeded")
    return None

# Usage
url = generate_with_retries("A mystery novel")
if url:
    print(f"✅ Success: {url}")

Pre-flight Validation #

from nellie_api import Nellie, APIError

def validate_and_generate(
    prompt: str,
    style: str,
    output_format: str,
    required_credits: int = 500
) -> dict:
    ""Validate parameters before starting generation.""
    client = Nellie()
    errors = []
    
    # 1. Check configuration
    config = client.get_configuration()
    
    if style not in config.styles:
        errors.append(f"Invalid style '{style}'. Valid: {config.styles}")
    
    if output_format not in config.formats:
        errors.append(f"Invalid format '{output_format}'. Valid: {config.formats}")
    
    # 2. Check credits
    usage = client.get_usage()
    # Assuming 10000 credit plan
    available = 10000 - usage.total_credits_used
    
    if available < required_credits:
        errors.append(f"Insufficient credits. Need {required_credits}, have {available}")
    
    # 3. Return errors or start generation
    if errors:
        return {"success": False, "errors": errors}
    
    book = client.books.create(
        prompt=prompt,
        style=style,
        output_format=output_format
    )
    
    return {
        "success": True,
        "request_id": book.request_id,
        "status_url": book.status_url
    }

# Usage
result = validate_and_generate(
    prompt="A mystery novel",
    style="mystery",
    output_format="epub",
    required_credits=500
)

if result["success"]:
    print(f"Started: {result['request_id']}")
else:
    for error in result["errors"]:
        print(f"❌ {error}")

Download & Storage #

Download to File #

import requests
from pathlib import Path

def download_book(result_url: str, filename: str) -> Path:
    ""Download generated book to a file.""
    response = requests.get(result_url, stream=True)
    response.raise_for_status()
    
    filepath = Path(filename)
    with open(filepath, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)
    
    print(f"Downloaded: {filepath} ({filepath.stat().st_size} bytes)")
    return filepath

# Usage
from nellie_api import Nellie

client = Nellie()
book = client.books.create(prompt="A mystery", output_format="pdf")
result = client.books.wait_for_completion(book.request_id)

if result.is_successful():
    download_book(result.result_url, f"book_{result.request_id[:8]}.pdf")

Upload to S3 #

import boto3
import requests
from io import BytesIO

s3 = boto3.client('s3')

def save_to_s3(result_url: str, bucket: str, key: str) -> str:
    ""Download book and upload to S3.""
    # Download
    response = requests.get(result_url)
    response.raise_for_status()
    
    # Upload to S3
    s3.upload_fileobj(
        BytesIO(response.content),
        bucket,
        key,
        ExtraArgs={'ContentType': 'application/pdf'}
    )
    
    # Return S3 URL
    return f"s3://{bucket}/{key}"

# Usage
s3_url = save_to_s3(
    result.result_url,
    "my-books-bucket",
    f"books/{result.request_id}.pdf"
)
print(f"Saved to: {s3_url}")

Upload to Google Cloud Storage #

from google.cloud import storage
import requests

def save_to_gcs(result_url: str, bucket_name: str, blob_name: str) -> str:
    ""Download book and upload to Google Cloud Storage.""
    # Download
    response = requests.get(result_url)
    response.raise_for_status()
    
    # Upload to GCS
    client = storage.Client()
    bucket = client.bucket(bucket_name)
    blob = bucket.blob(blob_name)
    blob.upload_from_string(response.content, content_type='application/pdf')
    
    return f"gs://{bucket_name}/{blob_name}"

UI Integration #

React Hook #

// useNellieBook.ts
import { useState, useCallback } from 'react';

interface BookStatus {
  requestId: string;
  status: 'idle' | 'queued' | 'processing' | 'completed' | 'failed';
  progress: number;
  resultUrl?: string;
  error?: string;
}

export function useNellieBook(apiKey: string) {
  const [book, setBook] = useState({
    requestId: '',
    status: 'idle',
    progress: 0
  });

  const generate = useCallback(async (prompt: string) => {
    // Start generation
    const response = await fetch('https://api.nelliewriter.com/v1/book', {
      method: 'POST',
      headers: {
        'X-API-Key': apiKey,
        'Content-Type': 'application/json'
      },
      body: JSON.stringify({ prompt })
    });
    
    const data = await response.json();
    setBook({ requestId: data.requestId, status: 'queued', progress: 0 });
    
    // Start polling
    const pollInterval = setInterval(async () => {
      const statusRes = await fetch(
        `https://api.nelliewriter.com/v1/status/${data.requestId}`
      );
      const status = await statusRes.json();
      
      setBook({
        requestId: status.requestId,
        status: status.status,
        progress: status.progress,
        resultUrl: status.resultUrl,
        error: status.error
      });
      
      if (status.status === 'completed' || status.status === 'failed') {
        clearInterval(pollInterval);
      }
    }, 120000); // Poll every 2 minutes
    
  }, [apiKey]);

  return { book, generate };
}

// Usage in component
function BookGenerator() {
  const { book, generate } = useNellieBook(process.env.NELLIE_API_KEY!);
  
  return (
    

      
      
      
Status: {book.status}

      
      
      {book.resultUrl && (
        Download Book
      )}
    

  );
}

Python CLI Tool #

#!/usr/bin/env python3
""Simple CLI for generating books with Nellie.""

import argparse
import sys
from nellie_api import Nellie, NellieError

def main():
    parser = argparse.ArgumentParser(description="Generate books with Nellie")
    parser.add_argument("prompt", help="Book prompt")
    parser.add_argument("--style", default="automatic", help="Book style")
    parser.add_argument("--format", dest="output_format", default="pdf")
    parser.add_argument("--images", action="store_true")
    parser.add_argument("--output", "-o", help="Output filename")
    
    args = parser.parse_args()
    
    try:
        client = Nellie()  # Uses NELLIE_API_KEY env var
        
        print(f"🚀 Starting generation...")
        book = client.books.create(
            prompt=args.prompt,
            style=args.style,
            output_format=args.output_format,
            images=args.images
        )
        
        print(f"📝 Request ID: {book.request_id}")
        
        result = client.books.wait_for_completion(
            book.request_id,
            on_progress=lambda s: print(f"r⏳ Progress: {s.progress}%", end=")
        )
        print()  # Newline after progress
        
        if result.is_successful():
            print(f"✅ Complete! Credits used: {result.credits_used}")
            print(f"📥 Download: {result.result_url}")
            
            if args.output:
                import requests
                response = requests.get(result.result_url)
                with open(args.output, "wb") as f:
                    f.write(response.content)
                print(f"💾 Saved to: {args.output}")
        else:
            print(f"❌ Failed: {result.error}")
            sys.exit(1)
            
    except NellieError as e:
        print(f"❌ Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Usage:

# Basic usage
python nellie_cli.py "A mystery novel"

# With options
python nellie_cli.py "A sci-fi thriller" --style sci_fi --format pdf --images -o story.pdf

Related Documentation #