Crochet

Crochet allows blocking and threaded applications (such as Flask or Django) to utilize the Twisted networking framework without manually managing the Twisted reactor or event loop. It provides a completely blocking interface to Twisted operations.

# Example usage pattern (based on blockingdns.py)
# This allows you to perform Twisted-based network operations
# in a blocking manner.
import crochet
crochet.setup()

# After setup, you can use Twisted components directly
# without calling reactor.run()

While Python's async/await syntax provides an illusion of blocking, it requires the async keyword to percolate through your entire call stack. You can only use await inside functions marked as async.

In contrast, Crochet allows for true blocking integration. You can block on an asynchronous event using a standard, normal Python function call without needing to mark the function or its callers as async. This allows you to restrict asynchronous behavior to specific parts of your program without changing the signature of your entire codebase.

Because uWSGI, multiprocessing, and Celery use fork() by default on Unix, they clone the parent process in a way that breaks Crochet. You can resolve this using one of two strategies:

Strategy 1: Avoid fork() (Recommended)

• uWSGI: Use the --lazy-apps command-line option.
• multiprocessing: Use the spawn or forkserver start methods (available in Python 3).
• Celery: (N/A - see Strategy 2)

Strategy 2: Defer crochet.setup() to the child process

If you cannot avoid fork(), ensure crochet.setup() is only called after the process has branched:

• uWSGI: Call crochet.setup() inside the WSGI application function.
• multiprocessing: Call crochet.setup() only inside the child process.
• Celery: Call crochet.setup() only inside your tasks.

To use Crochet, you must first start the Twisted reactor in a background thread managed by Crochet. Call crochet.setup() to perform this initialization.

Since Crochet is designed to be used as a library, it is safe to call setup() multiple times; only the first call will actually perform the setup, and subsequent calls will be ignored. This allows multiple libraries to call crochet.setup() without conflict.

from crochet import setup
setup()

When testing code decorated with @wait_for or @run_in_reactor, you can bypass the Crochet layer to test the underlying Twisted logic directly. The decorated function exposes the original, undecorated function via the __wrapped__ attribute.

from crochet import wait_for

@wait_for
def add(a, b):
    # Twisted logic here
    return a + b

# In your test:
# This returns an EventualResult (requires crochet setup)
res = add(1, 2)

# This calls the underlying function directly (no crochet needed)
val = add.__wrapped__(1, 2)
assert val == 3

To improve testability and maintainability, keep the code inside @wait_for and @run_in_reactor wrappers to a minimum.

Because this code straddles two different threads (the application thread and the reactor thread), it is inherently more difficult to unit test. The recommended pattern is to push the core Twisted logic into a low-level layer and use a separate higher-level layer to hide those Twisted details from your application code. This separation makes the core logic easier to test in isolation.

Crochet provides type hints for Python 3.6+, but because of limitations in how Callable generics are constructed, the arguments of functions decorated with @run_in_reactor cannot be automatically checked. To enable proper type checking for these decorated functions, you must install the crochet.mypy plugin in your mypy configuration (e.g., in mypy.ini).

[mypy]
plugins = crochet.mypy

If your application is already running the Twisted reactor (for example, using Twisted as a WSGI container), Crochet's default behavior of starting a new reactor in a thread will cause conflicts.

To prevent Crochet from starting its own reactor, call crochet.no_setup() before importing any libraries that might call crochet.setup().

from crochet import no_setup
no_setup()

# Now import libraries that use crochet internally
import blocking_library

from twisted.internet import reactor
# ... setup app ...
reactor.run()

You can install Crochet using pip to enable the use of Twisted from blocking code.

$ pip install crochet

Twisted can produce a high volume of log messages. To limit the output to error messages only, configure the twisted logger using the standard logging module.

import logging
logging.getLogger('twisted').setLevel(logging.ERROR)