Metadata-Version: 2.4
Name: bean-core
Version: 0.3.0
Summary: Tiny framework for bootstrapping apps
Author-email: numen-0 <numen.0x1dea@gmail.com>
License: MIT License
        
        Copyright (c) `2026` `numen-0`
        
        Permission is hereby granted, free of charge, to any person obtaining a copy of
        this software and associated documentation files (the "Software"), to deal in
        the Software without restriction, including without limitation the rights to
        use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
        of the Software, and to permit persons to whom the Software is furnished to do
        so, subject to the following conditions:
        
        The above copyright notice and this permission notice shall be included in all
        copies or substantial portions of the Software.
        
        THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
        IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
        FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
        AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
        LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
        OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
        SOFTWARE.
        
Project-URL: homepage, https://github.com/numen-0/bean
Project-URL: repository, https://github.com/numen-0/bean
Project-URL: issues, https://github.com/numen-0/bean/issues
Keywords: bean,bean.core
Classifier: Development Status :: 2 - Pre-Alpha
Classifier: Environment :: Console
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.15
Classifier: Topic :: Scientific/Engineering
Classifier: Topic :: Software Development
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Typing :: Typed
Requires-Python: >=3.14
Description-Content-Type: text/markdown
License-File: LICENSE
Dynamic: license-file

# 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`:

```sh
pip install --upgrade bean-core
```

Using `curl` (direct download):

```sh
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](/bean-core/src/bean/core.py) `:)`.

### 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.

```py
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

```py
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:
>
> ```py
> 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:

```py
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": "alex@example.com"
    })
 ).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**.

```py
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:

```py
from bean.core import Predicate

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

Predicates support logical composition:

```py
@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:

```py
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.

```py
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:

```py
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:

```py
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`.

```py
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.

```py
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:

```py
from bean.core import Scheduler

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

Periodic Job:

```py
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:

```py
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](/LICENSE).

