Tutorial

7. A configuration processor¶

The two programs in this chapter combine lazy iteration, Result composition and pattern matching. The first parses and validates configuration, stopping when a line or setting is invalid. The second routes a queue of commands with guarded clauses and handles a typed dispatch error. Both express their parsing, validation and dispatch rules as plain functions.

Parse and validate configuration¶

The configuration processor follows five steps:

Parse each key=value line into Result[Entry, ConfigError].
Stop at the first parse error with collect_results.
Validate the complete entry list with parsed^.flat_map(validate).
Format only a valid Config with checked^.map(describe).
Fold the final Ok or Err into one message with a match.

config_parser.mojo Download

"""Configuration parsing: lazy parsing, Result aggregation and a validating fold.

Grammar: batch=<integer> and workers=<integer>, exactly once each. Parsing stops
at the first malformed line; validation runs only after every line has parsed.
"""
import fp
from fp.data import Result, Ok, Err, Break, Continue, ControlFlow, collect_results
from fp.functions import pipe
from fp.iteration import map, fold_until
from std.testing import assert_equal


@fieldwise_init
struct Entry(Movable):
    var key: String
    var value: Int


@fieldwise_init
struct Config(Copyable):
    var batch: Int
    var workers: Int


@fieldwise_init
struct ConfigError(Copyable):
    var message: String


comptime Parsed = Result[List[Entry], ConfigError]
comptime Checked = Result[Config, ConfigError]


def parse(var line: String) -> Result[Entry, ConfigError]:
    var parts = line.split("=")
    if len(parts) != 2:
        return Err(ConfigError("syntax"))
    try:
        return Ok(Entry(String(parts[0]), Int(String(parts[1]))))
    except:
        # A native conversion error becomes a domain error at this boundary.
        return Err(ConfigError("integer"))


def record(var seen: Config, var entry: Entry) -> ControlFlow[ConfigError, Config]:
    """Take one entry into the configuration, or stop at the first invalid one."""
    if entry.value <= 0:
        return Break(ConfigError("positive values required"))
    if entry.key == "batch" and seen.batch == 0:
        seen.batch = entry.value
    elif entry.key == "workers" and seen.workers == 0:
        seen.workers = entry.value
    else:
        return Break(ConfigError("unknown or duplicate key"))
    return Continue(seen^)


def complete(config: Config) -> Checked:
    if config.batch == 0 or config.workers == 0:
        return Err(ConfigError("missing key"))
    if config.workers > config.batch:
        return Err(ConfigError("workers exceeds batch"))
    return Ok(config.copy())


def validate(var entries: List[Entry]) -> Checked:
    return fp.match(fold_until(record, Config(0, 0), entries^),
        lambda (stop: Break[ConfigError]) -> Checked: Err(stop.value.copy()),
        lambda (done: Continue[Config]) -> Checked: complete(done.value))


def check(var parsed: Parsed) -> Checked:
    return parsed^.flat_map(validate)


def describe(var config: Config) -> String:
    return "batch=" + String(config.batch) + ", workers=" + String(config.workers)


def render(var checked: Checked) -> Result[String, ConfigError]:
    return checked^.map(describe)


def finish(var rendered: Result[String, ConfigError]) -> String:
    return fp.match(rendered,
        lambda (ok: Ok[String]) -> String: "ready: " + ok.value,
        lambda (err: Err[ConfigError]) -> String: "error: " + err.value.message)


def configure(var lines: List[String], mut parsed: Int) -> String:
    """Parse lazily, stopping at the first bad line; `parsed` counts the lines read."""
    def counted(var line: String) {mut parsed} -> Result[Entry, ConfigError]:
        parsed += 1
        return parse(line^)
    return pipe(collect_results(map(counted, lines^)), check, render, finish)


def scenario(var lines: List[String], expected: String, parsed_lines: Int) raises:
    var parsed = 0
    assert_equal(configure(lines^, parsed), expected)
    assert_equal(parsed, parsed_lines)


def main() raises:
    scenario(["batch=12", "workers=4"], "ready: batch=12, workers=4", 2)
    scenario(["bad", "workers=4"], "error: syntax", 1)
    scenario(["batch=12", "workers=no", "ignored=99"], "error: integer", 2)
    scenario(["batch=12", "batch=4"], "error: unknown or duplicate key", 2)
    scenario(["batch=12", "workers=0"], "error: positive values required", 2)
    scenario(["batch=2", "workers=4"], "error: workers exceeds batch", 2)
    scenario([], "error: missing key", 0)
    var parsed = 0
    print(configure(["batch=12", "workers=4"], parsed))

Run from the repository root pixi run mojo run -I src docs/examples/config_parser.mojo

Output

ready: batch=12, workers=4

For ["batch=12", "workers=4"], both lines parse, validate records each entry with fold_until, and complete accepts the result, so configure returns "ready: batch=12, workers=4" after reading two lines.

Each scenario in main stops at a different step:

Input	Result	Lines parsed
`["bad", "workers=4"]`	`error: syntax`	1
`["batch=12", "workers=no", "ignored=99"]`	`error: integer`	2
`["batch=12", "batch=4"]`	`error: unknown or duplicate key`	2
`["batch=12", "workers=0"]`	`error: positive values required`	2
`["batch=2", "workers=4"]`	`error: workers exceeds batch`	2
`[]`	`error: missing key`	0

collect_results stops at the first parse error, so the lazy map never reads "ignored=99". Validation runs only after every line has parsed: record returns Break at the first invalid entry, and the match on ControlFlow turns it into Err. parse converts the native error from Int(...) into ConfigError("integer") at its own boundary.

Route commands with guarded clauses¶

The job router dispatches a submitted or retried job when its guard allows it, and defers every other command. Guards and dispatches append to a trace that reaches every clause through the match's context, and one dispatch raises a typed error.

job_router.mojo Download

"""Route a queue of commands with guarded clauses, a trace context and a typed failure."""
import fp
from fp.adt import Data, Cases, Node, Value
from fp.data import Result, Ok, Err, attempt
from fp.iteration import map, collect_list
from std.memory import ArcPointer
from std.testing import assert_equal


@fieldwise_init
struct Job(Copyable):
    var id: Int
    var cost: Int


@fieldwise_init
struct Submit(Copyable):
    var job: Job


@fieldwise_init
struct Retry(Copyable):
    var job: Job


@fieldwise_init
struct Cancel(Copyable):
    var job: Job


struct Command(Data):
    comptime Layer[R: Value] = Cases[Submit, Retry, Cancel]


comptime C = Node[Command]
comptime Trace = ArcPointer[List[String]]


@fieldwise_init
struct DispatchFailure(Movable, Writable):
    var job_id: Int

    def write_to(self, mut writer: Some[Writer]):
        writer.write("dispatch failed for job ", self.job_id)


def allowed(job: Job, trace: Trace) -> Bool:
    trace[].append("check " + String(job.id))
    return job.cost <= 5


def dispatch(job: Job, trace: Trace) raises DispatchFailure -> Int:
    trace[].append("dispatch " + String(job.id))
    if job.id == 3:
        raise DispatchFailure(job.id)
    return job.id


def defer(trace: Trace) -> Int:
    trace[].append("defer")
    return 0


def route(command: C, trace: Trace) raises DispatchFailure -> Int:
    """Dispatch a submitted or retried job its guard allows; defer everything else."""
    return fp.match(command,
        fp.when[lambda (s: Submit, t: Trace) -> Bool: allowed(s.job, t),
                lambda (s: Submit, t: Trace) raises DispatchFailure -> Int: dispatch(s.job, t)],
        fp.when[lambda (r: Retry, t: Trace) -> Bool: allowed(r.job, t),
                lambda (r: Retry, t: Trace) raises DispatchFailure -> Int: dispatch(r.job, t)],
        lambda (c: C, t: Trace) -> Int: defer(t),
        context=trace)


def outcome(result: Result[Int, DispatchFailure]) -> String:
    return fp.match(result,
        lambda (ok: Ok[Int]) -> String: "deferred" if ok.value == 0 else "dispatched " + String(ok.value),
        lambda (err: Err[DispatchFailure]) -> String: String(err.value))


def main() raises:
    var trace = Trace(List[String]())
    var queue: List[C] = [Submit(Job(1, 4)), Retry(Job(2, 6)), Cancel(Job(4, 1)), Submit(Job(3, 5))]

    # A lazy callback cannot raise, so `attempt` keeps each failure as an `Err`.
    def routed(var command: C) {trace} -> String:
        return outcome(attempt(route, command, trace))
    var outcomes = collect_list(map(routed, queue^))

    assert_equal(outcomes, ["dispatched 1", "deferred", "deferred", "dispatch failed for job 3"])
    assert_equal(trace[], ["check 1", "dispatch 1", "check 2", "defer", "defer", "check 3", "dispatch 3"])
    print(String("; ").join(outcomes))

Run from the repository root pixi run mojo run -I src docs/examples/job_router.mojo

Output

dispatched 1; deferred; deferred; dispatch failed for job 3

The queue holds four commands, and the program prints dispatched 1; deferred; deferred; dispatch failed for job 3:

Submit(Job(1, 4)) passes the guard, because its cost is at most 5, and is dispatched.
Retry(Job(2, 6)) fails the guard, so the catch-all clause defers it.
Cancel(Job(4, 1)) has no guarded clause and is deferred.
Submit(Job(3, 5)) passes the guard, and dispatch raises DispatchFailure(3).

A lazy map callback cannot raise, so routed calls route through attempt, which keeps the failure as an Err that outcome formats. The trace records the order of the calls: check 1, dispatch 1, check 2, defer, defer, check 3, dispatch 3.

To look up any operation used here, start from the API reference. The example gallery lists every program on the site.

Next, use the same operations across Optional, Result and List in functors, applicatives and monads, then compose computations with Reader, State and Writer.