"""Typer CLI: ``fluxlit dev``, ``run``, ``workbench``, ``shutdown``, ``doctor``, ``config``,
``build``, ``new``.
The ``fluxlit`` setuptools entrypoint calls :func:`main`. Commands resolve defaults from
:func:`fluxlit.config.load_project_config` and :class:`~fluxlit.config.FluxlitSettings`.
"""
from __future__ import annotations
import json
import os
import platform
import socket
import sys
from importlib.util import find_spec
from pathlib import Path
from typing import Annotated, Literal
from urllib.parse import urlparse
import typer
from fluxlit.cli_doctor_verbose import build_doctor_verbose_detail, format_doctor_verbose_human
from fluxlit.config import load_project_config, resolve_binding, resolve_target
from fluxlit.config.config_print import build_config_payload
from fluxlit.config.project import ProjectConfig
from fluxlit.runtime import run_unified, shutdown_unified_process
app = typer.Typer(no_args_is_help=True, add_completion=False)
pages_cli = typer.Typer(no_args_is_help=True, add_completion=False, help="Streamlit page tooling.")
[docs]
@pages_cli.command("manifest")
def pages_manifest(
target: str | None = typer.Option(
None,
"--target",
help="module:attr (default: project target from fluxlit.toml or app:app).",
),
) -> None:
"""Print a JSON page manifest for the resolved FluxLit app (``manifest_version`` 1, stable)."""
pc = load_project_config()
resolved = resolve_target(target, pc)
from fluxlit.runtime import load_fluxlit
fl = load_fluxlit(resolved)
typer.echo(json.dumps(fl.build_page_manifest(), indent=2))
[docs]
@pages_cli.command("validate")
def pages_validate(
target: str | None = typer.Option(
None,
"--target",
help="module:attr (default: project target from fluxlit.toml or app:app).",
),
strict: bool = typer.Option(
False,
"--strict",
help="Run strict page signature checks even if settings.strict_page_signatures is off.",
),
) -> None:
"""Validate page handlers and manifest JSON-serializability (exit 1 on errors)."""
pc = load_project_config()
resolved = resolve_target(target, pc)
from fluxlit.pages.validate import validate_fluxlit_pages
from fluxlit.runtime import load_fluxlit
fl = load_fluxlit(resolved)
strict_sig = True if strict else None
errs = validate_fluxlit_pages(fl, strict_signatures=strict_sig)
if errs:
for line in errs:
typer.echo(line)
raise typer.Exit(code=1)
app.add_typer(pages_cli, name="pages")
CheckStatus = Literal["PASS", "WARN", "FAIL"]
def _fluxlit_auth_env_present() -> bool:
"""True when JWT/OIDC-related ``FLUXLIT_*`` variables look configured."""
prefixes = ("FLUXLIT_JWT_", "FLUXLIT_OIDC_")
for key, val in os.environ.items():
if any(key.startswith(p) for p in prefixes) and (val or "").strip():
return True
return False
def _dockerfile_body(target: str) -> str:
# Digest for python:3.12-slim (supply-chain pin); bump when rebuilding base images.
_py_slim_digest = "sha256:ec948fa5f90f4f8907e89f4800cfd2d2e91e391a4bce4a6afa77ba265bc3a2fe"
return (
"# Generated by `fluxlit build` — adjust COPY/RUN for your app layout.\n"
f"# python:3.12-slim @ {_py_slim_digest}\n"
f"FROM python@{_py_slim_digest}\n"
"WORKDIR /app\n"
'RUN pip install --no-cache-dir "fluxlit>=0.13,<1.0"\n'
"COPY . .\n"
"RUN useradd --create-home --uid 1000 appuser \\\n"
" && chown -R appuser:appuser /app\n"
"USER appuser\n"
"ENV FLUXLIT_GATEWAY_HOST=0.0.0.0\n"
"EXPOSE 8000\n"
f'CMD ["fluxlit", "run", "{target}"]\n'
)
def _dockerignore_body() -> str:
return (
"__pycache__\n"
"*.py[cod]\n"
".git\n"
".venv\n"
"venv\n"
".pytest_cache\n"
".mypy_cache\n"
".ruff_cache\n"
"*.egg-info\n"
"dist\n"
"build\n"
".env\n"
)
def _scaffold_app_body(profile: Literal["minimal", "auth-ready", "deploy"]) -> str:
if profile == "auth-ready":
return '''"""FluxLit auth-ready demo app.
Install auth helpers with: pip install "fluxlit[auth]"
"""
from fluxlit import FluxLit
app = FluxLit(title="FluxLit Auth Demo")
@app.api.get("/public")
def public():
return {"message": "public"}
@app.page("/")
def home(st, client):
st.title("Auth-ready Dashboard")
st.write(client.get("/public").json())
st.info("Add app.make_jwt_bearer() or app.attach_oidc_login(...) when ready.")
if __name__ == "__main__":
import subprocess
import sys
subprocess.run([sys.executable, "-m", "fluxlit", "dev"], check=True)
'''
if profile == "deploy":
return '''"""FluxLit deployment-ready starter."""
from fluxlit import FluxLit
app = FluxLit(
title="FluxLit Deploy Demo",
streamlit_page_config={"layout": "wide"},
)
@app.api.get("/items")
def items():
return [{"name": "Ada"}, {"name": "Grace"}]
@app.page("/")
def home(st, client):
st.title("Deployment Dashboard")
st.write(client.get("/items").json())
st.caption("Run `fluxlit doctor` before deploying behind a proxy.")
if __name__ == "__main__":
import subprocess
import sys
subprocess.run([sys.executable, "-m", "fluxlit", "dev"], check=True)
'''
return '''"""FluxLit demo app."""
from typing import Any
from fluxlit import Depends, FluxLit
from fluxlit.client import ApiClient
app = FluxLit(title="FluxLit Demo")
@app.api.get("/users")
def users():
return [{"name": "Ada"}]
def _demo_user() -> int:
"""Example dependency; see docs/streamlit-pages-typing."""
return 1
@app.page("/")
def home(st: Any, client: ApiClient, user_id: int = Depends(_demo_user)): # noqa: B008
st.title("Dashboard")
r = client.get("/users")
st.write(r.json())
if __name__ == "__main__":
import subprocess
import sys
subprocess.run([sys.executable, "-m", "fluxlit", "dev"], check=True)
'''
def _tcp_url_reachable(url: str, *, timeout_s: float = 0.25) -> tuple[bool, str]:
parsed = urlparse(url)
if parsed.scheme not in {"http", "https"} or not parsed.hostname:
return False, "not a valid http(s) URL"
port = parsed.port or (443 if parsed.scheme == "https" else 80)
try:
with socket.create_connection((parsed.hostname, port), timeout=timeout_s):
return True, f"{parsed.hostname}:{port} reachable"
except OSError as exc:
return False, f"{parsed.hostname}:{port} unreachable ({exc})"
def _execute_unified_cli(
*,
target: str | None,
host: str | None,
port: int | None,
log_level: str | None,
proxy_headers: bool,
forwarded_allow_ips: str | None,
pidfile: Path | None,
no_pidfile: bool,
reload: bool,
reload_scope: str,
workbench: bool,
debug: bool = False,
) -> None:
"""Shared entry for ``fluxlit dev``, ``run``, and ``workbench``."""
if reload and reload_scope not in {"gateway", "full"}:
typer.echo(
"--reload-scope must be 'gateway' or 'full'.",
err=True,
)
raise typer.Exit(code=2)
if debug:
os.environ["FLUXLIT_DEBUG"] = "1"
from fluxlit.runtime import load_fluxlit
pc = load_project_config()
resolved_target = resolve_target(target, pc)
fl = load_fluxlit(resolved_target)
host_r, port_r, log_r = resolve_binding(
cli_host=host,
cli_port=port,
cli_log_level=log_level,
pc=pc,
settings_gateway_host=fl.settings.gateway_host,
settings_gateway_port=fl.settings.gateway_port,
settings_log_level=fl.settings.log_level,
)
run_unified(
resolved_target,
host=host_r,
port=port_r,
reload=reload,
reload_scope=reload_scope,
log_level=log_r,
proxy_headers=proxy_headers or workbench,
forwarded_allow_ips=forwarded_allow_ips,
pidfile=pidfile,
write_pidfile=not no_pidfile,
workbench_mode=workbench,
)
[docs]
@app.command()
def dev(
target: str | None = typer.Argument(
default=None,
help="Import path to your FluxLit instance (default: from fluxlit.toml or app:app).",
),
host: str | None = typer.Option(None, help="Bind address for the unified gateway."),
port: int | None = typer.Option(None, help="Port for the unified gateway."),
log_level: str | None = typer.Option(
None, help="Uvicorn log level (debug, info, warning, error)."
),
proxy_headers: bool = typer.Option(False, help="Trust X-Forwarded-* headers from a proxy."),
forwarded_allow_ips: str | None = typer.Option(
None,
help="Comma-separated IPs to trust for forwarded headers (uvicorn forwarded_allow_ips).",
),
reload: bool = typer.Option(
False,
help="Reload on code changes; use --reload-scope for gateway-only vs full stack.",
),
reload_scope: str = typer.Option(
"gateway",
"--reload-scope",
help="'gateway' reloads the ASGI app only; 'full' also restarts Streamlit on file changes.",
),
pidfile: Annotated[
Path | None,
typer.Option(
"--pidfile",
help="Where to write the PID file (default: .fluxlit-dev.pid or FLUXLIT_PIDFILE).",
),
] = None,
no_pidfile: bool = typer.Option(
False,
"--no-pidfile",
help="Do not write a PID file (also respect FLUXLIT_NO_PIDFILE=1).",
),
workbench: bool = typer.Option(
False,
"--workbench",
help=(
"Posit Workbench/Connect-style: enable Uvicorn proxy_headers and print a "
"loopback browser URL hint (set FLUXLIT_ROOT_PATH when using a subpath)."
),
),
debug: bool = typer.Option(
False,
"--debug",
help=(
"Diagnostics: set FLUXLIT_DEBUG=1 (gateway access logs, API request logging, "
"GET /__fluxlit/debug JSON, verbose gateway path splits). Not for production."
),
),
) -> None:
"""Run the unified stack for local development (Streamlit subprocess + Uvicorn gateway).
Resolves ``target``, bind address, port, and log level from CLI, project file, and
:class:`~fluxlit.config.FluxlitSettings`. See :func:`fluxlit.runtime.run_unified`.
"""
_execute_unified_cli(
target=target,
host=host,
port=port,
log_level=log_level,
proxy_headers=proxy_headers,
forwarded_allow_ips=forwarded_allow_ips,
pidfile=pidfile,
no_pidfile=no_pidfile,
reload=reload,
reload_scope=reload_scope,
workbench=workbench,
debug=debug,
)
[docs]
@app.command()
def shutdown(
pidfile: Annotated[
Path | None,
typer.Option(
"--pidfile",
help="PID file path (default: .fluxlit-dev.pid in cwd or FLUXLIT_PIDFILE).",
),
] = None,
force: bool = typer.Option(
False,
"--force",
"-f",
help="After wait, send SIGKILL (POSIX only) if still running.",
),
wait_s: float = typer.Option(
5.0,
"--wait",
help="Seconds to wait after SIGTERM before --force kicks in.",
),
) -> None:
"""Stop ``fluxlit dev`` or ``fluxlit run`` using the PID file they write.
Run this from the same working directory (or pass ``--pidfile`` / set ``FLUXLIT_PIDFILE``)
as the server process. Normal exit removes the PID file; this command also deletes it when
the process is already gone.
"""
code, msg = shutdown_unified_process(pidfile, force=force, wait_s=wait_s)
typer.echo(msg)
raise typer.Exit(code=code)
[docs]
@app.command("run")
def run_cmd(
target: str | None = typer.Argument(
default=None,
help="Import path to your FluxLit instance (default: from fluxlit.toml or app:app).",
),
host: str | None = typer.Option(None, help="Bind address for the unified gateway."),
port: int | None = typer.Option(None, help="Port for the unified gateway."),
log_level: str | None = typer.Option(
None, help="Uvicorn log level (debug, info, warning, error)."
),
proxy_headers: bool = typer.Option(False, help="Trust X-Forwarded-* headers from a proxy."),
forwarded_allow_ips: str | None = typer.Option(
None,
help="Comma-separated IPs to trust for forwarded headers (uvicorn forwarded_allow_ips).",
),
pidfile: Annotated[
Path | None,
typer.Option(
"--pidfile",
help="Where to write the PID file (default: .fluxlit-dev.pid or FLUXLIT_PIDFILE).",
),
] = None,
no_pidfile: bool = typer.Option(
False,
"--no-pidfile",
help="Do not write a PID file (also respect FLUXLIT_NO_PIDFILE=1).",
),
workbench: bool = typer.Option(
False,
"--workbench",
help=(
"Posit Workbench/Connect-style: enable Uvicorn proxy_headers and print a "
"loopback browser URL hint (set FLUXLIT_ROOT_PATH when using a subpath)."
),
),
debug: bool = typer.Option(
False,
"--debug",
help=(
"Diagnostics: set FLUXLIT_DEBUG=1 (gateway access logs, API request logging, "
"GET /__fluxlit/debug JSON, verbose gateway path splits). Not for production."
),
),
) -> None:
"""Run the unified stack for production-style use (no Uvicorn reload).
Same resolution rules as :func:`dev`; always calls :func:`fluxlit.runtime.run_unified`
with ``reload=False``.
"""
_execute_unified_cli(
target=target,
host=host,
port=port,
log_level=log_level,
proxy_headers=proxy_headers,
forwarded_allow_ips=forwarded_allow_ips,
pidfile=pidfile,
no_pidfile=no_pidfile,
reload=False,
reload_scope="gateway",
workbench=workbench,
debug=debug,
)
[docs]
@app.command("workbench")
def workbench_cmd(
target: str | None = typer.Argument(
default=None,
help="Import path to your FluxLit instance (default: from fluxlit.toml or app:app).",
),
host: str | None = typer.Option(None, help="Bind address for the unified gateway."),
port: int | None = typer.Option(None, help="Port for the unified gateway."),
log_level: str | None = typer.Option(
None, help="Uvicorn log level (debug, info, warning, error)."
),
forwarded_allow_ips: str | None = typer.Option(
None,
help="Comma-separated IPs to trust for forwarded headers (uvicorn forwarded_allow_ips).",
),
pidfile: Annotated[
Path | None,
typer.Option(
"--pidfile",
help="Where to write the PID file (default: .fluxlit-dev.pid or FLUXLIT_PIDFILE).",
),
] = None,
no_pidfile: bool = typer.Option(
False,
"--no-pidfile",
help="Do not write a PID file (also respect FLUXLIT_NO_PIDFILE=1).",
),
debug: bool = typer.Option(
False,
"--debug",
help="Same as ``fluxlit run --debug`` (sets FLUXLIT_DEBUG=1).",
),
) -> None:
"""Run the unified stack for Posit Workbench / Posit Connect-style path proxies.
Equivalent to ``fluxlit run`` with ``--workbench``: Uvicorn ``proxy_headers`` is enabled,
``forwarded_allow_ips`` defaults follow :class:`~fluxlit.config.FluxlitSettings`, and a
startup banner prints suggested loopback URLs (set ``FLUXLIT_ROOT_PATH`` for subpaths).
"""
_execute_unified_cli(
target=target,
host=host,
port=port,
log_level=log_level,
proxy_headers=False,
forwarded_allow_ips=forwarded_allow_ips,
pidfile=pidfile,
no_pidfile=no_pidfile,
reload=False,
reload_scope="gateway",
workbench=True,
debug=debug,
)
def _doctor_collect(
target: str,
pc: ProjectConfig | None,
*,
verbose: bool,
check_pages: bool = False,
strict: bool = False,
) -> tuple[list[tuple[str, CheckStatus, str]], dict[str, object] | None]:
"""Run static checks; each row is ``(name, PASS|WARN|FAIL, message)``.
When ``verbose`` is True and the app imports, also return a redacted configuration
snapshot for JSON / human ``--verbose`` output.
When ``check_pages`` is True or ``verbose`` is True, runs the same validation as
``fluxlit pages validate`` (non-strict signatures unless settings enable strict).
"""
from fluxlit.runtime import load_fluxlit
from fluxlit.runtime.import_target import import_target_candidates
rows: list[tuple[str, CheckStatus, str]] = []
rows.append(("python_version", "PASS", platform.python_version()))
sys_path_head = ", ".join((p or ".") for p in sys.path[:3])
rows.append(("sys_path_head", "PASS", sys_path_head or "(empty)"))
fl = None
host_r: str | None = None
port_r: int | None = None
mod_name = target.partition(":")[0]
candidates = import_target_candidates(mod_name)
if len(candidates) > 1:
shown = ", ".join(str(p) for p in candidates[:4])
more = f" (+{len(candidates) - 4} more)" if len(candidates) > 4 else ""
rows.append(
(
"import_shadowing",
"WARN",
f"multiple import candidates for {mod_name!r}: {shown}{more}; "
"run from the intended project root or trim PYTHONPATH",
)
)
elif candidates:
rows.append(("import_shadowing", "PASS", f"{mod_name!r} resolves from {candidates[0]}"))
else:
rows.append(
("import_shadowing", "PASS", f"no top-level file/package candidates for {mod_name!r}")
)
try:
fl = load_fluxlit(target)
rows.append(("import_target", "PASS", target))
module_key = Path(mod_name).stem if Path(mod_name).suffix == ".py" else mod_name
module_file = getattr(sys.modules.get(module_key), "__file__", "")
rows.append(("import_module_file", "PASS", str(module_file or "(not available)")))
except Exception as e:
rows.append(("import_target", "FAIL", str(e)))
try:
import fastapi # noqa: F401
import httpx # noqa: F401
import streamlit # noqa: F401
import uvicorn # noqa: F401
except ImportError as e:
rows.append(("dependencies", "FAIL", str(e)))
else:
rows.append(("dependencies", "PASS", "fastapi, streamlit, uvicorn, httpx importable"))
try:
import streamlit as st
parts = st.__version__.split(".")
major, minor = int(parts[0]), int(parts[1])
if (major, minor) < (1, 33):
rows.append(
(
"streamlit_version",
"WARN",
f"{st.__version__} (fluxlit declares streamlit>=1.36; upgrade recommended)",
)
)
else:
rows.append(("streamlit_version", "PASS", st.__version__))
except Exception as e:
rows.append(("streamlit_version", "WARN", str(e)))
if fl is not None:
host_r, port_r, _ = resolve_binding(
cli_host=None,
cli_port=None,
cli_log_level=None,
pc=pc,
settings_gateway_host=fl.settings.gateway_host,
settings_gateway_port=fl.settings.gateway_port,
settings_log_level=fl.settings.log_level,
)
if host_r in {"0.0.0.0", ""}:
bind_host = "127.0.0.1"
elif host_r == "::":
bind_host = "::1"
else:
bind_host = host_r
try:
infos = socket.getaddrinfo(bind_host, port_r, type=socket.SOCK_STREAM)
if not infos:
raise OSError(f"could not resolve bind host {bind_host!r}")
family, socktype, proto, _, sockaddr = infos[0]
with socket.socket(family, socktype, proto) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(sockaddr)
rows.append(("gateway_bind", "PASS", f"{host_r}:{port_r} is available"))
except OSError as e:
rows.append(("gateway_bind", "FAIL", str(e)))
internal = os.environ.get("FLUXLIT_INTERNAL_API_BASE", "").strip()
if internal:
parsed = urlparse(internal)
if not parsed.scheme or not parsed.netloc:
rows.append(("FLUXLIT_INTERNAL_API_BASE", "FAIL", "not a valid absolute URL"))
elif fl is not None:
url_path = (parsed.path or "/").rstrip("/")
expected = fl.settings.api_mount_path.rstrip("/")
if url_path != expected:
rows.append(
(
"FLUXLIT_INTERNAL_API_BASE",
"WARN",
f"path {url_path!r} should match api_mount_path {expected!r}",
)
)
else:
rows.append(("FLUXLIT_INTERNAL_API_BASE", "PASS", "set and matches api_mount_path"))
else:
rows.append(
(
"FLUXLIT_INTERNAL_API_BASE",
"PASS",
"set (import failed; path not compared to api_mount_path)",
)
)
else:
rows.append(
(
"FLUXLIT_INTERNAL_API_BASE",
"PASS",
"unset (normal when not inside Streamlit subprocess)",
)
)
if _fluxlit_auth_env_present():
rows.append(
(
"jwt_clock_skew",
"WARN",
"JWT/OIDC env detected — synchronize clocks (NTP); validation uses exp/nbf.",
)
)
else:
rows.append(
(
"jwt_clock_skew",
"PASS",
"no JWT/OIDC FLUXLIT_* env vars detected",
)
)
if _fluxlit_auth_env_present():
try:
import jwt # noqa: F401
except ImportError:
rows.append(
(
"fluxlit_auth_extra",
"FAIL",
"JWT/OIDC env vars set but PyJWT is missing; pip install 'fluxlit[auth]'",
)
)
else:
rows.append(("fluxlit_auth_extra", "PASS", "PyJWT importable (fluxlit[auth])"))
if fl is not None and fl.settings.enable_gateway_prometheus_metrics:
if find_spec("prometheus_client") is None:
rows.append(
(
"fluxlit_metrics_extra",
"FAIL",
"Prometheus metrics enabled but prometheus-client is missing; "
"pip install 'fluxlit[metrics]'",
)
)
else:
rows.append(("fluxlit_metrics_extra", "PASS", "prometheus-client importable"))
if fl is not None:
rows.append(("config.api_mount_path", "PASS", fl.settings.api_mount_path))
rows.append(
(
"config.url_session",
"PASS",
f"param={fl.settings.url_session_query_param!r}; "
f"tests={bool(os.environ.get('FLUXLIT_TESTS'))}; "
f"disabled={bool(os.environ.get('FLUXLIT_DISABLE_URL_SESSION'))}",
)
)
rows.append(("config.root_path", "PASS", fl.settings.root_path or "(empty)"))
rows.append(("config.trust_proxy", "PASS", str(fl.settings.trust_proxy)))
rows.append(
(
"config.public_base_url",
"PASS",
fl.settings.public_base_url.strip() or "(empty)",
)
)
legacy_public = os.environ.get("PUBLIC_BASE_URL", "").strip()
fluxlit_public = os.environ.get("FLUXLIT_PUBLIC_BASE_URL", "").strip()
if legacy_public and fluxlit_public and legacy_public != fluxlit_public:
precedence_status: CheckStatus = (
"FAIL" if (fl.settings.strict_public_base_url or strict) else "WARN"
)
rows.append(
(
"public_base_url_precedence",
precedence_status,
"PUBLIC_BASE_URL and FLUXLIT_PUBLIC_BASE_URL differ; "
"FluxLit uses FLUXLIT_PUBLIC_BASE_URL",
)
)
elif fluxlit_public:
rows.append(("public_base_url_precedence", "PASS", "using FLUXLIT_PUBLIC_BASE_URL"))
elif legacy_public:
rows.append(("public_base_url_precedence", "PASS", "using PUBLIC_BASE_URL fallback"))
if (
fl is not None
and fl.settings.public_mount_path()
and not fl.settings.public_base_url.strip()
):
oauth_status: CheckStatus = "FAIL" if strict else "WARN"
rows.append(
(
"oauth_public_base_url",
oauth_status,
"subpath set but public_base_url empty — set FLUXLIT_PUBLIC_BASE_URL for correct "
"OAuth redirect_uri behind a reverse proxy (see docs/configuration.html)",
)
)
if fl is not None and fl.settings.public_mount_path():
if fl.settings.trust_proxy:
rows.append(
(
"proxy_headers",
"PASS",
"trust_proxy enabled for subpath deployment",
)
)
else:
ph_status: CheckStatus = "FAIL" if strict else "WARN"
rows.append(
(
"proxy_headers",
ph_status,
"subpath/root_path set but trust_proxy false — behind Posit Connect/nginx set "
"FLUXLIT_TRUST_PROXY=1 or pass --proxy-headers",
)
)
if fl is not None and fl.settings.trust_proxy:
allow = (fl.settings.forwarded_allow_ips or "").strip()
if not allow or allow == "*":
allow_status: CheckStatus = "FAIL" if strict else "WARN"
rows.append(
(
"forwarded_allow_ips",
allow_status,
"trust_proxy enabled but forwarded_allow_ips is broad; set "
"FLUXLIT_FORWARDED_ALLOW_IPS to your proxy IP/CIDR in production",
)
)
else:
rows.append(("forwarded_allow_ips", "PASS", allow))
if (
fl is not None
and fl.settings.trust_proxy
and fl.settings.gateway_max_proxy_request_body_bytes == 0
):
mb_status: CheckStatus = "FAIL" if strict else "WARN"
rows.append(
(
"gateway_max_proxy_body",
mb_status,
"trust_proxy enabled but gateway_max_proxy_request_body_bytes is 0 (unlimited); "
"set FLUXLIT_GATEWAY_MAX_PROXY_REQUEST_BODY_BYTES for production",
)
)
if fl is not None and fl.settings.public_base_url.strip():
parsed_public = urlparse(fl.settings.public_base_url.strip())
root = fl.settings.public_mount_path().rstrip("/")
public_path = (parsed_public.path or "").rstrip("/")
if root and public_path and public_path != root:
pub_status: CheckStatus = "FAIL" if strict else "WARN"
rows.append(
(
"public_base_url",
pub_status,
f"path {public_path!r} does not match public mount {root!r}",
)
)
elif not parsed_public.scheme or not parsed_public.netloc:
rows.append(("public_base_url", "FAIL", "not a valid absolute URL"))
else:
rows.append(("public_base_url", "PASS", fl.settings.public_base_url.strip()))
if fl is not None:
api_ready = f"{fl.settings.api_mount_path.rstrip('/')}/readyz"
rows.append(
(
"readiness_route",
"PASS",
f"Streamlit readiness probe: GET {api_ready} (hidden from OpenAPI)",
)
)
rows.append(
(
"l7_websocket",
"PASS",
"Ensure L7 proxies allow Upgrade/WebSocket for /_stcore/* (see docs/runbooks.md)",
)
)
if fl.settings.gateway_upstream_read_timeout_s < 5.0:
to_status: CheckStatus = "FAIL" if strict else "WARN"
rows.append(
(
"gateway_upstream_timeouts",
to_status,
"gateway_upstream_read_timeout_s is very low; "
"long Streamlit responses may fail",
)
)
else:
rows.append(
(
"gateway_upstream_timeouts",
"PASS",
f"HTTP read timeout={fl.settings.gateway_upstream_read_timeout_s}s; "
f"ws_open={fl.settings.gateway_ws_open_timeout_s}s",
)
)
if fl.settings.async_page_depends:
rows.append(
(
"async_depends_streamlit",
"WARN",
"Async Depends enabled — deps may run on an isolated thread loop; avoid "
"thread-unsafe globals (docs/streamlit-pages-typing.html)",
)
)
else:
rows.append(
(
"async_depends_streamlit",
"PASS",
"async page Depends disabled (FLUXLIT_ASYNC_PAGE_DEPENDS unset)",
)
)
fwd_n = len(fl.settings.gateway_forward_client_headers_to_streamlit or [])
if fwd_n:
rows.append(
(
"gateway_forward_client_headers",
"WARN",
f"{fwd_n} header name(s) forwarded to Streamlit HTTP — allowlist only "
"non-secrets; see docs/configuration.html",
)
)
else:
rows.append(
(
"gateway_forward_client_headers",
"PASS",
"no browser→Streamlit HTTP header forwarding (default); "
"set_page_header_context / set_page_cookie_context for explicit injection",
)
)
rejected = tuple(getattr(fl.settings, "_rejected_forward_headers", ()))
if rejected:
rej_status: CheckStatus = "FAIL" if strict else "WARN"
rows.append(
(
"gateway_forward_rejected_names",
rej_status,
f"allowlist contained rejected names {list(rejected)} — never forwarded; "
"see docs/security.html",
)
)
upstream_file = os.environ.get("FLUXLIT_STREAMLIT_UPSTREAM_FILE", "").strip()
upstream_env = os.environ.get("FLUXLIT_STREAMLIT_UPSTREAM", "").strip()
if upstream_file:
fp = Path(upstream_file)
if not fp.exists():
rows.append(("streamlit_upstream_state", "FAIL", f"state file missing: {fp}"))
else:
raw = fp.read_text(encoding="utf-8").strip()
if not raw:
rows.append(("streamlit_upstream_state", "WARN", f"state file is empty: {fp}"))
else:
ok, msg = _tcp_url_reachable(raw)
rows.append(("streamlit_upstream_state", "PASS" if ok else "WARN", msg))
elif upstream_env:
ok, msg = _tcp_url_reachable(upstream_env)
rows.append(("streamlit_upstream", "PASS" if ok else "WARN", msg))
cors_on = fl is not None and bool(fl.settings.cors_allow_origins)
sec_off = fl is not None and not fl.settings.enable_security_headers
if cors_on and sec_off:
sec_status: CheckStatus = "FAIL" if strict else "WARN"
rows.append(
(
"security_headers",
sec_status,
"CORS on, security headers off — consider FLUXLIT_ENABLE_SECURITY_HEADERS=1",
)
)
if fl is not None and (check_pages or verbose):
from fluxlit.pages.flags import FluxlitFeatureFlags
from fluxlit.pages.validate import validate_fluxlit_pages
errs = validate_fluxlit_pages(fl, strict_signatures=None)
if errs:
rows.append(("pages_validate", "FAIL", "; ".join(errs)))
else:
rows.append(
(
"pages_validate",
"PASS",
"manifest JSON-serializable and page signature checks passed",
)
)
if FluxlitFeatureFlags.from_environ().experimental_yield_pages or (
fl.settings.experimental_yield_pages
):
rows.append(
(
"experimental_yield_pages",
"WARN",
"FLUXLIT_EXPERIMENTAL_YIELD_PAGES is on — generator pages are experimental",
)
)
verbose_detail: dict[str, object] | None = None
if verbose:
if fl is not None and host_r is not None and port_r is not None:
verbose_detail = build_doctor_verbose_detail(
fl,
resolved_target=target,
bind_host=host_r,
bind_port=port_r,
pc=pc,
)
else:
verbose_detail = {"resolved_target": target, "import_failed": True}
return rows, verbose_detail
def _doctor_checks(
target: str, *, verbose: bool = False, check_pages: bool = False, strict: bool = False
) -> list[tuple[str, CheckStatus, str]]:
"""Backward-compatible wrapper used by tests."""
rows, _ = _doctor_collect(
target,
load_project_config(),
verbose=verbose,
check_pages=check_pages,
strict=strict,
)
return rows
def _doctor_payload(
rows: list[tuple[str, CheckStatus, str]],
*,
target: str,
warnings_only: bool,
verbose_detail: dict[str, object] | None = None,
) -> dict[str, object]:
has_fail = any(status == "FAIL" for _, status, _ in rows)
has_warn = any(status == "WARN" for _, status, _ in rows)
status = (
"fail" if has_fail and not warnings_only else "warn" if has_fail or has_warn else "pass"
)
out: dict[str, object] = {
"status": status,
"target": target,
"warnings_only": warnings_only,
"checks": [
{"name": name, "status": check_status, "detail": detail}
for name, check_status, detail in rows
],
}
if verbose_detail is not None:
out["verbose"] = verbose_detail
return out
[docs]
@app.command()
def doctor(
target: str | None = typer.Argument(
default=None,
help="Import path to your FluxLit instance (default: from fluxlit.toml or app:app).",
),
warnings_only: bool = typer.Option(
False,
"--warnings-only",
help="Always exit 0 (still print FAIL lines).",
),
json_output: bool = typer.Option(
False,
"--json",
help="Emit machine-readable JSON instead of human-readable text.",
),
verbose: bool = typer.Option(
False,
"--verbose",
"-v",
help="After checks, print a redacted effective-config snapshot (also included as "
"``verbose`` in ``--json`` output).",
),
check_pages: bool = typer.Option(
False,
"--check-pages",
help="Run ``fluxlit pages validate``-style checks (manifest + signatures per settings).",
),
strict: bool = typer.Option(
False,
"--strict",
help="Treat broad forwarded_allow_ips and public_base_url path mismatch as FAIL.",
),
) -> None:
"""Print PASS/WARN/FAIL diagnostics (imports, deps, bind, env).
Exits with code ``1`` if any check fails, unless ``--warnings-only`` is set.
"""
pc = load_project_config()
resolved_target = resolve_target(target, pc)
rows, verbose_detail = _doctor_collect(
resolved_target,
pc,
verbose=verbose,
check_pages=check_pages,
strict=strict,
)
if json_output:
typer.echo(
json.dumps(
_doctor_payload(
rows,
target=resolved_target,
warnings_only=warnings_only,
verbose_detail=verbose_detail if verbose else None,
)
)
)
else:
typer.echo("FluxLit doctor")
typer.echo("")
for name, status, message in rows:
typer.echo(f"{status:4} {name}: {message}")
typer.echo("")
if verbose and verbose_detail is not None:
typer.echo("--- Verbose (effective configuration; secrets redacted) ---")
for line in format_doctor_verbose_human(verbose_detail):
typer.echo(line)
typer.echo("")
if not warnings_only and any(s == "FAIL" for _, s, _ in rows):
raise typer.Exit(code=1)
[docs]
@app.command()
def config(
target: str | None = typer.Argument(
default=None,
help="Import path to your FluxLit instance (default: from fluxlit.toml or app:app).",
),
json_output: bool = typer.Option(
False,
"--json",
help="Emit machine-readable JSON (secrets redacted).",
),
strict: bool = typer.Option(
False,
"--strict",
help="Exit with code 1 when any warning is emitted (errors always exit 1).",
),
) -> None:
"""Print effective FluxLit configuration after precedence rules (env, project file, app).
Shows resolved bind defaults, redacted settings, derived internal API base, and warnings
with documentation links. Does not start the server.
"""
from fluxlit.runtime import load_fluxlit
pc = load_project_config()
resolved_target = resolve_target(target, pc)
fl = load_fluxlit(resolved_target)
host_r, port_r, log_r = resolve_binding(
cli_host=None,
cli_port=None,
cli_log_level=None,
pc=pc,
settings_gateway_host=fl.settings.gateway_host,
settings_gateway_port=fl.settings.gateway_port,
settings_log_level=fl.settings.log_level,
)
payload = build_config_payload(
target=resolved_target,
bind_host=host_r,
bind_port=port_r,
log_level=log_r,
pc=pc,
fl=fl,
)
if json_output:
typer.echo(json.dumps(payload, indent=2))
else:
typer.echo("FluxLit effective configuration")
typer.echo("")
typer.echo(f" target: {payload['target']}")
b = payload["binding"]
typer.echo(
f" binding (resolved): host={b['host']} port={b['port']} log_level={b['log_level']}"
)
comp = payload["computed"]
typer.echo(f" public_mount_path: {comp['public_mount_path'] or '(empty)'}")
typer.echo(f" derived_internal_api_base: {comp['derived_internal_api_base']}")
if comp.get("ambient_internal_api_base"):
typer.echo(f" FLUXLIT_INTERNAL_API_BASE: {comp['ambient_internal_api_base']}")
pf = payload.get("project_file")
if pf:
typer.echo(f" project_file: {json.dumps(pf)}")
warns = payload["warnings"]
if warns:
typer.echo("")
typer.echo("Warnings:")
for w in warns:
lvl = str(w.get("level", "warn")).upper()
typer.echo(f" [{lvl}] {w.get('code', '')}: {w.get('message', '')}")
if w.get("doc"):
typer.echo(f" Docs: {w['doc']}")
typer.echo("")
typer.echo("Settings (redacted, JSON):")
typer.echo(json.dumps(payload["settings"], indent=2))
typer.echo("")
warns = payload["warnings"]
if any(w.get("level") == "error" for w in warns):
raise typer.Exit(code=1)
if strict and any(w.get("level") == "warn" for w in warns):
raise typer.Exit(code=1)
[docs]
@app.command()
def build(
output: Annotated[
Path | None,
typer.Option(
"--output",
"-o",
help="Directory for Dockerfile and .dockerignore (default: current directory).",
),
] = None,
target: str | None = typer.Argument(
default=None,
help="Import path for CMD (default: from fluxlit.toml or app:app).",
),
force: bool = typer.Option(
False,
"--force",
"-f",
help="Overwrite existing Dockerfile / .dockerignore.",
),
) -> None:
"""Emit a minimal ``Dockerfile`` and ``.dockerignore`` for container deployment.
Refuses to overwrite existing files unless ``--force``. Adjust generated files for
your layout (dependencies, non-root user, etc.).
"""
pc = load_project_config()
resolved_target = resolve_target(target, pc)
out_dir = output or Path(".")
out_dir.mkdir(parents=True, exist_ok=True)
dockerfile = out_dir / "Dockerfile"
dockerignore = out_dir / ".dockerignore"
if not force:
if dockerfile.exists() or dockerignore.exists():
typer.echo(
f"Refusing to overwrite {dockerfile} / {dockerignore} (use --force).",
err=True,
)
raise typer.Exit(code=1)
dockerfile.write_text(_dockerfile_body(resolved_target), encoding="utf-8")
dockerignore.write_text(_dockerignore_body(), encoding="utf-8")
typer.echo(f"Wrote {dockerfile} and {dockerignore}")
[docs]
@app.command()
def new(
name: str = typer.Argument(..., help="Project directory name."),
profile: Literal["minimal", "auth-ready", "deploy"] = typer.Option(
"minimal",
"--profile",
help="Scaffold profile: minimal, auth-ready, or deploy.",
),
) -> None:
"""Create ``<name>/app.py`` with a sample API route and Streamlit home page."""
root = Path(name)
if root.exists():
typer.echo(f"Destination already exists: {root}", err=True)
raise typer.Exit(code=1)
root.mkdir(parents=True)
(root / "fluxlit.toml").write_text(
'target = "app:app"\ngateway_port = 8000\n',
encoding="utf-8",
)
(root / "app.py").write_text(_scaffold_app_body(profile), encoding="utf-8")
typer.echo(
f"Created {name}/ ({profile}) — run: cd {name} && fluxlit dev "
f"(or: uvicorn app:app --reload --port 8000)"
)
[docs]
def main() -> None:
"""Invoke the root Typer application (console script entrypoint)."""
app()