12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273 |
- import json
- import asyncclick as click
- from asyncclick import pass_context
- from cli.utils.timer import timer
- from r2r import R2RAsyncClient, R2RException
- @click.group()
- def prompts():
- """Prompts commands."""
- pass
- @prompts.command()
- @pass_context
- async def list(ctx: click.Context):
- """Get an overview of prompts."""
- client: R2RAsyncClient = ctx.obj
- try:
- with timer():
- response = await client.prompts.list()
- for prompt in response["results"]: # type: ignore
- click.echo(json.dumps(prompt, indent=2))
- except R2RException as e:
- click.echo(str(e), err=True)
- except Exception as e:
- click.echo(str(f"An unexpected error occurred: {e}"), err=True)
- @prompts.command()
- @click.argument("name", type=str)
- @click.option("--inputs", default=None, type=str)
- @click.option("--prompt-override", default=None, type=str)
- @pass_context
- async def retrieve(ctx: click.Context, name, inputs, prompt_override):
- """Retrieve an prompts by name."""
- client: R2RAsyncClient = ctx.obj
- try:
- with timer():
- response = await client.prompts.retrieve(
- name=name,
- inputs=inputs,
- prompt_override=prompt_override,
- )
- click.echo(json.dumps(response, indent=2))
- except R2RException as e:
- click.echo(str(e), err=True)
- except Exception as e:
- click.echo(str(f"An unexpected error occurred: {e}"), err=True)
- @prompts.command()
- @click.argument("name", required=True, type=str)
- @pass_context
- async def delete(ctx: click.Context, name):
- """Delete a prompt by name."""
- client: R2RAsyncClient = ctx.obj
- try:
- with timer():
- response = await client.prompts.delete(
- name=name,
- )
- click.echo(json.dumps(response, indent=2))
- except R2RException as e:
- click.echo(str(e), err=True)
- except Exception as e:
- click.echo(str(f"An unexpected error occurred: {e}"), err=True)
|