Callback

Extensible callback system for augmenting objects and tracking iterations

---

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

def Callback(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Base class of callbacks.

---

source

run_cbs

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

def EchoCB(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Print the arguments.

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) {}

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')

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

def FuncCB(
    kwargs:VAR_KEYWORD
):

Store functions that can be called as callbacks.

cb = FuncCB(on_count=lambda ctx: print(ctx.count))
run_cbs((cb,), 'on_count', o := AD(count=3))

3

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

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() - Temporary callbacks via this_cbs() context manager - Automatic method delegation for names in cbs_names

---

source

HasCallbacks

def HasCallbacks(
    cbs:Sequence[Callback]=()
):

Base for classes that can be augmented with callbacks.

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

def with_cbs(
    nm:str | None=None
):

Decorator to add callbacks to a method.

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 by determining iteration counts and monitoring state.

CollBack

Iterator wrapper with progress tracking and callbacks

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:

for item in items:
    process(item)  # No visibility into progress, timing, or position

With CollBack:

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

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

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 is a drop-in replacement for any iterable:

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

tuple(CollBack(open('static/file.txt')))  # Line iterator

('line 1\n', 'line 2\n', 'line 3\n')

tuple(CollBack(open('static/file.txt').read(3)))  # Char iterator

('l', 'i', 'n')

tuple(CollBack(os.scandir()))  # Directory iterator

(<DirEntry '10_callback.ipynb'>,
 <DirEntry '_quarto.yml'>,
 <DirEntry 'sidebar.yml'>,
 <DirEntry 'styles.css'>,
 <DirEntry '15_config.ipynb'>,
 <DirEntry 'nbdev.yml'>,
 <DirEntry '00_basic.ipynb'>,
 <DirEntry 'static'>,
 <DirEntry '05_test.ipynb'>,
 <DirEntry '.ipynb_checkpoints'>,
 <DirEntry '20_widgets.ipynb'>,
 <DirEntry '17_display.ipynb'>,
 <DirEntry '00_project.ipynb'>,
 <DirEntry 'index.ipynb'>)

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:

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},) {}

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}

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})

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)

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])

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})

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_?

Instead of manually consuming an iterator just to trigger side effects:

for item in CollBack(items[1:10], cbs=[logger]):
    if predicate(item):
        pass  # Just consuming for side effects

Use process_ for cleaner intent:

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_

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 kwargs

Usage Example

Process even numbers from positions 1-9:

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},) {}

# 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

---