Python run a block of code in a background thread immediately

thread_start_immed.py

"""Technically, you don't need this library. You *could* just DIY it with a clever pile of `lambda` decorators:

>>> import threading, functools

>>> @lambda t: t.start() or t
... @lambda f: threading.Thread(target=f, daemon=True)
... def worker_1():
...   import time
...   time.sleep(1)
...   print(threading.current_thread())

>>> worker_1.join(); print(worker_1) #doctest: +ELLIPSIS
<Thread(... (worker_1), started ...)>
<Thread(... (worker_1), stopped ...)>

>>> @lambda t: t.start() or t
... @lambda f: threading.Thread(target=f, daemon=True)
... @lambda f: functools.wraps(f)(lambda *a, **k: asyncio.run(f(*a, **k)))
... async def worker_2():
...   import asyncio
...   await asyncio.sleep(1)
...   print(threading.current_thread())

>>> worker_2.join(); print(worker_2) #doctest: +ELLIPSIS
<Thread(... (worker_2), started ...)>
<Thread(... (worker_2), stopped ...)>
"""

import asyncio
from functools import update_wrapper
from inspect import iscoroutinefunction
import threading

__all__ = ["background_thread", "thread_start_immed"]

if __debug__:
    from collections.abc import Callable


def thread_start_immed(*args, **kwargs):
    # Kick the tires by running `python -m doctest "thread_start_immed.py"`!
    """Decorator to kick off blocks of code in a background thread.

    >>> from thread_start_immed import * #doctest: +SKIP

    >>> @thread_start_immed("one thread", 3, _threading_daemon=False)
    ... def worker_1(name, time_):
    ...     import time
    ...     time.sleep(time_)
    ...     print(f"Hello from {name}!")

    >>> print(worker_1) #doctest: +ELLIPSIS
    <Thread(... (worker_1), started ...)>

    >>> @thread_start_immed("a second thread... sooner", 1, _threading_daemon=False)
    ... async def worker_2(name, time_):
    ...     import asyncio
    ...     await asyncio.sleep(time_)
    ...     print(f"Hello from {name}??")

    >>> print(worker_2) #doctest: +ELLIPSIS
    <Thread(... (worker_2), started ...)>

    >>> worker_1.join(); worker_2.join()
    Hello from a second thread... sooner??
    Hello from one thread!

    The worker function may be `def` or `def async`, it makes no difference.
    (However, remember that you **MUST NOT** `await` any object which is
    owned by a different event loop; if you need to synchronize multiple
    event loops in a nonblocking way, consider the excellent `aiologic`
    library, https://pypi.org/project/aiologic/ .)

    While it is technically legal not to use arguments, and to pass all
    worker parameters as closes, explicitly passing anything the worker
    encapsulates can make your code more self-documenting thus is *highly*
    recommended.
    """
    return lambda f: _thread_start_immed(f, *args, **kwargs)


def _thread_start_immed(func, /, *args, _coroutine_runner=asyncio.run, **kwargs):
    thread_kwargs = {
        key[11:]: kwargs.pop(key)
        for key in [
            key for key in kwargs.keys()
            if key.startswith('_threading_')
        ]
    }
    thread_kwargs.setdefault('daemon', True)

    if _coroutine_runner is None:
        target = func
    else:
        assert isinstance(_coroutine_runner, Callable), "_coroutine_runner must be Callable"

        # Essentially 100% of the time, this is what you want.
        # If you don't want it to be wrapped (which should should virtually never
        # be the case, as a thread will just die if its target returns an unexecuted
        # coroutine,) then pass _coroutine_runner=None,
        # and if you want your function to be wrapped in some more complex way than this,
        # then just do that yourself.
        if iscoroutinefunction(func):
            runner_kwargs = {
                key[9:]: kwargs.pop(key)
                for key in [
                    key for key in kwargs.keys()
                    if key.startswith('_asyncio_')
                ]
            }

            target = lambda *a, **k: _coroutine_runner(func(*a, **k, **runner_kwargs))
            target = update_wrapper(target, original_target) # on a different line to make tracebacks prettier
        else:
            target = func

    t = threading.Thread(target=target, args=args, kwargs=kwargs, **thread_kwargs)
    t.start()
    return t