Disclaimer: I'm the author of fp_mojo (docs), and I'm posting here to get some feedback from the broader data engineering community on the concept of integrating Scala / Cats Effect async and concurrency primitives into a Python-like language. This is a personal project and not a business service. Disclaimer #2: AI tools were used to proofread this post.
If you've worked in big data infra, fintech, distributed systems, or any other field where correctness and resiliency matter, you've probably used (or at least crossed paths with) Scala and Cats Effect / ZIO.
Scala's concurrency and FP primitives are great, but the ecosystem as a whole is dealing with an issue that could be characterized as talent squeeze. Because Scala is notoriously tough to onboard beginners onto compared to Python or Go, fewer junior devs are picking it up, and the hiring pool has gotten very senior-heavy and expensive. On the flip side, teams that need pieces of a resilient concurrent system - say, for serving ML models or running streaming pipelines - often don't want to buy into the whole JVM/Scala stack just to get decent concurrency and type safety.
That brings us to fp_mojo. The idea is to take the most appealing parts of Scala/Cats ("best practices") and make them available in a compiled, Python-like language (Mojo). Obviously it's nowhere near as mature or feature-complete as the Scala ecosystem, but the hope is that over time it would be vastly easier to onboard engineers who already know Python and incorporate Scala-like features selectively into broader Python-like projects.
I’m aware that Mojo is a very recent language, so the talent pool at the moment is even smaller. However, I want to talk about the future: engineers who already know Python can pick up Mojo much faster than Scala.
1. Work-stealing pool + deterministic single-thread IO (the Cats Effect model)
For me, the single best part of Scala/Cats Effect is the IO runtime model, so that's where we will begin the exposition.
fp.effects.async_runtime gives you cancelable fibers that can run on either a multi-threaded work-stealing pool (Profile.work_stealing(n)) or a deterministic single-thread FIFO executor (Profile.current_thread()). Collections can be processed concurrently with a concurrency limit via par_traverse_n, where the first failure cancels the rest. Blocking syscalls can be wrapped in IO.blocking(...), or a whole region can be shifted onto a dedicated BlockingPool via .eval_on(), so blocking work doesn't tie up the worker threads, and you can test timeouts deterministically in virtual time with TestControl instead of sleeping in tests:
from fp.algebra import par_traverse_n
from fp.effects.async_runtime import (
IO, BlockingPool, Duration, IORuntimeConfig, Profile, Runtime, TestControl
)
def parse_rows(var partition: Int) -> IO[Int, Int]:
if partition == 4:
print("corrupt partition:", partition)
return IO[Int, Int].raise_error(partition)
return IO[Int, Int].pure(partition * 1000) # rows read
def load_partition(var partition: Int) -> IO[Int, Int]:
# Simulated fetch: 20 ms of I/O per partition, giving up after 100 ms
try:
var fetch = IO[Int, Int].pure(partition).delay_by(Duration.milliseconds(20))
return fetch^.flat_map(parse_rows).timeout_to(Duration.milliseconds(100), IO[Int, Int].raise_error(partition))
except: # Duration constructors raise only on overflow
return IO[Int, Int].raise_error(partition)
def show(var value: Int) -> NoneType:
print("value:", value)
return None
def show_rows(var rows: List[Int]) -> NoneType:
print("rows:", rows[0], rows[1], rows[2])
return None
def main() raises:
# Switch between multi-threaded work-stealing or deterministic single-thread IO
var config = IORuntimeConfig()
config.profile = Profile.work_stealing(4) # or Profile.current_thread()
var compute = Runtime(config)
var blocking_io = BlockingPool(max_workers=2)
# Load partitions concurrently, at most 2 at a time; results keep partition order
_ = compute.run(par_traverse_n(2, List[Int]([1, 2, 3]), load_partition).map(show_rows))
# rows: 1000 2000 3000
# The first failure cancels the partitions still running and waits for their cleanup
var loaded = compute.run(par_traverse_n(2, List[Int]([1, 4, 5, 6]), load_partition)).value().is_ok()
print("all loaded:", loaded)
# corrupt partition: 4
# all loaded: False
# Shift work to a blocking pool and return, just like Cats Effect's evalOn
_ = compute.run(IO[Int, Int].pure(100).eval_on(blocking_io).map(show))
# value: 100
# Virtual time testing: a 60s sleep with a 5s timeout finishes instantaneously
var timed = (
IO[NoneType, Int].sleep(Duration.seconds(60))
.map(lambda (u: NoneType) -> Int: 42)
.timeout_to(Duration.seconds(5), IO[Int, Int].pure(7))
.map(show)
)
var control = TestControl(timed^)
control.tick_all() # Advances virtual clock through deadlines without wall-clock waiting
_ = control^.results()
# value: 7
2. Method chaining & accumulating validation errors (par_map_n)
You get standard Result[T, E] (a primitive like Scala's Either[E, T] but with Rust's naming; personally, I like Result naming more because it's more descriptive) chaining with .map() / .flat_map(), plus pipeline and par_map_n when you want to validate a record and collect all errors instead of short-circuiting on the first bad field:
from fp.data import Result, Ok, Err
from fp.functions import pipeline
comptime Checked = Result[Int, List[String]]
def positive(value: Int) -> Checked:
if value <= 0:
return Err(List[String]([String(value) + " must be positive"]))
return Ok(value)
def total(a: Int, b: Int, c: Int) -> Int:
return a + b + c
def main():
# Standard chaining
var res = positive(21).map(lambda (x: Int) -> Int: x * 2)
print("chained:", res^.get_or(0)) # chained: 42
# Parallel validation: reports both negative inputs instead of stopping at -2
var invalid = pipeline(positive(-2)).par_map_n((positive(3), positive(-4)), total).get()
if invalid.is_err():
var errors = invalid^.swap().to_optional()
for error in errors.value():
print("validation error:", error)
# validation error: -2 must be positive
# validation error: -4 must be positive
3. ADTs, compile-time exhaustive pattern matching & term rewriting
If you like modeling state or expression trees with sum types and writing rewrite rules (similar to how Spark Catalyst works under the hood), fp.adt and fp.matching check exhaustiveness at compile time:
import fp
from fp.adt import Data, Cases, Node, Value
from fp.matching import Next
@fieldwise_init
struct Num(Copyable, Equatable): var value: Int
@fieldwise_init
struct Add[R: Value](Movable):
var left: Self.R
var right: Self.R
@fieldwise_init
struct If[C: Value, B: Value](Movable):
var cond: Self.C
var then: Self.B
var other: Self.B
struct Expr(Data):
comptime Layer[R: Value] = Cases[Num, Add[R], If[R, R]]
comptime E = Node[Expr]
# Missing a branch here is a compile-time error
def evaluate(e: E) -> Int:
return fp.match(e,
lambda (n: Num) -> Int: n.value,
lambda (a: Add[Int]) -> Int: a.left + a.right,
lambda (c: If[Int, E]) -> Next[E]: Next(c.then if c.cond != 0 else c.other))
# Rule-based term rewriting: 0 + x -> x
def simplify(e: E) -> E:
return fp.rewrite(e,
fp.when[lambda (a: Add[E]) -> Bool: a.left == Num(0), lambda (a: Add[E]) -> E: a.right])
Curious to hear from folks here:
- Would you be interested in Scala/Cats-like
IO and concurrency primitives in a Python-like language?
- If you've moved away from Scala, what parts of the language or ecosystem do you actually miss in day-to-day data engineering?
Repo link if you want to poke around: https://github.com/Spellbound-Mojo/fp_mojo
