Fal.ai Examples¶
Code examples for common Fal.ai integration patterns.
Image Generation¶
Basic Generation¶
import fal_client
result = fal_client.subscribe(
"fal-ai/flux/schnell",
arguments={
"prompt": "Professional fashion photograph, female model, studio lighting",
"image_size": "portrait_4_3",
"num_outputs": 4,
},
with_logs=False,
)
for image in result["images"]:
print(f"Generated: {image['url']}")
Generation with LoRAs¶
result = fal_client.subscribe(
"workflows/Shootify/flux-krea-lora-i2i",
arguments={
"prompt": "sks person wearing elegant black dress, studio lighting",
"loras": [
{
"path": "https://storage.sartiq.com/loras/subject-001.safetensors",
"scale": 0.8
},
{
"path": "https://storage.sartiq.com/loras/style-editorial.safetensors",
"scale": 0.6
}
],
"num_outputs": 4,
"guidance_scale": 7.5,
"num_inference_steps": 30,
},
with_logs=False,
)
Image-to-Image with Reference¶
result = fal_client.subscribe(
"fal-ai/flux-pro/kontext/max",
arguments={
"prompt": "Same person wearing the red dress from reference",
"image_url": "https://storage.sartiq.com/products/dress.jpg",
"num_outputs": 2,
},
with_logs=False,
)
LoRA Training¶
Train Subject LoRA¶
# Submit training job
handle = fal_client.submit(
"fal-ai/flux-lora-fast-training",
arguments={
"images_data_url": "https://storage.sartiq.com/datasets/model-001.zip",
"steps": 1000,
"trigger_word": "sks",
"create_masks": True,
"is_style": False, # Subject training
"is_input_format_already_preprocessed": True,
}
)
request_id = handle.request_id
print(f"Training started: {request_id}")
# Poll for completion
import time
while True:
status = fal_client.status("fal-ai/flux-lora-fast-training", request_id)
if hasattr(status, 'completed') and status.completed:
break
print(f"Training in progress...")
time.sleep(30)
# Get result
result = fal_client.result("fal-ai/flux-lora-fast-training", request_id)
print(f"LoRA saved to: {result['lora_url']}")
Train Style LoRA¶
result = fal_client.subscribe(
"fal-ai/flux-lora-fast-training",
arguments={
"images_data_url": "https://storage.sartiq.com/datasets/style-editorial.zip",
"steps": 800,
"trigger_word": "editorial_style",
"is_style": True, # Style training
},
with_logs=True,
)
Vision Tasks¶
Face Detection¶
def detect_face(image_path: str) -> dict:
"""Detect face bounding box."""
url = fal_client.upload_file(image_path)
result = fal_client.subscribe(
"fal-ai/moondream3-preview/detect",
arguments={
"image_url": url,
"prompt": "face"
},
with_logs=False,
)
detections = result.get("objects", [])
if detections:
bbox = detections[0]["bbox"]
return {
"x1": bbox[0],
"y1": bbox[1],
"x2": bbox[2],
"y2": bbox[3],
"confidence": detections[0].get("confidence", 1.0)
}
return None
Person Segmentation¶
def segment_person(image_url: str) -> str:
"""Segment person and return mask URL."""
result = fal_client.subscribe(
"fal-ai/sam-3/image",
arguments={
"image_url": image_url,
"text_prompt": "human person",
"prompt": "human person",
"apply_mask": False,
"output_format": "png",
},
with_logs=False,
)
masks = result.get("masks", [])
if masks:
return masks[0]["url"]
return None
Garment Segmentation¶
def segment_garment(image_url: str, garment_label: str) -> str:
"""Segment specific garment from image."""
result = fal_client.subscribe(
"fal-ai/sam-3/image",
arguments={
"image_url": image_url,
"text_prompt": garment_label,
"prompt": garment_label,
"apply_mask": False,
"output_format": "png",
},
with_logs=False,
)
return result["masks"][0]["url"] if result.get("masks") else None
Image Processing¶
Upscale Image¶
def upscale_image(image_url: str, scale: int = 2) -> str:
"""Upscale image by given factor."""
result = fal_client.subscribe(
"fal-ai/seedvr/upscale/image",
arguments={
"image_url": image_url,
"scale": scale,
},
with_logs=False,
)
return result["image"]["url"]
Expand Image (Outpainting)¶
def expand_image(image_url: str, direction: str = "all") -> str:
"""Expand image boundaries."""
result = fal_client.subscribe(
"fal-ai/bria/expand",
arguments={
"image_url": image_url,
"direction": direction, # "left", "right", "top", "bottom", "all"
},
with_logs=False,
)
return result["image"]["url"]
Video Generation¶
Image to Video¶
result = fal_client.subscribe(
"fal-ai/video-gen",
arguments={
"image_url": "https://storage.sartiq.com/images/model-pose.jpg",
"prompt": "Model walking forward, subtle movement",
"duration": 3.0,
"fps": 24,
},
with_logs=False,
)
print(f"Video: {result['video']['url']}")
Async Processing¶
Submit and Track Multiple Tasks¶
import asyncio
from typing import List
async def generate_batch(prompts: List[str]) -> List[dict]:
"""Generate images for multiple prompts."""
handles = []
# Submit all tasks
for prompt in prompts:
handle = fal_client.submit(
"fal-ai/flux/schnell",
arguments={
"prompt": prompt,
"num_outputs": 1,
}
)
handles.append(handle)
# Collect results
results = []
for handle in handles:
result = handle.get() # Blocks until complete
results.append(result)
return results
With Progress Tracking¶
def generate_with_progress(prompt: str, on_progress: callable) -> dict:
"""Generate image with progress callbacks."""
handle = fal_client.submit(
"fal-ai/flux/schnell",
arguments={"prompt": prompt}
)
for event in handle.iter_events(with_logs=False, interval=1):
if hasattr(event, 'progress'):
on_progress(event.progress)
return handle.get()
# Usage
def print_progress(p):
print(f"Progress: {p * 100:.0f}%")
result = generate_with_progress("A beautiful sunset", print_progress)
Error Handling¶
Robust Task Execution¶
import time
from typing import Optional
def execute_with_retry(
app_id: str,
arguments: dict,
max_retries: int = 3,
timeout: int = 300
) -> Optional[dict]:
"""Execute Fal task with retry logic."""
for attempt in range(max_retries):
try:
handle = fal_client.submit(app_id, arguments)
start_time = time.time()
while time.time() - start_time < timeout:
try:
return handle.get()
except Exception as e:
if "pending" in str(e).lower():
time.sleep(5)
continue
raise
raise TimeoutError(f"Task timed out after {timeout}s")
except Exception as e:
if attempt < max_retries - 1:
print(f"Attempt {attempt + 1} failed: {e}, retrying...")
time.sleep(2 ** attempt)
else:
raise
return None
Cancel Long-Running Tasks¶
def execute_with_timeout(app_id: str, arguments: dict, timeout: int = 120):
"""Execute with automatic cancellation on timeout."""
handle = fal_client.submit(app_id, arguments)
request_id = handle.request_id
try:
start = time.time()
while time.time() - start < timeout:
status = fal_client.status(app_id, request_id)
if hasattr(status, 'completed') and status.completed:
return fal_client.result(app_id, request_id)
time.sleep(2)
# Timeout - cancel the task
fal_client.cancel(app_id, request_id)
raise TimeoutError("Task cancelled due to timeout")
except Exception as e:
# Ensure cleanup on error
try:
fal_client.cancel(app_id, request_id)
except:
pass
raise
Complete Workflow: Garment Fixing¶
def fix_garment(
model_image_url: str,
reference_garment_url: str
) -> str:
"""Complete garment fixing workflow."""
# Step 1: Detect garment label from reference
label_result = fal_client.subscribe(
"openrouter/router/vision",
arguments={
"image_urls": [reference_garment_url],
"prompt": "Extract a label (maximum 6 words) for this garment",
"model": "qwen/qwen3-vl-235b-a22b-instruct",
},
with_logs=False,
)
garment_label = label_result["text"]
# Step 2: Segment garment from both images
model_mask = segment_garment(model_image_url, garment_label)
ref_mask = segment_garment(reference_garment_url, garment_label)
# Step 3: Apply garment fix (inpainting)
result = fal_client.subscribe(
"workflows/Shootify/editing-workflow",
arguments={
"prompt": f"Replace garment with {garment_label} from reference",
"base_image_url": model_image_url,
"reference_image_urls": [reference_garment_url],
"mask_url": model_mask,
"edit_strength": 0.85,
},
with_logs=False,
)
return result["images"][0]["url"]