prefect.cli.root

Base prefect command-line application

display(object, nesting=0)

Recursive display of a dictionary with nesting.

Source code in src/prefect/cli/root.py

def display(object: Dict[str, Any], nesting: int = 0):
    """Recursive display of a dictionary with nesting."""
    for key, value in object.items():
        key += ":"
        if isinstance(value, dict):
            app.console.print(" " * nesting + key)
            display(value, nesting + 2)
        else:
            prefix = " " * nesting
            app.console.print(f"{prefix}{key.ljust(20 - len(prefix))} {value}")

get_prefect_integrations()

Get information about installed Prefect integrations.

Source code in src/prefect/cli/root.py

def get_prefect_integrations() -> Dict[str, str]:
    """Get information about installed Prefect integrations."""
    from importlib.metadata import distributions

    integrations = {}
    for dist in distributions():
        if dist.metadata["Name"].startswith("prefect-"):
            author_email = dist.metadata.get("Author-email", "").strip()
            if author_email.endswith("@prefect.io>"):
                integrations[dist.metadata["Name"]] = dist.version

    return integrations

version(omit_integrations=typer.Option(False, '--omit-integrations', help='Omit integration information')) async

Get the current Prefect version and integration information.

Source code in src/prefect/cli/root.py

@app.command()
async def version(
    omit_integrations: bool = typer.Option(
        False, "--omit-integrations", help="Omit integration information"
    ),
):
    """Get the current Prefect version and integration information."""
    import sqlite3

    from prefect.server.utilities.database import get_dialect
    from prefect.settings import PREFECT_API_DATABASE_CONNECTION_URL

    version_info = {
        "Version": prefect.__version__,
        "API version": SERVER_API_VERSION,
        "Python version": platform.python_version(),
        "Git commit": prefect.__version_info__["full-revisionid"][:8],
        "Built": pendulum.parse(
            prefect.__version_info__["date"]
        ).to_day_datetime_string(),
        "OS/Arch": f"{sys.platform}/{platform.machine()}",
        "Profile": prefect.context.get_settings_context().profile.name,
    }
    server_type = determine_server_type()

    version_info["Server type"] = server_type.lower()

    try:
        pydantic_version = import_version("pydantic")
    except PackageNotFoundError:
        pydantic_version = "Not installed"

    version_info["Pydantic version"] = pydantic_version

    if server_type == ServerType.EPHEMERAL.value:
        database = get_dialect(PREFECT_API_DATABASE_CONNECTION_URL.value()).name
        version_info["Server"] = {"Database": database}
        if database == "sqlite":
            version_info["Server"]["SQLite version"] = sqlite3.sqlite_version

    if not omit_integrations:
        integrations = get_prefect_integrations()
        if integrations:
            version_info["Integrations"] = integrations

    display(version_info)