bean-core 0.3.1

Tiny framework for bootstrapping apps

bean.core

bean.core is a tiny Python toolkit to bootstrap small apps.

It gives you ready-to-use building blocks so you can focus on the business logic, instead of boilerplate.

---

Overview

With bean.core you get:

• App lifecycle: start, run, shutdown
• Config: load from multiple sources (env, .py, ...) with validation.
• Logging.
• Pipes: composable data flows.
• Shell commands: run commands directly in your flows.
• Scheduler: jobs & tasks.

Just enough to cook some beans.

Installation

Requirements:

• Python 3.14+

Using pip:

pip install --upgrade bean-core

Using curl (direct download):

FILE="src/bean/core.py"
mkdir -p "$(dirname "$FILE")"
curl -Lso "$FILE" \
    https://raw.githubusercontent.com/numen-0/bean/refs/heads/main/bean-core/src/bean/core.py

Quick Examples

This is a quick reference for the main API.

For full details, peek at the source code :).

Simple Loop App

In this example:

• define a custom app by overriding the run method and optionally startup and shutdown.
• set up logging.
• script a infinite loop on run.
• run bean.main with our App()

So, the app will loop until a SIGINT or SIGTERM signal is received. On shutdown, the shutdown method is called, but you can force termination by sending another signal.

from time import sleep
from bean.core import BeanApp, Log, Logger, main, shutdown_requested

class App(BeanApp):
    def startup(self):
        Log.init(
            self.name,
            level=Logger.Level.from_debug(True),
            handlers=[Logger.TermHandler(Logger.fmt(color=True))]
        )
        Log.debug("starting...")

    def shutdown(self):
        Log.debug("ending...")

    def run(self):
        while not shutdown_requested():
            Log.info(":D doing nothing")
            sleep(2)

        Log.warning("xD shutdown requested, exiting run loop")
        return 0

if __name__ == "__main__":
    main(App("bean-app"))

Config

BeanConfig can be populated from multiple sources, with a defined priority. Available sources:

• .from_args(): parse command-line arguments with argparse
• .from_dict(): load configuration from a Python dictionary
• .from_env(): load environment variables matching prefix + filedname
• .from_ini(): Load from a INI file
• .from_json(): load from a JSON file
• .from_py(): load from a Python file exposing a Config object by default
• .from_toml(): Load from a TOML file

Note: File loaders will skip missing files unless force=True is passed

from enum import Enum
from bean.core import BeanConfig, ConfigField, isHost, isPort, isEmail

class Color(Enum):
    RED   = "#ff0000"
    GREEN = "#00ff00"
    BLUE  = "#0000ff"

class MyConfig(BeanConfig):
    NAME   = ConfigField(str)
    DEBUG  = ConfigField(bool, default=False, short_flag="-d")
    PORT   = ConfigField(int, default=8080, validator=isPort)
    HOST   = ConfigField(str, default="localhost", validator=isHost)
    EMAIL  = ConfigField(str, validator=isEmail)
    COLORS = ConfigField(list[Color], default=[Color.RED])

    @BeanConfig.validate("NAME")
    def check_empty_name(name):
        return len(name) > 0


( MyConfig.load()            # load priority:
    .from_env("APP_")        # 1. environment variables
    .from_py("./config.py")  # 2. Python file (ignored if not found)
    .from_args()             # 3. command-line arguments (auto --help)
).build()

MyConfig.print_config()

Note: If your type checker complains about check_empty_name, add @staticmethod, # type: ignore or declare it outside the class:

class MyConfig(BeanConfig):
    NAME   = ConfigField(str)
    DEBUG  = ConfigField(bool, default=False, short_flag="-d")
    PORT   = ConfigField(int, default=8080, validator=isPort)
    HOST   = ConfigField(str, default="localhost", validator=isHost)
    EMAIL  = ConfigField(str, validator=isEmail)
    COLORS = ConfigField(list[Color], default=[Color.RED])


@MyConfig.validate("NAME")
def check_empty_name(name):
    return len(name) > 0

Note: Validators declared this way only affect MyConfig or its subclasses, unless explicitly overridden (field + function names must match).

Simpler Dataclass-Style Config

For simpler configs, use the BeanConfig.dataclass decorator. It generates a dataclass-style config class while maintaining full BeanConfig features:

from dataclasses import field
from enum import Enum
from bean.core import BeanConfig

class Color(Enum):
    RED   = "#ff0000"
    GREEN = "#00ff00"
    BLUE  = "#0000ff"

@BeanConfig.dataclass
class MySimpleConfig(BeanConfig):
    NAME: str
    EMAIL: str
    DEBUG: bool = False
    PORT: int = 8080
    HOST: str = "localhost"
    COLORS: list[Color] = field(default_factory=lambda: [Color.RED])

( MySimpleConfig.load()
    .from_dict({
        "NAME": "alex",
        "EMAIL": "[e-mail]"
    })
 ).build().print_config()

• Required fields are inferred from the type annotations.
• Fields with defaults or default_factory are optional.
• After .build(), values can be accessed via instance or class-level.

Predicates

Predicate is a composable T -> bool function.

A predicate:

• Receives a value
• Returns True or False
• Can be combined with other predicates to build complex logic

Predicates are useful for validation, filtering and guards.

from bean.core import Predicate

positive = Predicate(lambda x: x > 0, name="positive")

print(positive(5))      # True
print(positive(-1))     # False

Predicates can also be created using a decorator:

from bean.core import Predicate

@Predicate
def positive(x: int):
    return x > 0

Predicates support logical composition:

@Predicate
def positive(x: int): return x > 0

@Predicate
def even(x: int): return x % 2 == 0

p = positive & even

print(p(4))     # True
print(p(3))     # False
print(p(-2))    # False

Available operators:

• p1 & p2: logical AND
• p1 | p2: logical OR
• ~p: logical NOT

Predicates can be traced or hooked:

from bean.core import Predicate

count = 0
def watcher(value, result):
    global count
    count += 1
    print(f"{count:02d}: {value} -> {result}")

@Predicate
def positive(x: int):
    return x > 0

traced = positive.trace(watcher)

traced(5)       # 01: 5 -> True
traced(-2)      # 02: -2 -> False
traced(4)       # 03: 4 -> True

Pipes

Pipe is a composable transformation pipeline.

Each stage:

• Receives a value
• Returns either:
  • Success(value) or tuple(value, True)
  • tuple(value, ok)
• If ok == False, the pipeline short-circuits

This makes it easy to build safe, expressive and composable data flows.

from bean.core import Pipe

result = (
    Pipe()
        .guard(lambda x: x != 0)
        .map(lambda x: 10 / x)
)(5)

print(result)   # Success(2.0, ok=True)

Note: pipes can be typed, e.g: Pipe[float, float]()

If the guard fails:

result = (
    Pipe()
        .guard(lambda x: x != 0)
        .map(lambda x: 10 / x)
)(0)

print(result)   # Success(value=0, ok=False)

The pipe short-circuits and the division step is never executed.

Available Pipe helpers:

• .map(fn): transform value
• .guard(fn): validate value (may short-circuit)
• .peek(fn): side-effect without modifying value
• .retry(fn, attempts, delay): retry a stage
• .fallback(fn, fallback_value): recover from failure
• .branch(cond, success_fn, fail_fn): conditional logic
• .trigger(fn, ex, msg): raise exception if condition matches

Pipes are fully composable using the | operator.

Functions can be wrapped using Pipe as decorator:

from bean.core import Pipe

@Pipe
def dup(x: int|float):
    return (x * 2, True)

@Pipe
def half(x: int|float):
    return (x / 2, True)

pipe = (dup | half).map(int)

print(pipe(10))     # Success(10, ok=True)

Using predicates with pipes

Predicates integrate naturally with Pipe.guard.

from bean.core import Pipe, Predicate

@Predicate
def positive(x: int):
    return x > 0

pipe = (
    Pipe[int, int]()
        .guard(positive)
        .map(lambda x: x * 2)
)

print(pipe(5))      # Success(10, ok=True)
print(pipe(-1))     # Success(-1, ok=False)

Shell Commands

Shell commands integrate directly into pipes.

sh(cmd) returns a Pipe that executes a shell command.

from bean.core import sh, cat, tee, stdout

res = ()
print(
    (sh("echo 'hello bean'")
        | sh("grep -F 'hello'")
        | tee("copy.txt")
        | stdout()
     )(None)
)

Available shell helpers:

• sh(cmd): run command
• cat(*paths): read files
• tee(*paths): write to files
• stdout(): extract stdout
• stderr(): extract stderr

Scheduler

Scheduler provides a minimal threaded task runner for delayed and periodic execution.

It supports:

• One-shot tasks
• Delayed tasks
• Periodic jobs
• Limited or infinite runs
• Graceful shutdown

Basic example:

from bean.core import Scheduler

( Scheduler()
    .task(lambda: print("Bean task once"))
).start()

Periodic Job:

from bean.core import Scheduler

schr = (
    Scheduler()
        .job(
            fn=lambda: print("Bean job running!"),
            interval=2.0,
            runs=5
        )
).start()

# ...

schr.join()  # wait until finite jobs complete

Mixed example:

import time
from bean.core import Scheduler

with Scheduler()
    .task(lambda: (print("Init task"), time.sleep(5), print("Done")))
    .job(lambda: print("Heartbeat"), interval=1.0)
) as schr:
    ...

Note: When exiting the with block, the scheduler automatically calls schr.stop().join() to gracefully stop infinite jobs.

API Overview:

• .task(fn, runs=1, delay=0) -> Schedule a task that runs:

  • Runs runs times (default: once) or indefinitely if runs=None
  • Starts after an optional delay

• .job(fn, interval, runs=None) -> Schedule a periodic job:

  • Runs every interval seconds
  • Runs indefinitely by default (runs=None)

Lifecycle Methods:

• .start(): start all tasks
• .join(timeout=None): wait for completion (non-infinite only)
• .stop(): signal stop
• .clear(): remove all tasks

All methods return Self for chaining.

Behavior Notes:

• Each task runs in its own daemon thread
• Finite tasks (runs != None) blocks .join() until completion or timeout
• Infinite tasks (runs=None) are ignored by .join()
• Tasks must exit voluntarily, blocking calls may delay shutdown
• Because threads are daemon, all tasks terminate when the main program exits

License

All the repo falls under the MIT License.