Refactored engine and removed feature store

.github/workflows/bandit.yml

@@ -19,3 +19,5 @@ jobs:
     steps:
     - name: Perform Bandit Analysis
       uses: PyCQA/bandit-action@v1
+      with:
+        targets: "src"

.gitignore

@@ -34,9 +34,13 @@ _version.py
 .mypy_cache/
 .ruff_cache/
 reports/
+.vscode/*
+!.vscode/settings.json
+.idea/
 
 # Misc
 .DS_Store
 ~*
-.vscode/*
-!.vscode/settings.json
+
+# Batchwise
+batchwise_checkpoints/

README.md

@@ -13,19 +13,157 @@
 [![Security - Bandit](https://img.shields.io/badge/security-Bandit-yellow.svg)](https://github.com/PyCQA/bandit)
 
 
-Lightweight batch processing engine and feature store.
+Lightweight batch processing engine.
 
 ## Table of Contents
 
 - [Getting Started](#getting_started)
+- [Examples](#examples)
 - [License](#license)
 
 ## Getting Started
 
+### Installation
+
 ```console
 pip install batchwise
 ```
 
+### Basic engine and processor
+
+The core of `batchwise` consists of an `Engine` that manages one or more `Processor` functions. Each processor operates on time-based windows determined by cron expressions.
+
+```python
+import datetime
+from batchwise import Engine, Window
+
+# Initialize the engine (defaults are shown here)
+engine = Engine(
+    checkpoint_path="./batchwise_checkpoints",  # path or uri to checkpoint directory
+    logger_name="batchwise",                    # name of logger instance
+    timezone=datetime.timezone.utc,             # timezone (datetime.tzinfo object)
+)
+
+# Register a processor using the decorator
+@engine.processor(
+    interval="*/30 * * * *",   # Run every 30 minutes
+    delay="1h",                # Wait 1 hour before considering window complete
+    lookback="1d",             # Look back 1 day for processing windows
+    include_incomplete=False,  # Only process complete windows
+)
+def my_processor(window: Window):
+    print(f"Processing window: {window.start} to {window.end}")
+    print(f"Window complete: {window.complete}")
+    # Your processing logic here
+    pass
+
+# Run the engine
+engine()
+```
+
+The `Window` object provides:
+- `window_id`: Unique identifier for the window
+- `start`: Start time of the window (datetime.datetime)
+- `end`: End time of the window (datetime.datetime)
+- `complete`: Boolean indicating if the window is complete
+
+### Using an fsspec-compatible filesystem
+
+`batchwise` supports any `fsspec`-compatible filesystem for checkpoint storage, enabling cloud storage integration:
+
+```python
+from pathlib import Path
+from fsspec.implementations.dirfs import DirFileSystem
+from fsspec.implementations.local import LocalFileSystem
+from batchwise import Engine, Window
+
+# For example, construct a DirFileSystem to wrap a base filesystem
+fs = LocalFileSystem()
+dir_fs = DirFileSystem(path=str(Path.cwd()), fs=fs)
+
+engine = Engine(
+    checkpoint_path="batchwise_checkpoints",  # sub-path on filesystem
+    fs=dir_fs,
+)
+
+@engine.processor(
+    interval="0 * * * *",
+    delay="2h",
+    lookback="1d",
+)
+def my_processor(window: Window):
+    # Your processing logic here
+    pass
+
+engine()
+```
+
+For cloud storage, use the appropriate fsspec implementation (e.g., `s3fs`, `adlfs`, `gcsfs`).
+
+### Additional context
+
+You can pass additional context to your processors:
+
+```python
+from batchwise import Engine, Window
+
+engine = Engine()
+
+# Define a dictionary to be passed to the processor
+config = {
+    "database_url": "postgresql://...",
+    "threshold": 100,
+}
+
+@engine.processor(
+    interval="0 0 * * *",
+    delay="1d",
+    lookback="7d",
+    context=config
+)
+def processor_with_context(window: Window, context: dict):
+    # Access context parameters
+    database_url = context["database_url"]
+    threshold = context["threshold"]
+    # Your processing logic here
+    pass
+
+engine()
+```
+
+**Important:** When using `context`, your processor function must accept a `context` parameter.
+
+### Parallel and/or continuous processing
+
+Run multiple processors in parallel using multiprocessing:
+
+```python
+from batchwise import Engine, Window
+
+engine = Engine()
+
+@engine.processor(interval="*/15 * * * *", delay="30m", lookback="2h")
+def processor_1(window: Window):
+    pass
+
+@engine.processor(interval="*/30 * * * *", delay="1h", lookback="4h")
+def processor_2(window: Window):
+    pass
+
+@engine.processor(interval="0 * * * *", delay="2h", lookback="12h")
+def processor_3(window: Window):
+    pass
+
+# Run sequentially (default)
+engine()
+
+# Run with 3 parallel processes
+engine(num_processes=3)
+
+# Or run continuously with a minimum time between full cycles
+engine(num_processes=3, every_seconds=60)  # Check and run every 60 seconds
+```
+
 ## License
 
 `batchwise` is distributed under the terms of the [MIT](https://spdx.org/licenses/MIT.html) license.

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "hatchling.build"
 [project]
 name = "batchwise"
 dynamic = ["version"]
-description = "Lightweight batch processing engine and feature store."
+description = "Lightweight batch processing engine."
 readme = "README.md"
 requires-python = ">=3.10"
 license = "MIT"
@@ -25,22 +25,14 @@ classifiers = [
   "Programming Language :: Python :: 3.13",
   "Programming Language :: Python :: Implementation :: CPython",
 ]
-dependencies = [
-  "pandas>=1,<3",
-  "pyyaml~=6.0",
-  "pydantic~=2.3",
-  "pydantic-settings~=2.3",
-  "pyarrow>=11.0",
-  "pillow>=11.0",
-  "numpy>=1,<3",
-  "fsspec>=2022",
-]
+dependencies = []
 
 [project.optional-dependencies]
 test = [
   "pytest~=8.3.3",
   "pytest-cov~=6.0.0",
   "pytest-html~=4.1.1",
+  "fsspec>=2024",
 ]
 dev = [
   "pre-commit~=4.0.1",
@@ -57,9 +49,6 @@ Documentation = "https://github.com/manuelkonrad/batchwise#readme"
 Issues = "https://github.com/manuelkonrad/batchwise/issues"
 Source = "https://github.com/manuelkonrad/batchwise"
 
-[project.scripts]
-batchwise = "batchwise.cli:cli"
-
 # ~~~~~ #
 # Hatch #
 # ~~~~~ #
@@ -198,4 +187,5 @@ ignore_missing_imports = true
 [tool.bandit]
 exclude_dirs = [
   "scripts",
+  "tests",
 ]

src/batchwise/__init__.py

@@ -3,6 +3,6 @@
 # SPDX-License-Identifier: MIT
 
 from batchwise.engine import Engine
-from batchwise.store import FeatureStore
+from batchwise.processor import PreventCompletion, Window
 
-__all__ = ["Engine", "FeatureStore"]
+__all__ = ["Engine", "PreventCompletion", "Window"]