Coverage for /usr/local/lib/python3.12/site-packages/prefect/cli/root.py: 29%
96 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 11:21 +0000
1"""
2Base `prefect` command-line application
3"""
5import asyncio 1a
6import platform 1a
7import sys 1a
8from importlib.metadata import PackageNotFoundError 1a
9from importlib.metadata import version as import_version 1a
10from typing import Any 1a
12import typer 1a
14import prefect 1a
15import prefect.context 1a
16import prefect.settings 1a
17from prefect.cli._types import PrefectTyper, SettingsOption 1a
18from prefect.cli._utilities import with_cli_exception_handling 1a
19from prefect.client.base import determine_server_type 1a
20from prefect.client.constants import SERVER_API_VERSION 1a
21from prefect.logging.configuration import setup_logging 1a
22from prefect.settings import get_current_settings 1a
23from prefect.types._datetime import parse_datetime 1a
25app: PrefectTyper = PrefectTyper(add_completion=True, no_args_is_help=True) 1a
28def version_callback(value: bool) -> None: 1a
29 if value: 29 ↛ 30line 29 didn't jump to line 30 because the condition on line 29 was never true1a
30 print(prefect.__version__)
31 raise typer.Exit()
34def is_interactive() -> bool: 1a
35 return app.console.is_interactive 1a
38@app.callback() 1a
39@with_cli_exception_handling 1a
40def main( 1a
41 ctx: typer.Context,
42 version: bool = typer.Option(
43 None,
44 "--version",
45 "-v",
46 # A callback is necessary for Typer to call this without looking for additional
47 # commands and erroring when excluded
48 callback=version_callback,
49 help="Display the current version.",
50 is_eager=True,
51 ),
52 profile: str = typer.Option(
53 None,
54 "--profile",
55 "-p",
56 help="Select a profile for this CLI run.",
57 is_eager=True,
58 ),
59 prompt: bool = SettingsOption(
60 prefect.settings.PREFECT_CLI_PROMPT,
61 help="Force toggle prompts for this CLI run.",
62 ),
63):
64 if profile and not prefect.context.get_settings_context().profile.name == profile: 64 ↛ 67line 64 didn't jump to line 67 because the condition on line 64 was never true1a
65 # Generally, the profile should entered by `enter_root_settings_context`.
66 # In the cases where it is not (i.e. CLI testing), we will enter it here.
67 settings_ctx = prefect.context.use_profile(
68 profile, override_environment_variables=True
69 )
70 try:
71 ctx.with_resource(settings_ctx)
72 except KeyError:
73 print(f"Unknown profile {profile!r}.")
74 exit(1)
76 # Configure the output console after loading the profile
77 app.setup_console(soft_wrap=get_current_settings().cli.wrap_lines, prompt=prompt) 1a
79 if not get_current_settings().testing.test_mode: 79 ↛ 90line 79 didn't jump to line 90 because the condition on line 79 was always true1a
80 # When testing, this entrypoint can be called multiple times per process which
81 # can cause logging configuration conflicts. Logging is set up in conftest
82 # during tests.
83 setup_logging() 1a
85 # When running on Windows we need to ensure that the correct event loop policy is
86 # in place or we will not be able to spawn subprocesses. Sometimes this policy is
87 # changed by other libraries, but here in our CLI we should have ownership of the
88 # process and be able to safely force it to be the correct policy.
89 # https://github.com/PrefectHQ/prefect/issues/8206
90 if sys.platform == "win32": 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true1a
91 asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
94@app.command() 1a
95async def version( 1a
96 omit_integrations: bool = typer.Option(
97 False, "--omit-integrations", help="Omit integration information"
98 ),
99):
100 """Get the current Prefect version and integration information."""
101 import sqlite3
103 import sqlalchemy as sa
105 from prefect.server.database.dependencies import provide_database_interface
106 from prefect.server.utilities.database import get_dialect
108 if build_date_str := prefect.__version_info__.get("date", None):
109 build_date = parse_datetime(build_date_str).strftime("%a, %b %d, %Y %I:%M %p")
110 else:
111 build_date = "unknown"
113 version_info: dict[str, Any] = {
114 "Version": prefect.__version__,
115 "API version": SERVER_API_VERSION,
116 "Python version": platform.python_version(),
117 "Git commit": prefect.__version_info__["full-revisionid"][:8],
118 "Built": build_date,
119 "OS/Arch": f"{sys.platform}/{platform.machine()}",
120 "Profile": prefect.context.get_settings_context().profile.name,
121 }
122 server_type = determine_server_type()
124 version_info["Server type"] = server_type.lower()
126 try:
127 pydantic_version = import_version("pydantic")
128 except PackageNotFoundError:
129 pydantic_version = "Not installed"
131 version_info["Pydantic version"] = pydantic_version
133 if connection_url_setting := get_current_settings().server.database.connection_url:
134 connection_url = connection_url_setting.get_secret_value()
135 database = get_dialect(connection_url).name
136 version_info["Server"] = {"Database": database}
137 if database == "sqlite":
138 version_info["Server"]["SQLite version"] = sqlite3.sqlite_version
139 elif database == "postgresql":
140 try:
141 db = provide_database_interface()
142 async with db.session_context() as session:
143 result = await session.execute(sa.text("SHOW server_version"))
144 if postgres_version := result.scalar():
145 version_info["Server"]["PostgreSQL version"] = postgres_version
146 except Exception:
147 # If we can't get the PostgreSQL version, don't print anything
148 pass
150 if not omit_integrations:
151 integrations = get_prefect_integrations()
152 if integrations:
153 version_info["Integrations"] = integrations
155 display(version_info)
158def get_prefect_integrations() -> dict[str, str]: 1a
159 """Get information about installed Prefect integrations."""
160 from importlib.metadata import distributions
162 integrations: dict[str, str] = {}
163 for dist in distributions():
164 name = dist.metadata.get("Name")
165 if name and name.startswith("prefect-"):
166 author_email = dist.metadata.get("Author-email", "").strip()
167 if author_email.endswith("@prefect.io>"):
168 if ( # TODO: remove clause after updating `prefect-client` packaging config
169 name == "prefect-client" and dist.version == "0.0.0"
170 ):
171 continue
172 integrations[name] = dist.version
174 return integrations
177def display(object: dict[str, Any], nesting: int = 0) -> None: 1a
178 """Recursive display of a dictionary with nesting."""
179 for key, value in object.items():
180 key += ":"
181 if isinstance(value, dict):
182 app.console.print(" " * nesting + key)
183 display(value, nesting + 2)
184 else:
185 prefix = " " * nesting
186 app.console.print(f"{prefix}{key.ljust(21 - len(prefix))} {value}")