finance-query 3.0.0

# Code generated by `cargo soothfast sdk gen`. DO NOT EDIT.
"""Embedded server lifecycle.

The bundled server binds a port of its own choosing and announces it on
stdout (``soothfast-ready {"base_url":...}``); this module spawns it, waits
for that line, and hands the client a base URL. Other stdout lines are
ignored, so the server is free to log.

Startup is synchronous — spawning a process and reading one line is fast,
and it lets ``Client()`` and ``AsyncClient()`` alike resolve their base URL
in ``__init__`` instead of deferring it into every request.
"""
from __future__ import annotations

import atexit
import dataclasses
import json
import os
import pathlib
import queue
import subprocess
import threading
import time
import typing

READY_PREFIX = "soothfast-ready "

_STARTUP_TIMEOUT = 30.0
_SHUTDOWN_GRACE = 5.0
_STDERR_TAIL = 8192


@dataclasses.dataclass(frozen=True)
class EmbedConfig:
    """Everything package-specific about the bundled server."""

    binary: str
    """Binary name, looked up on ``PATH`` when nothing else resolves it."""
    args: tuple[str, ...]
    base_url_env: str
    """Env var holding a base URL to use *instead of* spawning."""
    bin_env: str
    """Env var holding an explicit path to the server binary."""
    env: tuple[tuple[str, str], ...] = ()
    """Environment this package launches its server with, over the ambient
    one. Pairs rather than a dict so the config stays hashable."""


def environment(
    config: EmbedConfig, overrides: typing.Mapping[str, str] | None = None
) -> dict[str, str]:
    """The environment a launch runs with.

    Ambient process environment first, then the package's own launch
    settings, then the caller's. The package layer beats the ambient one on
    purpose: an inherited ``PORT`` from whatever runs the consumer's app
    must not decide where the bundled server binds.
    """
    env = dict(os.environ)
    env.update(config.env)
    if overrides:
        env.update(overrides)
    return env


def bundled_binary(config: EmbedConfig) -> str | None:
    """The binary this wheel shipped, if it was a platform wheel.

    An sdist install has no binary — that is a supported outcome, not an
    error; the caller falls back to ``PATH``.
    """
    name = config.binary + (".exe" if os.name == "nt" else "")
    path = pathlib.Path(__file__).parent / "bin" / name
    return str(path) if path.is_file() else None


class EmbeddedServer:
    """A running server subprocess."""

    def __init__(self, base_url: str, process: subprocess.Popen[str]) -> None:
        self.base_url = base_url
        self._process = process

    @classmethod
    def start(
        cls,
        config: EmbedConfig,
        *,
        bin: str | None = None,
        args: typing.Sequence[str] | None = None,
        timeout: float = _STARTUP_TIMEOUT,
        env: typing.Mapping[str, str] | None = None,
    ) -> EmbeddedServer:
        """Spawn the server and return once it has announced its address.

        ``env`` holds *overrides*, not a replacement environment: the
        server still inherits this process's, with these applied on top.
        """
        environ = environment(config, env)
        # Explicit path, then the env override, then the binary this wheel
        # shipped for the running platform, then whatever is on PATH.
        executable = (
            bin or environ.get(config.bin_env) or bundled_binary(config) or config.binary
        )
        argv = [executable, *(args if args is not None else config.args)]
        try:
            process = subprocess.Popen(
                argv,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                bufsize=1,
                env=environ,
            )
        except OSError as exc:
            raise RuntimeError(f"cannot spawn embedded server {executable!r}: {exc}") from exc

        # Both pipes are drained from the moment the process exists: a
        # server that logs on every request would otherwise fill its pipe
        # buffer and wedge on the write.
        stderr = _Tail(process.stderr)
        try:
            base_url = _read_ready_line(process, executable, timeout, stderr)
        except BaseException:
            process.kill()
            process.wait()
            raise
        return cls(base_url, process)

    def stop(self) -> None:
        """Stop the server, escalating to a kill if it does not go quietly."""
        if self._process.poll() is not None:
            return
        self._process.terminate()
        try:
            self._process.wait(timeout=_SHUTDOWN_GRACE)
        except subprocess.TimeoutExpired:
            self._process.kill()
            self._process.wait()


class _Tail:
    """The end of a stream, kept drained.

    Reading has to continue for the life of the process — a server free to
    log is a server that fills a 64 KiB pipe buffer and blocks on write —
    but only the tail is worth keeping, since all it is ever used for is
    explaining a failure.
    """

    def __init__(self, stream: typing.IO[str] | None) -> None:
        self._text = ""
        self._lock = threading.Lock()
        if stream is not None:
            # Daemon: a wedged server must never block interpreter exit.
            threading.Thread(target=self._pump, args=(stream,), daemon=True).start()

    def _pump(self, stream: typing.IO[str]) -> None:
        try:
            for chunk in stream:
                with self._lock:
                    self._text = (self._text + chunk)[-_STDERR_TAIL:]
        except ValueError:
            pass  # the pipe closed under us; nothing left to read

    def detail(self) -> str:
        with self._lock:
            text = self._text.strip()
        return "" if not text else f"\n--- server stderr ---\n{text}"


def _read_ready_line(
    process: subprocess.Popen[str], executable: str, timeout: float, stderr: _Tail
) -> str:
    """Wait for the readiness line, failing loudly on an early exit."""
    lines: queue.Queue[str | None] = queue.Queue()
    announced_already = threading.Event()

    def pump() -> None:
        assert process.stdout is not None
        for line in process.stdout:
            # Past the handshake the server's own logging is none of our
            # business — but it still has to be read, or the pipe fills.
            if not announced_already.is_set():
                lines.put(line)
        lines.put(None)

    # Daemon thread: a wedged server must never block interpreter exit.
    threading.Thread(target=pump, daemon=True).start()

    # One deadline for the whole handshake: a server that logs while starting
    # would otherwise re-arm a per-line timeout and never trip it.
    deadline = time.monotonic() + timeout
    try:
        while True:
            try:
                line = lines.get(timeout=max(0.0, deadline - time.monotonic()))
            except queue.Empty:
                raise RuntimeError(
                    f"embedded server {executable!r} did not announce a base URL "
                    f"within {timeout}s{stderr.detail()}"
                ) from None
            if line is None:
                process.wait()
                raise RuntimeError(
                    f"embedded server {executable!r} exited (code {process.returncode}) "
                    f"before announcing a base URL{stderr.detail()}"
                )
            line = line.strip()
            if not line.startswith(READY_PREFIX):
                continue
            try:
                announced = json.loads(line[len(READY_PREFIX) :])
            except ValueError as exc:
                raise RuntimeError(
                    f"unparseable readiness line from {executable!r}: {exc}"
                ) from exc
            base_url = announced.get("base_url") if isinstance(announced, dict) else None
            if not isinstance(base_url, str):
                raise RuntimeError(f"embedded server announced no base_url: {line}")
            return base_url
    finally:
        announced_already.set()


_running: dict[tuple[typing.Any, ...], EmbeddedServer] = {}
_lock = threading.Lock()


def embedded_base_url(
    config: EmbedConfig, env: typing.Mapping[str, str] | None = None
) -> str:
    """Resolve the base URL for a client constructed without one.

    Resolution order: the ``base_url_env`` override (no subprocess at all),
    then a shared server, started on first use. Clients asking for
    different ``env`` get different servers — one process cannot hold two
    configurations, so sharing across them would hand the second caller a
    server it did not ask for.
    """
    override = os.environ.get(config.base_url_env)
    if override:
        return override
    key = (config.binary, tuple(sorted((env or {}).items())))
    with _lock:
        server = _running.get(key)
        if server is None:
            server = EmbeddedServer.start(config, env=env)
            _running[key] = server
        return server.base_url


def stop_embedded_servers() -> None:
    """Stop every shared server this process started."""
    with _lock:
        servers = list(_running.values())
        _running.clear()
    for server in servers:
        server.stop()


atexit.register(stop_embedded_servers)