1import os
2import subprocess
3import sys
4import time
5from importlib.metadata import entry_points
6
7import click
8
9from plain.cli import register_cli
10from plain.cli.runtime import common_command
11from plain.runtime import PLAIN_TEMP_PATH
12
13from .alias import AliasManager
14from .core import ENTRYPOINT_GROUP, DevProcess
15from .services import ServicesProcess
16
17
18@common_command
19@register_cli("dev")
20@click.group(invoke_without_command=True)
21@click.pass_context
22@click.option(
23 "--port",
24 "-p",
25 default="",
26 type=str,
27 help=(
28 "Port to run the web server on. "
29 "If omitted, tries 8443 and picks the next free port."
30 ),
31)
32@click.option(
33 "--hostname",
34 "-h",
35 default=None,
36 type=str,
37 help="Hostname to run the web server on",
38)
39@click.option(
40 "--log-level",
41 "-l",
42 default="",
43 type=click.Choice(["debug", "info", "warning", "error", "critical", ""]),
44 help="Log level",
45)
46@click.option(
47 "--start",
48 is_flag=True,
49 default=False,
50 help="Start in the background",
51)
52@click.option(
53 "--stop",
54 is_flag=True,
55 default=False,
56 help="Stop the background process",
57)
58@click.option(
59 "--reinstall-ssl",
60 is_flag=True,
61 default=False,
62 help="Reinstall SSL certificates (updates mkcert, reinstalls CA, regenerates certs)",
63)
64def cli(
65 ctx: click.Context,
66 port: str,
67 hostname: str | None,
68 log_level: str,
69 start: bool,
70 stop: bool,
71 reinstall_ssl: bool,
72) -> None:
73 """Local development server"""
74 if ctx.invoked_subcommand:
75 return
76
77 if start and stop:
78 raise click.UsageError(
79 "You cannot use both --start and --stop at the same time."
80 )
81
82 os.environ["PLAIN_DEV_SERVICES_AUTO"] = "false"
83
84 dev = DevProcess()
85
86 if stop:
87 if ServicesProcess.running_pid():
88 ServicesProcess().stop_process()
89 click.secho("Services stopped.", fg="green")
90
91 if not dev.running_pid():
92 click.secho("No development server running.", fg="yellow")
93 return
94
95 dev.stop_process()
96 click.secho("Development server stopped.", fg="green")
97 return
98
99 if running_pid := dev.running_pid():
100 click.secho(f"`plain dev` already running (pid={running_pid})", fg="yellow")
101 sys.exit(1)
102
103 if start:
104 args = [sys.executable, "-m", "plain", "dev"]
105 if port:
106 args.extend(["--port", port])
107 if hostname:
108 args.extend(["--hostname", hostname])
109 if log_level:
110 args.extend(["--log-level", log_level])
111
112 result = subprocess.Popen(
113 args=args,
114 start_new_session=True,
115 stdout=subprocess.DEVNULL,
116 stderr=subprocess.DEVNULL,
117 )
118 click.secho(
119 f"Development server started in the background (pid={result.pid}).",
120 fg="green",
121 )
122 return
123
124 # Check and prompt for alias setup
125 AliasManager().check_and_prompt()
126
127 dev.setup(
128 port=int(port) if port else None,
129 hostname=hostname,
130 log_level=log_level if log_level else None,
131 )
132 returncode = dev.run(reinstall_ssl=reinstall_ssl)
133 if returncode:
134 sys.exit(returncode)
135
136
137@cli.command()
138def debug() -> None:
139 """Connect to the remote debugger"""
140
141 def _connect() -> subprocess.CompletedProcess[bytes]:
142 if subprocess.run(["which", "nc"], capture_output=True).returncode == 0:
143 return subprocess.run(["nc", "-C", "localhost", "4444"])
144 else:
145 raise OSError("nc not found")
146
147 result = _connect()
148
149 # Try again once without a message
150 if result.returncode == 1:
151 time.sleep(1)
152 result = _connect()
153
154 # Keep trying...
155 while result.returncode == 1:
156 click.secho(
157 "Failed to connect. Make sure remote pdb is ready. Retrying...", fg="red"
158 )
159 result = _connect()
160 time.sleep(1)
161
162
163@cli.command()
164@click.option("--start", is_flag=True, help="Start in the background")
165@click.option("--stop", is_flag=True, help="Stop the background process")
166def services(start: bool, stop: bool) -> None:
167 """Start additional development services"""
168
169 if start and stop:
170 raise click.UsageError(
171 "You cannot use both --start and --stop at the same time."
172 )
173
174 if stop:
175 if not ServicesProcess.running_pid():
176 click.secho("No services running.", fg="yellow")
177 return
178 ServicesProcess().stop_process()
179 click.secho("Services stopped.", fg="green")
180 return
181
182 if running_pid := ServicesProcess.running_pid():
183 click.secho(f"Services already running (pid={running_pid})", fg="yellow")
184 sys.exit(1)
185
186 if start:
187 result = subprocess.Popen(
188 args=[sys.executable, "-m", "plain", "dev", "services"],
189 start_new_session=True,
190 stdout=subprocess.DEVNULL,
191 stderr=subprocess.DEVNULL,
192 )
193 click.secho(
194 f"Services started in the background (pid={result.pid}).", fg="green"
195 )
196 return
197
198 ServicesProcess().run()
199
200
201@cli.command()
202@click.option("--follow", "-f", is_flag=True, help="Follow log output")
203@click.option("--pid", type=int, help="PID to show logs for")
204@click.option("--path", is_flag=True, help="Output log file path")
205@click.option("--services", is_flag=True, help="Show logs for services")
206def logs(follow: bool, pid: int | None, path: bool, services: bool) -> None:
207 """Show recent development logs"""
208
209 if services:
210 log_dir = PLAIN_TEMP_PATH / "dev" / "logs" / "services"
211 else:
212 log_dir = PLAIN_TEMP_PATH / "dev" / "logs" / "run"
213
214 if pid:
215 log_path = log_dir / f"{pid}.log"
216 if not log_path.exists():
217 click.secho(f"No log found for pid {pid}", fg="red")
218 return
219 else:
220 logs = sorted(log_dir.glob("*.log"), key=lambda p: p.stat().st_mtime)
221 if not logs:
222 click.secho("No logs found", fg="yellow")
223 return
224 log_path = logs[-1]
225
226 if path:
227 click.echo(str(log_path))
228 return
229
230 if follow:
231 subprocess.run(["tail", "-f", str(log_path)])
232 else:
233 with log_path.open() as f:
234 click.echo(f.read())
235
236
237@cli.command()
238@click.option(
239 "--list", "-l", "show_list", is_flag=True, help="List available entrypoints"
240)
241@click.argument("entrypoint", required=False)
242def entrypoint(show_list: bool, entrypoint: str | None) -> None:
243 """Run registered development entrypoints"""
244 if not show_list and not entrypoint:
245 raise click.UsageError("Please provide an entrypoint name or use --list")
246
247 for entry_point in entry_points().select(group=ENTRYPOINT_GROUP):
248 if show_list:
249 click.echo(entry_point.name)
250 elif entrypoint == entry_point.name:
251 entry_point.load()()