Add parse_endpoints helper function that handles complicated listen structures alike server.run() already did, returning a simple list of every endpoint.
This commit is contained in:
@@ -12,6 +12,7 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
|
|||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
List of dicts with uvicorn bind kwargs (host/port or uds).
|
List of dicts with uvicorn bind kwargs (host/port or uds).
|
||||||
|
Two entries may be returned for IPv4 and IPv6 (all interaces).
|
||||||
|
|
||||||
Supported forms:
|
Supported forms:
|
||||||
- None or empty -> [{host: "localhost", port: default_port}]
|
- None or empty -> [{host: "localhost", port: default_port}]
|
||||||
@@ -65,3 +66,19 @@ def parse_endpoint(value: str | None, default_port: int = 0) -> list[dict]:
|
|||||||
ipaddress.ip_address(host)
|
ipaddress.ip_address(host)
|
||||||
|
|
||||||
return [{"host": host, "port": port}]
|
return [{"host": host, "port": port}]
|
||||||
|
|
||||||
|
|
||||||
|
def parse_endpoints(
|
||||||
|
listen: str | list[str] | None = None, default_port: int = 8000
|
||||||
|
) -> list[dict]:
|
||||||
|
"""Parse listen strings into a list of endpoint dicts.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
listen: Endpoint string(s) (see parse_endpoint for formats).
|
||||||
|
default_port: Port to use when not specified in listen args.
|
||||||
|
"""
|
||||||
|
if listen is None:
|
||||||
|
listen = [f"localhost:{default_port}"]
|
||||||
|
elif isinstance(listen, str):
|
||||||
|
listen = [listen]
|
||||||
|
return [ep for s in listen for ep in parse_endpoint(s, default_port)]
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ from contextlib import suppress
|
|||||||
import uvicorn
|
import uvicorn
|
||||||
from uvicorn import Config, Server
|
from uvicorn import Config, Server
|
||||||
|
|
||||||
from .hostutil import parse_endpoint
|
from .hostutil import parse_endpoints
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -30,13 +30,9 @@ def run(
|
|||||||
workers: Number of worker processes (requires uvicorn.run, single endpoint only).
|
workers: Number of worker processes (requires uvicorn.run, single endpoint only).
|
||||||
**uvicorn_config: Additional uvicorn config options (overrides all other settings).
|
**uvicorn_config: Additional uvicorn config options (overrides all other settings).
|
||||||
"""
|
"""
|
||||||
if listen is None:
|
endpoints = parse_endpoints(listen, default_port)
|
||||||
listen = [f"localhost:{default_port}"]
|
if not endpoints:
|
||||||
elif isinstance(listen, str):
|
raise ValueError("No endpoints to serve; check listen configuration")
|
||||||
listen = [listen]
|
|
||||||
endpoints: list[dict] = []
|
|
||||||
for ep in listen:
|
|
||||||
endpoints.extend(parse_endpoint(ep, default_port))
|
|
||||||
|
|
||||||
conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers}
|
conf: dict[str, object] = {"app": app, "reload": reload, "workers": workers}
|
||||||
proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1")
|
proxy = os.getenv("FORWARDED_ALLOW_IPS", "127.0.0.1,::1")
|
||||||
|
|||||||
Reference in New Issue
Block a user