router_serve

fastvideo router-serve CLI subcommand.

Launches the streaming router from a YAML config. Separate from fastvideo serve because the router is an orthogonal process: it fronts one or more running servers rather than hosting a generator itself.

`fastvideo.entrypoints.cli.router_serve.load_raw_config`

load_raw_config(path: str | Path) -> dict[str, Any]

`fastvideo.entrypoints.cli.router_serve.CLISubcommand`

CLISubcommand()

Base class for CLI subcommands

`fastvideo.entrypoints.cli.router_serve.CLISubcommand.cmd`

cmd(args: argparse.Namespace) -> None

Execute the command with the given arguments

`fastvideo.entrypoints.cli.router_serve.CLISubcommand.validate`

validate(args: argparse.Namespace) -> None

Validate the arguments for this command

`fastvideo.entrypoints.cli.router_serve.CLISubcommand.subparser_init`

subparser_init(subparsers: argparse._SubParsersAction) -> FlexibleArgumentParser

Initialize the subparser for this command

`fastvideo.entrypoints.cli.router_serve.ReplicaEndpoint`

ReplicaEndpoint(url: str, name: str | None = None, primary: bool = False, weight: float = 1.0)

One backend replica the router can route to.

`fastvideo.entrypoints.cli.router_serve.ReplicaEndpoint.url`

url: str

HTTP base URL, e.g. http://host:8000. WebSocket URL is derived automatically by replacing the scheme.

`fastvideo.entrypoints.cli.router_serve.ReplicaEndpoint.primary`

primary: bool = False

True = prefer this replica over others in steady state.

`fastvideo.entrypoints.cli.router_serve.RouterConfig`

RouterConfig(host: str = '0.0.0.0', port: int = 9000, replicas: list[ReplicaEndpoint] = list(), health_check_path: str = '/health', health_check_interval_seconds: float = 5.0, health_check_timeout_seconds: float = 2.0, failure_threshold: int = 3, recovery_threshold: int = 2)

Typed router config loaded from a YAML file.

Example::

router:
  host: 0.0.0.0
  port: 9000
  replicas:
    - url: http://streamer-a:8000
      primary: true
    - url: http://streamer-b:8000
  health_check:
    path: /health
    interval_seconds: 5
    failure_threshold: 3

Validation runs in __post_init__: empty replicas, non-positive intervals/timeouts, thresholds < 1, non-http(s) URLs, and more than one primary all raise ValueError so misconfigurations surface at load time rather than as confusing runtime failures.

`fastvideo.entrypoints.cli.router_serve.init_logger`

init_logger(name: str) -> _FastvideoLogger

The main purpose of this function is to ensure that loggers are retrieved in such a way that we can be sure the root fastvideo logger has already been configured.

`fastvideo.entrypoints.cli.router_serve.FlexibleArgumentParser`

FlexibleArgumentParser(*args, **kwargs)

ArgumentParser that allows both underscore and dash in names.

`fastvideo.entrypoints.cli.router_serve.FlexibleArgumentParser.parse_args`

parse_args(args = None, namespace = None) -> argparse.Namespace

`fastvideo.entrypoints.cli.router_serve.FlexibleArgumentParser.parse_known_args`

parse_known_args(args = None, namespace = None) -> tuple[argparse.Namespace, list[str]]

`fastvideo.entrypoints.cli.router_serve.RouterServeSubcommand`

RouterServeSubcommand()

Start the multi-replica WebSocket router.

Source

class RouterServeSubcommand(CLISubcommand):
    """Start the multi-replica WebSocket router."""
def __init__(self) -&gt; None:
    self.name = &quot;router-serve&quot;
    super().__init__()

def cmd(self, args: argparse.Namespace) -&gt; None:
    config = _load_router_config(args.config)
    logger.info(
        &quot;router listening on %s:%d (%d replicas, %d primary)&quot;,
        config.host,
        config.port,
        len(config.replicas),
        sum(1 for r in config.replicas if r.primary),
    )
    from fastvideo.entrypoints.streaming.router.main import run_router

    run_router(config)

def validate(self, args: argparse.Namespace) -&gt; None:
    if not args.config:
        raise ValueError(&quot;fastvideo router-serve requires --config PATH&quot;)
    if not os.path.exists(args.config):
        raise ValueError(f&quot;Router config file not found: &#123;args.config&#125;&quot;)

def subparser_init(
    self,
    subparsers: argparse._SubParsersAction,
) -&gt; FlexibleArgumentParser:
    parser = subparsers.add_parser(
        &quot;router-serve&quot;,
        help=&quot;Start the streaming router (multi-replica load balancer)&quot;,
        usage=&quot;fastvideo router-serve --config ROUTER_CONFIG&quot;,
    )
    parser.add_argument(
        &quot;--config&quot;,
        type=str,
        default=&quot;&quot;,
        required=False,
        help=&quot;Path to a YAML/JSON router config. Required.&quot;,
    )
    return cast(FlexibleArgumentParser, parser)</code></pre></details>
fastvideo.entrypoints.cli.router_serve.RouterServeSubcommand.cmd
cmd(args: argparse.Namespace) -> None
Source
def cmd(self, args: argparse.Namespace) -> None:
    config = _load_router_config(args.config)
    logger.info(
        "router listening on %s:%d (%d replicas, %d primary)",
        config.host,
        config.port,
        len(config.replicas),
        sum(1 for r in config.replicas if r.primary),
    )
    from fastvideo.entrypoints.streaming.router.main import run_router
run_router(config)</code></pre></details>
fastvideo.entrypoints.cli.router_serve.RouterServeSubcommand.validate
validate(args: argparse.Namespace) -> None
Source
def validate(self, args: argparse.Namespace) -> None:
    if not args.config:
        raise ValueError("fastvideo router-serve requires --config PATH")
    if not os.path.exists(args.config):
        raise ValueError(f"Router config file not found: {args.config}")
fastvideo.entrypoints.cli.router_serve.RouterServeSubcommand.subparser_init
subparser_init(subparsers: argparse._SubParsersAction) -> FlexibleArgumentParser
Source
def subparser_init(
    self,
    subparsers: argparse._SubParsersAction,
) -> FlexibleArgumentParser:
    parser = subparsers.add_parser(
        "router-serve",
        help="Start the streaming router (multi-replica load balancer)",
        usage="fastvideo router-serve --config ROUTER_CONFIG",
    )
    parser.add_argument(
        "--config",
        type=str,
        default="",
        required=False,
        help="Path to a YAML/JSON router config. Required.",
    )
    return cast(FlexibleArgumentParser, parser)
fastvideo.entrypoints.cli.router_serve.cmd_init
cmd_init() -> list[CLISubcommand]
Source
def cmd_init() -> list[CLISubcommand]:
    return [RouterServeSubcommand()]