feat(pool_executors): add serial execution support and related components

MIT License

Copyright (c) 2025 Jan Reimes

Permission is hereby granted, free of charge, to any person obtaining a copy

of this software and associated documentation files (the "Software"), to deal

in the Software without restriction, including without limitation the rights

to use, copy, modify, merge, publish, distribute, sublicense, and/or sell

copies of the Software, and to permit persons to whom the Software is

furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all

copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR

IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,

FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE

AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER

LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,

OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE

SOFTWARE.

# pool-executors

Lightweight executor pool extensions for Python 3.13+, providing a unified interface for different execution strategies including serial (non-parallel) execution.

## Features

- **SerialPoolExecutor**: Execute tasks sequentially in the main thread (useful for debugging or when parallelism is not needed)

- **Unified factory**: Create any executor type with a consistent interface

- **Type-safe**: Full type hints for mypy strict mode

- **Case-insensitive**: Accept executor types in any case (e.g., "serial", "SERIAL", "Serial")

- **Alias support**: Use short aliases like "mp" for "multiprocessing"

## Installation

```bash

pip install pool-executors

```

## Usage Examples

### Using SerialPoolExecutor Directly

```python

from pool_executors import SerialPoolExecutor

def task(x: int) -> int:

    return x * 2

with SerialPoolExecutor() as executor:

    futures = [executor.submit(task, i) for i in range(5)]

    results = [f.result() for f in futures]

    print(results)  # [0, 2, 4, 6, 8]

```

### Using the Factory with Case-Insensitive Types

```python

from pool_executors import create_executor, ExecutorType

# All of these are equivalent:

executor = create_executor("serial")

executor = create_executor("SERIAL")

executor = create_executor(ExecutorType.SERIAL)

# Use aliases for convenience:

executor = create_executor("mp", max_workers=4)  # ProcessPoolExecutor

executor = create_executor("thread", max_workers=4)  # ThreadPoolExecutor

```

### Factory with Keyword Arguments

```python

from pool_executors import create_executor

# All kwargs are forwarded to the underlying executor

executor = create_executor(

    "multiprocessing",

    max_workers=4,

    initializer=setup_fn,

    initargs=(arg1, arg2)

)

```

### Executor Type Reference

| Type | Aliases | Description |

|------|---------|-------------|

| `serial` | - | Sequential execution in main thread (SerialPoolExecutor) |

| `multiprocessing` | `mp` | True parallelism with separate processes (ProcessPoolExecutor) |

| `threading` | `thread` | Concurrent execution with threads (ThreadPoolExecutor) |

| `subinterpreter` | `sub`, `si` | Subinterpreter-based parallelism (Python 3.14+, fallback to multiprocessing) |

## Design Rationale

This package was created to:

1. **Provide serial execution**: Sometimes you need to disable parallelism for debugging or when overhead outweighs benefits

2. **Unify executor creation**: Single factory function for all executor types with consistent interface

3. **Enable future extraction**: Designed as standalone package for easy repository separation

## Requirements

- Python 3.13 or higher

- No external dependencies (stdlib only)

## License

MIT License - see LICENSE file for details.

"""Executor pool extensions with serial execution support."""

from pool_executors.factory import create_executor

from pool_executors.serial import SerialPoolExecutor

from pool_executors.types import ExecutorType

__version__ = "0.1.0"

__all__ = ["ExecutorType", "SerialPoolExecutor", "create_executor"]

"""Executor factory with case-insensitive type selection."""

from __future__ import annotations

import logging

from concurrent.futures import Executor, ProcessPoolExecutor, ThreadPoolExecutor

from typing import Any

from pool_executors.serial import SerialPoolExecutor

from pool_executors.types import ExecutorType

logger = logging.getLogger("pool_executors.factory")

def create_executor(executor_type: ExecutorType | str, **kwargs: Any) -> Executor:

    """Create executor instance based on type.

    Supports case-insensitive type names and aliases:

    - "serial" / "SERIAL" → SerialPoolExecutor

    - "multiprocessing" / "mp" / "MP" → ProcessPoolExecutor

    - "threading" / "thread" / "THREAD" → ThreadPoolExecutor

    - "subinterpreter" / "sub" / "si" / "SI" → InterpreterPoolExecutor (Python 3.14+)

    Args:

        executor_type: Executor type (case-insensitive)

        **kwargs: Arguments forwarded to executor constructor

    Returns:

        Executor instance

    Raises:

        ValueError: If executor_type is invalid

    Example:

        >>> executor = create_executor("mp", max_workers=4)

        >>> executor = create_executor("SERIAL")

        >>> executor = create_executor(ExecutorType.THREADING, max_workers=2)

    """

    # Normalize to lowercase string

    type_str = str(executor_type).lower()

    # Convert to ExecutorType enum

    try:

        exec_type = ExecutorType(type_str)

    except ValueError as e:

        valid_types = ", ".join(sorted({member.value for member in ExecutorType}))

        msg = f"Invalid executor type '{executor_type}'. Valid types: {valid_types}"

        raise ValueError(msg) from e

    # Dispatch to appropriate executor

    if exec_type in (ExecutorType.SERIAL,):

        return SerialPoolExecutor(**kwargs)

    if exec_type in (ExecutorType.MULTIPROCESSING, ExecutorType.MP):

        return ProcessPoolExecutor(**kwargs)

    if exec_type in (ExecutorType.THREADING, ExecutorType.THREAD):

        return ThreadPoolExecutor(**kwargs)

    if exec_type in (ExecutorType.SUBINTERPRETER, ExecutorType.SUB, ExecutorType.SI):

        # Try to import InterpreterPoolExecutor (Python 3.14+)

        try:

            from concurrent.futures import InterpreterPoolExecutor  # type: ignore[attr-defined]

            return InterpreterPoolExecutor(**kwargs)

        except (ImportError, AttributeError):

            logger.warning("InterpreterPoolExecutor not available (Python 3.14+ required), falling back to ProcessPoolExecutor")

            return ProcessPoolExecutor(**kwargs)

    # Should never reach here due to enum validation above

    msg = f"Unhandled executor type: {exec_type}"

    raise RuntimeError(msg)