duck.contrib.sync.smart_async

Smart module for high-concurrency async execution of synchronous callables, with intelligent handling of transactional/atomic operations.

Features:

• Runs regular sync code on any available thread for maximal concurrency.

• Routes transactional/atomic database operations to a pool of specialized threads.

• Ensures thread (and connection) affinity for sync-to-async calls within a transaction context.

• Dynamically scales up transactional threads as needed, up to max_threads.

• Reliable queueing and error handling; safe for production usage.

Usage Example:

import time
import asyncio

def regular_func(x):
    time.sleep(1)
    return x * x

async def atomic_func(num_times):
    results = []
    x = 0

    def some_db_func():
        nonlocal x
        time.sleep(1)
        results.append(x + 1)
        x += 1

    with transaction_context():
        for i in range(num_times):
            await smart_sync_to_async(some_db_func)()
    return results

async def main():
    print("Regular ops concurrently:")
    st_time = time.time()
    results = await asyncio.gather(
        *(smart_sync_to_async(regular_func)(i) for i in range(8))
    )
    print(results)

    st_time = time.time()
    print("
Atomic ops sequentially (single transaction context):")
    atomic_results = await atomic_func(8)
    print(atomic_results)

asyncio.run(main())

The core principal of async responsiveness, use of small tasks rather than awaiting long running task or even converting to async.

Module Contents

Classes

TransactionThread	Dedicated thread for executing atomic/transactional database operations. Each thread maintains its own DB connection context. All tasks submitted are executed serially, preserving transaction context.
TransactionThreadPool	Dynamically scalable pool of TransactionThread objects.
disable_transaction_context	Context manager that temporarily disables any active transaction context. While inside this block, in_transaction_context() returns None.
transaction_context	Custom transaction context manager for testing purposes. This context manager simulates Django’s transaction.atomic() for testing. Sets a contextvar so that in_transaction_context() can detect when code is running inside a transaction context.

Functions

in_transaction_context	Returns a unique ID (str) if currently inside a DB transaction context/atomic block. Notes:
is_transactional	Heuristically determine if a function is transactional/atomic. For Django: checks for ‘transaction.atomic’ in source or ‘is_atomic’ attribute.
sync_to_async	High-concurrency async wrapper for synchronous functions.

Data

T
_TRANSACTION_THREAD_POOL
_transaction_context_id_var

API

duck.contrib.sync.smart_async.T

‘TypeVar(…)’

exception duck.contrib.sync.smart_async.TaskTookTooLongWarning

Bases: UserWarning

Warning when a task took too much time executing as this might exhaust the threadpool and might cause significant performance degradation (hangs subsequently).

Initialization

Initialize self. See help(type(self)) for accurate signature.

class duck.contrib.sync.smart_async.TransactionThread(context_id=None)

Bases: threading.Thread

Dedicated thread for executing atomic/transactional database operations. Each thread maintains its own DB connection context. All tasks submitted are executed serially, preserving transaction context.

Notes:

• This thread can also be used in non-transactional general contexts.

Initialization

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.init()) before doing anything else to the thread.

__repr__()

__str__()

current_task_executing() → Optional[Any]

Returns the current task/callable being executed.

is_free() → bool

Returns True if the thread is idle (no task running and queue empty).

run()

shutdown()

Shutsdown the thread but if no task is being executed or after current task finishes.

submit(func: Callable[..., duck.contrib.sync.smart_async.T], *args, **kwargs) → asyncio.Future

Puts the task in queue for execution.

Returns:

An asynchronous future you can wait for in async context.

Return type:

asyncio.Fututure

class duck.contrib.sync.smart_async.TransactionThreadPool(max_threads: Optional[int] = None, auto_free_general_threads: bool = True, general_threads_free_level: Optional[int] = None, headroom_factor: float = 1.5, idle_grace_period: float = 30.0, load_decay: float = 0.9, max_free_per_cycle: int = 1, load_check_interval: float = 0.05, enable_background_sweep: bool = True, sweep_interval: Optional[float] = None)

Dynamically scalable pool of TransactionThread objects.

Features:

• Context-affinity: threads created with a context_id are stored and reused for that context. These threads are NOT auto-freed, since callers may rely on persistent context-affinity.

• General threads: threads without context_id are pooled and freed adaptively, based on recent concurrency rather than a fixed percentage, so the pool stays warm through a busy period and only shrinks once load has genuinely settled.

• Thread creation is bounded by max_threads, which applies to both context and general threads.

Initialization

Initialize the thread pool.

Parameters:

• max_threads – Maximum number of threads to create (per context/general threads). If None, uses a reasonable default from duck.utils.threading.get_max_workers().

• auto_free_general_threads – If True, idle general threads beyond the target size are automatically shut down.

• general_threads_free_level – Percentage (0-100) of general threads to keep free before freeing extra idle ones. If None (default), the pool instead sizes itself adaptively from recent load (see headroom_factor, idle_grace_period, load_decay). Set this to an int to opt back into the old fixed-percentage behavior.

• headroom_factor – Only used when general_threads_free_level is None. Multiplier applied to the estimated concurrent load to size the pool, e.g. 1.5 keeps 50% spare capacity above recent usage so a following burst doesn’t immediately need new threads.

• idle_grace_period – Only used when general_threads_free_level is None. Seconds a general thread must sit idle before it becomes eligible for freeing. Protects against freeing a thread that a following request would have reused.

• load_decay – Only used when general_threads_free_level is None. Decay factor (0-1) for the load estimate once demand drops. Closer to 1 means the pool stays warm longer after a peak.

• max_free_per_cycle – Maximum number of general threads freed per call, to avoid thrashing (rapid free/recreate cycles) under bursty load.

• load_check_interval – Minimum seconds between recomputing the load estimate and running a free check. get_thread is called on every request, so under high call rates this throttles that O(n) work to at most once per interval instead of every single call. Thread selection itself is never throttled, only this bookkeeping.

• enable_background_sweep – If True, a daemon thread periodically runs the free check on its own, so idle general threads still get freed after a load spike even if no further get_thread calls arrive to trigger it. Set False to only free on demand, as part of get_thread.

• sweep_interval – Seconds between background sweeps. If None, defaults to max(idle_grace_period, 5.0).

free_general_threads_by_level(ignore_threads: Optional[List[duck.contrib.sync.smart_async.TransactionThread]] = None) → None

Free idle general threads to respect the fixed general_threads_free_level percentage.

Calculation:

desired_free_threads = ceil(max_threads * (free_level / 100))

if current_free_threads > desired_free_threads:
    free (current_free_threads - desired_free_threads)  # keep >= 1

Parameters:

ignore_threads – Threads to exclude from freeing.

free_general_threads_by_load(ignore_threads: Optional[List[duck.contrib.sync.smart_async.TransactionThread]] = None) → None

Free idle general threads that exceed the load-adjusted target size.

The target pool size is load_estimate * headroom_factor, clamped to [1, max_threads]. Only threads idle for at least idle_grace_period seconds are eligible, and at most max_free_per_cycle threads are freed per call, so the pool shrinks gradually rather than in one big drop.

Parameters:

ignore_threads – Threads to exclude from freeing, e.g. a thread about to be returned to a caller.

get_thread(context_id: Optional[str] = None) → duck.contrib.sync.smart_async.TransactionThread

Return a TransactionThread appropriate for the provided context_id.

If a context_id is given, the pool attempts to return a dedicated thread for that context, creating one if necessary (and if under max_threads). Context-bound threads are not auto-freed by the pool.

If context_id is None, returns an available general thread if free, otherwise may create a new one (subject to max_threads). This may trigger freeing of extra general threads once load has settled.

Parameters:

context_id – The context ID of the thread that is needed to run the task.

Returns:

The thread matching the context ID, or any free/appropriate general thread if no context_id was provided.

Return type:

TransactionThread

mark_active(thread: duck.contrib.sync.smart_async.TransactionThread) → None

Record that a general thread was just handed out to a caller.

Parameters:

thread – The general thread that is about to be used.

maybe_free_general_threads(ignore_threads: Optional[List[duck.contrib.sync.smart_async.TransactionThread]] = None) → None

Free extra idle general threads, using whichever sizing strategy is configured.

If general_threads_free_level was set, the pool targets a fixed percentage of free threads. Otherwise it sizes itself adaptively from recent concurrency via free_general_threads_by_load.

Parameters:

ignore_threads – Threads to exclude from freeing, e.g. a thread about to be returned to a caller.

run_background_sweep() → None

Loop that recomputes the load estimate and runs a free check every sweep_interval seconds, until sweep_stop_event is set.

should_run_load_check() → bool

Decide whether enough time has passed to recompute load and run a free check, per load_check_interval.

Returns:

True if the check should run now, in which case the internal timestamp is advanced so the next call waits a full interval.

Return type:

bool

shutdown(wait: bool = True)

Stop all running threads attached to this pool.

Parameters:

wait – Whether to wait for all threads to stop. Defaults to True.

start_background_sweep() → None

Start the daemon thread that periodically frees idle general threads.

This runs independently of get_thread calls, so a pool that grew during a burst still shrinks back down even if no further requests arrive to trigger the check.

submit(func: Callable[..., duck.contrib.sync.smart_async.T], *args, context_id=None, **kwargs) → asyncio.Future

Queue a task for execution on the thread matching context_id, or any free/appropriate thread if no context_id is provided.

Parameters:

• func – The callable to run on the selected thread.

• *args – Positional arguments passed through to func.

• context_id – Context ID used to pick a dedicated thread. If None, a general thread is used instead.

• **kwargs – Keyword arguments passed through to func.

Returns:

An asynchronous future you can await in async context.

Return type:

asyncio.Future

Raises:

AssertionError – If the returned thread from get_thread is dead.

update_load_estimate() → None

Update the EWMA estimate of concurrent general-thread usage.

Rises immediately to match a new peak, so a burst is captured without lag, then decays slowly via load_decay once demand drops. This keeps extra threads warm for a while after a spike instead of freeing them right before the next spike arrives.

duck.contrib.sync.smart_async._TRANSACTION_THREAD_POOL

‘TransactionThreadPool(…)’

duck.contrib.sync.smart_async._transaction_context_id_var

‘ContextVar(…)’

class duck.contrib.sync.smart_async.disable_transaction_context

Context manager that temporarily disables any active transaction context. While inside this block, in_transaction_context() returns None.

Usage:

with transaction_context():
    print(in_transaction_context())  # Not None
    with disable_transaction_context():
        print(in_transaction_context())  # None
    print(in_transaction_context())  # Not None

__enter__()

__exit__(exc_type, exc_val, exc_tb)

duck.contrib.sync.smart_async.in_transaction_context() → Optional[str]

Returns a unique ID (str) if currently inside a DB transaction context/atomic block. Notes:

• For Django: True if in transaction.atomic, returns thread identity.

• For unsupported ORMs: returns None.

duck.contrib.sync.smart_async.is_transactional(func: Callable) → bool

Heuristically determine if a function is transactional/atomic. For Django: checks for ‘transaction.atomic’ in source or ‘is_atomic’ attribute.

duck.contrib.sync.smart_async.sync_to_async(func: Callable[..., duck.contrib.sync.smart_async.T], *outer_args, **outer_kwargs) → Callable[..., asyncio.Future]

High-concurrency async wrapper for synchronous functions.

• Runs sync code in any available thread for maximum concurrency.

• Detects atomic/transactional operations and routes them to a pool of specialized threads, ensuring all DB operations within a transaction run on the same thread/connection.

• If called inside a transaction context, all sync_to_async calls for that transaction use the same thread.

• Returns an awaitable Future with the result.

class duck.contrib.sync.smart_async.transaction_context

Custom transaction context manager for testing purposes. This context manager simulates Django’s transaction.atomic() for testing. Sets a contextvar so that in_transaction_context() can detect when code is running inside a transaction context.

__enter__()

__exit__(exc_type, exc_val, exc_tb)