Overview
Firestone generates complete, production-ready CLI applications using the Click framework. Understanding the generated structure helps you customize, extend, and troubleshoot your CLIs.
File Structure
Single File Mode (Default)
#!/usr/bin/env python
"""
Main entry point for a click based CLI.
"""
# 1. Imports
# 2. Exception Handler
# 3. Main Command Group
# 4. Resource Command Groups
# 5. Operation Commands
# 6. Entry Point
Module Mode (--as-modules)
cli/
├── __init__.py # You create this
├── tasks.py # Generated
├── projects.py # Generated
└── users.py # Generated
Each module:
# 1. Imports
# 2. Exception Handler
# 3. init() Function
# ├── Resource Command Group
# ├── Operation Commands
# └── Return Command Group
Section Breakdown
1. Shebang and Docstring
#!/usr/bin/env python
"""
Main entry point for a click based CLI.
"""
Purpose:
- Makes file executable on Unix systems
- Documents the script purpose
Usage:
chmod +x cli.py
./cli.py --help
2. Imports
import functools
import json
import logging
import os
import sys
import click
from firestone_lib import cli
from firestone_lib import utils as firestone_utils
from {{ client_pkg }} import api_client
from {{ client_pkg }} import configuration
from {{ client_pkg }} import exceptions
# Resource-specific imports
from {{ client_pkg }}.api import tasks_api
from {{ client_pkg }}.models import task as task_model
from {{ client_pkg }}.models import create_task as create_task_model
from {{ client_pkg }}.models import update_task as update_task_model
Import groups:
- Standard library - Python built-ins
- Third-party - Click framework
- Firestone - CLI utilities
- Client - OpenAPI-generated client
- Resource models - API data models
Conditional imports:
create_*models only if resource has POSTupdate_*models only if resource has PUT/PATCH
3. Logger
_LOGGER = logging.getLogger(__name__)
Usage in commands:
_LOGGER.debug(f"resp: {resp}")
_LOGGER.info("Calling API...")
Control with --debug:
python cli.py --debug tasks list
# Enables DEBUG level logging
4. Exception Handler
def api_exc(func):
"""Handle ApiExceptions in all functions."""
async def wrapper(*args, **kwargs):
resp = None
try:
return await func(*args, **kwargs)
except exceptions.ApiException as apie:
if apie.body:
click.echo(apie.body)
else:
click.echo(apie.reason)
api_obj = args[0].get("api_obj")
if api_obj:
await api_obj.api_client.close()
sys.exit(-1)
return functools.update_wrapper(wrapper, func)
Purpose:
- Catches API exceptions from client library
- Displays error messages to user
- Properly closes API client
- Exits with error code
Applied to all commands:
@api_exc
async def tasks_get(ctx_obj):
# If API raises exception, api_exc catches it
5. Main Command Group
@click.group()
@click.option("--debug", help="Turn on debugging", is_flag=True)
@click.option(
"--api-key",
help="The API key to authorize against API",
envvar="API_KEY",
)
@click.option(
"--api-url",
help="The API url, e.g. https://localhost",
envvar="API_URL",
)
@click.option(
"--client-cert",
help="Path to the client cert for mutual TLS",
envvar="CLIENT_CERT",
)
@click.option(
"--client-key",
help="Path to the client key for mutual TLS",
envvar="CLIENT_KEY",
)
@click.option("--trust-proxy", help="Trust the proxy env vars", is_flag=True, default=False)
@click.pass_context
def main(ctx, debug, api_key, api_url, client_cert, client_key, trust_proxy):
"""{{ title }}
{{ description }}
"""
Global options:
--debug- Enable debug logging--api-key- API authentication (orAPI_KEYenv var)--api-url- Base API URL (orAPI_URLenv var)--client-cert- mTLS certificate (orCLIENT_CERTenv var)--client-key- mTLS key (orCLIENT_KEYenv var)--trust-proxy- Respect HTTP proxy environment variables
Main function body:
# Remove proxy env vars unless --trust-proxy
if not trust_proxy:
for prefix in ["http", "https", "all_http", "all_https"]:
env_var = f"{prefix}_proxy"
if env_var in os.environ:
del os.environ[env_var]
if env_var.upper() in os.environ:
del os.environ[env_var.upper()]
# Setup logging
try:
cli.init_logging("{{ pkg }}.resources.logging", "cli.conf")
except Exception:
logging.basicConfig(
level=logging.INFO,
format="# %(asctime)s - [%(threadName)s] %(name)s:%(lineno)d %(levelname)s - %(message)s",
)
# Enable debug if requested
logging.getLogger("asyncio").setLevel(logging.CRITICAL)
if debug:
_LOGGER.setLevel(logging.DEBUG)
logging.getLogger().setLevel(logging.DEBUG)
logging.getLogger("{{ pkg }}").setLevel(logging.DEBUG)
logging.getLogger("aiohttp").setLevel(logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)
logging.getLogger("httplib").setLevel(logging.DEBUG)
# Configure API client
config = configuration.Configuration(host=api_url)
config.debug = debug
if api_key:
config.access_token = api_key
if client_cert:
config.cert_file = client_cert
if client_key:
config.key_file = client_key
if "SSL_CA_CERT" in os.environ:
config.ssl_ca_cert = os.environ["SSL_CA_CERT"]
if "REQUESTS_CA_BUNDLE" in os.environ:
config.ssl_ca_cert = os.environ["REQUESTS_CA_BUNDLE"]
# Store config in context
ctx.obj = {
"api_client_config": config,
}
6. Resource Command Groups
One group per resource:
@main.group()
@firestone_utils.click_coro
@click.pass_obj
async def tasks(ctx_obj):
"""High level command for tasks."""
_LOGGER.debug(f"ctx_obj: {ctx_obj}")
config = ctx_obj["api_client_config"]
aclient = api_client.ApiClient(configuration=config)
ctx_obj["api_obj"] = tasks_api.TasksApi(api_client=aclient)
Key components:
@main.group()- Nest under main@firestone_utils.click_coro- Enable async@click.pass_obj- Receive context from main- Initialize resource-specific API client
- Store API client in context for commands
Usage:
python cli.py tasks --help
7. Resource Operations (Collection)
Operations on the resource collection (/tasks):
@tasks.command("list")
@click.option("--limit", help="Limit the number of responses back", type=int, show_default=True, required=False)
@click.option("--offset", help="The offset to start returning resources", type=int, show_default=True, required=False)
@click.pass_obj
@firestone_utils.click_coro
@api_exc
async def tasks_get(ctx_obj, limit, offset):
"""List all tasks in this collection"""
api_obj = ctx_obj["api_obj"]
params = {
"limit": limit,
"offset": offset,
}
resp = await api_obj.tasks_get(**params)
_LOGGER.debug(f"resp: {resp}")
if isinstance(resp, list):
click.echo(json.dumps([obj.to_dict() for obj in resp]))
return
if resp:
click.echo(json.dumps(resp.to_dict()))
return
click.echo("No data returned")
Command name mapping:
GET /tasks→listPOST /tasks→create
Response handling:
- List of objects → JSON array
- Single object → JSON object
- No data → "No data returned"
8. Resource Operations (Instance)
Operations on individual resources (/tasks/{task_id}):
@tasks.command("get")
@click.argument("task_id", type=str)
@click.pass_obj
@firestone_utils.click_coro
@api_exc
async def tasks_task_id_get(ctx_obj, task_id):
"""Get a specific task from this collection"""
api_obj = ctx_obj["api_obj"]
params = {}
resp = await api_obj.tasks_task_id_get(task_id, **params)
_LOGGER.debug(f"resp: {resp}")
if isinstance(resp, list):
print(json.dumps([obj.to_dict() for obj in resp]))
return
print(json.dumps(resp.to_dict()) if resp else "None")
Command name mapping:
GET /tasks/{id}→getPUT /tasks/{id}→updateDELETE /tasks/{id}→delete
Key differences:
- Uses
@click.argumentfor ID (positional) - Uses
@click.optionfor other parameters - ID is passed as first argument to API method
9. Create Operation
@tasks.command("create")
@click.option("--title", help="Task title", type=str, show_default=True, required=True)
@click.option("--completed/--no-completed", help="Task completion status", is_flag=True, show_default=True, required=False)
@click.pass_obj
@firestone_utils.click_coro
@api_exc
async def tasks_post(ctx_obj, title, completed):
"""Create a new task in this collection"""
api_obj = ctx_obj["api_obj"]
params = {
"title": title,
"completed": completed,
}
req_body = create_task_model.CreateTask(**params)
resp = await api_obj.tasks_post(req_body)
_LOGGER.debug(f"resp: {resp}")
if isinstance(resp, list):
click.echo(json.dumps([obj.to_dict() for obj in resp]))
return
if resp:
click.echo(json.dumps(resp.to_dict()))
return
click.echo("No data returned")
Special handling:
- Creates model instance from parameters
- Passes model as request body
- Uses
CreateTaskmodel (from OpenAPI spec)
10. Update Operation
@tasks.command("update")
@click.option("--title", help="Task title", type=str, required=False)
@click.option("--completed/--no-completed", help="Task completion status", is_flag=True, required=False)
@click.argument("task_id", type=str)
@click.pass_obj
@firestone_utils.click_coro
@api_exc
async def tasks_task_id_put(ctx_obj, title, completed, task_id):
"""Update an existing task in this collection"""
api_obj = ctx_obj["api_obj"]
params = {
"title": title,
"completed": completed,
}
req_body = update_task_model.UpdateTask(**params)
resp = await api_obj.tasks_task_id_put(task_id, req_body)
_LOGGER.debug(f"resp: {resp}")
if isinstance(resp, list):
print(json.dumps([obj.to_dict() for obj in resp]))
return
print(json.dumps(resp.to_dict()) if resp else "None")
Special handling:
- Combines ID argument with optional update fields
- Creates
UpdateTaskmodel - Fields typically not required (partial update)
11. Entry Point
if __name__ == "__main__":
# pylint: disable=no-value-for-parameter
main()
Allows direct execution:
python cli.py --help
Module Structure (--as-modules)
Each module exports an init() function:
def init():
"""Initialize tasks resource CLI."""
@click.group()
@firestone_utils.click_coro
@click.pass_obj
async def tasks(ctx_obj):
"""High level command for tasks."""
# ... setup ...
@tasks.command("list")
@click.pass_obj
@firestone_utils.click_coro
@api_exc
async def tasks_get(ctx_obj):
# ... operation ...
# ... more operations ...
return tasks
Integration example:
# main.py
import click
from myapi.cli import tasks
from myapi.cli import projects
@click.group()
@click.option("--api-url", envvar="API_URL")
@click.pass_context
def main(ctx, api_url):
"""My API CLI"""
# ... setup config ...
# Register resource command groups
tasks_cli = tasks.init()
projects_cli = projects.init()
main.add_command(tasks_cli)
main.add_command(projects_cli)
if __name__ == "__main__":
main()
Code Patterns
Consistent Decorator Order
All commands follow this pattern:
@<group>.command("<name>")
@click.option(...) / @click.argument(...) # Options/args first
@click.pass_obj # Pass context
@firestone_utils.click_coro # Enable async
@api_exc # Exception handling
async def command_name(ctx_obj, ...):
# Implementation
Consistent Response Handling
if isinstance(resp, list):
click.echo(json.dumps([obj.to_dict() for obj in resp]))
return
if resp:
click.echo(json.dumps(resp.to_dict()))
return
click.echo("No data returned")
Context Usage
# Main stores config
ctx.obj = {"api_client_config": config}
# Resource group adds API client
ctx_obj["api_obj"] = tasks_api.TasksApi(...)
# Commands use API client
api_obj = ctx_obj["api_obj"]
resp = await api_obj.tasks_get()
Customization Points
1. Add Custom Commands
@tasks.command("export")
@click.option("--format", type=click.Choice(["csv", "json"]))
@click.pass_obj
@firestone_utils.click_coro
@api_exc
async def export_tasks(ctx_obj, format):
"""Export tasks to file"""
# Custom implementation
2. Modify Response Formatting
# Instead of JSON
click.echo(json.dumps(resp.to_dict()))
# Use table format
from tabulate import tabulate
table = [[t.id, t.title, t.completed] for t in resp]
click.echo(tabulate(table, headers=["ID", "Title", "Done"]))
3. Add Global Options
@main.option("--timeout", type=int, default=30)
def main(ctx, api_url, timeout, ...):
config = configuration.Configuration(host=api_url)
config.timeout = timeout
Next Steps
- CRUD Operations - Use the generated commands
- Integration - Connect with OpenAPI clients
- Customization - Modify templates
- Usage Examples - Real-world scenarios