Skip to content

Python API

Most people only need the CLI. The Python API is for embedding a shell in your own code, building tools on top of the namespace, and writing adapters.

Starting a shell

embed

embed(
    namespace: dict[str, Any] | None = None,
    *,
    interface: str | None = None,
    config_file: str | Path | None = None,
    include_caller: bool = True,
    banner: bool = True,
    **overrides: Any,
) -> None

Open a fastapi-repl shell right here, like IPython.embed().

The project configuration is discovered as usual, then the caller's local and global variables (and namespace) are added on top, so you can poke at the state of a script or a debugging session with your models and session at hand.

Parameters:

Name Type Description Default
namespace dict[str, Any] | None

Extra names to add last.

None
interface str | None

"ipython", "ptpython", "python"... Defaults to the config.

None
config_file str | Path | None

Use this config file instead of discovering one.

None
include_caller bool

Add the caller's globals and locals.

True
banner bool

Print the startup banner.

True
**overrides Any

Any :class:~fastapi_repl.config.ReplConfig setting, e.g. print_sql=True.

{}

Raises:

Type Description
ReplError

If called while an event loop is running (for example inside an async def endpoint). The shell needs to own the loop.

Example
from fastapi_repl import embed

def debug_user(user_id: int) -> None:
    user = load_user(user_id)
    embed()  # `user`, `user_id`, your models and `session` are available

build_namespace

build_namespace(
    config_file: str | Path | None = None, **overrides: Any
) -> tuple[ReplSession, dict[str, Any]]

Build a session without starting a shell.

Useful in notebooks, tests and scripts. Remember to call session.close() (or use the session as a context manager) when done.

Example
session, ns = build_namespace(print_sql=True)
try:
    users = session.runtime.run(ns["session"].scalars(ns["select"](ns["User"])))
finally:
    session.close()

start_shell

start_shell(
    loaded: LoadedConfig,
    *,
    extra: dict[str, Any] | None = None,
    banner: bool = True,
    console: Console | None = None,
) -> None

Build a session from loaded and run the configured interface until exit.

Configuration

load_config

load_config(
    *,
    cli: Mapping[str, Any] | None = None,
    config_file: str | Path | None = None,
    cwd: Path | None = None,
    environ: Mapping[str, str] | None = None,
    load_env_file: bool = True,
) -> LoadedConfig

Discover, merge and validate the configuration.

Parameters:

Name Type Description Default
cli Mapping[str, Any] | None

Values from command-line flags. None values are ignored.

None
config_file str | Path | None

An explicit config file. It may be a pyproject.toml (the [tool.fastapi-repl] table is read) or any other TOML file (read as a top-level table, like fastapi-repl.toml).

None
cwd Path | None

Where to start looking for the project root. Defaults to the cwd.

None
environ Mapping[str, str] | None

Environment variables. Defaults to os.environ.

None
load_env_file bool

Load env_file into os.environ before reading FASTAPI_REPL_* variables. Disable in tests.

True

Raises:

Type Description
ConfigError

If a file cannot be parsed or a value is invalid.

LoadedConfig dataclass

The result of :func:load_config.

root instance-attribute

root: Path

The project root. Relative paths in the config are resolved from here.

files class-attribute instance-attribute

files: list[Path] = field(default_factory=list)

Config files that were read, lowest precedence first.

sources class-attribute instance-attribute

sources: dict[str, str] = field(default_factory=dict)

Maps dotted setting names (print_sql, sqlalchemy.engine) to where they came from.

config_file property

config_file: Path | None

The highest-precedence config file, if any.

Sessions and the event loop

ReplSession

A fully prepared shell session.

Use it as a context manager::

with ReplSession(load_config()) as session:
    session.namespace.to_dict()

Parameters:

Name Type Description Default
loaded LoadedConfig

The resolved configuration.

required
console Console | None

Console for the banner and warnings. Defaults to stdout.

None
extra dict[str, Any] | None

Extra names added last (embed() uses this for the caller's locals).

None

start

start() -> None

Import everything and build the namespace.

close

close() -> None

Run shutdown hooks, close created objects, adapters and the lifespan.

Runtime

Owns the event loop used by the shell session.

error_hooks instance-attribute

error_hooks: list[Callable[[BaseException], None]] = []

Called with each exception raised by code run on the loop (see :meth:report_error).

run

run(awaitable: Awaitable[T]) -> T

Run an awaitable to completion on the session loop and return its result.

Context variables set by the awaitable are propagated back to the caller. Ctrl+C cancels the awaitable and re-raises KeyboardInterrupt.

Raises:

Type Description
RuntimeError

If called while the loop is already running (for example from inside ptpython, where you can await directly).

report_error

report_error(error: BaseException) -> None

Tell the error hooks that user code raised error.

Called automatically for awaitables run through :meth:run. Interfaces also call it for errors in synchronous code. Each exception is reported once, and a failing hook never hides the original error.

resolve

resolve(value: T | Awaitable[T]) -> T

Return value, awaiting it first if it is awaitable.

await_

await_(awaitable: Awaitable[T]) -> T

Run an awaitable and return its result. Exposed in the shell as await_.

Useful in interfaces without top-level await support (bpython) and in synchronous helper code.

adopt_context staticmethod

adopt_context(context: Context) -> None

Copy variables set in context into the current context.

close

close() -> None

Cancel leftover tasks and close the loop.

LifespanManager

Drives an ASGI application's lifespan protocol.

This speaks raw ASGI rather than calling framework internals, so it works with FastAPI, Starlette, Litestar, Quart and any other ASGI 3 app.

The app runs in a background task on the session loop. After :meth:startup, the task sits waiting for the shutdown message, which :meth:shutdown sends when the shell exits.

startup async

startup() -> dict[str, Any]

Send lifespan.startup and wait for the app to finish starting.

Returns:

Type Description
dict[str, Any]

The lifespan state dict (what request.state would see).

Raises:

Type Description
LifespanError

If the app reports a startup failure or times out.

shutdown async

shutdown() -> None

Send lifespan.shutdown and wait for the app to finish.

Raises:

Type Description
LifespanError

If the app reports a shutdown failure.

Adapters

ORMAdapter

Base class for ORM adapters.

Lifecycle, in order:

  1. detect() (class method) decides whether the adapter applies.
  2. setup() runs on the session loop, e.g. to initialise connections.
  3. enable_read_only() if read_only is on.
  4. default_imports(), discover_models() and objects() feed the namespace.
  5. enable_sql_echo() if --print-sql is on, after startup hooks ran.
  6. on_error() after each failed statement, while the shell is open.
  7. teardown() runs on the session loop when the shell exits.

name class-attribute

name: str = ''

Unique identifier, used in the adapters setting and plugin options.

display_name class-attribute

display_name: str = ''

Human friendly name for the banner.

package class-attribute

package: str | None = None

Import name of the ORM package, used by :meth:is_installed.

replaces class-attribute

replaces: tuple[str, ...] = ()

Adapters made redundant by this one (SQLModel replaces SQLAlchemy).

is_installed classmethod

is_installed() -> bool

Return True if the ORM can be imported.

detect classmethod

detect(context: AdapterContext) -> bool

Return True if this project uses the ORM.

Called after the app, models and base have been imported, so checking sys.modules is usually enough. Only used when the adapters setting is empty (auto-detection).

describe

describe() -> str

One-line description for the banner, e.g. SQLAlchemy 2.0 (async).

database

database() -> str | None

Where the adapter is connected, for the banner. Never include passwords.

Use :func:mask_url to render a URL safely.

setup async

setup() -> None

Prepare the ORM. Runs once, before the namespace is built.

enable_read_only async

enable_read_only() -> None

Make every connection read-only (the read_only setting).

Raise an exception if that cannot be guaranteed: the shell then refuses to start rather than silently giving write access.

on_error

on_error(error: BaseException) -> None

Called after a statement typed in the shell raised error.

Use it to recover, e.g. roll back a session that can no longer be used. Only called when the rollback_on_error setting is on.

discover_models

discover_models() -> Iterable[ModelInfo]

Yield the project's models.

default_imports

default_imports() -> list[str]

Import specs for helpers such as select or Q.

Skipped when the default_imports setting is false.

objects

objects() -> dict[str, Any]

Ready-made objects such as engine and session.

enable_sql_echo

enable_sql_echo(printer: SQLPrinter) -> None

Start printing SQL through printer.

teardown async

teardown() -> None

Release resources. Runs once when the shell exits.

tip

tip(namespace: dict[str, Any]) -> str | None

An example line of code for the banner, using names in namespace.

AdapterContext dataclass

Everything an adapter can use. Passed to the adapter's constructor.

root class-attribute instance-attribute

root: Path = field(default_factory=Path.cwd)

The project root.

modules class-attribute instance-attribute

modules: list[ModuleType] = field(default_factory=list)

Modules imported from the models setting, including submodules.

bases class-attribute instance-attribute

bases: list[Any] = field(default_factory=list)

Objects imported from the base setting.

app class-attribute instance-attribute

app: Any = None

The ASGI app, if configured.

options class-attribute instance-attribute

options: dict[str, Any] = field(default_factory=dict)

This adapter's [tool.fastapi-repl.plugins.<name>] table.

warn class-attribute instance-attribute

warn: Callable[[str], None] = print

Report a non-fatal problem to the user.

ModelInfo dataclass

A model class discovered by an adapter.

name instance-attribute

name: str

The name the model gets in the shell (before aliases and collisions).

module instance-attribute

module: str

The module the class is defined in.

adapter instance-attribute

adapter: str

Name of the adapter that found it.

prefix class-attribute instance-attribute

prefix: str = ''

Prefix used when two models share a name (collision = "prefix").

label class-attribute instance-attribute

label: str = ''

Optional group label, such as a Tortoise app label. Usable in dont_load.

SQLPrinter

Prints SQL statements with syntax highlighting.

Adapters call the printer with each statement they see. Custom adapters can use it too; see "Writing adapters" in the docs.

Parameters:

Name Type Description Default
console Console | None

Where to print. Defaults to stderr, so piped output stays clean.

None
truncate int | None

Maximum number of characters of SQL to show.

None
location bool

Also show the line of user code that triggered the query.

False

__call__

__call__(
    statement: str,
    params: Any = None,
    *,
    duration: float | None = None,
    many: bool = False,
    caller: str | None = None,
) -> None

Print one statement.

Parameters:

Name Type Description Default
statement str

The SQL text.

required
params Any

Bound parameters, shown dimmed.

None
duration float | None

Execution time in seconds.

None
many bool

The statement ran with executemany.

False
caller str | None

file:line to show when location is enabled. Found automatically when omitted.

None

Errors

ReplError

Bases: Exception

Base class for all fastapi-repl errors.

The CLI catches these and prints the message without a traceback, so the message should tell the user what went wrong and how to fix it.