[pynumaflow-lite] : Use callable syntax for handlers instead of class inheritance

[BulkBeing]

NOTE: The proposed changes are only for the new sdk (not yet published to pypi). Existing pure-python version won't have any breaking API changes.

In new version of the Rust FFI based Python SDK, we intend to provide most Pythonic APIs.
Currently, we accept either a class (with single method) or a function as the handler to the UDF Grpc server constructor.

Current syntax:

class SimpleCat(mapper.Mapper):
    async def handler(self, keys: list[str], payload: mapper.Datum) -> mapper.Messages:
        ...

# Or a plain function
async def mapper(self, keys: list[str], payload: mapper.Datum) -> mapper.Messages:
    ...

And the server signature will look like:

handler: Mapper | Callable[[list[str], Datum], Awaitable[Messages]]

Instead, we can simplify the handler definition to accept any callable.

handler: Callable[[Datum], Awaitable[list[Message]]],

I made 2 more changes:

• Removed our own list like type Messages and used builtin list instead
• Removed keys as input argument. This is already available as Datum.keys.

So, instead of this:

class SimpleCat(mapper.Mapper):
    async def handler(self, keys: list[str], payload: mapper.Datum) -> mapper.Messages:
        messages = mapper.Messages()

        if payload.value == b"bad world":
            messages.append(mapper.Message.message_to_drop())
            return messages
        
        messages.append(mapper.Message(payload.value, keys))
        return messages

Users can write:

class SimpleCat:
    async def handler(self, datum: mapper.Datum) -> list[mapper.Message]:

        if datum.value == b"bad world":
            return [mapper.Message.to_drop()]

        return [mapper.Message(datum.value, keys=datum.keys)]

Main difference is that, instead of passing the object, we need to pass the handler method (can be any name) to the constructor:

mapper_obj = SimpleCat()

mapper.MapAsyncServer(
    mapper_obj.handler,
).run()

If the user named the handler method as __call__, the object can be passed directly:

class SimpleCat:
    async def __call__(self, datum: mapper.Datum) -> list[mapper.Message]:
        ...


mapper.MapAsyncServer(
   SimpleCat() 
).run()

Since we only need a handler, the state can be in class or even in a closure:

def simple_batch_cat(drop_value: bytes):
    stats = {"processed": 0, "dropped": 0}

    async def handler(batch: AsyncIterator[Datum]) -> list[BatchResponse]:
        responses = []

        async for datum in batch:
            stats["processed"] += 1

            if datum.value == drop_value:
                stats["dropped"] += 1
                messages = [Message.to_drop()]
            else:
                messages = [Message(datum.value, keys=datum.keys)]

            responses.append(BatchResponse(datum.id, messages))

        print(f"Stats: {stats}")
        return responses

    return handler


batchmapper.BatchMapAsyncServer(
    simple_batch_cat(drop_value=b"bad world"),
).run()

This pattern will be followed for all UDFs that doesn't need more than 1 method. The source UDF will still need a class.

---

Message from the maintainers:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.

[vigith]

@th0ger @kohlisid

Please review the proposal

[vigith]

1. Keep a typed contract, just not via inheritance. Removing the ABC also removes IDE autocomplete and abstractmethod enforcement, so a wrong signature would only fail at runtime. Replace each ABC with a typing.Protocol (or at minimum a public type alias, e.g. MapHandler = Callable[[Datum], Awaitable[list[Message]]]). Users get type-checking and discoverability without being forced to subclass.
2. Validate the handler at server construction. Since abstractmethod no longer enforces the contract, *AsyncServer(...) should check callable(handler) (and ideally inspect.iscoroutinefunction, unwrapping __call__) and raise a clear TypeError immediately instead of failing deep in the gRPC loop.
3. Be explicit about the multi-method exception. Sourcer has 5 methods, so it stays a class — correct, a callable can't express that. But the SDK now has two shapes ("pass a callable" vs "pass an object"), so docs/examples should state the rule plainly: single-method UDF -> callable, multi-method UDF -> class/Protocol.