Skip to content

Workers API Reference

Base Worker Class

concurry.core.worker.base_worker.Worker

Base class for workers in concurry.

This class provides the foundation for user-defined workers. Users should inherit from this class and implement their worker logic. The worker will be automatically managed by the executor.

The Worker class implements the actor pattern, allowing you to run methods in different execution contexts (sync, thread, process, asyncio, ray) while maintaining state isolation and providing a unified Future-based API.

Important Design Note:

The Worker class itself does NOT inherit from morphic.Typed. This design choice allows you complete freedom in defining your __init__ method - you can use any signature with any combination of positional arguments, keyword arguments, args, and *kwargs. The Typed integration is applied at the WorkerProxy layer, which wraps your worker and provides validation for worker configuration (mode, blocking, etc.) but not for worker initialization.

Model Inheritance Support:

Worker supports cooperative multiple inheritance, allowing you to combine Worker with model classes for automatic field validation and serialization:

  • ✅ morphic.Typed: Full support (sync, thread, process, asyncio)
  • ✅ pydantic.BaseModel: Full support (sync, thread, process, asyncio)
  • ❌ Ray mode limitation: Ray mode is NOT compatible with Typed/BaseModel workers

Validation Decorators (Works with ALL modes including Ray):

  • ✅ @morphic.validate: Works on methods and init (all modes including Ray)
  • ✅ @pydantic.validate_call: Works on methods and init (all modes including Ray)

These decorators provide runtime validation without class inheritance, making them compatible with Ray mode.

This means you can use: - Plain Python classes (all modes including Ray) - Worker + morphic.Typed for validation and hooks (all modes EXCEPT Ray) - Worker + pydantic.BaseModel for Pydantic validation (all modes EXCEPT Ray) - @validate or @validate_call decorators on methods (all modes including Ray) - Dataclasses, Attrs, or any other class structure (all modes)

The only requirement is that your worker class is instantiable via __init__ with the arguments you pass to .init().

Basic Usage 
from concurry import Worker

class DataProcessor(Worker):
    def __init__(self, multiplier: int):
        self.multiplier = multiplier
        self.count = 0

    def process(self, value: int) -> int:
        self.count += 1
        return value * self.multiplier

# Initialize worker with thread execution
worker = DataProcessor.options(mode="thread").init(3)
future = worker.process(10)
result = future.result()  # 30
worker.stop()

Context Manager (Automatic Cleanup): Workers and pools support context manager protocol for automatic cleanup:

```python
from concurry import Worker

class DataProcessor(Worker):
    def __init__(self, multiplier: int):
        self.multiplier = multiplier

    def process(self, value: int) -> int:
        return value * self.multiplier

# Context manager automatically calls .stop() on exit
with DataProcessor.options(mode="thread").init(3) as worker:
    future = worker.process(10)
    result = future.result()  # 30
# Worker is automatically stopped here

# Works with pools too
with DataProcessor.options(mode="thread", max_workers=5).init(3) as pool:
    results = [pool.process(i).result() for i in range(10)]
# All workers in pool are automatically stopped here

# Cleanup happens even on exceptions
with DataProcessor.options(mode="thread").init(3) as worker:
    if some_error:
        raise ValueError("Error occurred")
# Worker is still stopped despite exception
```
Model Inheritance Usage 
from concurry import Worker
from morphic import Typed
from pydantic import BaseModel, Field
from typing import List, Optional

# Worker + Typed for validation and lifecycle hooks
class TypedWorker(Worker, Typed):
    name: str
    value: int = Field(default=0, ge=0)
    tags: List[str] = []

    @classmethod
    def pre_initialize(cls, data: dict) -> None:
        # Normalize data before validation
        if 'name' in data:
            data['name'] = data['name'].strip().title()

    def compute(self, x: int) -> int:
        return self.value * x

# Initialize with validated fields
worker = TypedWorker.options(mode="thread").init(
    name="processor",
    value=10,
    tags=["ml", "preprocessing"]
)
result = worker.compute(5).result()  # 50
worker.stop()

# Worker + Pydantic BaseModel for validation
class PydanticWorker(Worker, BaseModel):
    name: str = Field(..., min_length=1, max_length=50)
    age: int = Field(..., ge=0, le=150)
    email: Optional[str] = None

    def get_info(self) -> dict:
        return {"name": self.name, "age": self.age, "email": self.email}

worker = PydanticWorker.options(mode="process").init(
    name="Alice",
    age=30,
    email="alice@example.com"
)
info = worker.get_info().result()
worker.stop()

Validation Decorators (Ray-Compatible): 

from concurry import Worker
from morphic import validate
from pydantic import validate_call

# @validate decorator works with ALL modes including Ray
class ValidatedWorker(Worker):
    def __init__(self, multiplier: int):
        self.multiplier = multiplier

    @validate
    def process(self, value: int, scale: float = 1.0) -> float:
        '''Process with automatic type validation and coercion.'''
        return (value * self.multiplier) * scale

# Works with Ray mode!
worker = ValidatedWorker.options(mode="ray").init(multiplier=5)
result = worker.process("10", scale="2.0").result()  # "10" -> 10, "2.0" -> 2.0
# result = 100.0
worker.stop()

# @validate_call also works with ALL modes including Ray
class PydanticValidatedWorker(Worker):
    def __init__(self, base: int):
        self.base = base

    @validate_call
    def compute(self, x: int, y: int = 0) -> int:
        '''Compute with Pydantic validation.'''
        return (x + y) * self.base

# Also works with Ray mode!
worker = PydanticValidatedWorker.options(mode="ray").init(base=3)
result = worker.compute("5", y="2").result()  # Strings coerced to ints
# result = 21
worker.stop()

Ray Mode Limitations and Workarounds 
# ❌ BAD: Typed/BaseModel workers don't work with Ray
class TypedWorker(Worker, Typed):
    name: str
    value: int = 0

# This will raise ValueError with Ray mode
try:
    worker = TypedWorker.options(mode="ray").init(name="test", value=10)
except ValueError as e:
    print(e)  # "Cannot create Ray worker with Pydantic-based class..."

# ✅ GOOD: Use composition instead of inheritance for Ray
class RayCompatibleWorker(Worker):
    def __init__(self, name: str, value: int = 0):
        self.name = name
        self.value = value

    def compute(self, x: int) -> int:
        return self.value * x

# This works with Ray!
worker = RayCompatibleWorker.options(mode="ray").init(name="test", value=10)
result = worker.compute(5).result()  # 50
worker.stop()

# ✅ EVEN BETTER: Use validation decorators for type checking
class ValidatedRayWorker(Worker):
    @validate
    def __init__(self, name: str, value: int = 0):
        self.name = name
        self.value = value

    @validate
    def compute(self, x: int) -> int:
        return self.value * x

# Validation + Ray compatibility!
worker = ValidatedRayWorker.options(mode="ray").init(name="test", value="10")
result = worker.compute("5").result()  # Types coerced, result = 50
worker.stop()

Why Ray + Typed/BaseModel doesn't work:

Ray's ray.remote() wraps classes as actors and modifies their __setattr__ behavior, which conflicts with Pydantic's frozen model implementation. When you try to create a Ray actor from a Pydantic-based class, Ray attempts to set internal attributes that trigger Pydantic's validation, causing AttributeError.

Automatic Error Detection:

Concurry automatically detects this incompatibility and raises a clear error: - ValueError: When attempting to create a Ray worker/pool with Typed/BaseModel - UserWarning: When creating non-Ray workers (if Ray is installed)

The warning helps you know that your worker won't be compatible with Ray mode if you later decide to switch execution modes.

Different Execution Modes 
# Synchronous (for testing/debugging)
worker = DataProcessor.options(mode="sync").init(2)

# Thread-based (good for I/O-bound tasks)
worker = DataProcessor.options(mode="thread").init(2)

# Process-based (good for CPU-bound tasks)
worker = DataProcessor.options(mode="process").init(2)

# Asyncio-based (good for async I/O)
worker = DataProcessor.options(mode="asyncio").init(2)

# Ray-based (distributed computing)
import ray
ray.init()
worker = DataProcessor.options(mode="ray", actor_options={"num_cpus": 1}).init(2)
Async Function Support

All workers can execute both sync and async functions. Async functions are automatically detected and executed correctly across all modes.

import asyncio

class AsyncWorker(Worker):
    def __init__(self):
        self.count = 0

    async def async_method(self, x: int) -> int:
        await asyncio.sleep(0.01)  # Simulate async I/O
        self.count += 1
        return x * 2

    def sync_method(self, x: int) -> int:
        return x + 10

# Use asyncio mode for best async performance
worker = AsyncWorker.options(mode="asyncio").init()
result1 = worker.async_method(5).result()  # 10
result2 = worker.sync_method(5).result()  # 15
worker.stop()

# Submit async functions via TaskWorker
from concurry import TaskWorker
import asyncio

async def compute(x, y):
    await asyncio.sleep(0.01)
    return x ** 2 + y ** 2

task_worker = TaskWorker.options(mode="asyncio").init()
result = task_worker.submit(compute, 3, 4).result()  # 25
task_worker.stop()

Performance: AsyncioWorkerProxy provides significant speedup (5-15x) for I/O-bound async operations by enabling true concurrent execution. Other modes execute async functions correctly but without concurrency benefits.

Blocking Mode 
# Returns results directly instead of futures
worker = DataProcessor.options(mode="thread", blocking=True).init(5)
result = worker.process(10)  # Returns 50 directly, not a future
worker.stop()

# With context manager (recommended)
with DataProcessor.options(mode="thread", blocking=True).init(5) as worker:
    result = worker.process(10)  # Returns 50 directly
# Worker automatically stopped
Submitting Arbitrary Functions with TaskWorker 
# Use TaskWorker for Executor-like interface
from concurry import TaskWorker

def compute(x, y):
    return x ** 2 + y ** 2

task_worker = TaskWorker.options(mode="process").init()

# Submit arbitrary functions
future = task_worker.submit(compute, 3, 4)
result = future.result()  # 25

# Use map() for multiple tasks
results = list(task_worker.map(lambda x: x * 100, [1, 2, 3, 4, 5]))

task_worker.stop()
State Management 
class Counter(Worker):
    def __init__(self):
        self.count = 0

    def increment(self):
        self.count += 1
        return self.count

# Each worker maintains its own state
with Counter.options(mode="thread").init() as worker1:
    with Counter.options(mode="thread").init() as worker2:
        print(worker1.increment().result())  # 1
        print(worker1.increment().result())  # 2
        print(worker2.increment().result())  # 1 (separate state)
# Both workers automatically stopped

Submission Queue (Client-Side Task Queuing): Workers support client-side submission queuing via the max_queued_tasks parameter. This prevents overloading worker backends when submitting large batches of tasks.

**Key Benefits:**
- Prevents memory exhaustion from thousands of pending futures
- Avoids backend overload (especially Ray actors)
- Reduces network saturation for distributed workers
- Works transparently with your submission loops

**How it works:**
The submission queue limits how many tasks can be "in-flight" (submitted but not completed)
per worker. When the queue is full, further submissions block until a task completes.

```python
# Create worker with submission queue
worker = MyWorker.options(
    mode="thread",
    max_queued_tasks=10  # Max 10 tasks in-flight
).init()

# Submit 1000 tasks - automatically blocks when queue is full
futures = [worker.process(item) for item in range(1000)]
results = gather(futures)  # Submission queue prevents overload
worker.stop()
```

**Default values by mode:**
- sync/asyncio: None (bypassed) - immediate execution or event loop handles concurrency
- thread: 100 - high concurrency, large queue
- process: 5 - limited by CPU cores
- ray: 2 - minimize data transfer overhead

**Integration with other features:**
- **Limits**: Submission queue (client-side) + resource limits (worker-side) work together
- **Retries**: Only original submissions count, not retry attempts
- **Load Balancing**: Each worker in a pool has its own independent queue
- **On-Demand Workers**: Automatically bypass submission queue

For comprehensive documentation and examples, see the user guide:
`/docs/user-guide/limits.md#submission-queue`
Resource Protection with Limits

Workers support resource protection and rate limiting via the limits parameter. Limits enable control over API rates, resource pools, and call frequency.

Important: Workers always have self.limits available, even when no limits are configured. If no limits parameter is provided, workers get an empty LimitSet that always allows acquisition without blocking. This means your code can safely call self.limits.acquire() without checking if limits exist.

from concurry import Worker, LimitSet, RateLimit, CallLimit, ResourceLimit
from concurry import RateLimitAlgorithm

# Define limits
limits = LimitSet(limits=[
    CallLimit(window_seconds=60, capacity=100),  # 100 calls/min
    RateLimit(
        key="api_tokens",
        window_seconds=60,
        algorithm=RateLimitAlgorithm.TokenBucket,
        capacity=1000
    ),
    ResourceLimit(key="connections", capacity=10)
])

class APIWorker(Worker):
    def __init__(self, api_key: str):
        self.api_key = api_key

    def call_api(self, prompt: str):
        # Acquire limits before operation
        # CallLimit automatically acquired with default of 1
        with self.limits.acquire(requested={"api_tokens": 100}) as acq:
            result = external_api_call(prompt)
            # Update with actual usage
            acq.update(usage={"api_tokens": result.tokens_used})
            return result.response

# Option 1: Share limits across workers
worker1 = APIWorker.options(mode="thread", limits=limits).init("key1")
worker2 = APIWorker.options(mode="thread", limits=limits).init("key2")
# Both workers share the 1000 token/min pool

# Option 2: Private limits per worker
limit_defs = [
    RateLimit(key="tokens", window_seconds=60, capacity=1000)
]
worker = APIWorker.options(mode="thread", limits=limit_defs).init("key")
# This worker has its own private 1000 token/min pool

# Option 3: No limits (always succeeds)
worker = APIWorker.options(mode="thread").init("key")
# self.limits.acquire() always succeeds immediately, no blocking

Limit Types: - CallLimit: Count calls (usage always 1, no update needed) - RateLimit: Token/bandwidth limiting (requires update() call) - ResourceLimit: Semaphore-based resources (no update needed)

Key Behaviors: - Passing LimitSet: Workers share the same limit pool - Passing List[Limit]: Each worker gets private limits - No limits parameter: Workers get empty LimitSet (always succeeds) - CallLimit/ResourceLimit auto-acquired with default of 1 - RateLimits must be explicitly specified in requested dict - RateLimits require update() call (raises RuntimeError if missing) - Empty LimitSet has zero overhead (no synchronization, no waiting)

See user guide for more: /docs/user-guide/limits.md

Source code in src/concurry/core/worker/base_worker.py 
 911
 912
 913
 914
 915
 916
 917
 918
 919
 920
 921
 922
 923
 924
 925
 926
 927
 928
 929
 930
 931
 932
 933
 934
 935
 936
 937
 938
 939
 940
 941
 942
 943
 944
 945
 946
 947
 948
 949
 950
 951
 952
 953
 954
 955
 956
 957
 958
 959
 960
 961
 962
 963
 964
 965
 966
 967
 968
 969
 970
 971
 972
 973
 974
 975
 976
 977
 978
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
class Worker:
    """Base class for workers in concurry.

    This class provides the foundation for user-defined workers. Users should inherit from this class
    and implement their worker logic. The worker will be automatically managed by the executor.

    The Worker class implements the actor pattern, allowing you to run methods in different execution
    contexts (sync, thread, process, asyncio, ray) while maintaining state isolation and providing
    a unified Future-based API.

    **Important Design Note:**

    The Worker class itself does NOT inherit from morphic.Typed. This design choice allows you
    complete freedom in defining your `__init__` method - you can use any signature with any
    combination of positional arguments, keyword arguments, *args, and **kwargs. The Typed
    integration is applied at the WorkerProxy layer, which wraps your worker and provides
    validation for worker configuration (mode, blocking, etc.) but not for worker initialization.

    **Model Inheritance Support:**

    Worker supports cooperative multiple inheritance, allowing you to combine Worker with
    model classes for automatic field validation and serialization:

    - ✅ **morphic.Typed**: Full support (sync, thread, process, asyncio)
    - ✅ **pydantic.BaseModel**: Full support (sync, thread, process, asyncio)
    - ❌ **Ray mode limitation**: Ray mode is NOT compatible with Typed/BaseModel workers

    **Validation Decorators (Works with ALL modes including Ray):**

    - ✅ **@morphic.validate**: Works on methods and __init__ (all modes including Ray)
    - ✅ **@pydantic.validate_call**: Works on methods and __init__ (all modes including Ray)

    These decorators provide runtime validation without class inheritance, making them
    compatible with Ray mode.

    This means you can use:
    - Plain Python classes (all modes including Ray)
    - Worker + morphic.Typed for validation and hooks (all modes EXCEPT Ray)
    - Worker + pydantic.BaseModel for Pydantic validation (all modes EXCEPT Ray)
    - @validate or @validate_call decorators on methods (all modes including Ray)
    - Dataclasses, Attrs, or any other class structure (all modes)

    The only requirement is that your worker class is instantiable via `__init__` with the
    arguments you pass to `.init()`.

    Basic Usage:
        ```python
        from concurry import Worker

        class DataProcessor(Worker):
            def __init__(self, multiplier: int):
                self.multiplier = multiplier
                self.count = 0

            def process(self, value: int) -> int:
                self.count += 1
                return value * self.multiplier

        # Initialize worker with thread execution
        worker = DataProcessor.options(mode="thread").init(3)
        future = worker.process(10)
        result = future.result()  # 30
        worker.stop()
        ```

    Context Manager (Automatic Cleanup):
        Workers and pools support context manager protocol for automatic cleanup:

        ```python
        from concurry import Worker

        class DataProcessor(Worker):
            def __init__(self, multiplier: int):
                self.multiplier = multiplier

            def process(self, value: int) -> int:
                return value * self.multiplier

        # Context manager automatically calls .stop() on exit
        with DataProcessor.options(mode="thread").init(3) as worker:
            future = worker.process(10)
            result = future.result()  # 30
        # Worker is automatically stopped here

        # Works with pools too
        with DataProcessor.options(mode="thread", max_workers=5).init(3) as pool:
            results = [pool.process(i).result() for i in range(10)]
        # All workers in pool are automatically stopped here

        # Cleanup happens even on exceptions
        with DataProcessor.options(mode="thread").init(3) as worker:
            if some_error:
                raise ValueError("Error occurred")
        # Worker is still stopped despite exception
        ```

    Model Inheritance Usage:
        ```python
        from concurry import Worker
        from morphic import Typed
        from pydantic import BaseModel, Field
        from typing import List, Optional

        # Worker + Typed for validation and lifecycle hooks
        class TypedWorker(Worker, Typed):
            name: str
            value: int = Field(default=0, ge=0)
            tags: List[str] = []

            @classmethod
            def pre_initialize(cls, data: dict) -> None:
                # Normalize data before validation
                if 'name' in data:
                    data['name'] = data['name'].strip().title()

            def compute(self, x: int) -> int:
                return self.value * x

        # Initialize with validated fields
        worker = TypedWorker.options(mode="thread").init(
            name="processor",
            value=10,
            tags=["ml", "preprocessing"]
        )
        result = worker.compute(5).result()  # 50
        worker.stop()

        # Worker + Pydantic BaseModel for validation
        class PydanticWorker(Worker, BaseModel):
            name: str = Field(..., min_length=1, max_length=50)
            age: int = Field(..., ge=0, le=150)
            email: Optional[str] = None

            def get_info(self) -> dict:
                return {"name": self.name, "age": self.age, "email": self.email}

        worker = PydanticWorker.options(mode="process").init(
            name="Alice",
            age=30,
            email="alice@example.com"
        )
        info = worker.get_info().result()
        worker.stop()
        ```

    Validation Decorators (Ray-Compatible):
        ```python
        from concurry import Worker
        from morphic import validate
        from pydantic import validate_call

        # @validate decorator works with ALL modes including Ray
        class ValidatedWorker(Worker):
            def __init__(self, multiplier: int):
                self.multiplier = multiplier

            @validate
            def process(self, value: int, scale: float = 1.0) -> float:
                '''Process with automatic type validation and coercion.'''
                return (value * self.multiplier) * scale

        # Works with Ray mode!
        worker = ValidatedWorker.options(mode="ray").init(multiplier=5)
        result = worker.process("10", scale="2.0").result()  # "10" -> 10, "2.0" -> 2.0
        # result = 100.0
        worker.stop()

        # @validate_call also works with ALL modes including Ray
        class PydanticValidatedWorker(Worker):
            def __init__(self, base: int):
                self.base = base

            @validate_call
            def compute(self, x: int, y: int = 0) -> int:
                '''Compute with Pydantic validation.'''
                return (x + y) * self.base

        # Also works with Ray mode!
        worker = PydanticValidatedWorker.options(mode="ray").init(base=3)
        result = worker.compute("5", y="2").result()  # Strings coerced to ints
        # result = 21
        worker.stop()
        ```

    Ray Mode Limitations and Workarounds:
        ```python
        # ❌ BAD: Typed/BaseModel workers don't work with Ray
        class TypedWorker(Worker, Typed):
            name: str
            value: int = 0

        # This will raise ValueError with Ray mode
        try:
            worker = TypedWorker.options(mode="ray").init(name="test", value=10)
        except ValueError as e:
            print(e)  # "Cannot create Ray worker with Pydantic-based class..."

        # ✅ GOOD: Use composition instead of inheritance for Ray
        class RayCompatibleWorker(Worker):
            def __init__(self, name: str, value: int = 0):
                self.name = name
                self.value = value

            def compute(self, x: int) -> int:
                return self.value * x

        # This works with Ray!
        worker = RayCompatibleWorker.options(mode="ray").init(name="test", value=10)
        result = worker.compute(5).result()  # 50
        worker.stop()

        # ✅ EVEN BETTER: Use validation decorators for type checking
        class ValidatedRayWorker(Worker):
            @validate
            def __init__(self, name: str, value: int = 0):
                self.name = name
                self.value = value

            @validate
            def compute(self, x: int) -> int:
                return self.value * x

        # Validation + Ray compatibility!
        worker = ValidatedRayWorker.options(mode="ray").init(name="test", value="10")
        result = worker.compute("5").result()  # Types coerced, result = 50
        worker.stop()
        ```

        **Why Ray + Typed/BaseModel doesn't work:**

        Ray's `ray.remote()` wraps classes as actors and modifies their `__setattr__`
        behavior, which conflicts with Pydantic's frozen model implementation. When you
        try to create a Ray actor from a Pydantic-based class, Ray attempts to set
        internal attributes that trigger Pydantic's validation, causing AttributeError.

        **Automatic Error Detection:**

        Concurry automatically detects this incompatibility and raises a clear error:
        - **ValueError**: When attempting to create a Ray worker/pool with Typed/BaseModel
        - **UserWarning**: When creating non-Ray workers (if Ray is installed)

        The warning helps you know that your worker won't be compatible with Ray mode
        if you later decide to switch execution modes.

    Different Execution Modes:
        ```python
        # Synchronous (for testing/debugging)
        worker = DataProcessor.options(mode="sync").init(2)

        # Thread-based (good for I/O-bound tasks)
        worker = DataProcessor.options(mode="thread").init(2)

        # Process-based (good for CPU-bound tasks)
        worker = DataProcessor.options(mode="process").init(2)

        # Asyncio-based (good for async I/O)
        worker = DataProcessor.options(mode="asyncio").init(2)

        # Ray-based (distributed computing)
        import ray
        ray.init()
        worker = DataProcessor.options(mode="ray", actor_options={"num_cpus": 1}).init(2)
        ```

    Async Function Support:
        All workers can execute both sync and async functions. Async functions are
        automatically detected and executed correctly across all modes.

        ```python
        import asyncio

        class AsyncWorker(Worker):
            def __init__(self):
                self.count = 0

            async def async_method(self, x: int) -> int:
                await asyncio.sleep(0.01)  # Simulate async I/O
                self.count += 1
                return x * 2

            def sync_method(self, x: int) -> int:
                return x + 10

        # Use asyncio mode for best async performance
        worker = AsyncWorker.options(mode="asyncio").init()
        result1 = worker.async_method(5).result()  # 10
        result2 = worker.sync_method(5).result()  # 15
        worker.stop()

        # Submit async functions via TaskWorker
        from concurry import TaskWorker
        import asyncio

        async def compute(x, y):
            await asyncio.sleep(0.01)
            return x ** 2 + y ** 2

        task_worker = TaskWorker.options(mode="asyncio").init()
        result = task_worker.submit(compute, 3, 4).result()  # 25
        task_worker.stop()
        ```

        **Performance:** AsyncioWorkerProxy provides significant speedup (5-15x) for
        I/O-bound async operations by enabling true concurrent execution. Other modes
        execute async functions correctly but without concurrency benefits.

    Blocking Mode:
        ```python
        # Returns results directly instead of futures
        worker = DataProcessor.options(mode="thread", blocking=True).init(5)
        result = worker.process(10)  # Returns 50 directly, not a future
        worker.stop()

        # With context manager (recommended)
        with DataProcessor.options(mode="thread", blocking=True).init(5) as worker:
            result = worker.process(10)  # Returns 50 directly
        # Worker automatically stopped
        ```

    Submitting Arbitrary Functions with TaskWorker:
        ```python
        # Use TaskWorker for Executor-like interface
        from concurry import TaskWorker

        def compute(x, y):
            return x ** 2 + y ** 2

        task_worker = TaskWorker.options(mode="process").init()

        # Submit arbitrary functions
        future = task_worker.submit(compute, 3, 4)
        result = future.result()  # 25

        # Use map() for multiple tasks
        results = list(task_worker.map(lambda x: x * 100, [1, 2, 3, 4, 5]))

        task_worker.stop()
        ```

    State Management:
        ```python
        class Counter(Worker):
            def __init__(self):
                self.count = 0

            def increment(self):
                self.count += 1
                return self.count

        # Each worker maintains its own state
        with Counter.options(mode="thread").init() as worker1:
            with Counter.options(mode="thread").init() as worker2:
                print(worker1.increment().result())  # 1
                print(worker1.increment().result())  # 2
                print(worker2.increment().result())  # 1 (separate state)
        # Both workers automatically stopped
        ```

    Submission Queue (Client-Side Task Queuing):
        Workers support client-side submission queuing via the `max_queued_tasks` parameter.
        This prevents overloading worker backends when submitting large batches of tasks.

        **Key Benefits:**
        - Prevents memory exhaustion from thousands of pending futures
        - Avoids backend overload (especially Ray actors)
        - Reduces network saturation for distributed workers
        - Works transparently with your submission loops

        **How it works:**
        The submission queue limits how many tasks can be "in-flight" (submitted but not completed)
        per worker. When the queue is full, further submissions block until a task completes.

        ```python
        # Create worker with submission queue
        worker = MyWorker.options(
            mode="thread",
            max_queued_tasks=10  # Max 10 tasks in-flight
        ).init()

        # Submit 1000 tasks - automatically blocks when queue is full
        futures = [worker.process(item) for item in range(1000)]
        results = gather(futures)  # Submission queue prevents overload
        worker.stop()
        ```

        **Default values by mode:**
        - sync/asyncio: None (bypassed) - immediate execution or event loop handles concurrency
        - thread: 100 - high concurrency, large queue
        - process: 5 - limited by CPU cores
        - ray: 2 - minimize data transfer overhead

        **Integration with other features:**
        - **Limits**: Submission queue (client-side) + resource limits (worker-side) work together
        - **Retries**: Only original submissions count, not retry attempts
        - **Load Balancing**: Each worker in a pool has its own independent queue
        - **On-Demand Workers**: Automatically bypass submission queue

        For comprehensive documentation and examples, see the user guide:
        `/docs/user-guide/limits.md#submission-queue`

    Resource Protection with Limits:
        Workers support resource protection and rate limiting via the `limits` parameter.
        Limits enable control over API rates, resource pools, and call frequency.

        **Important: Workers always have `self.limits` available, even when no limits
        are configured.** If no limits parameter is provided, workers get an empty
        LimitSet that always allows acquisition without blocking. This means your
        code can safely call `self.limits.acquire()` without checking if limits exist.

        ```python
        from concurry import Worker, LimitSet, RateLimit, CallLimit, ResourceLimit
        from concurry import RateLimitAlgorithm

        # Define limits
        limits = LimitSet(limits=[
            CallLimit(window_seconds=60, capacity=100),  # 100 calls/min
            RateLimit(
                key="api_tokens",
                window_seconds=60,
                algorithm=RateLimitAlgorithm.TokenBucket,
                capacity=1000
            ),
            ResourceLimit(key="connections", capacity=10)
        ])

        class APIWorker(Worker):
            def __init__(self, api_key: str):
                self.api_key = api_key

            def call_api(self, prompt: str):
                # Acquire limits before operation
                # CallLimit automatically acquired with default of 1
                with self.limits.acquire(requested={"api_tokens": 100}) as acq:
                    result = external_api_call(prompt)
                    # Update with actual usage
                    acq.update(usage={"api_tokens": result.tokens_used})
                    return result.response

        # Option 1: Share limits across workers
        worker1 = APIWorker.options(mode="thread", limits=limits).init("key1")
        worker2 = APIWorker.options(mode="thread", limits=limits).init("key2")
        # Both workers share the 1000 token/min pool

        # Option 2: Private limits per worker
        limit_defs = [
            RateLimit(key="tokens", window_seconds=60, capacity=1000)
        ]
        worker = APIWorker.options(mode="thread", limits=limit_defs).init("key")
        # This worker has its own private 1000 token/min pool

        # Option 3: No limits (always succeeds)
        worker = APIWorker.options(mode="thread").init("key")
        # self.limits.acquire() always succeeds immediately, no blocking
        ```

        **Limit Types:**
        - `CallLimit`: Count calls (usage always 1, no update needed)
        - `RateLimit`: Token/bandwidth limiting (requires update() call)
        - `ResourceLimit`: Semaphore-based resources (no update needed)

        **Key Behaviors:**
        - Passing `LimitSet`: Workers share the same limit pool
        - Passing `List[Limit]`: Each worker gets private limits
        - No limits parameter: Workers get empty LimitSet (always succeeds)
        - CallLimit/ResourceLimit auto-acquired with default of 1
        - RateLimits must be explicitly specified in `requested` dict
        - RateLimits require `update()` call (raises RuntimeError if missing)
        - Empty LimitSet has zero overhead (no synchronization, no waiting)

        See user guide for more: `/docs/user-guide/limits.md`
    """

    @classmethod
    @validate
    def options(
        cls: Type[T],
        *,
        mode: ExecutionMode,
        blocking: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
        max_workers: Union[conint(ge=0), None, _NO_ARG_TYPE] = _NO_ARG,
        load_balancing: Union[LoadBalancingAlgorithm, _NO_ARG_TYPE] = _NO_ARG,
        on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
        max_queued_tasks: Union[conint(ge=0), None, _NO_ARG_TYPE] = _NO_ARG,
        # Retry parameters
        num_retries: Union[conint(ge=0), _NO_ARG_TYPE] = _NO_ARG,
        retry_on: Optional[Any] = None,
        retry_algorithm: Union[RetryAlgorithm, _NO_ARG_TYPE] = _NO_ARG,
        retry_wait: Union[confloat(ge=0), _NO_ARG_TYPE] = _NO_ARG,
        retry_jitter: Union[confloat(ge=0, le=1), _NO_ARG_TYPE] = _NO_ARG,
        retry_until: Optional[Any] = None,
        **kwargs: Any,
    ) -> WorkerBuilder:
        """Configure worker execution options.

        Returns a WorkerBuilder that can be used to create worker instances
        with .init(*args, **kwargs).

        **Type Validation:**

        This method uses the `@validate` decorator from morphic, providing:
        - Automatic type checking and conversion
        - String-to-bool coercion (e.g., "true" → True)
        - AutoEnum fuzzy matching for mode parameter
        - Enhanced error messages for invalid inputs

        Args:
            mode: Execution mode (sync, thread, process, asyncio, ray)
                Accepts string or ExecutionMode enum value
            blocking: If True, method calls return results directly instead of futures
                Accepts bool or string representation ("true", "false", "1", "0")
                Default value determined by global_config.<mode>.blocking
            max_workers: Maximum number of workers in pool (optional)
                - If None or 1: Creates single worker. If >1: Creates worker pool with specified size.
                - Sync/Asyncio: Must be 1 or None (raises error otherwise)
                - Default value determined by global_config.<mode>.max_workers
            load_balancing: Load balancing algorithm (optional)
                - "round_robin": Distribute requests evenly
                - "least_active": Select worker with fewest active calls
                - "least_total": Select worker with fewest total calls
                - "random": Random selection
                - Default value determined by global_config.<mode>.load_balancing (for pools)
                  or global_config.<mode>.load_balancing_on_demand (for on-demand pools)
            on_demand: If True, create workers on-demand per request (default: False)
                - Workers are created for each request and destroyed after completion
                - Useful for bursty workloads or resource-constrained environments
                - Cannot be used with Sync/Asyncio modes
                - With max_workers=0: Unlimited concurrent workers (Ray) or
                  limited to cpu_count()-1 (Thread/Process)
            max_queued_tasks: Maximum number of in-flight tasks per worker (default varies by mode)
                - Controls how many tasks can be submitted to a worker's backend before blocking
                - Per-worker limit: each worker in a pool has its own independent queue
                - Value of N means max N tasks submitted but not yet completed per worker
                - Automatically bypassed in blocking mode (unlimited submissions allowed)
                - Automatically bypassed in sync and asyncio modes
                - Prevents overload when submitting large batches (e.g., 5000+ tasks to Ray)
                - Default value determined by global_config.<mode>.max_queued_tasks
                - See user guide for detailed usage: /docs/user-guide/limits.md#submission-queue
            unwrap_futures: If True, automatically unwrap BaseFuture arguments
                by calling .result() on them before passing to worker methods. This enables
                seamless composition of workers. Set to False to pass futures as-is.
                Default value determined by global_config.<mode>.unwrap_futures
            limits: Resource protection and rate limiting (optional, defaults to empty LimitSet)
                - Pass LimitSet: Workers share the same limit pool
                - Pass List[Limit]: Each worker gets private limits (creates shared LimitSet for pools)
                - Omit parameter: Workers get empty LimitSet (self.limits.acquire() always succeeds)
                Workers always have self.limits available, even when no limits configured.
                See Worker docstring "Resource Protection with Limits" section for details.
            num_retries: Maximum number of retry attempts after initial failure
                Total attempts = num_retries + 1 (initial attempt).
                Set to 0 to disable retries (zero overhead).
                Default value determined by global_config.<mode>.num_retries
            retry_on: Exception types or callables that trigger retries (optional)
                - Single exception class: retry_on=ValueError
                - List of exceptions: retry_on=[ValueError, ConnectionError]
                - Callable filter: retry_on=lambda exception, **ctx: "retry" in str(exception)
                - Mixed list: retry_on=[ValueError, custom_filter]
                Default: [Exception] (retry on all exceptions when num_retries > 0)
            retry_algorithm: Backoff strategy for wait times
                Default value determined by global_config.<mode>.retry_algorithm
            retry_wait: Minimum wait time between retries in seconds
                Base wait time before applying strategy and jitter.
                Default value determined by global_config.<mode>.retry_wait
            retry_jitter: Jitter factor between 0 and 1
                Uses Full Jitter algorithm from AWS: sleep = random(0, calculated_wait).
                Set to 0 to disable jitter. Prevents thundering herd when many workers retry.
                Default value determined by global_config.<mode>.retry_jitter
            retry_until: Validation functions for output (optional)
                - Single validator: retry_until=lambda result, **ctx: result.get("status") == "success"
                - List of validators: retry_until=[validator1, validator2] (all must pass)
                Validators receive result and context as kwargs. Return True for valid output.
                If validation fails, triggers retry even without exception.
                Useful for LLM output validation (JSON schema, XML format, etc.)
            **kwargs: Additional options passed to the worker implementation
                - For ray: num_cpus, num_gpus, resources, etc.
                - For process: mp_context (fork, spawn, forkserver)

        Returns:
            A WorkerBuilder instance that can create workers via .init()

        Examples:
            Basic Usage:
                ```python
                # Configure and create worker
                worker = MyWorker.options(mode="thread").init(multiplier=3)
                ```

            Type Coercion:
                ```python
                # String booleans are automatically converted
                worker = MyWorker.options(mode="thread", blocking="true").init()
                assert worker.blocking is True
                ```

            Mode-Specific Options:
                ```python
                # Ray with resource requirements
                worker = MyWorker.options(
                    mode="ray",
                    num_cpus=2,
                    num_gpus=1
                ).init(multiplier=3)

                # Process with spawn context
                worker = MyWorker.options(
                    mode="process",
                    mp_context="spawn"
                ).init(multiplier=3)
                ```

            Future Unwrapping (Default Enabled):
                ```python
                # Automatic future unwrapping (default)
                producer = Worker1.options(mode="thread").init()
                consumer = Worker2.options(mode="thread").init()

                future = producer.compute(10)  # Returns BaseFuture
                result = consumer.process(future).result()  # future is auto-unwrapped

                # Disable unwrapping to pass futures as objects
                worker = MyWorker.options(mode="thread", unwrap_futures=False).init()
                result = worker.inspect_future(future).result()  # Receives BaseFuture object
                ```

            Worker Pools:
                ```python
                # Create a thread pool with 10 workers
                pool = MyWorker.options(mode="thread", max_workers=10).init(multiplier=3)
                future = pool.process(10)  # Dispatched to one of 10 workers

                # Process pool with load balancing
                pool = MyWorker.options(
                    mode="process",
                    max_workers=4,
                    load_balancing="least_active"
                ).init(multiplier=3)

                # On-demand workers for bursty workloads
                pool = MyWorker.options(
                    mode="ray",
                    on_demand=True,
                    max_workers=0  # Unlimited
                ).init(multiplier=3)
                ```

            Retries:
                ```python
                # Basic retry with exponential backoff
                worker = APIWorker.options(
                    mode="thread",
                    num_retries=3,
                    retry_algorithm="exponential",
                    retry_wait=1.0,
                    retry_jitter=0.3
                ).init()

                # Retry only on specific exceptions
                worker = APIWorker.options(
                    mode="thread",
                    num_retries=5,
                    retry_on=[ConnectionError, TimeoutError]
                ).init()

                # Custom exception filter
                worker = APIWorker.options(
                    mode="thread",
                    num_retries=3,
                    retry_on=lambda exception, **ctx: (
                        isinstance(exception, ValueError) and "retry" in str(exception)
                    )
                ).init()

                # Output validation for LLM responses
                worker = LLMWorker.options(
                    mode="thread",
                    num_retries=5,
                    retry_until=lambda result, **ctx: (
                        isinstance(result, dict) and "data" in result
                    )
                ).init()

                # Multiple validators (all must pass)
                worker = LLMWorker.options(
                    mode="thread",
                    num_retries=5,
                    retry_until=[
                        lambda result, **ctx: isinstance(result, str),
                        lambda result, **ctx: result.startswith("{"),
                        lambda result, **ctx: validate_json(result)
                    ]
                ).init()
                ```
        """
        # Import here to avoid circular imports
        from ...config import global_config

        # Get defaults for this mode from global config
        mode_defaults = global_config.get_defaults(mode)

        # Apply defaults for all parameters if not specified
        if blocking is _NO_ARG:
            blocking = mode_defaults.blocking

        if max_workers is _NO_ARG:
            max_workers = mode_defaults.max_workers

        if on_demand is _NO_ARG:
            on_demand = mode_defaults.on_demand

        if max_queued_tasks is _NO_ARG:
            max_queued_tasks = mode_defaults.max_queued_tasks

        if load_balancing is _NO_ARG:
            if on_demand:
                load_balancing = mode_defaults.load_balancing_on_demand
            else:
                load_balancing = mode_defaults.load_balancing

        if num_retries is _NO_ARG:
            num_retries = mode_defaults.num_retries

        if retry_algorithm is _NO_ARG:
            retry_algorithm = mode_defaults.retry_algorithm

        if retry_wait is _NO_ARG:
            retry_wait = mode_defaults.retry_wait

        if retry_jitter is _NO_ARG:
            retry_jitter = mode_defaults.retry_jitter

        # Apply default for unwrap_futures if not in kwargs
        if "unwrap_futures" not in kwargs:
            kwargs["unwrap_futures"] = mode_defaults.unwrap_futures

        # Apply default for retry_on if None (default is [Exception])
        if retry_on is None:
            retry_on = [Exception]

        return WorkerBuilder(
            worker_cls=cls,
            mode=mode,
            blocking=blocking,
            max_workers=max_workers,
            load_balancing=load_balancing,
            on_demand=on_demand,
            max_queued_tasks=max_queued_tasks,
            num_retries=num_retries,
            retry_on=retry_on,
            retry_algorithm=retry_algorithm,
            retry_wait=retry_wait,
            retry_jitter=retry_jitter,
            retry_until=retry_until,
            options=kwargs,  # Pass **kwargs as options dict
        )

    def __new__(cls, *args, **kwargs):
        """Override __new__ to support direct instantiation as sync mode."""
        # If instantiated directly (not via options), behave as sync mode
        if cls is Worker:
            raise TypeError("Worker cannot be instantiated directly. Subclass it or use @worker decorator.")

        # Check if this is being called from a proxy
        # This is a bit of a hack but allows: worker = MLModelWorker() to work
        instance = super().__new__(cls)
        return instance

    def __init__(self, *args, **kwargs):
        """Initialize the worker. Subclasses can override this freely.

        This method supports cooperative multiple inheritance, allowing Worker
        to be combined with model classes like morphic.Typed or pydantic.BaseModel.

        Examples:
            ```python
            # Regular Worker subclass
            class MyWorker(Worker):
                def __init__(self, value: int):
                    self.value = value

            # Worker + Typed
            class TypedWorker(Worker, Typed):
                name: str
                value: int = 0

            # Worker + BaseModel
            class PydanticWorker(Worker, BaseModel):
                name: str
                value: int = 0
            ```
        """
        # Support cooperative multiple inheritance with Typed/BaseModel
        # Try to call super().__init__() to propagate to other base classes
        try:
            super().__init__(*args, **kwargs)
        except TypeError as e:
            # object.__init__() doesn't accept arguments
            # This happens when Worker is the only meaningful base class
            if "object.__init__()" in str(e) or "no arguments" in str(e).lower():
                pass
            else:
                raise

options(*, mode: ExecutionMode, blocking: Union[bool, _NO_ARG_TYPE] = _NO_ARG, max_workers: Union[conint(ge=0), None, _NO_ARG_TYPE] = _NO_ARG, load_balancing: Union[LoadBalancingAlgorithm, _NO_ARG_TYPE] = _NO_ARG, on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG, max_queued_tasks: Union[conint(ge=0), None, _NO_ARG_TYPE] = _NO_ARG, num_retries: Union[conint(ge=0), _NO_ARG_TYPE] = _NO_ARG, retry_on: Optional[Any] = None, retry_algorithm: Union[RetryAlgorithm, _NO_ARG_TYPE] = _NO_ARG, retry_wait: Union[confloat(ge=0), _NO_ARG_TYPE] = _NO_ARG, retry_jitter: Union[confloat(ge=0, le=1), _NO_ARG_TYPE] = _NO_ARG, retry_until: Optional[Any] = None, **kwargs: Any) -> WorkerBuilder classmethod

Configure worker execution options.

Returns a WorkerBuilder that can be used to create worker instances with .init(args, *kwargs).

Type Validation:

This method uses the @validate decorator from morphic, providing: - Automatic type checking and conversion - String-to-bool coercion (e.g., "true" → True) - AutoEnum fuzzy matching for mode parameter - Enhanced error messages for invalid inputs

Parameters:

Name Type Description Default
mode ExecutionMode

Execution mode (sync, thread, process, asyncio, ray) Accepts string or ExecutionMode enum value

 required
blocking Union[bool, _NO_ARG_TYPE]

If True, method calls return results directly instead of futures Accepts bool or string representation ("true", "false", "1", "0") Default value determined by global_config..blocking

 _NO_ARG
max_workers Union[conint(ge=0), None, _NO_ARG_TYPE]

Maximum number of workers in pool (optional) - If None or 1: Creates single worker. If >1: Creates worker pool with specified size. - Sync/Asyncio: Must be 1 or None (raises error otherwise) - Default value determined by global_config..max_workers

 _NO_ARG
load_balancing Union[LoadBalancingAlgorithm, _NO_ARG_TYPE]

Load balancing algorithm (optional) - "round_robin": Distribute requests evenly - "least_active": Select worker with fewest active calls - "least_total": Select worker with fewest total calls - "random": Random selection - Default value determined by global_config..load_balancing (for pools) or global_config..load_balancing_on_demand (for on-demand pools)

 _NO_ARG
on_demand Union[bool, _NO_ARG_TYPE]

If True, create workers on-demand per request (default: False) - Workers are created for each request and destroyed after completion - Useful for bursty workloads or resource-constrained environments - Cannot be used with Sync/Asyncio modes - With max_workers=0: Unlimited concurrent workers (Ray) or limited to cpu_count()-1 (Thread/Process)

 _NO_ARG
max_queued_tasks Union[conint(ge=0), None, _NO_ARG_TYPE]

Maximum number of in-flight tasks per worker (default varies by mode) - Controls how many tasks can be submitted to a worker's backend before blocking - Per-worker limit: each worker in a pool has its own independent queue - Value of N means max N tasks submitted but not yet completed per worker - Automatically bypassed in blocking mode (unlimited submissions allowed) - Automatically bypassed in sync and asyncio modes - Prevents overload when submitting large batches (e.g., 5000+ tasks to Ray) - Default value determined by global_config..max_queued_tasks - See user guide for detailed usage: /docs/user-guide/limits.md#submission-queue

 _NO_ARG
unwrap_futures

If True, automatically unwrap BaseFuture arguments by calling .result() on them before passing to worker methods. This enables seamless composition of workers. Set to False to pass futures as-is. Default value determined by global_config..unwrap_futures

 required
limits

Resource protection and rate limiting (optional, defaults to empty LimitSet) - Pass LimitSet: Workers share the same limit pool - Pass List[Limit]: Each worker gets private limits (creates shared LimitSet for pools) - Omit parameter: Workers get empty LimitSet (self.limits.acquire() always succeeds) Workers always have self.limits available, even when no limits configured. See Worker docstring "Resource Protection with Limits" section for details.

 required
num_retries Union[conint(ge=0), _NO_ARG_TYPE]

Maximum number of retry attempts after initial failure Total attempts = num_retries + 1 (initial attempt). Set to 0 to disable retries (zero overhead). Default value determined by global_config..num_retries

 _NO_ARG
retry_on Optional[Any]

Exception types or callables that trigger retries (optional) - Single exception class: retry_on=ValueError - List of exceptions: retry_on=[ValueError, ConnectionError] - Callable filter: retry_on=lambda exception, **ctx: "retry" in str(exception) - Mixed list: retry_on=[ValueError, custom_filter] Default: [Exception] (retry on all exceptions when num_retries > 0)

 None
retry_algorithm Union[RetryAlgorithm, _NO_ARG_TYPE]

Backoff strategy for wait times Default value determined by global_config..retry_algorithm

 _NO_ARG
retry_wait Union[confloat(ge=0), _NO_ARG_TYPE]

Minimum wait time between retries in seconds Base wait time before applying strategy and jitter. Default value determined by global_config..retry_wait

 _NO_ARG
retry_jitter Union[confloat(ge=0, le=1), _NO_ARG_TYPE]

Jitter factor between 0 and 1 Uses Full Jitter algorithm from AWS: sleep = random(0, calculated_wait). Set to 0 to disable jitter. Prevents thundering herd when many workers retry. Default value determined by global_config..retry_jitter

 _NO_ARG
retry_until Optional[Any]

Validation functions for output (optional) - Single validator: retry_until=lambda result, **ctx: result.get("status") == "success" - List of validators: retry_until=[validator1, validator2] (all must pass) Validators receive result and context as kwargs. Return True for valid output. If validation fails, triggers retry even without exception. Useful for LLM output validation (JSON schema, XML format, etc.)

 None
**kwargs Any

Additional options passed to the worker implementation - For ray: num_cpus, num_gpus, resources, etc. - For process: mp_context (fork, spawn, forkserver)

 {}

Returns:

Type Description
WorkerBuilder

A WorkerBuilder instance that can create workers via .init()

Examples:

Basic Usage: 

# Configure and create worker
worker = MyWorker.options(mode="thread").init(multiplier=3)

Type Coercion: 

# String booleans are automatically converted
worker = MyWorker.options(mode="thread", blocking="true").init()
assert worker.blocking is True

Mode-Specific Options: 

# Ray with resource requirements
worker = MyWorker.options(
    mode="ray",
    num_cpus=2,
    num_gpus=1
).init(multiplier=3)

# Process with spawn context
worker = MyWorker.options(
    mode="process",
    mp_context="spawn"
).init(multiplier=3)

Future Unwrapping (Default Enabled): 

# Automatic future unwrapping (default)
producer = Worker1.options(mode="thread").init()
consumer = Worker2.options(mode="thread").init()

future = producer.compute(10)  # Returns BaseFuture
result = consumer.process(future).result()  # future is auto-unwrapped

# Disable unwrapping to pass futures as objects
worker = MyWorker.options(mode="thread", unwrap_futures=False).init()
result = worker.inspect_future(future).result()  # Receives BaseFuture object

Worker Pools: 

# Create a thread pool with 10 workers
pool = MyWorker.options(mode="thread", max_workers=10).init(multiplier=3)
future = pool.process(10)  # Dispatched to one of 10 workers

# Process pool with load balancing
pool = MyWorker.options(
    mode="process",
    max_workers=4,
    load_balancing="least_active"
).init(multiplier=3)

# On-demand workers for bursty workloads
pool = MyWorker.options(
    mode="ray",
    on_demand=True,
    max_workers=0  # Unlimited
).init(multiplier=3)

Retries: 

# Basic retry with exponential backoff
worker = APIWorker.options(
    mode="thread",
    num_retries=3,
    retry_algorithm="exponential",
    retry_wait=1.0,
    retry_jitter=0.3
).init()

# Retry only on specific exceptions
worker = APIWorker.options(
    mode="thread",
    num_retries=5,
    retry_on=[ConnectionError, TimeoutError]
).init()

# Custom exception filter
worker = APIWorker.options(
    mode="thread",
    num_retries=3,
    retry_on=lambda exception, **ctx: (
        isinstance(exception, ValueError) and "retry" in str(exception)
    )
).init()

# Output validation for LLM responses
worker = LLMWorker.options(
    mode="thread",
    num_retries=5,
    retry_until=lambda result, **ctx: (
        isinstance(result, dict) and "data" in result
    )
).init()

# Multiple validators (all must pass)
worker = LLMWorker.options(
    mode="thread",
    num_retries=5,
    retry_until=[
        lambda result, **ctx: isinstance(result, str),
        lambda result, **ctx: result.startswith("{"),
        lambda result, **ctx: validate_json(result)
    ]
).init()

Source code in src/concurry/core/worker/base_worker.py 
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663
@classmethod
@validate
def options(
    cls: Type[T],
    *,
    mode: ExecutionMode,
    blocking: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    max_workers: Union[conint(ge=0), None, _NO_ARG_TYPE] = _NO_ARG,
    load_balancing: Union[LoadBalancingAlgorithm, _NO_ARG_TYPE] = _NO_ARG,
    on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    max_queued_tasks: Union[conint(ge=0), None, _NO_ARG_TYPE] = _NO_ARG,
    # Retry parameters
    num_retries: Union[conint(ge=0), _NO_ARG_TYPE] = _NO_ARG,
    retry_on: Optional[Any] = None,
    retry_algorithm: Union[RetryAlgorithm, _NO_ARG_TYPE] = _NO_ARG,
    retry_wait: Union[confloat(ge=0), _NO_ARG_TYPE] = _NO_ARG,
    retry_jitter: Union[confloat(ge=0, le=1), _NO_ARG_TYPE] = _NO_ARG,
    retry_until: Optional[Any] = None,
    **kwargs: Any,
) -> WorkerBuilder:
    """Configure worker execution options.

    Returns a WorkerBuilder that can be used to create worker instances
    with .init(*args, **kwargs).

    **Type Validation:**

    This method uses the `@validate` decorator from morphic, providing:
    - Automatic type checking and conversion
    - String-to-bool coercion (e.g., "true" → True)
    - AutoEnum fuzzy matching for mode parameter
    - Enhanced error messages for invalid inputs

    Args:
        mode: Execution mode (sync, thread, process, asyncio, ray)
            Accepts string or ExecutionMode enum value
        blocking: If True, method calls return results directly instead of futures
            Accepts bool or string representation ("true", "false", "1", "0")
            Default value determined by global_config.<mode>.blocking
        max_workers: Maximum number of workers in pool (optional)
            - If None or 1: Creates single worker. If >1: Creates worker pool with specified size.
            - Sync/Asyncio: Must be 1 or None (raises error otherwise)
            - Default value determined by global_config.<mode>.max_workers
        load_balancing: Load balancing algorithm (optional)
            - "round_robin": Distribute requests evenly
            - "least_active": Select worker with fewest active calls
            - "least_total": Select worker with fewest total calls
            - "random": Random selection
            - Default value determined by global_config.<mode>.load_balancing (for pools)
              or global_config.<mode>.load_balancing_on_demand (for on-demand pools)
        on_demand: If True, create workers on-demand per request (default: False)
            - Workers are created for each request and destroyed after completion
            - Useful for bursty workloads or resource-constrained environments
            - Cannot be used with Sync/Asyncio modes
            - With max_workers=0: Unlimited concurrent workers (Ray) or
              limited to cpu_count()-1 (Thread/Process)
        max_queued_tasks: Maximum number of in-flight tasks per worker (default varies by mode)
            - Controls how many tasks can be submitted to a worker's backend before blocking
            - Per-worker limit: each worker in a pool has its own independent queue
            - Value of N means max N tasks submitted but not yet completed per worker
            - Automatically bypassed in blocking mode (unlimited submissions allowed)
            - Automatically bypassed in sync and asyncio modes
            - Prevents overload when submitting large batches (e.g., 5000+ tasks to Ray)
            - Default value determined by global_config.<mode>.max_queued_tasks
            - See user guide for detailed usage: /docs/user-guide/limits.md#submission-queue
        unwrap_futures: If True, automatically unwrap BaseFuture arguments
            by calling .result() on them before passing to worker methods. This enables
            seamless composition of workers. Set to False to pass futures as-is.
            Default value determined by global_config.<mode>.unwrap_futures
        limits: Resource protection and rate limiting (optional, defaults to empty LimitSet)
            - Pass LimitSet: Workers share the same limit pool
            - Pass List[Limit]: Each worker gets private limits (creates shared LimitSet for pools)
            - Omit parameter: Workers get empty LimitSet (self.limits.acquire() always succeeds)
            Workers always have self.limits available, even when no limits configured.
            See Worker docstring "Resource Protection with Limits" section for details.
        num_retries: Maximum number of retry attempts after initial failure
            Total attempts = num_retries + 1 (initial attempt).
            Set to 0 to disable retries (zero overhead).
            Default value determined by global_config.<mode>.num_retries
        retry_on: Exception types or callables that trigger retries (optional)
            - Single exception class: retry_on=ValueError
            - List of exceptions: retry_on=[ValueError, ConnectionError]
            - Callable filter: retry_on=lambda exception, **ctx: "retry" in str(exception)
            - Mixed list: retry_on=[ValueError, custom_filter]
            Default: [Exception] (retry on all exceptions when num_retries > 0)
        retry_algorithm: Backoff strategy for wait times
            Default value determined by global_config.<mode>.retry_algorithm
        retry_wait: Minimum wait time between retries in seconds
            Base wait time before applying strategy and jitter.
            Default value determined by global_config.<mode>.retry_wait
        retry_jitter: Jitter factor between 0 and 1
            Uses Full Jitter algorithm from AWS: sleep = random(0, calculated_wait).
            Set to 0 to disable jitter. Prevents thundering herd when many workers retry.
            Default value determined by global_config.<mode>.retry_jitter
        retry_until: Validation functions for output (optional)
            - Single validator: retry_until=lambda result, **ctx: result.get("status") == "success"
            - List of validators: retry_until=[validator1, validator2] (all must pass)
            Validators receive result and context as kwargs. Return True for valid output.
            If validation fails, triggers retry even without exception.
            Useful for LLM output validation (JSON schema, XML format, etc.)
        **kwargs: Additional options passed to the worker implementation
            - For ray: num_cpus, num_gpus, resources, etc.
            - For process: mp_context (fork, spawn, forkserver)

    Returns:
        A WorkerBuilder instance that can create workers via .init()

    Examples:
        Basic Usage:
            ```python
            # Configure and create worker
            worker = MyWorker.options(mode="thread").init(multiplier=3)
            ```

        Type Coercion:
            ```python
            # String booleans are automatically converted
            worker = MyWorker.options(mode="thread", blocking="true").init()
            assert worker.blocking is True
            ```

        Mode-Specific Options:
            ```python
            # Ray with resource requirements
            worker = MyWorker.options(
                mode="ray",
                num_cpus=2,
                num_gpus=1
            ).init(multiplier=3)

            # Process with spawn context
            worker = MyWorker.options(
                mode="process",
                mp_context="spawn"
            ).init(multiplier=3)
            ```

        Future Unwrapping (Default Enabled):
            ```python
            # Automatic future unwrapping (default)
            producer = Worker1.options(mode="thread").init()
            consumer = Worker2.options(mode="thread").init()

            future = producer.compute(10)  # Returns BaseFuture
            result = consumer.process(future).result()  # future is auto-unwrapped

            # Disable unwrapping to pass futures as objects
            worker = MyWorker.options(mode="thread", unwrap_futures=False).init()
            result = worker.inspect_future(future).result()  # Receives BaseFuture object
            ```

        Worker Pools:
            ```python
            # Create a thread pool with 10 workers
            pool = MyWorker.options(mode="thread", max_workers=10).init(multiplier=3)
            future = pool.process(10)  # Dispatched to one of 10 workers

            # Process pool with load balancing
            pool = MyWorker.options(
                mode="process",
                max_workers=4,
                load_balancing="least_active"
            ).init(multiplier=3)

            # On-demand workers for bursty workloads
            pool = MyWorker.options(
                mode="ray",
                on_demand=True,
                max_workers=0  # Unlimited
            ).init(multiplier=3)
            ```

        Retries:
            ```python
            # Basic retry with exponential backoff
            worker = APIWorker.options(
                mode="thread",
                num_retries=3,
                retry_algorithm="exponential",
                retry_wait=1.0,
                retry_jitter=0.3
            ).init()

            # Retry only on specific exceptions
            worker = APIWorker.options(
                mode="thread",
                num_retries=5,
                retry_on=[ConnectionError, TimeoutError]
            ).init()

            # Custom exception filter
            worker = APIWorker.options(
                mode="thread",
                num_retries=3,
                retry_on=lambda exception, **ctx: (
                    isinstance(exception, ValueError) and "retry" in str(exception)
                )
            ).init()

            # Output validation for LLM responses
            worker = LLMWorker.options(
                mode="thread",
                num_retries=5,
                retry_until=lambda result, **ctx: (
                    isinstance(result, dict) and "data" in result
                )
            ).init()

            # Multiple validators (all must pass)
            worker = LLMWorker.options(
                mode="thread",
                num_retries=5,
                retry_until=[
                    lambda result, **ctx: isinstance(result, str),
                    lambda result, **ctx: result.startswith("{"),
                    lambda result, **ctx: validate_json(result)
                ]
            ).init()
            ```
    """
    # Import here to avoid circular imports
    from ...config import global_config

    # Get defaults for this mode from global config
    mode_defaults = global_config.get_defaults(mode)

    # Apply defaults for all parameters if not specified
    if blocking is _NO_ARG:
        blocking = mode_defaults.blocking

    if max_workers is _NO_ARG:
        max_workers = mode_defaults.max_workers

    if on_demand is _NO_ARG:
        on_demand = mode_defaults.on_demand

    if max_queued_tasks is _NO_ARG:
        max_queued_tasks = mode_defaults.max_queued_tasks

    if load_balancing is _NO_ARG:
        if on_demand:
            load_balancing = mode_defaults.load_balancing_on_demand
        else:
            load_balancing = mode_defaults.load_balancing

    if num_retries is _NO_ARG:
        num_retries = mode_defaults.num_retries

    if retry_algorithm is _NO_ARG:
        retry_algorithm = mode_defaults.retry_algorithm

    if retry_wait is _NO_ARG:
        retry_wait = mode_defaults.retry_wait

    if retry_jitter is _NO_ARG:
        retry_jitter = mode_defaults.retry_jitter

    # Apply default for unwrap_futures if not in kwargs
    if "unwrap_futures" not in kwargs:
        kwargs["unwrap_futures"] = mode_defaults.unwrap_futures

    # Apply default for retry_on if None (default is [Exception])
    if retry_on is None:
        retry_on = [Exception]

    return WorkerBuilder(
        worker_cls=cls,
        mode=mode,
        blocking=blocking,
        max_workers=max_workers,
        load_balancing=load_balancing,
        on_demand=on_demand,
        max_queued_tasks=max_queued_tasks,
        num_retries=num_retries,
        retry_on=retry_on,
        retry_algorithm=retry_algorithm,
        retry_wait=retry_wait,
        retry_jitter=retry_jitter,
        retry_until=retry_until,
        options=kwargs,  # Pass **kwargs as options dict
    )

TaskWorker

concurry.core.worker.task_worker.TaskWorker

Bases: Worker

A generic worker for submitting arbitrary tasks.

TaskWorker is a concrete worker implementation that provides an Executor-like interface for executing arbitrary functions in different execution contexts (sync, thread, process, asyncio, ray). Unlike custom workers that define specific methods, TaskWorker is designed for general-purpose task execution.

This class implements the same interface as concurrent.futures.Executor: - submit(fn, args, kwargs): Submit a single task - map(fn, iterables, **kwargs): Submit multiple tasks with automatic iteration

This class is intended to be used by higher-level abstractions like WorkerExecutor and WorkerPool, or directly when you don't need custom worker methods.

Examples:

Basic Task Execution: 

from concurry import TaskWorker

# Create a task worker
worker = TaskWorker.options(mode="thread").init()

# Submit arbitrary functions
def compute(x, y):
    return x ** 2 + y ** 2

future = worker.submit(compute, 3, 4)
result = future.result()  # 25

worker.stop()

Using map() for Multiple Tasks: 

worker = TaskWorker.options(mode="process").init()

def square(x):
    return x ** 2

# Process multiple items
results = list(worker.map(square, range(10)))
print(results)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

worker.stop()

With Different Execution Modes: 

# Thread-based execution
thread_worker = TaskWorker.options(mode="thread").init()

# Process-based execution for CPU-intensive tasks
process_worker = TaskWorker.options(mode="process").init()

# Asyncio-based execution
async_worker = TaskWorker.options(mode="asyncio").init()

# Submit tasks to any of them
result1 = thread_worker.submit(lambda x: x * 2, 10).result()
result2 = process_worker.submit(lambda x: x ** 3, 5).result()
result3 = async_worker.submit(lambda x: x + 100, 7).result()

thread_worker.stop()
process_worker.stop()
async_worker.stop()

Blocking Mode: 

# Get results directly without futures
worker = TaskWorker.options(mode="thread", blocking=True).init()

result = worker.submit(lambda x: x * 10, 5)  # Returns 50 directly

worker.stop()

Multiple Tasks with map(): 

worker = TaskWorker.options(mode="process").init()

def factorial(n):
    if n <= 1:
        return 1
    return n * factorial(n - 1)

# Process multiple inputs concurrently
results = list(worker.map(factorial, range(1, 11)))
print(results)  # [1, 2, 6, 24, 120, 720, 5040, 40320, 362880, 3628800]

worker.stop()

With Timeout: 

import time
from concurry import TaskWorker

def slow_task(x):
    time.sleep(1)
    return x * 2

worker = TaskWorker.options(mode="thread").init()

# This will raise TimeoutError
try:
    results = list(worker.map(slow_task, range(5), timeout=0.5))
except TimeoutError:
    print("Task timed out!")

worker.stop()

With Bound Function: 

from concurry import TaskWorker

def compute(x, y):
    return x ** 2 + y ** 2

# Bind function during initialization
worker = TaskWorker.options(mode="thread").init(fn=compute)

# Submit without passing function
future = worker.submit(3, 4)
result = future.result()  # 25

# Map without passing function
results = list(worker.map([(1, 2), (3, 4), (5, 6)]))

# Call directly
result = worker(3, 4).result()  # 25

worker.stop()

With Progress Bar: 

from concurry import TaskWorker

def square(x):
    return x ** 2

worker = TaskWorker.options(mode="process").init(fn=square)

# Show progress bar during map
results = list(worker.map(range(100), progress=True))

# Custom progress bar configuration
results = list(worker.map(
    range(100),
    progress={"desc": "Processing", "ncols": 80}
))

worker.stop()

Source code in src/concurry/core/worker/task_worker.py 
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
class TaskWorker(Worker):
    """A generic worker for submitting arbitrary tasks.

    TaskWorker is a concrete worker implementation that provides an Executor-like
    interface for executing arbitrary functions in different execution contexts
    (sync, thread, process, asyncio, ray). Unlike custom workers that define
    specific methods, TaskWorker is designed for general-purpose task execution.

    This class implements the same interface as concurrent.futures.Executor:
    - submit(fn, *args, **kwargs): Submit a single task
    - map(fn, *iterables, **kwargs): Submit multiple tasks with automatic iteration

    This class is intended to be used by higher-level abstractions like
    WorkerExecutor and WorkerPool, or directly when you don't need custom worker methods.

    Examples:
        Basic Task Execution:
            ```python
            from concurry import TaskWorker

            # Create a task worker
            worker = TaskWorker.options(mode="thread").init()

            # Submit arbitrary functions
            def compute(x, y):
                return x ** 2 + y ** 2

            future = worker.submit(compute, 3, 4)
            result = future.result()  # 25

            worker.stop()
            ```

        Using map() for Multiple Tasks:
            ```python
            worker = TaskWorker.options(mode="process").init()

            def square(x):
                return x ** 2

            # Process multiple items
            results = list(worker.map(square, range(10)))
            print(results)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

            worker.stop()
            ```

        With Different Execution Modes:
            ```python
            # Thread-based execution
            thread_worker = TaskWorker.options(mode="thread").init()

            # Process-based execution for CPU-intensive tasks
            process_worker = TaskWorker.options(mode="process").init()

            # Asyncio-based execution
            async_worker = TaskWorker.options(mode="asyncio").init()

            # Submit tasks to any of them
            result1 = thread_worker.submit(lambda x: x * 2, 10).result()
            result2 = process_worker.submit(lambda x: x ** 3, 5).result()
            result3 = async_worker.submit(lambda x: x + 100, 7).result()

            thread_worker.stop()
            process_worker.stop()
            async_worker.stop()
            ```

        Blocking Mode:
            ```python
            # Get results directly without futures
            worker = TaskWorker.options(mode="thread", blocking=True).init()

            result = worker.submit(lambda x: x * 10, 5)  # Returns 50 directly

            worker.stop()
            ```

        Multiple Tasks with map():
            ```python
            worker = TaskWorker.options(mode="process").init()

            def factorial(n):
                if n <= 1:
                    return 1
                return n * factorial(n - 1)

            # Process multiple inputs concurrently
            results = list(worker.map(factorial, range(1, 11)))
            print(results)  # [1, 2, 6, 24, 120, 720, 5040, 40320, 362880, 3628800]

            worker.stop()
            ```

        With Timeout:
            ```python
            import time
            from concurry import TaskWorker

            def slow_task(x):
                time.sleep(1)
                return x * 2

            worker = TaskWorker.options(mode="thread").init()

            # This will raise TimeoutError
            try:
                results = list(worker.map(slow_task, range(5), timeout=0.5))
            except TimeoutError:
                print("Task timed out!")

            worker.stop()
            ```

        With Bound Function:
            ```python
            from concurry import TaskWorker

            def compute(x, y):
                return x ** 2 + y ** 2

            # Bind function during initialization
            worker = TaskWorker.options(mode="thread").init(fn=compute)

            # Submit without passing function
            future = worker.submit(3, 4)
            result = future.result()  # 25

            # Map without passing function
            results = list(worker.map([(1, 2), (3, 4), (5, 6)]))

            # Call directly
            result = worker(3, 4).result()  # 25

            worker.stop()
            ```

        With Progress Bar:
            ```python
            from concurry import TaskWorker

            def square(x):
                return x ** 2

            worker = TaskWorker.options(mode="process").init(fn=square)

            # Show progress bar during map
            results = list(worker.map(range(100), progress=True))

            # Custom progress bar configuration
            results = list(worker.map(
                range(100),
                progress={"desc": "Processing", "ncols": 80}
            ))

            worker.stop()
            ```
    """

    def __init__(self, fn: Optional[Callable] = None):
        """Initialize the TaskWorker.

        Args:
            fn: Optional callable to bind to this worker. When provided,
                submit() and map() can be called without passing a function.
        """
        super().__init__()
        self._bound_fn = fn

Task Decorator

concurry.core.worker.task_decorator.task(*, mode: ExecutionMode, on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG, **kwargs: Any) -> Callable

Decorator to create a TaskWorker bound to a function.

This decorator automatically creates and initializes a TaskWorker with the decorated function, enabling easy parallelization without manual worker management.

Parameters:

Name Type Description Default
mode ExecutionMode

Execution mode (sync, thread, process, asyncio, ray). Defaults to ExecutionMode.Sync

 required
on_demand Union[bool, _NO_ARG_TYPE]

Create workers on-demand. If not specified, uses global_config.defaults.task_decorator_on_demand (defaults to True). Note: on_demand is automatically set to False for Sync and Asyncio modes.

 _NO_ARG
**kwargs Any

All other Worker.options() parameters are supported.

 {}

Returns:

Type Description
Callable

Initialized TaskWorker instance bound to the decorated function

Examples:

Basic Usage: 

from concurry import task

@task(mode="thread", max_workers=4)
def process_item(x):
    return x ** 2

# Call like a function (returns future)
future = process_item(10)
result = future.result()  # 100

# Or use submit/map explicitly
future = process_item.submit(10)
results = list(process_item.map(range(10)))

With Limits: 

from concurry import task, RateLimit

limits = [RateLimit(key="api", capacity=100, window_seconds=60)]

@task(mode="thread", limits=limits)
def call_api(prompt, limits):  # limits param detected automatically
    with limits.acquire(requested={"api": 1}):
        return external_api(prompt)

result = call_api("Hello").result()

With Progress Bar: 

@task(mode="process", max_workers=4)
def compute(x):
    return x ** 2

results = list(compute.map(range(1000), progress=True))

Source code in src/concurry/core/worker/task_decorator.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
@validate
def task(
    *,
    mode: ExecutionMode,
    on_demand: Union[bool, _NO_ARG_TYPE] = _NO_ARG,
    **kwargs: Any,
) -> Callable:
    """Decorator to create a TaskWorker bound to a function.

    This decorator automatically creates and initializes a TaskWorker with the
    decorated function, enabling easy parallelization without manual worker management.

    Args:
        mode: Execution mode (sync, thread, process, asyncio, ray).
            Defaults to ExecutionMode.Sync
        on_demand: Create workers on-demand. If not specified, uses
            global_config.defaults.task_decorator_on_demand (defaults to True).
            Note: on_demand is automatically set to False for Sync and Asyncio modes.
        **kwargs: All other Worker.options() parameters are supported.

    Returns:
        Initialized TaskWorker instance bound to the decorated function

    Examples:
        Basic Usage:
            ```python
            from concurry import task

            @task(mode="thread", max_workers=4)
            def process_item(x):
                return x ** 2

            # Call like a function (returns future)
            future = process_item(10)
            result = future.result()  # 100

            # Or use submit/map explicitly
            future = process_item.submit(10)
            results = list(process_item.map(range(10)))
            ```

        With Limits:
            ```python
            from concurry import task, RateLimit

            limits = [RateLimit(key="api", capacity=100, window_seconds=60)]

            @task(mode="thread", limits=limits)
            def call_api(prompt, limits):  # limits param detected automatically
                with limits.acquire(requested={"api": 1}):
                    return external_api(prompt)

            result = call_api("Hello").result()
            ```

        With Progress Bar:
            ```python
            @task(mode="process", max_workers=4)
            def compute(x):
                return x ** 2

            results = list(compute.map(range(1000), progress=True))
            ```
    """
    # Import here to avoid circular imports
    from ...config import global_config

    local_config = global_config.clone()
    # Apply default for on_demand if not specified
    if on_demand is _NO_ARG:
        # on_demand is not supported for Sync and Asyncio modes
        if mode in (ExecutionMode.Sync, ExecutionMode.Asyncio):
            on_demand = False
        else:
            on_demand = local_config.defaults.task_decorator_on_demand
    on_demand: bool = bool(on_demand)

    def decorator(fn: Callable) -> TaskWorker:
        # Check if function accepts 'limits' parameter
        fn_args = get_fn_args(fn)
        has_limits_param = "limits" in fn_args

        # Extract limits from kwargs if present
        limits = kwargs.pop("limits", None)

        # If function has limits param and limits were provided, wrap function
        if has_limits_param and limits is not None:
            # Store original function
            original_fn = fn

            # Create wrapper that injects limits
            def wrapper_with_limits(*args, **fn_kwargs):
                # Only inject if not already provided
                if "limits" not in fn_kwargs:
                    # Get limits from worker instance (stored as closure variable)
                    fn_kwargs["limits"] = wrapper_with_limits._worker_limits
                return original_fn(*args, **fn_kwargs)

            # Mark wrapper to get limits later
            wrapper_with_limits._needs_limits = True
            wrapper_with_limits._original_fn = original_fn
            fn = wrapper_with_limits

        # Create worker options
        # Note: limits was already popped from kwargs earlier
        builder = TaskWorker.options(
            mode=mode,
            on_demand=on_demand,
            limits=limits,
            **kwargs,
        )

        # Initialize worker with the function
        worker = builder.init(fn=fn)

        # If we created a wrapper that needs limits, inject them now
        if hasattr(fn, "_needs_limits") and fn._needs_limits:
            # Access limits from the worker's underlying proxy
            # The worker is a WorkerProxy/WorkerProxyPool instance
            if hasattr(worker, "limits"):
                fn._worker_limits = worker.limits
            else:
                # For pools, we need to get limits from the first worker or the pool itself
                # Since limits are passed through, they should be accessible
                fn._worker_limits = None

        # Preserve function metadata
        if hasattr(fn, "_original_fn"):
            original = fn._original_fn
        else:
            original = fn

        worker.__name__ = original.__name__
        worker.__doc__ = original.__doc__
        worker.__module__ = original.__module__
        worker.__qualname__ = original.__qualname__

        # Add __del__ for cleanup
        def cleanup(self):
            try:
                if hasattr(self, "_stopped") and not self._stopped:
                    self.stop()
            except Exception:
                pass

        # Bind cleanup method to worker instance
        worker.__del__ = lambda: cleanup(worker)

        return worker

    return decorator