config

@fx.gen
def main():
    port = yield from config.port("PORT").with_default(8080)
    ...

all(*configs) -> Config[tuple[Any, ...]]

def all(*configs: Config[Any]) -> Config[tuple[Any, ...]]:
    """The configs' values as a tuple, in order (Effect's ``Config.all``
    of a tuple or an iterable: pass ``*configs``)."""
    for i, c in enumerate(configs):
        require_config(c, f"configs[{i}]")
    return Config(lambda provider, prefix: _collect(configs, provider, prefix).map(_tuple))

all_dict(configs) -> Config[dict[K, A]]

def all_dict[K, A](configs: Mapping[K, Config[A]]) -> Config[dict[K, A]]:
    """The configs' values under the same keys (Effect's ``Config.all`` of a
    record). The mapping is copied when the config is made."""
    keys = tuple(configs)
    values = tuple(configs[k] for k in keys)
    for k, c in zip(keys, values, strict=True):
        require_config(c, f"configs[{k!r}]")

    def to_dict(resolution: Resolution[list[A]]) -> Resolution[dict[K, A]]:
        match resolution:
            case Ok(items):
                return Ok(dict(zip(keys, items, strict=True)))
            case Err():
                return resolution

    return Config(lambda provider, prefix: _collect(values, provider, prefix).map(to_dict))

dict_of(value, name=None, *, separator=',', key_value_separator='=') -> Config[dict[str, A]]

def dict_of[A](
    value: Config[A],
    name: str | None = None,
    *,
    separator: str = ",",
    key_value_separator: str = "=",
) -> Config[dict[str, A]]:
    """A dict whose values are read with ``value``: from a record node
    (its keys that hold something), else from a separated string (``"a=1,b=2"``; a pair with
    no separator, an empty key or an empty value is skipped)."""
    require_config(value, "value")

    def from_record(provider: ConfigProvider, path: Path, node: Node | None) -> Effect[Any, Any]:
        if not isinstance(node, Record):
            return succeed(Err(at(path, InvalidType("object"))))
        children = _load_children(provider, [(*path, k) for k in sorted(node.keys)])
        return children.flat_map(lambda items: _read_present(value, provider, items))

    def from_string(provider: ConfigProvider, path: Path, node: Node | None) -> Effect[Any, Any]:
        text = None if node is None else node.value
        if text is None:
            return succeed(Err(at(path, InvalidType("string"))))
        pairs = _split_key_value(text, separator, key_value_separator)
        keys = list(pairs)
        return _read_pieces(value, [((*path, k), pairs[k]) for k in keys]).map(_keyed(keys))

    def has_input(node: Node | None) -> bool:
        return isinstance(node, Record) or (node is not None and node.value is not None)

    return _container(name, (from_record, from_string), has_input, _as_dict)

list_of(element, name=None, *, separator=',') -> Config[list[A]]

def list_of[A](
    element: Config[A], name: str | None = None, *, separator: str = ","
) -> Config[list[A]]:
    """A list whose items are read with ``element``: from a separated
    string (``"1,2"``; ``""`` is empty), else from an array node
    (``PORTS_0``, ``PORTS_1`` in ``from_env``)."""
    require_config(element, "element")

    def from_string(provider: ConfigProvider, path: Path, node: Node | None) -> Effect[Any, Any]:
        text = None if node is None else node.value
        if text is None:
            return succeed(Err(at(path, InvalidType("string"))))
        pieces = _split(text, separator) if text else []
        return _read_pieces(element, [((*path, i), piece) for i, piece in enumerate(pieces)])

    def from_array(provider: ConfigProvider, path: Path, node: Node | None) -> Effect[Any, Any]:
        if not isinstance(node, Array):
            return succeed(Err(at(path, InvalidType("array"))))
        children = _load_children(provider, [(*path, i) for i in range(node.length)])
        return children.flat_map(lambda items: _read_children(element, provider, items))

    def has_input(node: Node | None) -> bool:
        return isinstance(node, Array) or (node is not None and node.value is not None)

    return _container(name, (from_string, from_array), has_input, _as_list)

gen(f) -> Config[A]

@config.gen
def server():
    host = yield from config.string("HOST")
    port = yield from config.port("PORT").with_default(8080)
    return f"{host}:{port}"

def gen[A](f: Callable[[], Generator[Effect[Any, ConfigError, Never], Any, A]]) -> Config[A]:
    """Build a config from a generator: ``x = yield from cfg`` reads
    ``cfg`` under the result's ``nested`` prefix, and the result is absent
    as soon as one read is. Other effects run as they are (they may fail
    only with ``ConfigError`` and need no services)::

        @config.gen
        def server():
            host = yield from config.string("HOST")
            port = yield from config.port("PORT").with_default(8080)
            return f"{host}:{port}"
    """

    def run(provider: ConfigProvider, prefix: Path) -> Effect[Resolution[A], ConfigError]:
        def start() -> Effect[Resolution[A], ConfigError]:
            generator = f()

            def advance(send: Callable[[], Effect[Any, ConfigError, Never]]) -> Effect[Any, Any]:
                try:
                    yielded = send()
                except StopIteration as stop:
                    done: Resolution[A] = Ok(stop.value)
                    return succeed(done)

                def next_step(resolution: Resolution[Any]) -> Effect[Any, Any]:
                    match resolution:
                        case Ok(value):
                            return advance(lambda: generator.send(value))
                        case Err():
                            return succeed(resolution)

                return _evaluate_at(yielded, provider, prefix).flat_map(next_step)

            # Closing runs the generator's ``finally`` blocks however the
            # read ends: absent, failed, interrupted or done (a no-op then).
            return advance(lambda: next(generator)).ensuring(sync(generator.close))

        return _Suspend(start)

    return Config(run)

option(config) -> Config[Option[A]]

def option[A](config: Config[A]) -> Config[Option[A]]:
    """``Some(value)``, or ``None`` when the value is absent. Invalid data
    and source errors still fail. (``config.option()`` is
    ``Effect.option``, which turns every failure into ``None``.)"""
    return config.map(Some).with_default(None)

unwrap(wrapped) -> Config[Any]

def unwrap(wrapped: Config[Any] | Mapping[str, Any]) -> Config[Any]:
    """A config itself, or a (nested) mapping of configs read as one
    ``all_dict`` per level."""
    from pyfect.config._all import all_dict

    if isinstance(wrapped, Config):
        return wrapped
    return all_dict({key: unwrap(value) for key, value in wrapped.items()})

fail_config(error) -> Config[Never]

def fail_config(error: SourceError | SchemaError) -> Config[Never]:
    """A config that always fails with ``ConfigError(error)``."""
    failure: Effect[Never, ConfigError] = fail_cause(Cause((Fail(ConfigError(error)),)))
    return Config(lambda _provider, _prefix: failure)

succeed_config(value) -> Config[A]

def succeed_config[A](value: A) -> Config[A]:
    """A config that reads nothing and gives ``value``."""
    present: Resolution[A] = Ok(value)
    return Config(lambda _provider, _prefix: succeed(present))

boolean(name=None) -> Config[bool]

def boolean(name: str | None = None) -> Config[bool]:
    """``true`` / ``yes`` / ``on`` / ``1`` / ``y`` or ``false`` / ``no`` /
    ``off`` / ``0`` / ``n``, in lower case."""
    return scalar_config(_scalars.boolean, name)

byte_size(name=None) -> Config[ByteSize]

def byte_size(name: str | None = None) -> Config[ByteSize]:
    """An exact size such as ``"10 MB"`` or ``"1.5 KiB"``."""
    return scalar_config(_scalars.byte_size, name)

date_time(name=None) -> Config[datetime]

def date_time(name: str | None = None) -> Config[datetime]:
    """An ISO 8601 date or date-time (``datetime.fromisoformat``); one with
    no offset is UTC (Effect's ``Config.Date``)."""
    return scalar_config(_scalars.date_time, name)

duration(name=None) -> Config[Seconds]

def duration(name: str | None = None) -> Config[Seconds]:
    """A duration such as ``"10 seconds"`` or ``"500 millis"`` (``nanos``
    to ``weeks``, singular or plural), or ``"Infinity"`` / ``"-Infinity"``."""
    return scalar_config(_scalars.duration, name)

finite(name=None) -> Config[float]

def finite(name: str | None = None) -> Config[float]:
    """A finite number."""
    return scalar_config(_scalars.finite, name)

integer(name=None) -> Config[int]

def integer(name: str | None = None) -> Config[int]:
    """A whole number, read exactly (``"1e3"`` is ``1000``; Effect's
    ``Config.Int``)."""
    return scalar_config(_scalars.integer, name)

literal(*values, /, *, name=None) -> Config[L]

def literal[L: LiteralValue](*values: L, name: str | None = None) -> Config[L]:
    """One of ``values``, written as Effect writes literals (``True`` is
    ``"true"``, ``None`` is ``"null"``). Annotate the result to keep the
    literal types: ``env: Config[Literal["dev", "prod"]] = config.literal("dev", "prod")``."""
    return scalar_config(_scalars.literal(values), name)

log_level(name=None) -> Config[LogLevel]

def log_level(name: str | None = None) -> Config[LogLevel]:
    """A ``LogLevel`` by Effect's name: ``All``, ``Fatal``, ``Error``,
    ``Warn``, ``Info``, ``Debug``, ``Trace`` or ``None``."""
    return scalar_config(_scalars.log_level, name)

non_empty_string(name=None) -> Config[str]

def non_empty_string(name: str | None = None) -> Config[str]:
    """A string of at least one character."""
    return scalar_config(_scalars.non_empty_string, name)

number(name=None) -> Config[float]

def number(name: str | None = None) -> Config[float]:
    """A number, ``"Infinity"``, ``"-Infinity"`` and ``"NaN"`` included
    (Effect's ``Config.Number``)."""
    return scalar_config(_scalars.number, name)

port(name=None) -> Config[int]

def port(name: str | None = None) -> Config[int]:
    """An integer from 1 to 65535."""
    return scalar_config(_scalars.port, name)

redacted(name=None) -> Config[Redacted[str]]

def redacted(name: str | None = None) -> Config[Redacted[str]]:
    """A string hidden in a ``Redacted``."""
    return scalar_config(_scalars.redacted, name)

string(name=None) -> Config[str]

def string(name: str | None = None) -> Config[str]:
    """Any string."""
    return scalar_config(_scalars.string, name)

url(name=None) -> Config[SplitResult]

def url(name: str | None = None) -> Config[SplitResult]:
    """An absolute URL with a scheme (and a host for ``http``, ``https``,
    ``ws``, ``wss`` and ``ftp``), split by ``urllib.parse.urlsplit``."""
    return scalar_config(_scalars.url, name)