Add async generation commands to dvp.py — the commands that make it genuinely useful.
Day 74 of 80
Today you add the two most powerful commands:
generate — describe one shot, get prompts for all platforms in parallelbatch — pass a JSON file of shots, generate everything at onceBoth use asyncio.run() to run async code from inside a synchronous Typer command — the standard pattern for calling async code from CLI tools.
Add these two commands to dvp.py, before the if __name__ == "__main__": block:
@app.command()
def generate(
shot: str = typer.Argument(..., help="Shot description in plain English"),
platforms: str = typer.Option(
"Kling,Runway,Veo", "--platforms", "-p",
help="Comma-separated list of platforms"
),
save: bool = typer.Option(False, "--save", "-s", help="Save results to database"),
):
"""Generate AI prompts for a shot across platforms (in parallel)."""
platform_list = [p.strip().title() for p in platforms.split(",")]
typer.echo(f"\n Generating for {len(platform_list)} platforms...\n")
async def _run():
client = anthropic.AsyncAnthropic()
async def gen_one(platform):
message = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=150,
messages=[{
"role": "user",
"content": f"Write a {platform} AI video prompt for: {shot}. "
f"Under 75 words. Only the prompt."
}]
)
return {"platform": platform, "prompt": message.content[0].text}
return await asyncio.gather(*[gen_one(p) for p in platform_list])
# Typer commands are synchronous — use asyncio.run() to call async code.
results = asyncio.run(_run())
for r in results:
typer.echo(f" [{r['platform']}]")
typer.echo(f" {r['prompt']}\n")
if save:
db.add_prompt(r["platform"], shot, r["prompt"])
if save:
typer.echo(f" Saved {len(results)} prompts to database.")
@app.command()
def batch(
file: Path = typer.Argument(..., help="JSON file with list of shot descriptions"),
save: bool = typer.Option(False, "--save", "-s", help="Save all results to database"),
output: str = typer.Option(None, "--output", "-o", help="Export results to JSON file"),
):
"""Batch-generate prompts from a JSON file of shots.
The file should contain a JSON array of strings:
["Wide establishing shot", "Close-up hands", "Aerial reveal"]
"""
if not file.exists():
typer.echo(f" File not found: {file}")
raise typer.Exit(1)
with open(file, "r") as f:
shots = json.load(f)
if not isinstance(shots, list):
typer.echo(" Error: JSON file must contain a list of strings")
raise typer.Exit(1)
typer.echo(f"\n Processing {len(shots)} shots × 3 platforms = {len(shots) * 3} prompts...\n")
async def _run():
client = anthropic.AsyncAnthropic()
platforms = ["Kling", "Runway", "Veo"]
async def gen(shot, platform):
message = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=150,
messages=[{
"role": "user",
"content": f"Write a {platform} AI video prompt for: {shot}. "
f"Under 75 words. Only the prompt."
}]
)
return {"shot": shot, "platform": platform, "prompt": message.content[0].text}
all_tasks = [gen(s, p) for s in shots for p in platforms]
return await asyncio.gather(*all_tasks)
results = asyncio.run(_run())
# Group by shot for clean display
current = None
for r in sorted(results, key=lambda x: x["shot"]):
if r["shot"] != current:
current = r["shot"]
typer.echo(f" SHOT: {current}")
typer.echo(f" [{r['platform']}] {r['prompt'][:60]}...")
if save:
db.add_prompt(r["platform"], r["shot"], r["prompt"])
if save:
typer.echo(f"\n Saved {len(results)} prompts to database.")
if output:
grouped = {}
for r in results:
grouped.setdefault(r["shot"], []).append({
"platform": r["platform"], "prompt": r["prompt"]
})
output_data = [{"shot": shot, "prompts": prompts} for shot, prompts in grouped.items()]
with open(output, "w") as f:
json.dump(output_data, f, indent=2)
typer.echo(f" Exported to {output}")Typer commands are synchronous. Typer doesn't know about async def. The solution: define an inner async def _run() and call asyncio.run(_run()) inside the regular Typer command. This is the standard pattern — sync wrapper around async work.
Path as an argument type — Typer understands pathlib.Path. It automatically validates that a path argument looks like a valid path and converts the string to a Path object. You get .exists(), .open() etc. without manual conversion.
dict.setdefault(key, []).append() is a compact way to build a list-of-items per key. If the key doesn't exist yet, it creates it with an empty list. If it does exist, it returns the existing list. Either way, you can immediately append to it.
python dvp.py --help # shows all 7 commands
python dvp.py generate "Slow motion golf at sunset" # all 3 platforms
python dvp.py generate "Ocean aerial" --platforms Kling,Veo # 2 platforms
python dvp.py generate "Golf swing" --save # save to database
python dvp.py batch my_shots.json --save # batch generate and save
python dvp.py stats # see the bar chart updategenerate command works — parallel calls, results displayed--save flag saves to database correctlybatch command accepts a JSON file and generates in bulk--output flag exports results to a JSON filepython dvp.py --helpDay 75 automates a real workflow using the DVP CLI tool — creating a shots file from a real project and running the full batch pipeline.