20771719563

Committed 07 Jan 2026 05:32AM UTC coverage: 90.735%. First build

Build # 20771719563

Build Type

push

github

Committed by

web-flow

Commit Message

💥 Drop support for python3.9 (#39)

Run Details

7 of 8 new or added lines in 5 files covered. (87.5%)

617 of 680 relevant lines covered (90.74%)

9.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

65.5

/fastapi_cdn_host/cli.py

#!/usr/bin/env python
from __future__ import annotations

import functools
import os
import shlex
import subprocess  # nosec:B404
import sys
from collections.abc import AsyncGenerator, Callable, Generator
from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Annotated, Any

import anyio
import typer
from rich.progress import Progress, SpinnerColumn, TaskID

from .client import CdnHostBuilder, HttpSniff

if TYPE_CHECKING:
    if sys.version_info >= (3, 11):
        from typing import Self
    else:
        from typing_extensions import Self


app = typer.Typer()


def load_bool(name: str) -> bool:
    v = os.getenv(name)
    if not v:
        return False
    return v.lower() in ("1", "true", "t", "y", "yes", "on")


def run_shell(cmd: str) -> None:
    """Run cmd by subprocess

    Example::
        run_shell('PYTHONPATH=. python main.py')

    Will be convert to:
        subprocess.run(['python', 'main.py'], env={**os.environ, 'PYTHONPATH': '.'})
    """
    typer.echo(f"--> {cmd}")
    command = shlex.split(cmd)
    cmd_env = None
    index = 0
    for i, c in enumerate(command):
        if "=" not in c:
            index = i
            break
        name, value = c.split("=")
        if cmd_env is None:
            cmd_env = {**os.environ, name: value}
        else:
            cmd_env[name] = value
    if cmd_env is not None:
        command = command[index:]
    subprocess.run(command, env=cmd_env)  # nosec:B603


TEMPLATE = """
#!/usr/bin/env python
'''This file is auto generated by fastapi_cdn_host.
Feel free to change or remove it.
'''
import subprocess
import sys

import fastapi_cdn_host

from {} import app

fastapi_cdn_host.patch_docs(app)

def _runserver() -> int:
    r = subprocess.run(["fastapi", "dev", __file__, *sys.argv[2:]])
    return int(bool(r.returncode))

if __name__ == "__main__":
    sys.exit(_runserver())
"""


def write_app(dest: Path, from_path: str | Path) -> None:
    module = Path(from_path).stem
    size = dest.write_text(TEMPLATE.format(module).strip())
    typer.echo(f"Create {dest} with {size=}")


@contextmanager
def patch_app(path: str | Path, remove: bool = True) -> Generator[Path, None, None]:
    ident = f"{datetime.now():%Y%m%d%H%M%S}"
    app_file = Path(f"app_{ident}.py")
    write_app(app_file, path)
    try:
        yield app_file
    finally:
        if remove:
            app_file.unlink()
            typer.echo(f"Auto remove temp file: {app_file}")


class PercentBar(AbstractAsyncContextManager):
    default_threshold: Annotated[int, "0-100"] = 80
    total_seconds: Annotated[int, "Cancel task if reach this value"] = 5

    def __init__(
        self, msg: str, seconds: int | None = None, color: str = "", **kw: Any
    ) -> None:
        self.seconds = seconds or self.total_seconds
        self.prompt = self.build_prompt(msg, color)
        self.progress = Progress(**kw)

    @staticmethod
    def build_prompt(msg: str, color: str | None, suffix: str = ":") -> str:
        prompt = f"{msg}{suffix}"
        if color:
            prompt = f"[{color}]" + prompt
        return prompt

    async def play(
        self, progress: Progress, task: TaskID, threshold: int | None = None
    ) -> None:
        expected = 1 / 2
        progress_forward = functools.partial(progress.advance, task)

        # Play quickly: run 80% progress in 1/2 total seconds
        threshold = threshold or self.default_threshold
        cost = self.seconds * expected
        await self.percent_play(cost, threshold, progress_forward)

        # Play slow
        slow = 100 - threshold
        cost = self.seconds - cost
        await self.percent_play(cost, slow, progress_forward)

    @staticmethod
    async def percent_play(cost: float, percent: int, forward: Callable) -> None:
        delay = cost / percent
        for _ in range(percent):
            await anyio.sleep(delay)
            forward()

    async def __aenter__(self) -> Self:
        self.progress.start()
        self.progress_task = self.progress.add_task(self.prompt)
        self.task_group = await anyio.create_task_group().__aenter__()
        self.task_group.start_soon(self.play, self.progress, self.progress_task)
        return self

    async def __aexit__(self, *args, **kw) -> None:
        self.task_group.cancel_scope.cancel()
        self.progress.update(self.progress_task, completed=100)
        await self.task_group.__aexit__(*args, **kw)
        self.progress.__exit__(*args, **kw)


@asynccontextmanager
async def percentbar(msg: str, **kwargs: Any) -> AsyncGenerator[None, None]:
    """Progressbar with custom font color

    :param msg: prompt message.
    """
    async with PercentBar(msg, **kwargs):
        yield


class SpinnerProgress(Progress):
    def __init__(self, msg: str, color: str | None = None, **kwargs: Any) -> None:
        self.prompt = PercentBar.build_prompt(msg, color, "...")
        kwargs.setdefault("transient", True)
        super().__init__(SpinnerColumn(), *Progress.get_default_columns(), **kwargs)

    def start(self) -> None:
        super().start()
        self.add_task(self.prompt, total=None)


@contextmanager
def spinnerbar(
    msg: str, color: str | None = None, **kwargs: Any
) -> Generator[None, None, None]:
    with SpinnerProgress(msg, color, **kwargs):
        yield


async def download_offline_assets(dirname: str | Path, timeout: float = 30) -> None:
    cwd = await anyio.Path.cwd()
    static_root = (
        cwd / dirname if isinstance(dirname, str) else anyio.Path(dirname.resolve())
    )
    if not await static_root.exists():
        await static_root.mkdir(parents=True)
        typer.echo(f"Directory {static_root} created.")
    else:
        async for p in static_root.glob("swagger-ui*.js"):
            relative_path = p.relative_to(cwd)
            typer.echo(f"{relative_path} already exists. abort!")
            return
    async with percentbar("Comparing cdn hosts response speed"):
        urls = await CdnHostBuilder.sniff_the_fastest()
    typer.echo(f"Result: {urls}")
    with spinnerbar("Fetching files from cdn", color="yellow"):
        url_list = [urls.js, urls.css, urls.redoc]
        contents = await HttpSniff.bulk_fetch(
            url_list, get_content=True, total_seconds=timeout
        )
        for url, content in zip(url_list, contents, strict=False):
            if not content:
                red_head = typer.style("ERROR:", fg=typer.colors.RED)
                typer.echo(red_head + f" Failed to fetch content from {url}")
            else:
                path = static_root / Path(url).name
                size = await path.write_bytes(content)
                typer.echo(f"Write to {path} with {size=}")
    typer.secho("Done.", fg=typer.colors.GREEN)


def handle_cache() -> None:
    exists, cache_path = CdnHostBuilder.get_cache_file()
    if not exists:
        typer.echo("Cache not create yet.")
        return
    if "--remove" in sys.argv:
        cache_path.unlink()
        typer.secho(
            f"Success to remove cache file({cache_path})", fg=typer.colors.GREEN
        )
    else:
        typer.echo(f"Content of cache file({cache_path}):\n{cache_path.read_text()}")
        if "--reload" in sys.argv:
            cache_path.unlink()
            with spinnerbar("Refreshing cache", color="yellow"):
                CdnHostBuilder(cache=True).run()
            typer.echo(f"Cache file updated:\n{cache_path.read_text()}")


@app.command()
def dev(
    path: Annotated[
        Path,
        typer.Argument(
            help=(
                "A path to a Python file or package directory"
                " (with [blue]__init__.py[/blue] file)"
                " containing a [bold]FastAPI[/bold] app."
                " If not provided, a default set of paths will be tried."
            )
        ),
    ],
    port: Annotated[
        int,
        typer.Option(
            help=(
                "The port to serve on."
                " You would normally have a termination proxy on top (another program)"
                " handling HTTPS on port [blue]443[/blue] and HTTP on port [blue]80[/blue],"
                " transferring the communication to your app."
            )
        ),
    ] = 0,
    remove: Annotated[
        bool,
        typer.Option(
            help="Whether remove the temp app_<time>.py file after server stopped."
        ),
    ] = True,
    prod: Annotated[
        bool,
        typer.Option(help="Whether enable production mode."),
    ] = False,
    reload: Annotated[
        bool,
        typer.Option(help="Enable auto-reload of the server when (code) files change."),
    ] = True,
) -> None:
    if str(path) == "offline":
        anyio.run(download_offline_assets, "static")
        return
    elif str(path) == "cache":
        handle_cache()
        return
    with patch_app(path, remove) as file:
        runserver(file, prod, reload, port)


def runserver(file: Path, prod: bool, reload: bool, port: int) -> None:
    if load_bool("FASTCDN_UVICORN"):
        module = file.stem
        if file.parent != Path() and file.parent.resolve() != Path.cwd():
            os.chdir(file.parent)
        cmd = f"PYTHONPATH=. uvicorn {module}:app"
        if reload and not load_bool("FASTCDN_NORELOAD"):
            cmd += " --reload"
    else:
        mode = "run" if prod else "dev"
        cmd = f"PYTHONPATH=. fastapi {mode} {file}"
        if (not reload and not prod) or load_bool("FASTCDN_NORELOAD"):
            cmd += " --no-reload"
    if port:
        cmd += f" --{port=}"
    run_shell(cmd)


def main() -> None:
    app()


if __name__ == "__main__":
    main()

1	#!/usr/bin/env python
2	from __future__ import annotations	10✔
3
4	import functools	10✔
5	import os	10✔
6	import shlex	10✔
7	import subprocess # nosec:B404	10✔
8	import sys	10✔
9	from collections.abc import AsyncGenerator, Callable, Generator	10✔
10	from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager	10✔
11	from datetime import datetime	10✔
12	from pathlib import Path	10✔
13	from typing import TYPE_CHECKING, Annotated, Any	10✔
14
15	import anyio	10✔
16	import typer	10✔
17	from rich.progress import Progress, SpinnerColumn, TaskID	10✔
18
19	from .client import CdnHostBuilder, HttpSniff	10✔
20
21	if TYPE_CHECKING:
22	if sys.version_info >= (3, 11):
23	from typing import Self
24	else:
25	from typing_extensions import Self
26
27
28	app = typer.Typer()	10✔
29
30
31	def load_bool(name: str) -> bool:	10✔
32	v = os.getenv(name)	10✔
33	if not v:	10✔
34	return False	10✔
35	return v.lower() in ("1", "true", "t", "y", "yes", "on")	10✔
36
37
38	def run_shell(cmd: str) -> None:	10✔
39	"""Run cmd by subprocess
40
41	Example::
42	run_shell('PYTHONPATH=. python main.py')
43
44	Will be convert to:
45	subprocess.run(['python', 'main.py'], env={**os.environ, 'PYTHONPATH': '.'})
46	"""
47	typer.echo(f"--> {cmd}")	10✔
48	command = shlex.split(cmd)	10✔
49	cmd_env = None	10✔
50	index = 0	10✔
51	for i, c in enumerate(command):	10✔
52	if "=" not in c:	10✔
53	index = i	10✔
54	break	10✔
55	name, value = c.split("=")	10✔
56	if cmd_env is None:	10✔
57	cmd_env = {**os.environ, name: value}	10✔
58	else:
59	cmd_env[name] = value	10✔
60	if cmd_env is not None:	10✔
61	command = command[index:]	10✔
62	subprocess.run(command, env=cmd_env) # nosec:B603	10✔
63
64
65	TEMPLATE = """
66	#!/usr/bin/env python
67	'''This file is auto generated by fastapi_cdn_host.
68	Feel free to change or remove it.
69	'''
70	import subprocess
71	import sys
72
73	import fastapi_cdn_host
74
75	from {} import app
76
77	fastapi_cdn_host.patch_docs(app)
78
79	def _runserver() -> int:
80	r = subprocess.run(["fastapi", "dev", __file__, *sys.argv[2:]])
81	return int(bool(r.returncode))
82
83	if __name__ == "__main__":
84	sys.exit(_runserver())
85	"""
86
87
88	def write_app(dest: Path, from_path: str \| Path) -> None:	10✔
89	module = Path(from_path).stem	10✔
90	size = dest.write_text(TEMPLATE.format(module).strip())	10✔
91	typer.echo(f"Create {dest} with {size=}")	10✔
92
93
94	@contextmanager	10✔
95	def patch_app(path: str \| Path, remove: bool = True) -> Generator[Path, None, None]:	10✔
96	ident = f"{datetime.now():%Y%m%d%H%M%S}"	10✔
97	app_file = Path(f"app_{ident}.py")	10✔
98	write_app(app_file, path)	10✔
99	try:	10✔
100	yield app_file	10✔
101	finally:
102	if remove:	10✔
103	app_file.unlink()	10✔
104	typer.echo(f"Auto remove temp file: {app_file}")	10✔
105
106
107	class PercentBar(AbstractAsyncContextManager):	10✔
108	default_threshold: Annotated[int, "0-100"] = 80	10✔
109	total_seconds: Annotated[int, "Cancel task if reach this value"] = 5	10✔
110
111	def __init__(	10✔
112	self, msg: str, seconds: int \| None = None, color: str = "", **kw: Any
113	) -> None:
114	self.seconds = seconds or self.total_seconds	10✔
115	self.prompt = self.build_prompt(msg, color)	10✔
116	self.progress = Progress(**kw)	10✔
117
118	@staticmethod	10✔
119	def build_prompt(msg: str, color: str \| None, suffix: str = ":") -> str:	10✔
120	prompt = f"{msg}{suffix}"	10✔
121	if color:	10✔
122	prompt = f"[{color}]" + prompt	10✔
123	return prompt	10✔
124
125	async def play(	10✔
126	self, progress: Progress, task: TaskID, threshold: int \| None = None
127	) -> None:
128	expected = 1 / 2	10✔
129	progress_forward = functools.partial(progress.advance, task)	10✔
130
131	# Play quickly: run 80% progress in 1/2 total seconds
132	threshold = threshold or self.default_threshold	10✔
133	cost = self.seconds * expected	10✔
134	await self.percent_play(cost, threshold, progress_forward)	10✔
135
136	# Play slow
137	slow = 100 - threshold	10✔
138	cost = self.seconds - cost	10✔
139	await self.percent_play(cost, slow, progress_forward)	10✔
140
141	@staticmethod	10✔
142	async def percent_play(cost: float, percent: int, forward: Callable) -> None:	10✔
143	delay = cost / percent	10✔
144	for _ in range(percent):	10✔
145	await anyio.sleep(delay)	10✔
146	forward()	10✔
147
148	async def __aenter__(self) -> Self:	10✔
149	self.progress.start()	10✔
150	self.progress_task = self.progress.add_task(self.prompt)	10✔
151	self.task_group = await anyio.create_task_group().__aenter__()	10✔
152	self.task_group.start_soon(self.play, self.progress, self.progress_task)	10✔
153	return self	10✔
154
155	async def __aexit__(self, args, *kw) -> None:	10✔
156	self.task_group.cancel_scope.cancel()	10✔
157	self.progress.update(self.progress_task, completed=100)	10✔
158	await self.task_group.__aexit__(args, *kw)	10✔
159	self.progress.__exit__(args, *kw)	10✔
160
161
162	@asynccontextmanager	10✔
163	async def percentbar(msg: str, **kwargs: Any) -> AsyncGenerator[None, None]:	10✔
164	"""Progressbar with custom font color
165
166	:param msg: prompt message.
167	"""
168	async with PercentBar(msg, **kwargs):	10✔
169	yield	10✔
170
171
172	class SpinnerProgress(Progress):	10✔
173	def __init__(self, msg: str, color: str \| None = None, **kwargs: Any) -> None:	10✔
174	self.prompt = PercentBar.build_prompt(msg, color, "...")	10✔
175	kwargs.setdefault("transient", True)	10✔
176	super().__init__(SpinnerColumn(), Progress.get_default_columns(), *kwargs)	10✔
177
178	def start(self) -> None:	10✔
179	super().start()	10✔
180	self.add_task(self.prompt, total=None)	10✔
181
182
183	@contextmanager	10✔
184	def spinnerbar(	10✔
185	msg: str, color: str \| None = None, **kwargs: Any
186	) -> Generator[None, None, None]:
187	with SpinnerProgress(msg, color, **kwargs):	10✔
188	yield	10✔
189
190
191	async def download_offline_assets(dirname: str \| Path, timeout: float = 30) -> None:	10✔
192	cwd = await anyio.Path.cwd()	×
193	static_root = (	×
194	cwd / dirname if isinstance(dirname, str) else anyio.Path(dirname.resolve())
195	)
196	if not await static_root.exists():	×
197	await static_root.mkdir(parents=True)	×
198	typer.echo(f"Directory {static_root} created.")	×
199	else:
200	async for p in static_root.glob("swagger-ui*.js"):	×
201	relative_path = p.relative_to(cwd)	×
202	typer.echo(f"{relative_path} already exists. abort!")	×
203	return	×
204	async with percentbar("Comparing cdn hosts response speed"):	×
205	urls = await CdnHostBuilder.sniff_the_fastest()	×
206	typer.echo(f"Result: {urls}")	×
207	with spinnerbar("Fetching files from cdn", color="yellow"):	×
208	url_list = [urls.js, urls.css, urls.redoc]	×
209	contents = await HttpSniff.bulk_fetch(	×
210	url_list, get_content=True, total_seconds=timeout
211	)
NEW 212	for url, content in zip(url_list, contents, strict=False):	×
213	if not content:	×
214	red_head = typer.style("ERROR:", fg=typer.colors.RED)	×
215	typer.echo(red_head + f" Failed to fetch content from {url}")	×
216	else:
217	path = static_root / Path(url).name	×
218	size = await path.write_bytes(content)	×
219	typer.echo(f"Write to {path} with {size=}")	×
220	typer.secho("Done.", fg=typer.colors.GREEN)	×
221
222
223	def handle_cache() -> None:	10✔
224	exists, cache_path = CdnHostBuilder.get_cache_file()	×
225	if not exists:	×
226	typer.echo("Cache not create yet.")	×
227	return	×
228	if "--remove" in sys.argv:	×
229	cache_path.unlink()	×
230	typer.secho(	×
231	f"Success to remove cache file({cache_path})", fg=typer.colors.GREEN
232	)
233	else:
234	typer.echo(f"Content of cache file({cache_path}):\n{cache_path.read_text()}")	×
235	if "--reload" in sys.argv:	×
236	cache_path.unlink()	×
237	with spinnerbar("Refreshing cache", color="yellow"):	×
238	CdnHostBuilder(cache=True).run()	×
239	typer.echo(f"Cache file updated:\n{cache_path.read_text()}")	×
240
241
242	@app.command()	10✔
243	def dev(	10✔
244	path: Annotated[
245	Path,
246	typer.Argument(
247	help=(
248	"A path to a Python file or package directory"
249	" (with [blue]__init__.py[/blue] file)"
250	" containing a [bold]FastAPI[/bold] app."
251	" If not provided, a default set of paths will be tried."
252	)
253	),
254	],
255	port: Annotated[
256	int,
257	typer.Option(
258	help=(
259	"The port to serve on."
260	" You would normally have a termination proxy on top (another program)"
261	" handling HTTPS on port [blue]443[/blue] and HTTP on port [blue]80[/blue],"
262	" transferring the communication to your app."
263	)
264	),
265	] = 0,
266	remove: Annotated[
267	bool,
268	typer.Option(
269	help="Whether remove the temp app_<time>.py file after server stopped."
270	),
271	] = True,
272	prod: Annotated[
273	bool,
274	typer.Option(help="Whether enable production mode."),
275	] = False,
276	reload: Annotated[
277	bool,
278	typer.Option(help="Enable auto-reload of the server when (code) files change."),
279	] = True,
280	) -> None:
281	if str(path) == "offline":	×
282	anyio.run(download_offline_assets, "static")	×
283	return	×
284	elif str(path) == "cache":	×
285	handle_cache()	×
286	return	×
287	with patch_app(path, remove) as file:	×
288	runserver(file, prod, reload, port)	×
289
290
291	def runserver(file: Path, prod: bool, reload: bool, port: int) -> None:	10✔
292	if load_bool("FASTCDN_UVICORN"):	×
293	module = file.stem	×
294	if file.parent != Path() and file.parent.resolve() != Path.cwd():	×
295	os.chdir(file.parent)	×
296	cmd = f"PYTHONPATH=. uvicorn {module}:app"	×
297	if reload and not load_bool("FASTCDN_NORELOAD"):	×
298	cmd += " --reload"	×
299	else:
300	mode = "run" if prod else "dev"	×
301	cmd = f"PYTHONPATH=. fastapi {mode} {file}"	×
302	if (not reload and not prod) or load_bool("FASTCDN_NORELOAD"):	×
303	cmd += " --no-reload"	×
304	if port:	×
305	cmd += f" --{port=}"	×
306	run_shell(cmd)	×
307
308
309	def main() -> None:	10✔
310	app()	×
311
312
313	if __name__ == "__main__":
314	main()

waketzheng / fastapi-cdn-host / 20771719563

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous