Skip to content

bayinfosys/cliapi

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

3 Commits
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

cliapi

Write a Python function once. Expose it as a CLI command and a REST API endpoint, automatically and always in sync.

cliapi is a Python library for teams building tools that need to work both at the terminal and over HTTP. Define your command interface once in Python; cliapi derives the argparse CLI and the FastAPI HTTP API from the same source. The two surfaces cannot diverge because they share a single definition.

Useful for: inference pipelines, internal tooling, data processing services, and any project where a human operator and an automated client need to drive the same underlying logic.

Built and used by Bay Information Systems in Marigold (private model inference pipelines) and Reheat (search intent clustering).

Install

pip install cliapi

Getting started

Define a registry, decorate your command functions, then expose them.

from cliapi import CliApiRegistry, Resource, Payload

def get_db():
    db = Database.connect()
    yield db
    db.close()

registry = CliApiRegistry(
    prog="myapp",
    description="My application.",
    version="0.1.0",
    context_factory=get_db,
)

command = registry.command


@command(help="List items")
def cmd_items_list(db, *, limit: int = 10):
    return db.list_items(limit=limit)

@command(help="Create an item")
def cmd_items_create(db, *, name: Payload[str] = ""):
    return db.create_item(name=name)

@command(help="Read one item")
def cmd_item_read(db, *, item_id: Resource[str] = ""):
    return db.get_item(item_id)


# CLI entry point
if __name__ == "__main__":
    registry.run()

# ASGI entry point
app = registry.app()

Run via CLI:

python -m myapp items list --limit 5
python -m myapp items create --name "Widget"
python -m myapp item read --item-id abc123

Run via HTTP:

uvicorn myapp:app --reload

GET  /items?limit=5
POST /items          {"name": "Widget"}
GET  /item/abc123

API docs are available at /docs and /redoc.

Command naming

Function names encode the command path. The cmd_ prefix is stripped; remaining underscore-separated segments become the dotted path, which drives both the CLI subcommand tree and the HTTP route.

cmd_items_list      ->  items.list    ->  GET  /items
cmd_items_create    ->  items.create  ->  POST /items
cmd_item_read       ->  item.read     ->  GET  /item/{item_id}
cmd_reports_run_create -> reports.run.create -> POST /reports/run

Up to three levels of nesting are supported.

HTTP method is derived from the final path segment:

read, list, show    ->  GET
create, update, set ->  POST
delete              ->  DELETE

Parameter types

Keyword-only parameters on a command function are introspected to produce CLI flags, query parameters, path segments, and request body fields.

Plain annotation -- query parameter on HTTP, flag on CLI.

def cmd_items_list(db, *, limit: int = 10): ...

CLI:  myapp items list --limit 20
HTTP: GET /items?limit=20

Resource[T] -- path segment on HTTP, flag on CLI.

def cmd_item_read(db, *, item_id: Resource[str] = ""): ...

CLI:  myapp item read --item-id abc123
HTTP: GET /item/abc123

Payload[T] -- POST body field on HTTP, flag on CLI.

def cmd_items_create(db, *, name: Payload[str] = ""): ...

CLI:  myapp items create --name "Widget"
HTTP: POST /items   {"name": "Widget"}

Supported plain types: str, int, float, bool. Bool parameters become store_true flags on the CLI.

Context factory

Every command receives a context object as its first positional argument. The context factory is a zero-argument callable registered on the registry that produces it. It is called once per CLI invocation and once per HTTP request.

A plain function:

def get_db():
    return Database.connect()

registry = CliApiRegistry(..., context_factory=get_db)

A generator -- preferred when setup and teardown are needed:

def get_db():
    db = Database.connect()
    try:
        yield db
    finally:
        db.close()

The generator form maps directly to a FastAPI Depends generator on the HTTP path, and is wrapped with contextlib.contextmanager on the CLI path. Use it for logging, connection lifecycle, transaction management, or any per-request concern.

Global CLI flags and startup

To add flags that appear before the subcommand, supply a setup_parser callback. It receives the ArgumentParser and may mutate it freely.

def setup_parser(parser):
    parser.add_argument("--verbose", "-v", action="store_true")
    parser.add_argument("--config", metavar="PATH")

To act on those flags before the command runs, supply an on_startup callback. It receives the parsed argparse.Namespace.

def on_startup(args):
    if args.verbose:
        logging.basicConfig(level=logging.DEBUG)
    if args.config:
        load_config(args.config)

registry = CliApiRegistry(
    ...
    setup_parser=setup_parser,
    on_startup=on_startup,
)

There is no equivalent of setup_parser or on_startup on the HTTP path. Use FastAPI lifespan events or middleware for application-level concerns on that side.

CLI output

Command functions return a value. On the CLI path that value is passed to the output callback. The default renders JSON to stdout.

To customise:

def render(result, args):
    print(json.dumps(result, indent=2, default=str))

registry = CliApiRegistry(..., output=render)

The callback receives the return value and the parsed Namespace. A common pattern is a --json flag for machine-readable output alongside a human-readable default:

def setup_parser(parser):
    parser.add_argument(
        "--json",
        dest="as_json",
        action="store_true",
        default=False,
        help="Output as JSON.",
    )

def render(result, args):
    if getattr(args, "as_json", False):
        print(json.dumps(result, indent=2, default=str))
    else:
        pprint(result)

registry = CliApiRegistry(
    ...
    setup_parser=setup_parser,
    output=render,
)

The output callback is not invoked on the HTTP path. The FastAPI app returns command results directly as JSON responses.

interactive_only commands

Commands decorated with interactive_only=True appear in the CLI but are excluded from the HTTP API. Use this for commands that require a terminal.

@command(help="Open a shell", interactive_only=True)
def cmd_shell(db):
    ...

CliApiRegistry reference

CliApiRegistry(
    prog: str,
    description: str = "",
    version: str = "0.1.0",
    context_factory: Callable = None,
    setup_parser: Callable = None,
    on_startup: Callable = None,
    output: Callable = None,
    cmd_prefix: str = "cmd",
)

prog -- program name, used as the CLI prog and the FastAPI title. description -- used in CLI help text and the API docs. version -- exposed via --version on the CLI and in the API docs. context_factory -- zero-argument callable or generator, called per invocation. setup_parser -- receives the ArgumentParser, used to add global flags. on_startup -- receives the parsed Namespace, runs before CLI dispatch. output -- receives (result, args), renders CLI output. Default is JSON. cmd_prefix -- function name prefix stripped before path encoding. Default cmd.

Methods:

.command(help="", interactive_only=False)
    Decorator. Registers the function as a command.

.run()
    Parse sys.argv and dispatch. CLI entry point.

.parser() -> ArgumentParser
    Return the configured parser. Cached after first call.

.app() -> FastAPI
    Return the configured FastAPI app. Cached after first call.

.commands() -> dict[str, Callable]
    Return the registered command map keyed by path string.

About

Unified CLI and HTTP API based on argparse and fastapi

Resources

License

Stars

0 stars

Watchers

0 watching

Forks

Packages

 
 
 

Contributors

Languages