# Callback
------------------------------------------------------------------------
# Callback core
The callback pattern allows you to inject custom behavior at specific
points in an object’s lifecycle without modifying the object itself.
Think of callbacks as “hooks” where you can attach custom logic.
**Basic concept**: An object calls predefined method names (like
`before_fit`, `after_batch`) on its registered callbacks. Each callback
can implement whichever methods it needs and ignore the rest.
------------------------------------------------------------------------
source
### Callback
``` python
def Callback(
args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):
```
*Base class of callbacks.*
------------------------------------------------------------------------
source
### run_cbs
``` python
def run_cbs(
cbs:Iterable[Callback] | FC.L, method_nm:str, ctx:NoneType=None, args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):
```
*Run `method_nm(ctx, ...)` of each callback in `cbs` in order.*
# Examples
------------------------------------------------------------------------
source
### EchoCB
``` python
def EchoCB(
args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):
```
*Print the arguments.*
``` python
run_cbs((EchoCB(),), 'on_start', AD(count=3))
run_cbs((EchoCB(),), 'on_update', AD(count=3), 12)
```
on_start ({'count': 3},) {}
on_update ({'count': 3}, 12) {}
``` python
class VerboseCB(Callback):
"Inspect the arguments."
def before_fit(self, ctx): self.count = 0
def after_batch(self, ctx): self.count += 1
def after_fit(self, ctx): print(f'{ctx} Completed {self.count} batches')
```
``` python
cbs = [VerboseCB()]
run_cbs(cbs, 'before_fit')
test_eq(cbs[0].count, 0)
run_cbs(cbs, 'after_batch')
test_eq(cbs[0].count, 1)
test_stdout(lambda: run_cbs(cbs, 'after_fit'), 'None Completed 1 batches')
```
------------------------------------------------------------------------
source
### FuncCB
``` python
def FuncCB(
kwargs:VAR_KEYWORD
):
```
*Store functions that can be called as callbacks.*
``` python
cb = FuncCB(on_count=lambda ctx: print(ctx.count))
run_cbs((cb,), 'on_count', o := AD(count=3))
```
3
``` python
def inc(ctx): ctx.count += 1
cb = FuncCB(on_count=(inc, lambda ctx: print(ctx.count)))
run_cbs((cb,), 'on_count', o := AD(count=3))
test_eq(o.count, 4)
```
4
------------------------------------------------------------------------
source
### PassCB
``` python
def PassCB(
args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):
```
*A callback that does nothing.*
# Helpers
## HasCallbacks
Mixin class that adds callback support to any class. It manages a list
of callbacks and provides methods to run them at specific lifecycle
points.
**Key features**: - Register callbacks via constructor or
[`with_cbs()`](https://civvic.github.io/pote/callback.html#with_cbs) -
Temporary callbacks via `this_cbs()` context manager - Automatic method
delegation for names in `cbs_names`
------------------------------------------------------------------------
source
### HasCallbacks
``` python
def HasCallbacks(
cbs:Sequence[Callback]=()
):
```
*Base for classes that can be augmented with callbacks.*
``` python
class Test(HasCallbacks):
count = 0
cbs_names = ('on_think',)
def think(self):
self.callback('before_think')
self.count += 1
print('thinking...')
self.on_think()
self.callback('after_think')
def act(self):
self.callback('before_act')
self.count = 1
print('acting...')
self.think()
self.callback('after_act')
class VerboseCB(Callback):
def before_act(self, ctx): print('before_act count:', ctx.count)
def after_act(self, ctx): print('after_act count:', ctx.count)
test = Test()
test.act()
test_eq(test.count, 2)
print()
test = Test([VerboseCB()])
test.act()
test_eq(test.count, 2)
class ThinkCB(Callback):
def on_think(self, ctx): ctx.count += 1
print()
test = Test([VerboseCB(), ThinkCB()])
test.act()
test_eq(test.count, 3)
```
acting...
thinking...
before_act count: 0
acting...
thinking...
after_act count: 2
before_act count: 0
acting...
thinking...
after_act count: 3
------------------------------------------------------------------------
source
### with_cbs
``` python
def with_cbs(
nm:str | None=None
):
```
*Decorator to add callbacks to a method.*
``` python
class Test(HasCallbacks):
count = 0
@with_cbs()
def think(self):
print('thinking...')
self.count += 1
@with_cbs('act')
def act(self, cbs:Sequence[Callback]=()):
with self.this_cbs(cbs):
print('acting...')
self.count = 1
self.think()
class ActCB(Callback):
def before_act(self, ctx): print('before_act count:', ctx.count)
def after_act(self, ctx):
print('after_act count:', ctx.count)
ctx.count += 1
class ThinkCB(Callback):
def before_think(self, ctx): print('before_think count:', ctx.count)
def after_think(self, ctx):
print('after_think count:', ctx.count)
ctx.count += 1
test = Test()
test.act()
test_eq(test.count, 2)
print()
test = Test([ActCB()])
test.act()
test_eq(test.count, 3)
print()
test = Test([ActCB()])
test.act([ThinkCB()])
test_eq(test.count, 4)
```
acting...
thinking...
before_act count: 0
acting...
thinking...
after_act count: 2
before_act count: 0
acting...
before_think count: 1
thinking...
after_think count: 2
after_act count: 3
# Iteration
Infrastructure for tracking iteration progress. These helpers support
[`CollBack`](https://civvic.github.io/pote/callback.html#collback) by
determining iteration counts and monitoring state.
# CollBack
> Iterator wrapper with progress tracking and callbacks
[`CollBack`](https://civvic.github.io/pote/callback.html#collback) wraps
any iterable and provides:
1. **Progress tracking**: Current position, total count, percentage,
elapsed time
2. **State access**: Query iteration state at any point
3. **Callbacks**: Run custom code before/after/during iteration
4. **Drop-in replacement**: Works anywhere an iterable is expected
## CollBack vs Standard Iteration
**Standard iteration**:
``` python
for item in items:
process(item) # No visibility into progress, timing, or position
```
**With CollBack**:
``` python
for item in CollBack(items, cbs=[LogProgressCB()]):
process(item) # Auto-logging of progress, timing, position
```
## Common Use Cases
- File processing — Track progress through large files
- Data pipelines — Monitor throughput and estimate completion time
- Training loops — Report epoch/batch progress with callbacks
- API calls — Rate limit, retry, log responses without cluttering
business logic
- Testing — Inject behavior for debugging without modifying code
------------------------------------------------------------------------
source
### CollBack
``` python
def CollBack(
source:Iterable[Any]=(), total:int | None | Type[EmptyT]=EmptyT, context:Any=EmptyT, kwargs:VAR_KEYWORD
):
```
*Track iterables and extend them with callbacks.*
------------------------------------------------------------------------
source
### trackback
``` python
def trackback(
source:Iterable[Any], total:int | None | Type[EmptyT]=EmptyT, context:Any=EmptyT, cbs:Sequence[Callback]=()
)->Iterator[tuple[AD[Any], Any]]:
```
## Usage Examples
[`CollBack`](https://civvic.github.io/pote/callback.html#collback) is a
drop-in replacement for any iterable:
``` python
test_eq(deque([1,2,3], maxlen=3), deque(CollBack([1,2,3]), maxlen=3))
test_eq(list(ChainMap({'a':1}, {'b':2})), list(CollBack(ChainMap({'a':1}, {'b':2})))) # Search multiple dicts
test_eq(list(CollBack(Counter('hello').items())), [('h', 1), ('e', 1), ('l', 2), ('o', 1)])
```
You can use
[`CollBack`](https://civvic.github.io/pote/callback.html#collback) to
track the progress of any iterable with arbitrary callbacks. It can be
used in place of the iterable in any function that takes an iterable.
``` python
tuple(CollBack(open('static/file.txt'))) # Line iterator
```
('line 1\n', 'line 2\n', 'line 3\n')
``` python
tuple(CollBack(open('static/file.txt').read(3))) # Char iterator
```
('l', 'i', 'n')
``` python
tuple(CollBack(os.scandir())) # Directory iterator
```
(,
,
,
,
,
,
,
,
,
,
,
,
,
)
``` python
def print_progress(ctx, line):
if ctx.n % 100 == 0:
print(f"Processed {ctx.n} {ctx.elapsed_time:.2f}")
def process_line(line): time.sleep(random.uniform(0.0001, 0.0005))
with open('10_callback.ipynb') as f:
for st,line in trackback(f, cbs=[FuncCB(on_iter=print_progress)]):
process_line(line)
```
Processed 0 0.00
Processed 100 0.04
Processed 200 0.08
Processed 300 0.12
Processed 400 0.16
Processed 500 0.20
Processed 600 0.24
Processed 700 0.28
Processed 800 0.32
Processed 900 0.36
Processed 1000 0.40
Processed 1100 0.44
Processed 1200 0.48
Processed 1300 0.52
Processed 1400 0.56
Processed 1500 0.60
Processed 1600 0.64
## State Tracking and Validation
CollBack maintains iteration state and handles edge cases:
``` python
for o in trackback(range(3), cbs=[EchoCB()]): print(o)
```
before_iter ({'n': None, 'total': 3, 'progress': None, 'elapsed_time': None},) {}
({'item': 0, 'n': 0, 'total': 3, 'progress': 0.3333, 'elapsed_time': None}, 0)
on_iter ({'item': 0, 'n': 0, 'total': 3, 'progress': 0.3333, 'elapsed_time': 0.00010013580322265625}, 0) {}
({'item': 1, 'n': 1, 'total': 3, 'progress': 0.6667, 'elapsed_time': 0.00010013580322265625}, 1)
on_iter ({'item': 1, 'n': 1, 'total': 3, 'progress': 0.6667, 'elapsed_time': 0.0001499652862548828}, 1) {}
({'item': 2, 'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': 0.0001499652862548828}, 2)
on_iter ({'item': 2, 'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': 0.0001900196075439453}, 2) {}
after_iter ({'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': 0.0001900196075439453},) {}
``` python
t = CollBack(())
test_eq(t.total, 0)
with test_raises(StopIteration): next(iter(t))
for _ in (t := CollBack(range(3))):
print(t.state)
t = CollBack(range(3))
test_eq(t.state, {'n': None, 'total': 3, 'progress': None, 'elapsed_time': None})
test_eq(next(it := iter(t)), 0)
test_eq(t.state, {'item': 0, 'n': 0, 'total': 3, 'progress': 0.3333, 'elapsed_time': t.elapsed_time})
test_eq(next(it), 1)
test_eq(t.state, {'item': 1, 'n': 1, 'total': 3, 'progress': 0.6667, 'elapsed_time': t.elapsed_time})
test_eq(next(it), 2)
test_eq(t.state, {'item': 2, 'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
with test_raises(StopIteration): next(it)
test_eq(t.state, {'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
t = CollBack('abcdef')
test_eq([o for o in t], list('abcdef'))
test_eq(t.state, {'n': 5, 'total': 6, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
test_eq(t.active, False)
t = CollBack(repeat(1, 3))
test_eq(list(map(lambda x: x, t)), [1, 1, 1])
test_eq(t.state, {'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
```
{'item': 0, 'n': 0, 'total': 3, 'progress': 0.3333, 'elapsed_time': None}
{'item': 1, 'n': 1, 'total': 3, 'progress': 0.6667, 'elapsed_time': 4.1961669921875e-05}
{'item': 2, 'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': 6.699562072753906e-05}
``` python
t = CollBack((12, 56, -1, 2, 67), 3)
test_eq([o for o in t], (12, 56, -1))
test_eq(t.state, {'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
```
``` python
t = CollBack(c := (1, 2, -1, -2, 7))
test_eq(reduce(lambda x, y: x+y, t), 7)
test_eq(t.state, {'n': 4, 'total': 5, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
class CountCB(Callback):
def before_iter(self, ctx): self.count = 0
def on_iter(self, ctx, _): self.count += 1
with (t := CollBack(c)).this_cbs([cb := CountCB()]):
test_eq(reduce(lambda x, y: x+y, t), 7)
test_eq(t.state, {'n': 4, 'total': 5, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
test_eq(cb.count, 5)
test_eq(reduce(lambda x, y: x+y, (t := CollBack(c, cbs=[cb]))), 7)
test_eq(t.state, {'n': 4, 'total': 5, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
test_eq(cb.count, 5)
```
``` python
t = CollBack(repeat(7), 3)
oo = [o for o in t]
test_eq(oo, (7, 7, 7))
test_eq(t.state, {'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
t = CollBack(repeat(None), None)
for _ in t:
if t.state.n >= 10: break # type: ignore
test_eq(t.state, {'n': 10, 'total': 11, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
def fibonacci():
a, b = 0, 1
while True:
yield a
a, b = b, a + b
test_eq(CollBack(fibonacci(), 10), [0, 1, 1, 2, 3, 5, 8, 13, 21, 34])
```
``` python
t = CollBack(())
test_eq(t.total, 0)
with test_raises(StopIteration): next(iter(t))
for i in t: pass
t = CollBack(range(6))
test_eq(t.state, {'n': None, 'total': 6, 'progress': None, 'elapsed_time': None})
test_eq(next(iter(t)), 0)
test_eq(t.state, {'n': 0, 'total': 6, 'progress': 0.1667, 'elapsed_time': t.elapsed_time})
t = CollBack(range(6))
test_eq([t.n for i in t], range(6))
test_eq(t.state, {'n': 5, 'total': 6, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
```
``` python
t = CollBack('abc', None)
test_eq([_ for _ in t], ('a', 'b', 'c'))
test_eq(t.state, {'n': 2, 'total': 3, 'progress': 1.0, 'elapsed_time': t.elapsed_time})
```
# process\_
Convenience function for batch processing iterables with filtering,
slicing, and callbacks.
**Why use
[`process_`](https://civvic.github.io/pote/callback.html#process_)?**
Instead of manually consuming an iterator just to trigger side effects:
``` python
for item in CollBack(items[1:10], cbs=[logger]):
if predicate(item):
pass # Just consuming for side effects
```
Use [`process_`](https://civvic.github.io/pote/callback.html#process_)
for cleaner intent:
``` python
process_(items, logger, slice(1, 10), pred=predicate)
```
**Common patterns**: - Process a subset of data with progress tracking -
Apply callbacks without explicit loop - Filter and slice in one
expression - Collect callback results for inspection
https://stackoverflow.com/questions/50937966/fastest-most-pythonic-way-to-consume-an-iterator
------------------------------------------------------------------------
source
### process\_
``` python
def process_(
iterable:Iterable[_T], cbs:Callback | Sequence[Callback]=(), slc:slice | None=None,
pred:Callable[[_T], bool] | None=None, context:Any=EmptyT, kwargs:VAR_KEYWORD
)->tuple[Callback, ...]: # FuncCB kwargs
```
*Process a subset `slc` of `iterable` filtered by `pred` with callbacks
from `cbs` and
[`FuncCB`](https://civvic.github.io/pote/callback.html#funccb) `kwargs`*
### Usage Example
Process even numbers from positions 1-9:
``` python
process_(range(10), EchoCB(), slice(1,9), pred=lambda x: x%2==0);
```
before_iter ({'n': None, 'total': 4, 'progress': None, 'elapsed_time': None},) {}
on_iter ({'item': 2, 'n': 0, 'total': 4, 'progress': 0.25, 'elapsed_time': 5.1975250244140625e-05}, 2) {}
on_iter ({'item': 4, 'n': 1, 'total': 4, 'progress': 0.5, 'elapsed_time': 7.772445678710938e-05}, 4) {}
on_iter ({'item': 6, 'n': 2, 'total': 4, 'progress': 0.75, 'elapsed_time': 9.393692016601562e-05}, 6) {}
on_iter ({'item': 8, 'n': 3, 'total': 4, 'progress': 1.0, 'elapsed_time': 0.0001087188720703125}, 8) {}
after_iter ({'n': 3, 'total': 4, 'progress': 1.0, 'elapsed_time': 0.0001087188720703125},) {}
``` python
# Real-world pattern: process with inline callback functions
count = 0
def track_count(ctx, item):
global count
count += 1
process_(range(20), slc=slice(5, 15), pred=lambda x: x % 3 == 0, on_iter=track_count)
test_eq(count, 3) # 6, 9, 12
```
------------------------------------------------------------------------