### Example SideInput Implementation

Source: https://numaproj.io/numaflow-python/latest/api/sideinput

A complete example showing how to implement a custom SideInput class and start the server.

```python
import datetime
from pynumaflow.sideinput import Response, SideInputServer, SideInput

class ExampleSideInput(SideInput):
    def __init__(self):
        self.counter = 0

    def retrieve_handler(self) -> Response:
        time_now = datetime.datetime.now()
        # val is the value to be broadcasted
        val = f"an example: {str(time_now)}"
        self.counter += 1
        # broadcast every other time
        if self.counter % 2 == 0:
            # no_broadcast_message() is used to indicate that there is no broadcast
            return Response.no_broadcast_message()
        # broadcast_message() is used to indicate that there is a broadcast
        return Response.broadcast_message(val.encode("utf-8"))

if __name__ == "__main__":
    grpc_server = SideInputServer(ExampleSideInput())
    grpc_server.start()
```

--------------------------------

### Implement and Start PrimeMap Server

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Example implementation of a Mapper class and starting the MapMultiprocServer.

```python
import math
import os
from pynumaflow.mapper import Messages, Message, Datum, Mapper, MapMultiprocServer

def is_prime(n):
    for i in range(2, int(math.ceil(math.sqrt(n)))):
        if n % i == 0:
            return False
    else:
        return True

class PrimeMap(Mapper):
    def handler(self, keys: list[str], datum: Datum) -> Messages:
        val = datum.value
        _ = datum.event_time
        _ = datum.watermark
        messages = Messages()
        for i in range(2, 100000):
            is_prime(i)
        messages.append(Message(val, keys=keys))
        return messages

if __name__ == "__main__":
    server_count = 2
    prime_class = PrimeMap()
    # Server count is the number of server processes to start
    grpc_server = MapMultiprocServer(prime_class, server_count=server_count)
    grpc_server.start()
```

--------------------------------

### AsyncSource Implementation Example

Source: https://numaproj.io/numaflow-python/latest/api/sourcer

A complete example demonstrating how to implement a custom source by inheriting from Sourcer and starting it with SourceAsyncServer.

```python
from datetime import datetime
from pynumaflow.shared.asynciter import NonBlockingIterator
from pynumaflow.sourcer import (
    ReadRequest,
    Message,
    AckRequest,
    PendingResponse,
    Offset,
    PartitionsResponse,
    get_default_partitions,
    Sourcer,
    SourceAsyncServer,
    NackRequest,
)

class AsyncSource(Sourcer):
    # AsyncSource is a class for User Defined Source implementation.

    def __init__(self):
        # The offset idx till where the messages have been read
        self.read_idx: int = 0
        # Set to maintain a track of the offsets yet to be acknowledged
        self.to_ack_set: set[int] = set()
        # Set to maintain a track of the offsets that have been negatively acknowledged
        self.nacked: set[int] = set()

    async def read_handler(self, datum: ReadRequest, output: NonBlockingIterator):
        '''
        read_handler is used to read the data from the source and send the data forward
        for each read request we process num_records and increment the read_idx to
        indicate that the message has been read and the same is added to the ack set
        '''
        if self.to_ack_set:
            return

        for x in range(datum.num_records):
            # If there are any nacked offsets, re-deliver them
            if self.nacked:
                idx = self.nacked.pop()
            else:
                idx = self.read_idx
                self.read_idx += 1
            headers = {"x-txn-id": str(uuid.uuid4())}
            await output.put(
                Message(
                    payload=str(self.read_idx).encode(),
                    offset=Offset.offset_with_default_partition_id(str(idx).encode()),
                    event_time=datetime.now(),
                    headers=headers,
                )
            )
            self.to_ack_set.add(idx)

    async def ack_handler(self, ack_request: AckRequest):
        '''
        The ack handler is used acknowledge the offsets that have been read, and remove
        them from the to_ack_set
        '''
        for req in ack_request.offsets:
            offset = int(req.offset)
            self.to_ack_set.remove(offset)

    async def nack_handler(self, ack_request: NackRequest):
        '''
        Add the offsets that have been negatively acknowledged to the nacked set
        '''
        for req in ack_request.offsets:
            offset = int(req.offset)
            self.to_ack_set.remove(offset)
            self.nacked.add(offset)

    async def pending_handler(self) -> PendingResponse:
        '''
        The simple source always returns zero to indicate there is no pending record.
        '''
        return PendingResponse(count=0)

    async def partitions_handler(self) -> PartitionsResponse:
        '''
        The simple source always returns default partitions.
        '''
        return PartitionsResponse(partitions=get_default_partitions())


if __name__ == "__main__":
    ud_source = AsyncSource()
    grpc_server = SourceAsyncServer(ud_source)
    grpc_server.start()
```

--------------------------------

### ReduceAsyncServer Usage Example

Source: https://numaproj.io/numaflow-python/latest/api/reducer

Demonstrates how to implement a reducer using either a class or a function and start the server.

```python
import os
from collections.abc import AsyncIterable
from pynumaflow.reducer import Messages, Message, Datum, Metadata,
ReduceAsyncServer, Reducer

class ReduceCounter(Reducer):
    def __init__(self, counter):
        self.counter = counter

    async def handler(
        self, keys: list[str], datums: AsyncIterable[Datum], md: Metadata
    ) -> Messages:
        interval_window = md.interval_window
        self.counter = 0
        async for _ in datums:
            self.counter += 1
        msg = (
            f"counter:{self.counter} interval_window_start:{interval_window.start} "
            f"interval_window_end:{interval_window.end}"
        )
        return Messages(Message(str.encode(msg), keys=keys))

async def reduce_handler(
    keys: list[str], datums: AsyncIterable[Datum], md: Metadata
) -> Messages:
    interval_window = md.interval_window
    counter = 0
    async for _ in datums:
        counter += 1
    msg = (
        f"counter:{counter} interval_window_start:{interval_window.start} "
        f"interval_window_end:{interval_window.end}"
    )
    return Messages(Message(str.encode(msg), keys=keys))

if __name__ == "__main__":
    invoke = os.getenv("INVOKE", "func_handler")
    if invoke == "class":
        # Here we are using the class instance as the reducer_instance
        # which will be used to invoke the handler function.
        # We are passing the init_args for the class instance.
        grpc_server = ReduceAsyncServer(ReduceCounter, init_args=(0,))
    else:
        # Here we are using the handler function directly as the reducer_instance.
        grpc_server = ReduceAsyncServer(reduce_handler)
    grpc_server.start()
```

--------------------------------

### start

Source: https://numaproj.io/numaflow-python/latest/api/sinker

Starts the Async server instance.

```APIDOC
## start()

### Description
Starter function for the Async server class. This method is required to start all async coroutines from a single context.
```

--------------------------------

### start()

Source: https://numaproj.io/numaflow-python/latest/api/reducestreamer

Starts the Async Reduce Stream Server.

```APIDOC
## start()

### Description
Starter function for the Async server class. This method runs the server's execution context, allowing all async coroutines to be started from a single context.
```

--------------------------------

### start()

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

Starts the multi-process gRPC server.

```APIDOC
## start()

### Description
Starts the N gRPC servers on the configured socket path. The number of processes (N) is determined by the number of CPUs or the user-defined server_count, capped at 2 * CPU count.
```

--------------------------------

### Invoke ReduceStreamAsyncServer

Source: https://numaproj.io/numaflow-python/latest/api/reducestreamer

Example demonstrating how to use either a class or a function as the reducer instance to start the server.

```python
import os
from collections.abc import AsyncIterable
from pynumaflow.reducestreamer import Messages, Message, Datum, Metadata,
ReduceStreamAsyncServer, ReduceStreamer

class ReduceCounter(ReduceStreamer):
    def __init__(self, counter):
        self.counter = counter

    async def handler(
        self,
        keys: list[str],
        datums: AsyncIterable[Datum],
        output: NonBlockingIterator,
        md: Metadata,
    ):
        async for _ in datums:
            self.counter += 1
            if self.counter > 20:
                msg = f"counter:{self.counter}"
                await output.put(Message(str.encode(msg), keys=keys))
                self.counter = 0
        msg = f"counter:{self.counter}"
        await output.put(Message(str.encode(msg), keys=keys))

async def reduce_handler(
        keys: list[str],
        datums: AsyncIterable[Datum],
        output: NonBlockingIterator,
        md: Metadata,
    ):
    counter = 0
    async for _ in datums:
        counter += 1
        if counter > 20:
            msg = f"counter:{counter}"
            await output.put(Message(str.encode(msg), keys=keys))
            counter = 0
    msg = f"counter:{counter}"
    await output.put(Message(str.encode(msg), keys=keys))

if __name__ == "__main__":
    invoke = os.getenv("INVOKE", "func_handler")
    if invoke == "class":
        # Here we are using the class instance as the reducer_instance
        # which will be used to invoke the handler function.
        # We are passing the init_args for the class instance.
        grpc_server = ReduceStreamAsyncServer(ReduceCounter, init_args=(0,))
    else:
        # Here we are using the handler function directly as the reducer_instance.
        grpc_server = ReduceStreamAsyncServer(reduce_handler)
    grpc_server.start()
```

--------------------------------

### MapAsyncServer Usage Example

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Demonstrates how to define an asynchronous map handler and start the MapAsyncServer.

```python
from pynumaflow.mapper import Messages, Message, Datum, MapAsyncServer

async def async_map_handler(keys: list[str], datum: Datum) -> Messages:
    val = datum.value
    msg = (
        f"payload:{val.decode('utf-8')} "
        f"event_time:{datum.event_time} "
        f"watermark:{datum.watermark}"
    )
    return Messages(Message(value=msg.encode('utf-8'), keys=keys))

if __name__ == "__main__":
    grpc_server = MapAsyncServer(async_map_handler)
    grpc_server.start()
```

--------------------------------

### Source Transformer Example Invocation

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

A complete example demonstrating how to define a handler function and start a SourceTransformMultiProcServer.

```python
import datetime
import logging

from pynumaflow.sourcetransformer import Messages, Message, Datum, SourceTransformServer

# This is a simple User Defined Function example which receives a message,
# applies the following data transformation, and returns the message.
# If the message event time is before year 2022, drop the message
# with event time unchanged.
# If it's within year 2022, update the tag to "within_year_2022" and
# update the message event time to Jan 1st 2022.
# Otherwise, (exclusively after year 2022), update the tag to
# "after_year_2022" and update the


january_first_2022 = datetime.datetime.fromtimestamp(1640995200)
january_first_2023 = datetime.datetime.fromtimestamp(1672531200)


def my_handler(keys: list[str], datum: Datum) -> Messages:
    val = datum.value
    event_time = datum.event_time
    messages = Messages()

    if event_time < january_first_2022:
        logging.info("Got event time:%s, it is before 2022, so dropping", event_time)
        messages.append(Message.to_drop(event_time))
    elif event_time < january_first_2023:
        logging.info(
            "Got event time:%s, it is within year 2022, so
            forwarding to within_year_2022",
            event_time,
        )
        message = Message(
            value=val,
            event_time=january_first_2022,
            tags=["within_year_2022"],
        )
        messages.append(message)
    else:
        logging.info(
            "Got event time:%s, it is after year 2022, so forwarding to after_year_2022",
            event_time,
        )
        message = Message(
            value=val,
            event_time=january_first_2023,
            tags=["after_year_2022"],
        )
        messages.append(message)

    return messages

if __name__ == "__main__":
    grpc_server = SourceTransformMultiProcServer(
        source_transform_instance=my_handler,
        server_count=2,
    )
    grpc_server.start()
```

--------------------------------

### SinkServer invocation example

Source: https://numaproj.io/numaflow-python/latest/api/sinker

Demonstrates how to implement a sink handler using either a class-based or function-based approach and start the server.

```python
import os
import logging
from collections.abc import Iterator

from pynumaflow.sinker import Datum, Responses, Response, SinkServer
from pynumaflow.sinker import Sinker
from pynumaflow._constants import _LOGGER

logging.basicConfig(level=logging.INFO)

class UserDefinedSink(Sinker):
    def handler(self, datums: Iterator[Datum]) -> Responses:
        responses = Responses()
        for msg in datums:
            logging.info("User Defined Sink %s", msg.value.decode("utf-8"))
            responses.append(Response.as_success(msg.id))
        return responses

def udsink_handler(datums: Iterator[Datum]) -> Responses:
    responses = Responses()
    for msg in datums:
        logging.info("User Defined Sink %s", msg.value.decode("utf-8"))
        responses.append(Response.as_success(msg.id))
    return responses

if __name__ == "__main__":
    invoke = os.getenv("INVOKE", "func_handler")
    if invoke == "class":
        sink_handler = UserDefinedSink()
    else:
        sink_handler = udsink_handler
    grpc_server = SinkServer(sink_handler)
    grpc_server.start()
```

--------------------------------

### Implement and Run a User Defined Sink

Source: https://numaproj.io/numaflow-python/latest/api/sinker

Example showing how to define a sink using a class or function and start the gRPC server.

```python
import os
import logging
from collections.abc import AsyncIterable
from pynumaflow.sinker import Datum, Responses, Response, Sinker
from pynumaflow.sinker import SinkAsyncServer
from pynumaflow._constants import _LOGGER

logging.basicConfig(level=logging.INFO)

class UserDefinedSink(Sinker):
    async def handler(self, datums: AsyncIterable[Datum]) -> Responses:
        responses = Responses()
        async for msg in datums:
            logging.info("User Defined Sink %s", msg.value.decode("utf-8"))
            responses.append(Response.as_success(msg.id))
        return responses


async def udsink_handler(datums: AsyncIterable[Datum]) -> Responses:
    responses = Responses()
    async for msg in datums:
        logging.info("User Defined Sink %s", msg.value.decode("utf-8"))
        responses.append(Response.as_success(msg.id))
    return responses


if __name__ == "__main__":
    invoke = os.getenv("INVOKE", "func_handler")
    if invoke == "class":
        sink_handler = UserDefinedSink()
    else:
        sink_handler = udsink_handler
    grpc_server = SinkAsyncServer(sink_handler)
    grpc_server.start()
```

--------------------------------

### MapStreamAsyncServer Usage Example

Source: https://numaproj.io/numaflow-python/latest/api/mapstreamer

Demonstrates how to implement a map stream handler using either a class or a function and start the server.

```python
import os
from collections.abc import AsyncIterable
from pynumaflow.mapstreamer import Message, Datum, MapStreamAsyncServer, MapStreamer

class FlatMapStream(MapStreamer):
    async def handler(self, keys: list[str], datum: Datum) -> AsyncIterable[Message]:
        val = datum.value
        _ = datum.event_time
        _ = datum.watermark
        strs = val.decode("utf-8").split(",")

        if len(strs) == 0:
            yield Message.to_drop()
            return
        for s in strs:
            yield Message(str.encode(s))

async def map_stream_handler(_: list[str], datum: Datum) -> AsyncIterable[Message]:

    val = datum.value
    _ = datum.event_time
    _ = datum.watermark
    strs = val.decode("utf-8").split(",")

    if len(strs) == 0:
        yield Message.to_drop()
        return
    for s in strs:
        yield Message(str.encode(s))

if __name__ == "__main__":
    invoke = os.getenv("INVOKE", "func_handler")
    if invoke == "class":
        handler = FlatMapStream()
    else:
        handler = map_stream_handler
    grpc_server = MapStreamAsyncServer(handler)
    grpc_server.start()
```

--------------------------------

### Start Sink Server

Source: https://numaproj.io/numaflow-python/latest/api/sinker

Method to start the Async server.

```python
start()
```

```python
def start(self):
    """
    Starter function for the Async server class, need a separate caller
    so that all the async coroutines can be started from a single context
    """
    aiorun.run(self.aexec(), use_uvloop=True)
```

--------------------------------

### start()

Source: https://numaproj.io/numaflow-python/latest/api/reducer

Starts the ReduceAsyncServer instance. This method runs the server's async execution loop.

```APIDOC
## start()

### Description
Starts the Async Reduce Server. This method must be called to begin listening for requests.
```

--------------------------------

### Implement and Start BatchMapper

Source: https://numaproj.io/numaflow-python/latest/api/batchmapper

Example implementation of a BatchMapper class and its invocation using BatchMapAsyncServer.

```python
class Flatmap(BatchMapper):
   async def handler(
       self,
       datums: AsyncIterable[Datum],
   ) -> BatchResponses:
       batch_responses = BatchResponses()
       async for datum in datums:
           val = datum.value
           _ = datum.event_time
           _ = datum.watermark
           strs = val.decode("utf-8").split(",")
           batch_response = BatchResponse.from_id(datum.id)
           if len(strs) == 0:
               batch_response.append(Message.to_drop())
           else:
               for s in strs:
                   batch_response.append(Message(str.encode(s)))
           batch_responses.append(batch_response)

       return batch_responses
if __name__ == "__main__":
    grpc_server = BatchMapAsyncServer(Flatmap())
    grpc_server.start()
```

--------------------------------

### Example usage of AccumulatorAsyncServer

Source: https://numaproj.io/numaflow-python/latest/api/accumulator

A complete example showing how to implement a custom StreamSorter class inheriting from Accumulator and starting the server.

```python
import os
from collections.abc import AsyncIterable
from datetime import datetime

from pynumaflow.accumulator import Accumulator, AccumulatorAsyncServer
from pynumaflow.accumulator import Message, Datum
from pynumaflow.shared.asynciter import NonBlockingIterator

class StreamSorter(Accumulator):
    def __init__(self, counter):
        self.latest_wm = datetime.fromtimestamp(-1)
        self.sorted_buffer: list[Datum] = []

    async def handler(
        self,
        datums: AsyncIterable[Datum],
        output: NonBlockingIterator,
    ):
        async for _ in datums:
            # Process the datums and send output
            if datum.watermark and datum.watermark > self.latest_wm:
                self.latest_wm = datum.watermark
                await self.flush_buffer(output)

            self.insert_sorted(datum)

    def insert_sorted(self, datum: Datum):
        # Binary insert to keep sorted buffer in order
        left, right = 0, len(self.sorted_buffer)
        while left < right:
            mid = (left + right) // 2
            if self.sorted_buffer[mid].event_time > datum.event_time:
                right = mid
            else:
                left = mid + 1
        self.sorted_buffer.insert(left, datum)

    async def flush_buffer(self, output: NonBlockingIterator):
        i = 0
        for datum in self.sorted_buffer:
            if datum.event_time > self.latest_wm:
                break
            await output.put(Message.from_datum(datum))
            i += 1
        # Remove flushed items
        self.sorted_buffer = self.sorted_buffer[i:]


if __name__ == "__main__":
    grpc_server = AccumulatorAsyncServer(StreamSorter)
    grpc_server.start()
```

--------------------------------

### SinkServer.start()

Source: https://numaproj.io/numaflow-python/latest/api/sinker

Starts the gRPC server instance.

```APIDOC
## SinkServer.start()

### Description
Starts the gRPC server to begin listening for requests on the configured UNIX socket.
```

--------------------------------

### start()

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Starts the Synchronous gRPC server on the configured UNIX socket with the specified maximum number of threads.

```APIDOC
## start()

### Description
Starts the Synchronous gRPC server on the given UNIX socket with given max threads.

### Signature
`start() -> None`
```

--------------------------------

### Start SideInputServer

Source: https://numaproj.io/numaflow-python/latest/api/sideinput

The start method initializes the synchronous gRPC server on the configured UNIX socket.

```python
start()
```

```python
def start(self):
    """
    Starts the Synchronous gRPC server on the given UNIX socket with given max threads.
    """
    # Get the servicer instance based on the server type
    side_input_servicer = self.servicer

    _LOGGER.info(
        "Side Input GRPC Server listening on: %s with max threads: %s",
        self.sock_path,
        self.max_threads,
    )

    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[ContainerType.Sideinput]
    # Start the server
    sync_server_start(
        servicer=side_input_servicer,
        bind_address=self.sock_path,
        max_threads=self.max_threads,
        server_info_file=self.server_info_file,
        server_options=self._server_options,
        udf_type=UDFType.SideInput,
        server_info=serv_info,
    )
```

--------------------------------

### User Defined Function Example

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

A complete example showing how to define a handler that transforms messages based on event time and starts the server.

```python
import datetime
import logging
from pynumaflow.sourcetransformer import Messages, Message, Datum, SourceTransformServer

january_first_2022 = datetime.datetime.fromtimestamp(1640995200)
january_first_2023 = datetime.datetime.fromtimestamp(1672531200)


async def my_handler(keys: list[str], datum: Datum) -> Messages:
    val = datum.value
    event_time = datum.event_time
    messages = Messages()

    if event_time < january_first_2022:
        logging.info("Got event time:%s, it is before 2022, so dropping", event_time)
        messages.append(Message.to_drop(event_time))
    elif event_time < january_first_2023:
        logging.info(
            "Got event time:%s, it is within year 2022, so forwarding to within_year_2022",
            event_time,
        )
        messages.append(
            Message(value=val, event_time=january_first_2022,
                    tags=["within_year_2022"])
        )
    else:
        logging.info(
            "Got event time:%s, it is after year 2022, so forwarding to
            after_year_2022", event_time
        )
        messages.append(Message(value=val, event_time=january_first_2023,
                        tags=["after_year_2022"]))

    return messages


if __name__ == "__main__":
    grpc_server = SourceTransformAsyncServer(my_handler)
    grpc_server.start()
```

--------------------------------

### start()

Source: https://numaproj.io/numaflow-python/latest/api/mapstreamer

Starter function for the Async Map Stream server. It ensures all async coroutines are started from a single context.

```APIDOC
## start()

### Description
Starter function for the Async Map Stream server. It provides a separate caller to the aexec method so that all async coroutines can be started from a single context.

### Method
`start()`

### Usage
```python
server.start()
```
```

--------------------------------

### Install Numaflow Python SDK

Source: https://numaproj.io/numaflow-python/latest

Install the package using pip.

```bash
pip install pynumaflow
```

--------------------------------

### start()

Source: https://numaproj.io/numaflow-python/latest/api/accumulator

The start method initializes the Async server class. It acts as a caller to ensure all asynchronous coroutines are initiated within a single context.

```APIDOC
## start()

### Description
Starter function for the Async server class. It ensures that all async coroutines are started from a single context.

### Signature
`start(self)`
```

--------------------------------

### Start Async Server

Source: https://numaproj.io/numaflow-python/latest/api/reducestreamer

Method to initiate the server and run async coroutines.

```python
def start(self):
    """
    Starter function for the Async server class, need a separate caller
    so that all the async coroutines can be started from a single context
    """
    _LOGGER.info(
        "Starting Async Reduce Stream Server",
    )
    aiorun.run(self.aexec(), use_uvloop=True)
```

--------------------------------

### ReduceAsyncServer Start Method

Source: https://numaproj.io/numaflow-python/latest/api/reducer

Starts the asynchronous server using the configured event loop.

```python
start()
```

```python
def start(self):
    """
    Starter function for the Async server class, need a separate caller
    so that all the async coroutines can be started from a single context
    """
    _LOGGER.info(
        "Starting Async Reduce Server",
    )
    aiorun.run(self.aexec(), use_uvloop=True)
```

--------------------------------

### SourceTransformServer.start

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

Starts the server to process source transformations.

```APIDOC
## SourceTransformServer.start

### Description
Starts the synchronous server instance to begin processing source transformation requests.

### Signature
`start(self)`
```

--------------------------------

### BatchMapAsyncServer.start

Source: https://numaproj.io/numaflow-python/latest/api/batchmapper

Starts the Async Batch Map server.

```APIDOC
## start()

### Description
Starter function for the Async Batch Map server. It uses a separate caller to the aexec so that all the async coroutines can be started from a single context.
```

--------------------------------

### Start Async Server

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Initiates the async server with specified socket path, thread limits, and server information.

```python
# Start the async server
await start_async_server(
    server_async=server_new,
    sock_path=self.sock_path,
    max_threads=self.max_threads,
    cleanup_coroutines=list(),
    server_info_file=self.server_info_file,
    server_info=serv_info,
)
```

--------------------------------

### Start SourceTransformServer

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

Method to start the synchronous gRPC server on a UNIX socket.

```python
start()
```

```python
def start(self):
    """
    Starts the Synchronous gRPC server on the given UNIX socket with given max threads.
    """
    _LOGGER.info(
        "Sync GRPC Server listening on: %s with max threads: %s",
        self.sock_path,
        self.max_threads,
    )
    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[
        ContainerType.Sourcetransformer
    ]
    # Start the sync server
    sync_server_start(
        servicer=self.servicer,
        bind_address=self.sock_path,
        max_threads=self.max_threads,
        server_info_file=self.server_info_file,
        server_options=self._server_options,
        udf_type=UDFType.SourceTransformer,
        server_info=serv_info,
    )
```

--------------------------------

### MapAsyncServer.start

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Starts the asynchronous gRPC server.

```APIDOC
## MapAsyncServer.start()

### Description
Starter function for the Async server class. It runs the server in a single context to ensure all async coroutines are started correctly.
```

--------------------------------

### MapServer.start()

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Starts the gRPC server to begin processing requests.

```APIDOC
## start()

### Description
Starts the gRPC server. The number of servers started is determined by the number of CPUs or the user-defined server_count, capped at 2 * CPU count.
```

--------------------------------

### Initialize Datum instance

Source: https://numaproj.io/numaflow-python/latest/api/reducer

Example usage for creating a Datum object with required event metadata.

```python
from pynumaflow.reducer import Datum
from datetime import datetime, timezone

d = Datum(
      keys=["test_key"],
      value=b"test_mock_message",
      event_time=datetime.fromtimestamp(1662998400, timezone.utc),
      watermark=datetime.fromtimestamp(1662998460, timezone.utc),
      headers={"key1": "value1", "key2": "value2"},
   )
```

--------------------------------

### Start Multi-process Mapper Server

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Initializes and starts the gRPC server with a specified number of threads and processes.

```python
def start(self) -> None:
    """
    Starts the N grpc servers gRPC serves on the with given max threads.
    Here N = The number of CPUs or the value of the parameter server_count
    defined by the user. The max value is capped to 2 * CPU count.
    """

    # Create the server info file
    server_info = ServerInfo.get_default_server_info()
    server_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[ContainerType.Mapper]
    server_info.metadata = get_metadata_env(envs=METADATA_ENVS)
    # Add the MULTIPROC metadata using the number of servers to use
    server_info.metadata[MULTIPROC_KEY] = str(self._process_count)
    # Add the MAP_MODE metadata to the server info for the correct map mode
    server_info.metadata[MAP_MODE_KEY] = MapMode.UnaryMap

    # Start the multiproc server
    start_multiproc_server(
        max_threads=self.max_threads,
        servicer=self.servicer,
        process_count=self._process_count,
        server_info_file=self.server_info_file,
        server_options=self._server_options,
        udf_type=UDFType.Map,
        server_info=server_info,
    )
```

--------------------------------

### Start Async gRPC Server

Source: https://numaproj.io/numaflow-python/latest/api/batchmapper

Initializes and starts the asynchronous gRPC server on a UNIX socket. This method must be called within the appropriate event loop context.

```python
aexec()
```

```python
async def aexec(self):
    """
    Starts the Async gRPC server on the given UNIX socket with
    given max threads.
    """
    # As the server is async, we need to create a new server instance in the
    # same thread as the event loop so that all the async calls are made in the
    # same context
    # Create a new async server instance and add the servicer to it
    server = grpc.aio.server(options=self._server_options)
    server.add_insecure_port(self.sock_path)
    map_pb2_grpc.add_MapServicer_to_server(
        self.servicer,
        server,
    )
    _LOGGER.info("Starting Batch Map Server")
    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[ContainerType.Mapper]
    # Add the MAP_MODE metadata to the server info for the correct map mode
    serv_info.metadata[MAP_MODE_KEY] = MapMode.BatchMap

    # Start the async server
    await start_async_server(
        server_async=server,
        sock_path=self.sock_path,
        max_threads=self.max_threads,
        cleanup_coroutines=list(),
        server_info_file=self.server_info_file,
        server_info=serv_info,
    )
```

--------------------------------

### Example usage of Datum

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

Demonstrates how to instantiate a Datum object with required and optional fields.

```python
from pynumaflow.sourcetransformer import Datum
from datetime import datetime, timezone

d = Datum(
        keys=["test_key"],
        value=b"test_mock_message",
        event_time=datetime.fromtimestamp(1662998400, timezone.utc),
        watermark=datetime.fromtimestamp(1662998460, timezone.utc),
        headers={"key1": "value1", "key2": "value2"},
   )
```

--------------------------------

### SourceTransformAsyncServer.start

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

Starts the asynchronous server to process source transformations.

```APIDOC
## SourceTransformAsyncServer.start

### Description
Starts the asynchronous server instance to begin processing source transformation requests.

### Signature
`start(self)`
```

--------------------------------

### MapAsyncServer Start Method

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Starts the asynchronous server using a separate caller to manage the event loop context.

```python
def start(self) -> None:
    """
    Starter function for the Async server class, need a separate caller
    so that all the async coroutines can be started from a single context
    """
    aiorun.run(self.aexec(), use_uvloop=True)
```

--------------------------------

### Start Async Batch Map Server

Source: https://numaproj.io/numaflow-python/latest/api/batchmapper

Method to initiate the server, ensuring all async coroutines run within a single context.

```python
start()
```

```python
def start(self):
    """
    Starter function for the Async Batch Map server, we need a separate caller
    to the aexec so that all the async coroutines can be started from a single context
    """
    aiorun.run(self.aexec(), use_uvloop=True)
```

--------------------------------

### SourceTransformMultiProcServer.start

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

Starts the multi-processing server to process source transformations.

```APIDOC
## SourceTransformMultiProcServer.start

### Description
Starts the multi-processing server instance to handle source transformation requests in parallel.

### Signature
`start(self)`
```

--------------------------------

### SinkServer.start

Source: https://numaproj.io/numaflow-python/latest/api/sinker

Starts the Synchronous gRPC server on the configured UNIX socket.

```APIDOC
## SinkServer.start

### Description
Starts the Synchronous gRPC server on the given UNIX socket with the configured max threads.
```

--------------------------------

### SideInputServer.start

Source: https://numaproj.io/numaflow-python/latest/api/sideinput

Starts the gRPC server on the configured UNIX socket.

```APIDOC
## start()

### Description
Starts the Synchronous gRPC server on the given UNIX socket with the configured max threads.
```

--------------------------------

### aexec()

Source: https://numaproj.io/numaflow-python/latest/api/mapstreamer

Starts the Async gRPC server on the configured UNIX socket.

```APIDOC
## aexec()

### Description
Starts the Async gRPC server on the given UNIX socket with the configured max threads. It initializes the server instance, adds the MapServicer, and begins the async server execution.

### Method
`async aexec()`

### Usage
```python
await server.aexec()
```
```

--------------------------------

### Start Synchronous gRPC Server

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Starts the gRPC server on a specified UNIX socket. Requires configuration of server info and metadata before invocation.

```python
start() -> None
```

```python
def start(self) -> None:
    """
    Starts the Synchronous gRPC server on the given UNIX socket with given max threads.
    """
    _LOGGER.info(
        "Sync GRPC Server listening on: %s with max threads: %s",
        self.sock_path,
        self.max_threads,
    )

    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[ContainerType.Mapper]
    # Add the MAP_MODE metadata to the server info for the correct map mode
    serv_info.metadata[MAP_MODE_KEY] = MapMode.UnaryMap
    # Start the server
    sync_server_start(
        servicer=self.servicer,
        bind_address=self.sock_path,
        max_threads=self.max_threads,
        server_info_file=self.server_info_file,
        server_options=self._server_options,
        udf_type=UDFType.Map,
        server_info=serv_info,
    )
```

--------------------------------

### Start Async Map Stream Server

Source: https://numaproj.io/numaflow-python/latest/api/mapstreamer

Initializes the Async Map Stream server by running the execution coroutine in a single context.

```python
def start(self):
    """
    Starter function for the Async Map Stream server, we need a separate caller
    to the aexec so that all the async coroutines can be started from a single context
    """
    aiorun.run(self.aexec(), use_uvloop=True)
```

--------------------------------

### SourceTransformServer.start()

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

Starts the synchronous gRPC server on the configured UNIX socket with the specified thread limits.

```APIDOC
## SourceTransformServer.start()

### Description
Starts the Synchronous gRPC server on the given UNIX socket with given max threads. This method initializes the server with the provided handler and begins listening for incoming requests.

### Method
Method: `start()`

### Parameters
None (Uses instance configuration provided during initialization).

### Usage Example
```python
# Initialize the server with a handler
grpc_server = SourceTransformServer(my_handler)

# Start the server
grpc_server.start()
```
```

--------------------------------

### Async Server Implementation

Source: https://numaproj.io/numaflow-python/latest/api/reducer

Implementation of the aexec method which configures and starts the gRPC server.

```python
async def aexec(self):
    """
    Starts the Async gRPC server on the given UNIX socket with
    given max threads.
    """
    # As the server is async, we need to create a new server instance in the
    # same thread as the event loop so that all the async calls are made in the
    # same context
    # Create a new async server instance and add the servicer to it
    server = grpc.aio.server(options=self._server_options)
    server.add_insecure_port(self.sock_path)
    reduce_servicer = self.servicer
    reduce_pb2_grpc.add_ReduceServicer_to_server(reduce_servicer, server)

    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[ContainerType.Reducer]
    # Start the async server
    await start_async_server(
        server_async=server,
        sock_path=self.sock_path,
        max_threads=self.max_threads,
        cleanup_coroutines=list(),
        server_info_file=self.server_info_file,
        server_info=serv_info,
    )
```

--------------------------------

### Start the Multi-Processing Server

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

Method to initiate the gRPC server processes. It uses the configured server options and servicer instance.

```python
start()
```

```python
def start(self):
    """
    Starts the N gRPC servers on the given socket path with given max threads.
    Here N = The number of CPUs or the value of the parameter `server_count`
    defined by the user. The max value is capped to 2 * CPU count.
    """

    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[
        ContainerType.Sourcetransformer
    ]
    start_multiproc_server(
        max_threads=self.max_threads,
        servicer=self.servicer,
        process_count=self._process_count,
        server_info_file=self.server_info_file,
        server_options=self._server_options,
        udf_type=UDFType.SourceTransformer,
        server_info=serv_info,
    )
```

--------------------------------

### aexec()

Source: https://numaproj.io/numaflow-python/latest/api/batchmapper

Starts the Async gRPC server on the given UNIX socket with the configured maximum number of threads.

```APIDOC
## aexec()

### Description
Starts the Async gRPC server on the given UNIX socket with given max threads. This method initializes the server instance, adds the MapServicer, and begins listening for requests.

### Method
async def aexec(self)

### Usage
This method is called to initiate the server lifecycle within the BatchMapper context.
```

--------------------------------

### MapAsyncServer aexec Method

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Initializes and starts the asynchronous gRPC server on the specified UNIX socket.

```python
async def aexec(self) -> None:
    """
    Starts the Async gRPC server on the given UNIX socket with
    given max threads.
    """

    # As the server is async, we need to create a new server instance in the
    # same thread as the event loop so that all the async calls are made in the
    # same context

    server_new = grpc.aio.server(options=self._server_options)
    server_new.add_insecure_port(self.sock_path)
    map_pb2_grpc.add_MapServicer_to_server(self.servicer, server_new)

    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[ContainerType.Mapper]
    # Add the MAP_MODE metadata to the server info for the correct map mode
    serv_info.metadata[MAP_MODE_KEY] = MapMode.UnaryMap
```

--------------------------------

### SourceTransformServer Initialization

Source: https://numaproj.io/numaflow-python/latest/api/sourcetransformer

Internal configuration of the SourceTransformServer instance, including gRPC message size limits and servicer setup.

```python
    self.sock_path = f"unix://{sock_path}"
    self.max_threads = min(max_threads, MAX_NUM_THREADS)
    self.max_message_size = max_message_size
    self.server_info_file = server_info_file

    self.source_transform_instance = source_transform_instance

    self._server_options = [
        ("grpc.max_send_message_length", self.max_message_size),
        ("grpc.max_receive_message_length", self.max_message_size),
    ]
    self.servicer = SourceTransformServicer(handler=source_transform_instance)
```

--------------------------------

### SideInputServer

Source: https://numaproj.io/numaflow-python/latest/api/sideinput

Class used to initialize and start a gRPC server for Side Input UDFs.

```APIDOC
## SideInputServer(side_input_instance, sock_path, max_message_size, max_threads, ...)

### Description
Initializes a new Side Input Server instance to host the provided side input logic.

### Parameters
- **side_input_instance** (RetrieverCallable) - Required - The side input instance to be used for Side Input UDF.
- **sock_path** (str) - Optional - The UNIX socket path to be used for the server.
- **max_message_size** (int) - Optional - The max message size in bytes the server can receive and send.
- **max_threads** (int) - Optional - The max number of threads to be spawned.
```

--------------------------------

### Start the gRPC Sink Server

Source: https://numaproj.io/numaflow-python/latest/api/sinker

Initiates the synchronous gRPC server on the configured UNIX socket.

```python
def start(self):
    """
    Starts the Synchronous gRPC server on the
    given UNIX socket with given max threads.
    """
    _LOGGER.info(
        "Sync GRPC Sink listening on: %s with max threads: %s",
        self.sock_path,
        self.max_threads,
    )
    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[ContainerType.Sinker]
    # Start the server
    sync_server_start(
        servicer=self.servicer,
        bind_address=self.sock_path,
        max_threads=self.max_threads,
        server_info_file=self.server_info_file,
        server_options=self._server_options,
        udf_type=UDFType.Sink,
        server_info=serv_info,
    )
```

--------------------------------

### Initialize PartitionsResponse

Source: https://numaproj.io/numaflow-python/latest/api/sourcer

Create a response indicating available partitions.

```python
def __init__(self, partitions: list[int]):
    if not isinstance(partitions, list):
        raise TypeError(f"Wrong data type: {type(partitions)} for Partition.partitions")
    self._partitions = partitions
```

--------------------------------

### Execute Async gRPC Server

Source: https://numaproj.io/numaflow-python/latest/api/mapstreamer

Starts the gRPC server on a UNIX socket and configures the servicer with appropriate metadata.

```python
async def aexec(self):
    """
    Starts the Async gRPC server on the given UNIX socket with
    given max threads.
    """
    # As the server is async, we need to create a new server instance in the
    # same thread as the event loop so that all the async calls are made in the
    # same context
    # Create a new async server instance and add the servicer to it
    server = grpc.aio.server(options=self._server_options)
    server.add_insecure_port(self.sock_path)
    map_pb2_grpc.add_MapServicer_to_server(
        self.servicer,
        server,
    )
    _LOGGER.info("Starting Map Stream Server")
    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[ContainerType.Mapper]
    # Add the MAP_MODE metadata to the server info for the correct map mode
    serv_info.metadata[MAP_MODE_KEY] = MapMode.StreamMap

    # Start the async server
    await start_async_server(
        server_async=server,
        sock_path=self.sock_path,
        max_threads=self.max_threads,
        cleanup_coroutines=list(),
        server_info_file=self.server_info_file,
        server_info=serv_info,
    )
```

--------------------------------

### Execute Async gRPC Server

Source: https://numaproj.io/numaflow-python/latest/api/accumulator

Starts the gRPC server on a UNIX socket with specified thread limits.

```python
aexec()
```

```python
async def aexec(self):
    """
    Starts the Async gRPC server on the given UNIX socket with
    given max threads.
    """
    # As the server is async, we need to create a new server instance in the
    # same thread as the event loop so that all the async calls are made in the
    # same context
    # Create a new async server instance and add the servicer to it
    server = grpc.aio.server(options=self._server_options)
    server.add_insecure_port(self.sock_path)
    accumulator_pb2_grpc.add_AccumulatorServicer_to_server(self.servicer, server)

    serv_info = ServerInfo.get_default_server_info()
    serv_info.minimum_numaflow_version = MINIMUM_NUMAFLOW_VERSION[ContainerType.Accumulator]
    await start_async_server(
        server_async=server,
        sock_path=self.sock_path,
        max_threads=self.max_threads,
        cleanup_coroutines=list(),
        server_info_file=self.server_info_file,
        server_info=serv_info,
    )
```

--------------------------------

### Initialize and Invoke MapServer

Source: https://numaproj.io/numaflow-python/latest/api/mapper

Demonstrates how to instantiate a MapServer using either a class-based or function-based handler.

```python
from pynumaflow.mapper import Messages, Message, Datum, MapServer, Mapper

class MessageForwarder(Mapper):
    def handler(self, keys: list[str], datum: Datum) -> Messages:
        val = datum.value
        _ = datum.event_time
        _ = datum.watermark
        return Messages(Message(value=val, keys=keys))

def my_handler(keys: list[str], datum: Datum) -> Messages:
    val = datum.value
    _ = datum.event_time
    _ = datum.watermark
    return Messages(Message(value=val, keys=keys))


if __name__ == "__main__":
    # Use the class based approach or function based handler based on the env variable
    # Both can be used and passed directly to the server class

    invoke = os.getenv("INVOKE", "func_handler")
    if invoke == "class":
        handler = MessageForwarder()
    else:
        handler = my_handler
    grpc_server = MapServer(handler)
    grpc_server.start()
```

```python
def __init__(
    self,
    mapper_instance: MapSyncCallable,
    sock_path=MAP_SOCK_PATH,
    max_message_size=MAX_MESSAGE_SIZE,
    max_threads=NUM_THREADS_DEFAULT,
    server_info_file=MAP_SERVER_INFO_FILE_PATH,
):
    self.sock_path = f"unix://{sock_path}"
    self.max_threads = min(max_threads, MAX_NUM_THREADS)
    self.max_message_size = max_message_size
    self.server_info_file = server_info_file

    self.mapper_instance = mapper_instance

    self._server_options = [
        ("grpc.max_send_message_length", self.max_message_size),
        ("grpc.max_receive_message_length", self.max_message_size),
    ]
    # Get the servicer instance for the sync server
    self.servicer = SyncMapServicer(handler=mapper_instance)
```

--------------------------------

### Build and Test Locally

Source: https://numaproj.io/numaflow-python/latest

Commands for building the project, running tests, linting, and setting up development hooks.

```makefile
make setup
```

```makefile
make test
```

```makefile
make lint
```

```bash
pre-commit install
```