### Install Docker Compose

Source: https://github.com/avito-tech/aqueduct/blob/main/tests/benchmarks/bench_service/README.md

Downloads and installs Docker Compose version 1.29.2, which is required for managing the benchmark environment containers.

```shell
sudo curl -L "https://github.com/docker/compose/releases/download/1.29.2/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
sudo chmod +x /usr/local/bin/docker-compose
docker-compose --version
```

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

### Manage Benchmark Containers

Source: https://github.com/avito-tech/aqueduct/blob/main/tests/benchmarks/bench_service/README.md

Commands to start the benchmark container environment or stop and remove all associated containers.

```shell
docker-compose up -d maas-bench
docker-compose down
```

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

### BaseTaskHandler - Processing Step Handler

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Illustrates how to implement processing logic using BaseTaskHandler, including initialization in `on_start()` and task processing in `handle()`. Shows examples for CPU-bound and batched GPU-bound computations.

```APIDOC
## BaseTaskHandler - Processing Step Handler

### Description
`BaseTaskHandler` is the base class for implementing processing logic. Handlers define `on_start()` for initialization in worker processes and `handle()` for processing tasks. The handler runs in a separate OS process, isolating memory-heavy models from the main process.

### Method
```python
from aqueduct import BaseTaskHandler, BaseTask
from typing import Optional

class Task(BaseTask):
    def __init__(self, number: int):
        super().__init__()
        self.number = number
        self.result: Optional[int] = None

class HeavyComputationHandler(BaseTaskHandler):
    """Handler that loads a model once and processes multiple tasks."""

    def __init__(self):
        self._model = None

    def on_start(self):
        """Called once when worker process starts.
        Load heavy models here to keep parent process memory low."""
        # Simulate loading a heavy ML model
        self._model = lambda x: sum(i * i for i in range(x))
        print(f"Model loaded in process")

    def handle(self, *tasks: Task):
        """Process one or more tasks (batching support).
        Called each time tasks are available in the queue."""
        for task in tasks:
            task.result = self._model(task.number)

# Handler with batched processing for GPU efficiency
class BatchedMLHandler(BaseTaskHandler):
    def __init__(self):
        self._model = None

    def on_start(self):
        import numpy as np
        self._model = lambda batch: np.ones(len(batch))  # Simulated predictions

    def handle(self, *tasks: Task):
        # Process all tasks as a batch for better GPU utilization
        inputs = [task.number for task in tasks]
        results = self._model(inputs)
        for task, result in zip(tasks, results):
            task.result = float(result)
```
```

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

### Install Aqueduct with Optional Extras

Source: https://github.com/avito-tech/aqueduct/blob/main/README.rst

Installation command for Aqueduct including optional dependencies like numpy and aiohttp support.

```shell
pip install aqueduct[numpy,aiohttp]
```

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

### Execute Aqueduct Application

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/concurrency.rst

Provides the command-line instructions for starting the application. It highlights the migration from a standard Uvicorn execution to a Gunicorn configuration.

```bash
# Old command
uvicorn main:app

# New command
gunicorn -c gunicorn_config.py
```

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

### Install Aqueduct via Pip

Source: https://github.com/avito-tech/aqueduct/blob/main/README.rst

Standard installation command for the Aqueduct library using the Python package manager.

```shell
pip install aqueduct
```

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

### Install Uvicorn Worker for Gunicorn

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/concurrency.rst

Installs Uvicorn, a high-performance ASGI server, which can be used as a worker class for Gunicorn to handle asynchronous web requests efficiently.

```bash
pip install uvicorn
```

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

### Implementing Priority Queues in Aqueduct

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/priority_queues.rst

This example demonstrates how to set up and use priority queues in Aqueduct. It involves defining a task, a handler, and a flow with multiple priority levels. Tasks are then processed with different priorities to show the prioritization mechanism.

```python
import asyncio
from typing import Optional

from aqueduct.task import BaseTask
from aqueduct.task_handler import BaseTaskHandler
from aqueduct.flow import Flow, FlowStep

class Task(BaseTask):
    result: Optional[int]
    def __init__(self):
        super().__init__()
        self.result = None

class ProcessorHandler(BaseTaskHandler):
    def handle(self, *tasks: Task):
        for task in tasks:
            task.result = 42

def get_flow() -> Flow:
    return Flow(
        FlowStep(ProcessorHandler()),
        metrics_enabled=False,
        
        # Set how many priorities you are gonna need
        queue_priorities=2,
    )

async def main():
    flow = get_flow()
    flow.start()

    normal_task = Task()

    important_task = Task()

    # Task priority should be between 0 and queue_priorities - 1
    important_task.set_priority(1)

    normal_future = asyncio.create_task(flow.process(normal_task))
    normal_future.add_done_callback(lambda _: print('normal task finished'))

    important_future = asyncio.create_task(flow.process(important_task))
    important_future.add_done_callback(lambda _: print('important task finished'))
    await asyncio.gather(normal_future, important_future)

if __name__ == '__main__':
    asyncio.run(main())
```

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

### Python Example Handler for Task Result Calculation

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/fundamentals.rst

This Python example demonstrates a `SumHandler` inheriting from `BaseTaskHandler`. It initializes a model in `on_start` and processes tasks in the `handle` method, where it applies the model to a `number` attribute of each task and stores the result in `task.result`.

```python
class SumHandler(BaseTaskHandler):
    def __init__(self):
        self._model = None

    def on_start(self):
        self._model = MyModel()

    def handle(self, *tasks: Task):
        for task in tasks:
            task.result = self._model.process(task.number)
```

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

### Implement Batch Processing in Aqueduct

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/batching.rst

This example demonstrates how to define a custom task handler that processes batches of tasks using NumPy. It shows the configuration of FlowStep with a specific batch_size and the execution flow using asyncio.

```python
import asyncio
import time
from typing import List
import numpy as np
from aqueduct.flow import Flow, FlowStep
from aqueduct.handler import BaseTaskHandler
from aqueduct.task import BaseTask

TASKS_BATCH_SIZE = 20

class ArrayFieldTask(BaseTask):
	def __init__(self, array: np.array, *args, **kwargs):
		super().__init__(*args, **kwargs)
		self.array = array
		self.result = None

class CatDetectorHandler(BaseTaskHandler):
	def handle(self, *tasks: ArrayFieldTask):
		images = np.array([task.array for task in tasks])
		predicts = CatDetector().predict(images)
		for task, predict in zip(tasks, predicts):
			task.result = predict

flow_with_batch_handler = Flow(FlowStep(CatDetectorHandler(), batch_size=TASKS_BATCH_SIZE))
flow_with_batch_handler.start()
```

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

### Implementing CPU-bound task processing with Aqueduct

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/example.rst

This example defines a custom model, a task container, and a handler to process CPU-intensive operations in a separate process. It integrates these components into an aiohttp application to handle incoming requests asynchronously.

```python
import asyncio

from aiohttp import web
from aqueduct import Flow, FlowStep, BaseTaskHandler, BaseTask


class MyModel:
    """This is an example of a CPU-bound model"""

    def process(self, number):
        return sum(i * i for i in range(number))

class Task(BaseTask):
    """Container for sending arguments to the model."""
    def __init__(self, number):
        super().__init__()
        self.number = number
        self.sum = None  # result will be here

class SumHandler(BaseTaskHandler):
    """When using Aqueduct, we need to wrap your model."""
    def __init__(self):
        self._model = None

    def on_start(self):
        """Executed in a child process, so the parent process does not consume additional memory."""
        self._model = MyModel()

    def handle(self, *tasks: Task):
        """List of tasks because it can be batching."""
        for task in tasks:
            task.sum = self._model.process(task.number)


class SumView(web.View):
    """Simple aiohttp-view handler"""

    async def post(self):
        number = await self.request.read()
        task = Task(int(number))
        await self.request.app['flow'].process(task)
        return web.json_response(data={'result': task.sum})


def prepare_app() -> web.Application:
    app = web.Application()

    app['flow'] = Flow(
        FlowStep(SumHandler()),
    )
    app.router.add_post('/sum', SumView)

    app['flow'].start()
    return app


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    web.run_app(prepare_app(), loop=loop)
```

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

### Utilize Shared Memory for Large Data

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Explains how to use shared memory to transfer large data structures between processes without serialization. Includes examples for sharing existing data and streaming data directly from a request.

```python
class ImageProcessor(BaseTaskHandler):
    def handle(self, *tasks: ImageTask):
        for task in tasks:
            task.processed_image = task.image[::-1]
            task.share_value('processed_image')

# Streaming example
async def handle_upload(request):
    task = ImageTask()
    await task.share_value_with_data(
        field_name='image',
        content=request.content,
        size=request.content_length
    )
    await request.app['flow'].process(task)
    return task.processed_image
```

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

### Implement a CPU-Bound Task Handler with Aqueduct

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/fundamentals.rst

Provides an example of creating a custom task handler for Aqueduct by inheriting from `BaseTaskHandler`. This class is designed to wrap CPU-bound logic, such as model inference, preparing it for execution within an Aqueduct processing pipeline.

```python
from aqueduct import BaseTaskHandler

class MyModel:
    """This is an example of a CPU-bound model"""

    def process(self, image):
        """do something with image on CPU"""
        pass

class ImageHandler(BaseTaskHandler):
    """When using Aqueduct, we need to wrap your model."""
    def __init__(self):
        # Initialize your model here
        self.model = MyModel()
```

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

### Aqueduct Image Processing Flow with aiohttp

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/share_memory.rst

This Python code sets up an Aqueduct flow to handle image processing tasks. It defines a model, a task container, a task handler, and an aiohttp web view. The handler uses shared memory to process images efficiently. The flow is started, and an aiohttp server is run to expose an endpoint for image manipulation.

```python
import asyncio

from aiohttp import web
from aqueduct import Flow, FlowStep, BaseTaskHandler, BaseTask


class MyModel:
    """This is an example of a CPU-bound model"""

    def process(self, image):
        """do something with image on CPU"""
        pass

class Task(BaseTask):
    """Container for sending arguments to the model."""
    def __init__(self, image=None):
        super().__init__()
        self.image = image
        self.image_processed = None  # result will be here

class ImageHandler(BaseTaskHandler):
    """When using Aqueduct, we need to wrap your model."""
    def __init__(self):
        self._model = None

    def on_start(self):
        """Executed in a child process, so the parent process does not consume additional memory."""
        self._model = MyModel()

    def handle(self, *tasks: Task):
        """List of tasks because it can be batching."""
        for task in tasks:
            task.image_processed = self._model.process(task.image)


class ImageView(web.View):
    """Simple aiohttp-view handler"""

    async def post(self):
        task = Task()
        await task.share_value_with_data(
            field_name='image', 
            content=self.request.content,
            size=self.request.content_length,
        )

        await self.request.app['flow'].process(task)
        return web.json_response(data={'result': task.image_processed})


def prepare_app() -> web.Application:
    app = web.Application()

    app['flow'] = Flow(
        FlowStep(ImageHandler()),
    )
    app.router.add_post('/image_manipulation', ImageView)

    app['flow'].start()
    return app


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    web.run_app(prepare_app(), loop=loop)

```

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

### Prepare Benchmark Models

Source: https://github.com/avito-tech/aqueduct/blob/main/tests/benchmarks/bench_service/README.md

Creates the necessary directory structure and executes the model download script to optimize service startup times.

```shell
mkdir maas/data/models && cd maas/data && sh get_models.sh
```

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

### Orchestrate Multi-Step Pipelines with Flow

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Shows how to initialize a Flow, define custom task handlers, and manage task submission. It covers both single-task processing and concurrent execution using asyncio.

```python
import asyncio
from aqueduct import Flow, FlowStep, BaseTaskHandler, BaseTask

class DoubleHandler(BaseTaskHandler):
    def handle(self, *tasks: Task):
        for task in tasks:
            task.result = task.value * 2

async def main():
    flow = Flow(
        FlowStep(DoubleHandler(), nprocs=2),
        metrics_enabled=True,
        queue_size=100
    )
    flow.start(timeout=30)
    task = Task(5)
    await flow.process(task, timeout_sec=5.0)
    await flow.stop()
```

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

### Implement Processing Logic with BaseTaskHandler

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Shows how to implement custom handlers using BaseTaskHandler. The on_start method is used for heavy initialization like model loading, while handle supports batch processing for improved efficiency.

```python
from aqueduct import BaseTaskHandler, BaseTask
from typing import Optional

class Task(BaseTask):
    def __init__(self, number: int):
        super().__init__()
        self.number = number
        self.result: Optional[int] = None

class HeavyComputationHandler(BaseTaskHandler):
    """Handler that loads a model once and processes multiple tasks."""

    def __init__(self):
        self._model = None

    def on_start(self):
        """Called once when worker process starts."""
        self._model = lambda x: sum(i * i for i in range(x))
        print(f"Model loaded in process")

    def handle(self, *tasks: Task):
        for task in tasks:
            task.result = self._model(task.number)

class BatchedMLHandler(BaseTaskHandler):
    def __init__(self):
        self._model = None

    def on_start(self):
        import numpy as np
        self._model = lambda batch: np.ones(len(batch))

    def handle(self, *tasks: Task):
        inputs = [task.number for task in tasks]
        results = self._model(inputs)
        for task, result in zip(tasks, results):
            task.result = float(result)
```

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

### FlowStep - Pipeline Step Configuration

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Explains how to configure pipeline steps using `FlowStep`, including setting parallelism, batch sizes, and timeouts for efficient task processing.

```APIDOC
## FlowStep - Pipeline Step Configuration

### Description
`FlowStep` wraps a handler with configuration for parallelism, batching, and conditional execution. It defines how many worker processes run the handler, batch sizes for efficient processing, and optional conditions for task routing.

### Method
```python
import operator
from aqueduct import FlowStep, BaseTaskHandler, BaseTask

class PreprocessHandler(BaseTaskHandler):
    def handle(self, *tasks):
        for task in tasks:
            task.preprocessed = True

class ModelHandler(BaseTaskHandler):
    def on_start(self):
        self._model = lambda x: 0.95  # Simulated model

    def handle(self, *tasks):
        for task in tasks:
            task.score = self._model(task.data)

# Basic step with single worker
step1 = FlowStep(PreprocessHandler())

# Step with multiple parallel workers
step2 = FlowStep(
    handler=ModelHandler(),
    nprocs=4,  # Run 4 parallel worker processes
)

# Step with batching for GPU efficiency
step3 = FlowStep(
    handler=ModelHandler(),
    nprocs=2,
    batch_size=32,        # Collect up to 32 tasks before processing
    batch_timeout=0.1,    # Max wait time (seconds) to fill batch
)
```
```

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

### Configure Metrics and Monitoring

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Provides an overview of how to set up metrics exporters in Aqueduct to track system performance, such as transfer times and queue sizes.

```python
from aqueduct import Flow, FlowStep, BaseTaskHandler, BaseTask
from aqueduct.metrics import ToStatsDMetricsExporter

class Task(BaseTask):
    def __init__(self):
        super().__init__()
        self.result = None

class Handler(BaseTaskHandler):
    def handle(self, *tasks):
        for task in tasks:
            task.result = 42
```

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

### BaseTask - Task Container Class

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Demonstrates how to create and manage tasks using the BaseTask class, including setting priorities and utilizing shared memory for efficient data transfer.

```APIDOC
## BaseTask - Task Container Class

### Description
`BaseTask` is the base class for creating task containers that carry input data through the pipeline and store results. Tasks automatically get unique IDs, support priorities, expiration times, and shared memory fields for efficient data transfer between processes.

### Method
```python
from typing import Optional
from aqueduct import BaseTask

class ImageClassificationTask(BaseTask):
    """Task for image classification with shared memory support."""

    def __init__(self, image_data: bytes = None):
        super().__init__()
        self.image: Optional[bytes] = image_data
        self.prediction: Optional[str] = None
        self.confidence: Optional[float] = None

# Create a task with input data
task = ImageClassificationTask(image_data=b'\x89PNG...')

# Set task priority (0 is default, higher values = higher priority)
task.set_priority(1)

# Check if task has expired
if task.is_expired():
    print("Task timeout reached")

# For large data, use shared memory to avoid serialization overhead
task.share_value('image')  # Moves image bytes to shared memory
```
```

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

### POST /task/share

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/fundamentals.rst

Prepares a task for shared memory usage between OS processes by associating data with a specific task field.

```APIDOC
## POST /task/share

### Description
Configures a task field to use shared memory for large data transfers (e.g., bytes or numpy arrays) between pipeline steps.

### Method
POST

### Endpoint
/task/share

### Parameters
#### Request Body
- **field_name** (string) - Required - The name of the field in the Task object.
- **content** (StreamReader) - Required - The data source to be moved to shared memory.
- **size** (int) - Required - The size of the data in bytes.

### Request Example
{
  "field_name": "image",
  "size": 1024
}

### Response
#### Success Response (200)
- **success** (boolean) - Indicates if the shared memory allocation was successful.
```

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

### Execute Service Benchmarks

Source: https://github.com/avito-tech/aqueduct/blob/main/tests/benchmarks/bench_service/README.md

Runs the benchmark suites for service steps and configurations using Python modules within the containerized environment.

```shell
docker exec maas_bench_$USER python -m bench.bench_steps
python3 -m bench.bench_configurations bench/configs/image_small_3step.json
```

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

### Accessing the Default Aqueduct Logger

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/logging.rst

Demonstrates how to import the default preconfigured logger provided by Aqueduct, which defaults to DEBUG level and writes to stderr.

```python
from aqueduct.logger import log
```

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

### Integrating Custom Logger with Aiohttp

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/logging.rst

Demonstrates how to pass a custom logger instance to the AppIntegrator when setting up an aiohttp application.

```python
import logging
from aiohttp import web
from aqueduct.integrations.aiohttp import AppIntegrator
from .flow import get_flow

custom_logger = logging.getLogger(__name__)

def prepare_app() -> web.Application:
    app = web.Application(client_max_size=0)
    AppIntegrator(app).add_flow(get_flow(), logger=custom_logger)
    return app

if __name__ == '__main__':
    web.run_app(prepare_app())
```

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

### Python Handler Implementation for Model Processing

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/fundamentals.rst

This Python code defines a handler that loads a model during process startup and uses it to process images from incoming tasks. The `on_start` method initializes the model to avoid memory duplication in the parent process, and the `handle` method processes tasks in batches, updating each task with the processed image data.

```python
class Handler(BaseTaskHandler):
    def __init__(self):
        self._model = None

    def on_start(self):
        """Executed in a child process, so the parent process does not consume additional memory."""
        self._model = MyModel()

    def handle(self, *tasks: Task):
        """List of tasks because it can be batching."""
        for task in tasks:
            task.image_processed = self._model.process(task.image)
            task.share_value('image_processed')
```

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

### Configure Gunicorn Server Lifecycle Hooks

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/concurrency.rst

Defines on_starting and on_exit hooks to manage the lifecycle of flow socket server processes. The on_starting hook preloads the flow, while on_exit ensures all child processes are terminated upon server shutdown.

```python
def on_starting(server):
	global flow_socket_server_proc_ctx
	# Initialize Flow steps and Flow socket server with its steps only once
	flow_socket_server_proc_ctx = socket_flow.preload(build_flow())


def on_exit(server):
	if flow_socket_server_proc_ctx:
		for process in flow_socket_server_proc_ctx.processes:
			os.kill(process.pid, signal.SIGKILL)
```

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

### Configure Pipeline Steps with FlowStep

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Configures pipeline execution using FlowStep, allowing developers to define parallelism (nprocs), batch sizes, and timeouts to optimize resource utilization.

```python
import operator
from aqueduct import FlowStep, BaseTaskHandler, BaseTask

class PreprocessHandler(BaseTaskHandler):
    def handle(self, *tasks):
        for task in tasks:
            task.preprocessed = True

class ModelHandler(BaseTaskHandler):
    def on_start(self):
        self._model = lambda x: 0.95

    def handle(self, *tasks):
        for task in tasks:
            task.score = self._model(task.data)

# Basic step with single worker
step1 = FlowStep(PreprocessHandler())

# Step with multiple parallel workers
step2 = FlowStep(handler=ModelHandler(), nprocs=4)

# Step with batching for GPU efficiency
step3 = FlowStep(
    handler=ModelHandler(),
    nprocs=2,
    batch_size=32,
    batch_timeout=0.1
)
```

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

### Integrate Aqueduct with aiohttp

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Shows how to integrate Aqueduct into an aiohttp web application using AppIntegrator. This ensures the flow lifecycle is managed automatically alongside the web server.

```python
from aiohttp import web
from aqueduct import Flow, FlowStep, BaseTaskHandler, BaseTask
from aqueduct.integrations.aiohttp import AppIntegrator, FLOW_NAME

class Task(BaseTask):
    def __init__(self, number: int):
        super().__init__()
        self.number = number
        self.result = None

class ComputeHandler(BaseTaskHandler):
    def handle(self, *tasks: Task):
        for task in tasks:
            task.result = task.number ** 2

class ComputeView(web.View):
    @property
    def flow(self) -> Flow:
        return self.request.app[FLOW_NAME]

    async def post(self):
        data = await self.request.json()
        task = Task(data['number'])
        await self.flow.process(task)
        return web.json_response({'result': task.result})

def create_app() -> web.Application:
    app = web.Application()
    app.router.add_post('/compute', ComputeView)
    flow = Flow(FlowStep(ComputeHandler(), nprocs=2))
    AppIntegrator(app).add_flow(flow)
    return app
```

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

### Configure Aqueduct Source Directory

Source: https://github.com/avito-tech/aqueduct/blob/main/tests/benchmarks/bench_service/README.md

Creates a symbolic link to the aqueduct source code in the current directory to ensure the benchmark scripts can access the necessary modules.

```shell
ln -s `pwd`/../../../aqueduct ./aqueduct
```

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

### Define Custom Task with BaseTask

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Demonstrates how to extend BaseTask to create a data container for ML tasks. It includes support for shared memory to reduce serialization overhead and task priority management.

```python
from typing import Optional
from aqueduct import BaseTask

class ImageClassificationTask(BaseTask):
    """Task for image classification with shared memory support."""

    def __init__(self, image_data: bytes = None):
        super().__init__()
        self.image: Optional[bytes] = image_data
        self.prediction: Optional[str] = None
        self.confidence: Optional[float] = None

# Create a task with input data
task = ImageClassificationTask(image_data=b'\x89PNG...')

# Set task priority (0 is default, higher values = higher priority)
task.set_priority(1)

# Check if task has expired
if task.is_expired():
    print("Task timeout reached")

# For large data, use shared memory to avoid serialization overhead
task.share_value('image')  # Moves image bytes to shared memory
```

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

### Implement Task Prioritization in Aqueduct

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Shows how to configure multiple priority levels in a Flow and assign priorities to individual tasks. Higher priority tasks are processed first by available workers.

```python
import asyncio
from aqueduct import Flow, FlowStep, BaseTaskHandler, BaseTask
from typing import Optional

class Task(BaseTask):
    def __init__(self, name: str):
        super().__init__()
        self.name = name
        self.result: Optional[str] = None

class SlowHandler(BaseTaskHandler):
    def handle(self, *tasks: Task):
        import time
        for task in tasks:
            time.sleep(0.1)
            task.result = f"Processed: {task.name}"

async def priority_example():
    flow = Flow(
        FlowStep(SlowHandler(), nprocs=1),
        queue_priorities=3,
    )
    flow.start()
    low_priority = Task("low")
    low_priority.set_priority(0)
    medium_priority = Task("medium")
    medium_priority.set_priority(1)
    high_priority = Task("high")
    high_priority.set_priority(2)
    results = await asyncio.gather(
        flow.process(low_priority),
        flow.process(medium_priority),
        flow.process(high_priority),
    )
    print(f"High: {high_priority.result}")
    print(f"Medium: {medium_priority.result}")
    print(f"Low: {low_priority.result}")
    await flow.stop()

if __name__ == '__main__':
    asyncio.run(priority_example())
```

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

### Gunicorn Configuration for Preloaded App

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/concurrency.rst

Configures Gunicorn to run a FastAPI application with the `preload_app` flag enabled. This ensures the Aqueduct Flow is initialized once in the master process, allowing it to be shared across worker processes via Copy-on-Write.

```python
import os
import signal

from main import build_flow, socket_flow

# GUNICORN CONFIG
bind = os.getenv('SERVICE_HOST', '0.0.0.0') + ':' + os.getenv('SERVICE_PORT', '8890')
workers = os.getenv('WEB_CONCURRENCY', 2)
worker_class = 'uvicorn.workers.UvicornWorker'
wsgi_app = 'main:app'
preload_app = True

# PRELOAD APP
flow_socket_server_proc_ctx = None
```

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

### Implement StatsD Client Wrapper and Monitored Flow

Source: https://context7.com/avito-tech/aqueduct/llms.txt

This snippet demonstrates how to wrap the StatsD client to integrate with Aqueduct's metrics system and configure a Flow to export performance data. It uses the ToStatsDMetricsExporter to track metrics like transfer time and queue size.

```python
class AqueductStatsd(StatsDBuffer):
    def __init__(self, host: str = 'localhost', port: int = 8125):
        import statsd
        self.client = statsd.StatsClient(host, port)

    def count(self, name: str, value: Union[float, int]):
        self.client.incr(name, value)

    def timing(self, name: str, value: Union[float, int]):
        self.client.timing(name, value)

    def gauge(self, name: str, value: Union[float, int]):
        self.client.gauge(name, value)

def create_monitored_flow() -> Flow:
    statsd_client = AqueductStatsd('localhost', 8125)
    return Flow(
        FlowStep(Handler(), nprocs=2),
        metrics_enabled=True,
        metrics_exporter=ToStatsDMetricsExporter(
            target=statsd_client,
            prefix='myapp',
        ),
    )
```

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

### Implement Batched ML Inference with Aqueduct

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Demonstrates how to use BaseTaskHandler to process multiple tasks in a single batch, improving GPU utilization. The Flow configuration uses batch_size and batch_timeout to optimize throughput.

```python
import asyncio
import time
from typing import List, Optional
import numpy as np
from aqueduct import Flow, FlowStep, BaseTaskHandler, BaseTask

class ImageTask(BaseTask):
    def __init__(self, image: np.ndarray):
        super().__init__()
        self.image = image
        self.prediction: Optional[bool] = None

class CatDetectorHandler(BaseTaskHandler):
    def __init__(self):
        self._model = None

    def on_start(self):
        self._model = self._load_model()

    def _load_model(self):
        def predict_batch(images: np.ndarray) -> np.ndarray:
            batch_size = images.shape[0]
            time.sleep(0.01 * batch_size * 0.7)
            return np.ones(batch_size, dtype=bool)
        return predict_batch

    def handle(self, *tasks: ImageTask):
        images = np.array([task.image for task in tasks])
        predictions = self._model(images)
        for task, pred in zip(tasks, predictions):
            task.prediction = bool(pred)

async def batched_inference():
    flow = Flow(
        FlowStep(
            CatDetectorHandler(),
            nprocs=2,
            batch_size=16,
            batch_timeout=0.05,
        ),
    )
    flow.start()
    tasks = [ImageTask(np.random.rand(224, 224, 3)) for _ in range(50)]
    await asyncio.gather(*(flow.process(task) for task in tasks))
    await flow.stop()
```

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

### Configure Sentry Logging Handler in Python

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/sentry.rst

This Python code snippet demonstrates how to set up a Sentry handler for Aqueduct's logger. It checks for the SENTRY_ENABLED environment variable and configures the SentryClient with a DSN and HTTP transport if enabled. The SentryHandler is then added to the main log object.

```python
import logging
import os

from raven import Client
from raven.handlers.logging import SentryHandler
from raven.transport.http import HTTPTransport

from aqueduct.logger import log


if os.getenv('SENTRY_ENABLED') is True:
    dsn = os.getenv('SENTRY_DSN')
    sentry_handler = SentryHandler(client=Client(dsn=dsn, transport=HTTPTransport), level=logging.ERROR)
    log.addHandler(sentry_handler)
```

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

### Implement StatsDBuffer for Aqueduct Metrics

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/metrics.rst

This Python code demonstrates how to implement the StatsDBuffer protocol to create a custom StatsD client for Aqueduct metrics. It defines a class AqueductStatsd that wraps a statsd.StatsClient and provides count, timing, and gauge methods.

```python
import statsd

from typing import Union

from aqueduct import Flow, FlowStep
from aqueduct.metrics import ToStatsDMetricsExporter
from aqueduct.metrics.export import StatsDBuffer


class AqueductStatsd(StatsDBuffer):
    def __init__(self):
        self.statsd_client = statsd.StatsClient('localhost', 8125)

    def count(self, name: str, value: Union[float, int]):
        self.statsd_client.incr(name, value)

    def timing(self, name: str, value: Union[float, int]):
        self.statsd_client.timing(name, value)

    def gauge(self, name: str, value: Union[float, int]):
        self.statsd_client.gauge(name, value)
    
def create_flow() -> Flow:
    aqueduct_statsd = AqueductStatsd()
    return Flow(
        FlowStep(
            SumHandler() # Assuming SumHandler is defined elsewhere
        ),

        # Here we pass our metrics exporter and use our implementation of StatsDBuffer
        metrics_exporter=ToStatsDMetricsExporter(
            target=aqueduct_statsd,
            prefix='app_custom_prefix'
        ),
    )

```

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

### POST /flow/process

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/fundamentals.rst

Sends a task into the Aqueduct pipeline for processing. The flow manages the lifecycle of the task through defined FlowSteps.

```APIDOC
## POST /flow/process

### Description
Sends a custom Task object into the pipeline for asynchronous processing. The result is stored within the task object itself.

### Method
POST

### Endpoint
/flow/process

### Parameters
#### Request Body
- **task** (BaseTask) - Required - An instance of a class inheriting from BaseTask containing the data to be processed.
- **timeout_sec** (int) - Optional - Timeout in seconds for the process operation (default: 5).

### Request Example
{
  "task": {
    "number": 10,
    "result": null
  },
  "timeout_sec": 5
}

### Response
#### Success Response (200)
- **status** (string) - Confirmation that the task was accepted into the pipeline.

#### Response Example
{
  "status": "accepted"
}
```

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

### Implement Conditional Flow Steps

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Demonstrates how to define a conditional handler for a FlowStep. The handle_condition function determines whether a task should be processed by the specific step based on task attributes.

```python
def needs_processing(task: BaseTask) -> bool:
    return hasattr(task, 'requires_model') and task.requires_model

step4 = FlowStep(
    handler=ModelHandler(),
    handle_condition=needs_processing,
)
```

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

### Customizing the Default Aqueduct Logger

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/logging.rst

Shows how to modify the existing logger instance by changing the logging level and adding custom handlers like RotatingFileHandler.

```python
import logging
from logging.handlers import RotatingFileHandler
from aqueduct.logger import log

log.setLevel(logging.INFO)

handler = RotatingFileHandler('my_log.log', maxBytes=2000, backupCount=10)
log.addHandler(handler)
```

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

### Share Numpy Arrays with Zero-Copy Semantics

Source: https://context7.com/avito-tech/aqueduct/llms.txt

Demonstrates how to use the numpy extra in Aqueduct to share large arrays between processes without copying data. It uses the share_value method to transfer arrays to worker processes efficiently.

```python
import asyncio
import numpy as np
from aqueduct import Flow, FlowStep, BaseTaskHandler, BaseTask
from typing import Optional

class NumpyTask(BaseTask):
    def __init__(self, array: np.ndarray):
        super().__init__()
        self.array = array
        self.result: Optional[np.ndarray] = None

class NumpyProcessor(BaseTaskHandler):
    def handle(self, *tasks: NumpyTask):
        for task in tasks:
            task.result = task.array * 2
            task.share_value('result')

async def process_numpy_arrays():
    flow = Flow(FlowStep(NumpyProcessor(), nprocs=2))
    flow.start()
    large_array = np.random.rand(1000, 1000)
    task = NumpyTask(large_array)
    task.share_value('array')
    await flow.process(task)
    print(f"Result shape: {task.result.shape}")
    print(f"Result sum: {task.result.sum():.2f}")
    await flow.stop()

if __name__ == '__main__':
    asyncio.run(process_numpy_arrays())
```

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

### FastAPI App with Standard Aqueduct Flow

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/concurrency.rst

A standard FastAPI application that initializes and uses an Aqueduct Flow to process tasks. This version is suitable for single-process applications.

```python
from typing import Optional

from aqueduct.sockets.flow import SocketFlow
from fastapi import FastAPI
from pydantic import BaseModel

from flow import Handler, Task, build_flow


mp_flow = build_flow()

# Start Flow
# Initialize flow step processes. Initialize Flow utility (metrics, result awaiting, etc.)
app = FastAPI(on_startup=[mp_flow.start])


class GetResultResponse(BaseModel):
    result: Optional[str]


@app.get('/get_result')
async def get_result() -> GetResultResponse:
    task = Task()
    print(task.result)  # None
    await mp_flow.process(task)
    print(task.result)  # 'done'
    return GetResultResponse(result=task.result)
```

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

### Configure SocketFlow for Multi-Worker Concurrency

Source: https://context7.com/avito-tech/aqueduct/llms.txt

This snippet illustrates how to share a single Flow instance across multiple Gunicorn workers using SocketFlow and Unix Domain Sockets. It includes the flow definition, the FastAPI integration, and the Gunicorn configuration required for preloading the flow in the master process.

```python
# flow.py
class Handler(BaseTaskHandler):
    def handle(self, *tasks: Task):
        for task in tasks:
            task.result = 'processed'

# main.py
socket_flow = SocketFlow()
app = FastAPI(on_startup=[socket_flow.start])

@app.get('/process')
async def process() -> Response:
    task = Task()
    await socket_flow.process([task])
    return Response(result=task.result)

# gunicorn_config.py
def on_starting(server):
    global flow_server_ctx
    flow_server_ctx = socket_flow.preload(build_flow())
```

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

### Define Custom Task for Aqueduct

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/fundamentals.rst

Demonstrates how to create a custom Task class by inheriting from Aqueduct's BaseTask. This allows defining task-specific attributes and initializing them. The 'result' field is commonly used to store processing outcomes.

```python
from aqueduct import BaseTask
from typing import Optional

class MyTask(BaseTask):
    number: int
    result: Optional[int]

    def __init__(self, number: int):
        super().__init__()
        self.number = number
        self.result = None
```

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

### Replacing the Default Aqueduct Logger

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/logging.rst

Explains how to replace the default Aqueduct logger with a custom logger instance using the replace_logger function.

```python
import logging
from logging.handlers import RotatingFileHandler
from aqueduct.logger import replace_logger

custom_logger = logging.getLogger(__name__)
replace_logger(custom_logger)
```

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

### FastAPI App with Aqueduct SocketFlow for Concurrency

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/concurrency.rst

A FastAPI application refactored to use Aqueduct's SocketFlow. This enables communication with a separate Flow process, facilitating concurrent web worker management.

```python
from typing import Optional

from aqueduct.sockets.flow import SocketFlow
from fastapi import FastAPI
from pydantic import BaseModel

from flow import Handler, Task, build_flow


socket_flow = SocketFlow()

# Start SocketFlow wrapper
# Initialize socket connection pool to communicate with flow socket server
app = FastAPI(on_startup=[socket_flow.start])


class GetResultResponse(BaseModel):
    result: Optional[str]


@app.get('/get_result')
async def get_result() -> GetResultResponse:
    task = Task()
    print(task.result)  # None
    await socket_flow.process([task])  # Send batch of tasks via socket
    print(task.result)  # 'done'
    return GetResultResponse(result=task.result)
```

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

### Share Data via Shared Memory in Aqueduct Task

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/fundamentals.rst

Illustrates how to utilize Aqueduct's shared memory feature for efficient data transfer. The `share_value_with_data` method is used to move data, such as images from a request, into shared memory before processing. This is particularly useful for large data blobs.

```python
from aqueduct import BaseTask
from typing import Optional

class MyTask(BaseTask):
    image: Optional[bytes]
    result: Optional[int]

    def __init__(self, image: Optional[bytes] = None):
        super().__init__()
        self.image = image
        self.result = None

# Assuming 'request' is an aiohttp request object with 'content' and 'content_length'
task = MyTask()
await task.share_value_with_data(
    field_name='image',
    content=request.content,
    size=request.content_length,
)
await flow.process(task)
```

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

### Transfer Data Using Shared Memory in Aqueduct

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/fundamentals.rst

Shows how to use the `share_value` method to transfer data (bytes or numpy arrays) between Aqueduct steps via shared memory. This method is called on a task instance when the data is already available in a field.

```python
# Assuming 'task' is an instance of a BaseTask subclass with a 'data' field
# and 'flow' is an initialized Aqueduct Flow object
await task.share_value(field_name='data')
await flow.process(task)
```

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

### Aqueduct Flow Definition

Source: https://github.com/avito-tech/aqueduct/blob/main/docs/concurrency.rst

Defines a basic Aqueduct Flow with a simple task handler. This serves as the foundation for processing tasks within the Aqueduct framework.

```python
from aqueduct.flow import Flow, FlowStep
from aqueduct.handler import BaseTaskHandler
from aqueduct.task import BaseTask


class Task(BaseTask):
    def __init__(self):
        super().__init__()
        self.result = None


class Handler(BaseTaskHandler):
    def handle(self, *tasks: Task):
        for task in tasks:
            task.result = 'done'


def build_flow() -> Flow:
    return Flow(
        FlowStep(
            Handler(),
        ),
    )
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.