### Setup Crochet Reactor

Source: https://github.com/itamarst/crochet/blob/master/docs/api.md

Initialize Crochet by calling the setup() function. This starts Twisted's reactor in a managed thread. Multiple calls are safe; only the first one initializes the reactor.

```python
from crochet import setup
setup()
```

--------------------------------

### Initialize Crochet with setup()

Source: https://context7.com/itamarst/crochet/llms.txt

Initializes Crochet by starting the Twisted reactor in a background thread. This must be called before using Crochet decorators. Python logging is also configured.

```python
from crochet import setup
import logging

# Configure Python logging
logging.basicConfig(level=logging.DEBUG)

# Initialize Crochet - starts the reactor in a background thread
setup()

# Now you can use @wait_for and @run_in_reactor decorators
print("Crochet initialized and reactor running in background")
```

--------------------------------

### Run Twisted Reactor with WSGI Application

Source: https://github.com/itamarst/crochet/blob/master/docs/introduction.md

This snippet shows the standard setup for running a Twisted application. It configures logging, a thread pool for WSGI, and starts the TCP listener.

```python
def main():
    log.startLogging(sys.stdout)
    pool = reactor.getThreadPool()
    reactor.listenTCP(5000, Site(WSGIResource(reactor, pool, application)))
    reactor.run()

if __name__ == '__main__':
    main()
```

--------------------------------

### Install Crochet

Source: https://github.com/itamarst/crochet/blob/master/docs/introduction.md

Command to install the Crochet library via pip.

```bash
$ pip install crochet
```

--------------------------------

### DNS Lookup with Crochet (Blocking)

Source: https://github.com/itamarst/crochet/blob/master/docs/index.md

Performs a DNS lookup using Twisted's APIs, wrapped in a blocking call with a 5-second timeout. This example shows how to use Crochet to get a completely blocking interface to Twisted without running the reactor yourself. It requires the `crochet` library and Twisted's `twisted.names` module.

```python
#!/usr/bin/python
"""
Do a DNS lookup using Twisted's APIs.
"""
from __future__ import print_function

# The Twisted code we'll be using:
from twisted.names import client

from crochet import setup, wait_for
setup()


# Crochet layer, wrapping Twisted's DNS library in a blocking call.
@wait_for(timeout=5.0)
def gethostbyname(name):
    """Lookup the IP of a given hostname.

    Unlike socket.gethostbyname() which can take an arbitrary amount of time
    to finish, this function will raise crochet.TimeoutError if more than 5
    seconds elapse without an answer being received.
    """
    d = client.lookupAddress(name)
    d.addCallback(lambda result: result[0][0].payload.dottedQuad())
    return d


if __name__ == '__main__':
    # Application code using the public API - notice it works in a normal
    # blocking manner, with no event loop visible:
    import sys
    name = sys.argv[1]
    ip = gethostbyname(name)
    print(name, "->", ip)

```

--------------------------------

### Crochet Initialization

Source: https://context7.com/itamarst/crochet/llms.txt

Initializes the Crochet library, starting the Twisted reactor in a background thread. This must be called before using Crochet decorators.

```APIDOC
## setup()

### Description
Initializes the Crochet library by starting Twisted's reactor in a background thread and connecting Twisted's logging to Python's standard logging module. This function must be called before using any Crochet decorators. Multiple calls are safe and have no additional effect.

### Method
Function Call

### Endpoint
N/A

### Parameters
None

### Request Example
```python
from crochet import setup
import logging

# Configure Python logging
logging.basicConfig(level=logging.DEBUG)

# Initialize Crochet - starts the reactor in a background thread
setup()

# Now you can use @wait_for and @run_in_reactor decorators
print("Crochet initialized and reactor running in background")
```

### Response
None (initializes the reactor)

### Response Example
None
```

--------------------------------

### DNS Lookup with Crochet (Async/Await)

Source: https://github.com/itamarst/crochet/blob/master/docs/index.md

An equivalent DNS lookup example using async/await syntax, also wrapped with Crochet's `@wait_for` decorator for a blocking interface. This demonstrates how to integrate modern asynchronous Python features with Crochet.

```python
@wait_for(timeout=5.0)
async def gethostbyname(name):
    result = await client.lookupAddress(name)
    return result[0][0].payload.dottedQuad()

```

--------------------------------

### Configure Crochet with no_setup()

Source: https://context7.com/itamarst/crochet/llms.txt

Configures Crochet for applications that manage their own Twisted reactor. This prevents Crochet from starting its own reactor thread. Must be called before importing libraries that use `crochet.setup()`.

```python
from crochet import no_setup

# Must be called BEFORE importing any libraries that use crochet.setup()
no_setup()

# Now import libraries that might call crochet.setup()
import some_crochet_library

# Run your own Twisted reactor
from twisted.internet import reactor
from twisted.web.wsgi import WSGIResource
from twisted.web.server import Site

def my_wsgi_app(environ, start_response):
    start_response('200 OK', [])
    return [b"Hello from Twisted WSGI"]

pool = reactor.getThreadPool()
reactor.listenTCP(8080, Site(WSGIResource(reactor, pool, my_wsgi_app)))
reactor.run()
```

--------------------------------

### Integrate Crochet into Twisted Applications

Source: https://github.com/itamarst/crochet/blob/master/docs/introduction.md

Uses no_setup() to prevent Crochet from starting its own reactor when running within an existing Twisted application.

```python
#!/usr/bin/python
"""
An example of using Crochet from a normal Twisted application.
"""

import sys

from crochet import no_setup, wait_for
# Tell Crochet not to run the reactor:
no_setup()

from twisted.internet import reactor
from twisted.python import log
from twisted.web.wsgi import WSGIResource
from twisted.web.server import Site
from twisted.names import client

# A WSGI application, will be run in thread pool:
def application(environ, start_response):
    start_response('200 OK', [])
    try:
        ip = gethostbyname('twistedmatrix.com')
        return "%s has IP %s" % ('twistedmatrix.com', ip)
    except Exception, e:
        return 'Error doing lookup: %s' % (e,)
```

--------------------------------

### Implement a layered exchange rate service with Crochet

Source: https://github.com/itamarst/crochet/blob/master/docs/using.md

This example demonstrates separating Twisted-based background tasks from a blocking application API using Crochet's @run_in_reactor and @wait_for decorators.

```python
#!/usr/bin/python
"""
An example of scheduling time-based events in the background.

Download the latest EUR/USD exchange rate from Yahoo every 30 seconds in the
background; the rendered Flask web page can use the latest value without
having to do the request itself.

Note this is example is for demonstration purposes only, and is not actually
used in the real world. You should not do this in a real application without
reading Yahoo's terms-of-service and following them.
"""

from __future__ import print_function

from flask import Flask

from twisted.internet.task import LoopingCall
from twisted.web.client import getPage
from twisted.python import log

from crochet import wait_for, run_in_reactor, setup
setup()


# Twisted code:
class _ExchangeRate(object):
    """Download an exchange rate from Yahoo Finance using Twisted."""

    def __init__(self, name):
        self._value = None
        self._name = name

    # External API:
    def latest_value(self):
        """Return the latest exchange rate value.

        May be None if no value is available.
        """
        return self._value

    def start(self):
        """Start the background process."""
        self._lc = LoopingCall(self._download)
        # Run immediately, and then every 30 seconds:
        self._lc.start(30, now=True)

    def _download(self):
        """Download the page."""
        print("Downloading!")
        def parse(result):
            print("Got %r back from Yahoo." % (result,))
            values = result.strip().split(",")
            self._value = float(values[1])
        d = getPage(
            "http://download.finance.yahoo.com/d/quotes.csv?e=.csv&f=c4l1&s=%s=X"
            % (self._name,))
        d.addCallback(parse)
        d.addErrback(log.err)
        return d


# Blocking wrapper:
class ExchangeRate(object):
    """Blocking API for downloading exchange rate."""

    def __init__(self, name):
        self._exchange = _ExchangeRate(name)

    @run_in_reactor
    def start(self):
        self._exchange.start()

    @wait_for(timeout=1)
    def latest_value(self):
        """Return the latest exchange rate value.

        May be None if no value is available.
        """
        return self._exchange.latest_value()


EURUSD = ExchangeRate("EURUSD")
app = Flask(__name__)

@app.route('/')
def index():
    rate = EURUSD.latest_value()
    if rate is None:
        rate = "unavailable, please refresh the page"
    return "Current EUR/USD exchange rate is %s." % (rate,)


if __name__ == '__main__':
    import sys, logging
    logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
    EURUSD.start()
    app.run()
```

--------------------------------

### Preventing Crochet Reactor Setup in Twisted Apps

Source: https://github.com/itamarst/crochet/blob/master/docs/api.md

Call `no_setup()` before importing libraries that might run `crochet.setup()` to prevent conflicts when your application already manages the Twisted reactor.

```python
from crochet import no_setup
no_setup()
# Only now do we import libraries that might run crochet.setup():
import blockinglib

# ... setup application ...

from twisted.internet import reactor
reactor.run()
```

--------------------------------

### SSH into a Running Python Server

Source: https://github.com/itamarst/crochet/blob/master/docs/introduction.md

Starts an SSH server using Twisted Conch to provide a remote Python prompt for inspecting application state.

```python
#!/usr/bin/python
"""
A demonstration of Conch, allowing you to SSH into a running Python server and
inspect objects at a Python prompt.

If you're using the system install of Twisted, you may need to install Conch
separately, e.g. on Ubuntu:

   $ sudo apt-get install python-twisted-conch

Once you've started the program, you can ssh in by doing:

    $ ssh admin@localhost -p 5022

The password is 'secret'. Once you've reached the Python prompt, you have
access to the app object, and can import code, etc.:

    >>> 3 + 4
    7
    >>> print(app)
    <flask.app.Flask object at 0x18e1690>

"""

import logging

from flask import Flask
from crochet import setup, run_in_reactor
setup()

# Web server:
app = Flask(__name__)

@app.route('/')
def index():
    return "Welcome to my boring web server!"


@run_in_reactor
def start_ssh_server(port, username, password, namespace):
    """
    Start an SSH server on the given port, exposing a Python prompt with the
    given namespace.
    """
    # This is a lot of boilerplate, see http://tm.tl/6429 for a ticket to
    # provide a utility function that simplifies this.
    from twisted.internet import reactor
    from twisted.conch.insults import insults
    from twisted.conch import manhole, manhole_ssh
    from twisted.cred.checkers import (
        InMemoryUsernamePasswordDatabaseDontUse as MemoryDB)
    from twisted.cred.portal import Portal

    sshRealm = manhole_ssh.TerminalRealm()
    def chainedProtocolFactory():
        return insults.ServerProtocol(manhole.Manhole, namespace)
    sshRealm.chainedProtocolFactory = chainedProtocolFactory

    sshPortal = Portal(sshRealm, [MemoryDB(**{username: password})])
    reactor.listenTCP(port, manhole_ssh.ConchFactory(sshPortal),
                      interface="127.0.0.1")


if __name__ == '__main__':
    import sys
    logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
    start_ssh_server(5022, "admin", "secret", {"app": app})
    app.run()
```

--------------------------------

### Crochet Non-Initialization

Source: https://context7.com/itamarst/crochet/llms.txt

Configures Crochet for applications that manage their own Twisted reactor. This prevents Crochet from starting its own reactor thread.

```APIDOC
## no_setup()

### Description
Configures Crochet for use in applications that will run Twisted's reactor themselves (e.g., Twisted as a WSGI container). This prevents Crochet from starting its own reactor thread and makes subsequent `setup()` calls no-ops, allowing Crochet-based libraries to work within existing Twisted applications.

### Method
Function Call

### Endpoint
N/A

### Parameters
None

### Request Example
```python
from crochet import no_setup

# Must be called BEFORE importing any libraries that use crochet.setup()
no_setup()

# Now import libraries that might call crochet.setup()
import some_crochet_library

# Run your own Twisted reactor
from twisted.internet import reactor
from twisted.web.wsgi import WSGIResource
from twisted.web.server import Site

def my_wsgi_app(environ, start_response):
    start_response('200 OK', [])
    return [b"Hello from Twisted WSGI"]

pool = reactor.getThreadPool()
reactor.listenTCP(8080, Site(WSGIResource(reactor, pool, my_wsgi_app)))
reactor.run()
```

### Response
None (configures Crochet behavior)

### Response Example
None
```

--------------------------------

### Stash and retrieve results across web requests with Flask

Source: https://context7.com/itamarst/crochet/llms.txt

The `stash()` method on `EventualResult` stores the result in memory and returns an ID. `retrieve_result()` can then be used to get the result later, which is useful for web applications where results need to persist across HTTP requests.

```python
from flask import Flask, session
from crochet import setup, run_in_reactor, retrieve_result, TimeoutError

setup()

app = Flask(__name__)
app.secret_key = 'your-secret-key'

@run_in_reactor
def long_running_task(data):
    from twisted.internet import reactor, defer
    d = defer.Deferred()
    reactor.callLater(5, d.callback, f"Processed: {data}")
    return d

@app.route('/start')
def start_task():
    result = long_running_task("important data")
    # stash() stores EventualResult and returns an integer ID
    uid = result.stash()
    session['task_id'] = uid
    return "Task started! Check /status for progress."

@app.route('/status')
def check_status():
    if 'task_id' not in session:
        return "No task running"

    # retrieve_result() is a one-time operation per uid
    result = retrieve_result(session.pop('task_id'))
    try:
        data = result.wait(timeout=0.1)
        return f"Complete: {data}"
    except TimeoutError:
        # Re-stash for next request (gets new uid)
        session['task_id'] = result.stash()
        return "Still processing..."
    except Exception:
        return f"Error: {result.original_failure().getTraceback()}"
```

--------------------------------

### Perform DNS MX Queries

Source: https://github.com/itamarst/crochet/blob/master/docs/introduction.md

Demonstrates using Twisted's DNS library with a blocking wrapper to perform MX record lookups.

```python
#!/usr/bin/python
"""
A command-line application that uses Twisted to do an MX DNS query.
"""

from __future__ import print_function

from twisted.names.client import lookupMailExchange
from crochet import setup, wait_for
setup()


# Twisted code:
def _mx(domain):
    """
    Return Deferred that fires with a list of (priority, MX domain) tuples for
    a given domain.
    """
    def got_records(result):
        return sorted(
            [(int(record.payload.preference), str(record.payload.name))
             for record in result[0]])
    d = lookupMailExchange(domain)
    d.addCallback(got_records)
    return d


# Blocking wrapper:
@wait_for(timeout=5)
def mx(domain):
    """
    Return list of (priority, MX domain) tuples for a given domain.
    """
    return _mx(domain)


# Application code:
def main(domain):
    print("Mail servers for %s:" % (domain,))
    for priority, mailserver in mx(domain):
        print(priority, mailserver)


if __name__ == '__main__':
    import sys
    main(sys.argv[1])
```

--------------------------------

### Unit Testing with __wrapped__

Source: https://context7.com/itamarst/crochet/llms.txt

Demonstrates how to use the `__wrapped__` attribute to access the original Twisted functions for unit testing purposes.

```APIDOC
## Unit Testing with __wrapped__

Both `@wait_for` and `@run_in_reactor` decorators expose the underlying Twisted function via the `__wrapped__` attribute. This allows for direct testing of the Twisted code without the Crochet layer.

### Usage

Access `your_decorated_function.__wrapped__` to get the original function. If the decorator was `@run_in_reactor`, `__wrapped__` returns an `EventualResult`. If the decorator was `@wait_for`, `__wrapped__` returns a Twisted `Deferred`.

### Example Usage

```python
from crochet import setup, run_in_reactor, wait_for

setup()

@run_in_reactor
def add(x, y):
    return x + y

@wait_for(timeout=5.0)
def multiply(x, y):
    from twisted.internet import defer
    return defer.succeed(x * y)

# Normal usage returns EventualResult or blocks
result = add(2, 3)  # Returns EventualResult
print(result.wait(timeout=1.0))  # 5

product = multiply(4, 5)  # Blocks and returns 20
print(product)  # 20

# For testing, access the underlying function directly
print(add.__wrapped__(2, 3))  # 5 (direct result, no EventualResult)

# multiply.__wrapped__ returns a Deferred
d = multiply.__wrapped__(4, 5)
print(d)  # <Deferred at 0x...> 
```
```

--------------------------------

### Schedule Background Tasks with Crochet

Source: https://github.com/itamarst/crochet/blob/master/docs/introduction.md

Demonstrates using Crochet to run Twisted tasks in the background, such as periodic data fetching, without blocking the main application thread.

```python
#!/usr/bin/python
"""
An example of scheduling time-based events in the background.

Download the latest EUR/USD exchange rate from Yahoo every 30 seconds in the
background; the rendered Flask web page can use the latest value without
having to do the request itself.

Note this is example is for demonstration purposes only, and is not actually
used in the real world. You should not do this in a real application without
reading Yahoo's terms-of-service and following them.
"""

from __future__ import print_function

from flask import Flask

from twisted.internet.task import LoopingCall
from twisted.web.client import getPage
from twisted.python import log

from crochet import wait_for, run_in_reactor, setup
setup()


# Twisted code:
class _ExchangeRate(object):
    """Download an exchange rate from Yahoo Finance using Twisted."""

    def __init__(self, name):
        self._value = None
        self._name = name

    # External API:
    def latest_value(self):
        """Return the latest exchange rate value.

        May be None if no value is available.
        """
        return self._value

    def start(self):
        """Start the background process."""
        self._lc = LoopingCall(self._download)
        # Run immediately, and then every 30 seconds:
        self._lc.start(30, now=True)

    def _download(self):
        """Download the page."""
        print("Downloading!")
        def parse(result):
            print("Got %r back from Yahoo." % (result,))
            values = result.strip().split(",")
            self._value = float(values[1])
        d = getPage(
            "http://download.finance.yahoo.com/d/quotes.csv?e=.csv&f=c4l1&s=%s=X"
            % (self._name,))
        d.addCallback(parse)
        d.addErrback(log.err)
        return d
```

--------------------------------

### Integrate Crochet with Flask for Background Tasks

Source: https://context7.com/itamarst/crochet/llms.txt

Demonstrates using @run_in_reactor for periodic background tasks and @wait_for to retrieve data synchronously within a Flask route.

```python
from flask import Flask
from twisted.internet.task import LoopingCall
from twisted.web.client import Agent, readBody
from twisted.internet import reactor

from crochet import setup, run_in_reactor, wait_for

setup()

class BackgroundFetcher:
    """Periodically fetches data in the background using Twisted."""

    def __init__(self):
        self._latest_data = None
        self._agent = None

    def _fetch(self):
        """Runs in reactor thread - fetches data."""
        from twisted.internet import reactor
        if self._agent is None:
            self._agent = Agent(reactor)
        d = self._agent.request(b'GET', b'http://httpbin.org/uuid')
        d.addCallback(readBody)
        d.addCallback(self._store_result)
        d.addErrback(lambda f: None)  # Ignore errors
        return d

    def _store_result(self, body):
        self._latest_data = body.decode('utf-8')

    @run_in_reactor
    def start(self, interval=30):
        """Start periodic fetching."""
        self._lc = LoopingCall(self._fetch)
        self._lc.start(interval, now=True)

    @wait_for(timeout=1.0)
    def get_latest(self):
        """Return latest fetched data."""
        return self._latest_data

# Flask application
app = Flask(__name__)
fetcher = BackgroundFetcher()

@app.route('/')
def index():
    data = fetcher.get_latest()
    if data is None:
        return "Data not yet available, please refresh"
    return f"Latest data: {data}"

if __name__ == '__main__':
    import logging
    logging.basicConfig(level=logging.DEBUG)
    fetcher.start(interval=10)  # Fetch every 10 seconds
    app.run(port=5000)
```

--------------------------------

### EventualResult.stash() and retrieve_result() for Web Applications

Source: https://context7.com/itamarst/crochet/llms.txt

Explains how to use `stash()` and `retrieve_result()` to manage asynchronous results across HTTP requests in web applications.

```APIDOC
## EventualResult.stash() and retrieve_result()

For web applications, `EventualResult` objects can be stashed in memory and retrieved later using a unique ID. This is useful for storing results in session data across HTTP requests.

### Methods

- **EventualResult.stash()**: Stores the `EventualResult` and returns a unique integer ID.
- **retrieve_result(uid)**: Retrieves an `EventualResult` using its unique ID. This is a one-time operation per ID.

### Example Usage with Flask

```python
from flask import Flask, session
from crochet import setup, run_in_reactor, retrieve_result, TimeoutError

setup()

app = Flask(__name__)
app.secret_key = 'your-secret-key' # Replace with a real secret key

@run_in_reactor
def long_running_task(data):
    from twisted.internet import reactor, defer
    d = defer.Deferred()
    reactor.callLater(5, d.callback, f"Processed: {data}")
    return d

@app.route('/start')
def start_task():
    result = long_running_task("important data")
    # stash() stores EventualResult and returns an integer ID
    uid = result.stash()
    session['task_id'] = uid
    return "Task started! Check /status for progress."

@app.route('/status')
def check_status():
    if 'task_id' not in session:
        return "No task running"

    # retrieve_result() is a one-time operation per uid
    result = retrieve_result(session.pop('task_id'))
    try:
        data = result.wait(timeout=0.1) # Short timeout to check status
        return f"Complete: {data}"
    except TimeoutError:
        # Re-stash for next request (gets new uid)
        session['task_id'] = result.stash()
        return "Still processing..."
    except Exception:
        return f"Error: {result.original_failure().getTraceback()}"
```
```

--------------------------------

### Create a Blocking Wrapper for Exchange Rates

Source: https://github.com/itamarst/crochet/blob/master/docs/introduction.md

Uses @run_in_reactor and @wait_for decorators to expose asynchronous Twisted operations as a synchronous API.

```python
class ExchangeRate(object):
    """Blocking API for downloading exchange rate."""

    def __init__(self, name):
        self._exchange = _ExchangeRate(name)

    @run_in_reactor
    def start(self):
        self._exchange.start()

    @wait_for(timeout=1)
    def latest_value(self):
        """Return the latest exchange rate value.

        May be None if no value is available.
        """
        return self._exchange.latest_value()


EURUSD = ExchangeRate("EURUSD")
app = Flask(__name__)

@app.route('/')
def index():
    rate = EURUSD.latest_value()
    if rate is None:
        rate = "unavailable, please refresh the page"
    return "Current EUR/USD exchange rate is %s." % (rate,)


if __name__ == '__main__':
    import sys, logging
    logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
    EURUSD.start()
    app.run()
```

--------------------------------

### Configure mypy for Crochet

Source: https://github.com/itamarst/crochet/blob/master/docs/type-checking.md

Add the Crochet plugin to your mypy configuration file to enable proper type checking for decorated functions.

```default
[mypy]
plugins = crochet.mypy
```

--------------------------------

### Flask Web App with Background Downloads using EventualResult

Source: https://github.com/itamarst/crochet/blob/master/docs/api.md

This Flask application demonstrates using @run_in_reactor and EventualResult to perform background downloads. It uses session storage to track download progress via stashed EventualResult objects.

```python
#!/usr/bin/python
"""
A flask web application that downloads a page in the background.
"""

import logging
from flask import Flask, session, escape
from crochet import setup, run_in_reactor, retrieve_result, TimeoutError

# Can be called multiple times with no ill-effect:
setup()

app = Flask(__name__)


@run_in_reactor
def download_page(url):
    """
    Download a page.
    """
    from twisted.web.client import getPage
    return getPage(url)


@app.route('/')
def index():
    if 'download' not in session:
        # Calling an @run_in_reactor function returns an EventualResult:
        result = download_page('http://www.google.com')
        session['download'] = result.stash()
        return "Starting download, refresh to track progress."

    # Retrieval is a one-time operation, so the uid in the session cannot be
    # reused:
    result = retrieve_result(session.pop('download'))
    try:
        download = result.wait(timeout=0.1)
        return "Downloaded: " + escape(download)
    except TimeoutError:
        session['download'] = result.stash()
        return "Download in progress..."
    except:
        # The original traceback of the exception:
        return "Download failed:\n" + result.original_failure().getTraceback()


if __name__ == '__main__':
    import os, sys
    logging.basicConfig(stream=sys.stderr, level=logging.DEBUG)
    app.secret_key = os.urandom(24)
    app.run()
```

--------------------------------

### EventualResult - Core Functionality

Source: https://context7.com/itamarst/crochet/llms.txt

Demonstrates the basic usage of EventualResult, including waiting for results, handling timeouts, and canceling operations.

```APIDOC
## EventualResult

The object returned by `@run_in_reactor` decorated functions. Provides methods to wait for results, cancel operations, access failure tracebacks, and stash results for later retrieval across web requests.

### Methods

- **wait(timeout)**: Blocks until the result is available or the timeout is reached.
- **cancel()**: Attempts to cancel the underlying operation.
- **original_failure()**: Returns the Twisted Failure object if the operation failed.

### Example Usage

```python
from crochet import setup, run_in_reactor, TimeoutError

setup()

@run_in_reactor
def slow_operation(duration):
    from twisted.internet import reactor, defer
    d = defer.Deferred()
    reactor.callLater(duration, d.callback, f"Completed after {duration}s")
    return d

# Get an EventualResult
eventual = slow_operation(2.0)

# wait(timeout) - block until result or timeout
try:
    result = eventual.wait(timeout=3.0)
    print(result)  # "Completed after 2.0s"
except TimeoutError:
    print("Operation timed out")

# cancel() - attempt to cancel the underlying Deferred
eventual2 = slow_operation(10.0)
eventual2.cancel()  # Tries to cancel; may or may not succeed

# original_failure() - get Twisted Failure for error tracebacks
@run_in_reactor
def failing_operation():
    raise ValueError("Something went wrong")

result = failing_operation()
try:
    result.wait(timeout=1.0)
except Exception:
    failure = result.original_failure()
    if failure:
        print("Original traceback:")
        print(failure.getTraceback())
```
```

--------------------------------

### Handle TimeoutError and ReactorStopped Exceptions

Source: https://context7.com/itamarst/crochet/llms.txt

Shows how to catch Crochet-specific exceptions when performing blocking operations like DNS lookups.

```python
from crochet import setup, wait_for, TimeoutError, ReactorStopped

setup()

@wait_for(timeout=2.0)
def slow_dns_lookup(name):
    from twisted.names import client
    d = client.lookupAddress(name)
    d.addCallback(lambda r: r[0][0].payload.dottedQuad())
    return d

def robust_lookup(hostname):
    """Production-ready DNS lookup with proper error handling."""
    try:
        return slow_dns_lookup(hostname)
    except TimeoutError:
        # Operation took longer than 2 seconds
        print(f"Timeout looking up {hostname}")
        return None
    except ReactorStopped:
        # Reactor has shut down (e.g., during application exit)
        print("Application is shutting down")
        return None
    except Exception as e:
        # Other Twisted errors (DNS failures, network issues, etc.)
        print(f"Lookup failed: {e}")
        return None

# Usage
ip = robust_lookup("example.com")
if ip:
    print(f"Resolved to {ip}")
```

--------------------------------

### Use @wait_for for blocking DNS lookup

Source: https://context7.com/itamarst/crochet/llms.txt

Decorates a function to run in the Twisted reactor thread, blocking until a result is available. Handles Deferreds and async/await. Includes a 5-second timeout.

```python
from twisted.names import client
from crochet import setup, wait_for, TimeoutError

setup()

@wait_for(timeout=5.0)
def gethostbyname(name):
    """Lookup the IP of a given hostname with a 5-second timeout."""
    d = client.lookupAddress(name)
    d.addCallback(lambda result: result[0][0].payload.dottedQuad())
    return d

# Usage - blocks until result is ready or timeout
try:
    ip = gethostbyname("twistedmatrix.com")
    print(f"twistedmatrix.com -> {ip}")
    # Output: twistedmatrix.com -> 66.35.39.66
except TimeoutError:
    print("DNS lookup timed out")
except Exception as e:
    print(f"DNS lookup failed: {e}")
```

--------------------------------

### Blocking DNS Lookup with @wait_for

Source: https://github.com/itamarst/crochet/blob/master/docs/api.md

Wraps a Twisted DNS lookup function to make it behave like a blocking call. It includes a timeout to prevent indefinite waiting and handles Twisted Deferreds transparently. Use this when you need to call Twisted APIs from synchronous code.

```python
#!/usr/bin/python
"""
Do a DNS lookup using Twisted's APIs.
"""
from __future__ import print_function

# The Twisted code we'll be using:
from twisted.names import client

from crochet import setup, wait_for
setup()


# Crochet layer, wrapping Twisted's DNS library in a blocking call.
@wait_for(timeout=5.0)
def gethostbyname(name):
    """Lookup the IP of a given hostname.

    Unlike socket.gethostbyname() which can take an arbitrary amount of time
    to finish, this function will raise crochet.TimeoutError if more than 5
    seconds elapse without an answer being received.
    """
    d = client.lookupAddress(name)
    d.addCallback(lambda result: result[0][0].payload.dottedQuad())
    return d


if __name__ == '__main__':
    # Application code using the public API - notice it works in a normal
    # blocking manner, with no event loop visible:
    import sys
    name = sys.argv[1]
    ip = gethostbyname(name)
    print(name, "->", ip)
```

--------------------------------

### Use @wait_for with async/await

Source: https://context7.com/itamarst/crochet/llms.txt

Demonstrates the `@wait_for` decorator with async/await syntax for more readable asynchronous code. Callers interact with it as a regular blocking function.

```python
from twisted.names import client
from crochet import setup, wait_for

setup()

@wait_for(timeout=5.0)
async def gethostbyname(name):
    """Async/await version of DNS lookup."""
    result = await client.lookupAddress(name)
    return result[0][0].payload.dottedQuad()

# Called exactly like the Deferred-based version
ip = gethostbyname("example.com")
print(f"example.com -> {ip}")
```

--------------------------------

### Verify type checking with EventualResult

Source: https://github.com/itamarst/crochet/blob/master/docs/type-checking.md

Demonstrates how the mypy plugin ensures that the result of wait() matches the expected type.

```default
# OK
t1: float = get_time_in_x_seconds(2).wait(3)
print(f"The reactor time is {t1}")

# mypy error: Incompatible types in assignment
#   (expression has type "float", variable has type "str")
t2: str = get_time_in_x_seconds(2).wait(3)
print(f"The reactor time is {t2}")
```

--------------------------------

### Test Twisted code directly using __wrapped__

Source: https://context7.com/itamarst/crochet/llms.txt

Decorators like `@run_in_reactor` and `@wait_for` expose the original Twisted function via the `__wrapped__` attribute. This allows for direct testing of the underlying Twisted code without the Crochet layer.

```python
from crochet import setup, run_in_reactor, wait_for

setup()

@run_in_reactor
def add(x, y):
    return x + y

@wait_for(timeout=5.0)
def multiply(x, y):
    from twisted.internet import defer
    return defer.succeed(x * y)

# Normal usage returns EventualResult or blocks
result = add(2, 3)  # Returns EventualResult
print(result.wait(timeout=1.0))  # 5

product = multiply(4, 5)  # Blocks and returns 20
print(product)  # 20

# For testing, access the underlying function directly
print(add.__wrapped__(2, 3))  # 5 (direct result, no EventualResult)
# multiply.__wrapped__ returns a Deferred
d = multiply.__wrapped__(4, 5)
print(d)  # <Deferred at 0x...>
```

--------------------------------

### @wait_for Decorator (async/await)

Source: https://context7.com/itamarst/crochet/llms.txt

Runs an async function in the Twisted reactor thread and blocks until the result is available. Supports async/await syntax.

```APIDOC
## @wait_for(timeout) with async/await

### Description
The `@wait_for` decorator also supports async/await syntax for more readable asynchronous code. The decorated function can use `await` internally, but callers use it as a regular blocking function.

### Method
Decorator

### Endpoint
N/A

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from twisted.names import client
from crochet import setup, wait_for

setup()

@wait_for(timeout=5.0)
async def gethostbyname(name):
    """Async/await version of DNS lookup."""
    result = await client.lookupAddress(name)
    return result[0][0].payload.dottedQuad()

# Called exactly like the Deferred-based version
ip = gethostbyname("example.com")
print(f"example.com -> {ip}")
```

### Response
#### Success Response (200)
- **result** (any) - The resolved value of the async function.

#### Response Example
```json
{
  "result": "93.184.216.34"
}
```
```

--------------------------------

### Check if result is ready with short timeout

Source: https://context7.com/itamarst/crochet/llms.txt

Use `wait(timeout)` to block until a result is available or a timeout occurs. If the timeout is reached, the operation can be cancelled.

```python
try:
    content = result.wait(timeout=5.0)
    print(f"Downloaded {len(content)} bytes")
except TimeoutError:
    print("Still downloading...")
    # Can cancel if needed
    result.cancel()
```

--------------------------------

### Asynchronous Hostname Lookup with Timeout

Source: https://github.com/itamarst/crochet/blob/master/docs/api.md

Use @wait_for to decorate an async function for non-blocking hostname resolution. It raises crochet.TimeoutError if the operation exceeds the specified timeout.

```python
import crochet

crochet.setup()

@crochet.wait_for(timeout=5.0)
async def gethostbyname(name):
    """Lookup the IP of a given hostname.

    Unlike socket.gethostbyname() which can take an arbitrary amount of time
    to finish, this function will raise crochet.TimeoutError if more than 5
    seconds elapse without an answer being received.
    """
    result = await client.lookupAddress(name)
    return result[0][0].payload.dottedQuad()
```

--------------------------------

### Call Blocking DNS Lookup from Twisted

Source: https://github.com/itamarst/crochet/blob/master/docs/introduction.md

Use the @wait_for decorator to call a blocking function like gethostbyname from within a Twisted application. This ensures the reactor remains responsive.

```python
@wait_for(timeout=10)
def gethostbyname(name):
    d = client.lookupAddress(name)
    d.addCallback(lambda result: result[0][0].payload.dottedQuad())
    return d
```

--------------------------------

### Running Twisted APIs in a Reactor Thread with @run_in_reactor

Source: https://github.com/itamarst/crochet/blob/master/docs/api.md

Decorate functions calling Twisted APIs with @run_in_reactor. This ensures the code runs in the reactor thread and returns an EventualResult, which wraps results from Deferreds or async functions.

```python
from crochet import setup, run_in_reactor, retrieve_result, TimeoutError

# Can be called multiple times with no ill-effect:
setup()

@run_in_reactor
def download_page(url):
    """
    Download a page.
    """
    from twisted.web.client import getPage
    return getPage(url)
```

--------------------------------

### Unit Testing Wrapped Twisted Functions

Source: https://github.com/itamarst/crochet/blob/master/docs/api.md

Access the underlying Twisted function via the `__wrapped__` attribute for unit testing without the Crochet layer. Ensure `crochet.setup()` is called before defining wrapped functions.

```python
#!/usr/bin/python
"""
Demonstration of accessing wrapped functions for testing.
"""

from __future__ import print_function

from crochet import setup, run_in_reactor
setup()

@run_in_reactor
def add(x, y):
    return x + y


if __name__ == '__main__':
    print("add(1, 2) returns EventualResult:")
    print("    ", add(1, 2))
    print("add.__wrapped__(1, 2) is the result of the underlying function:")
    print("    ", add.__wrapped__(1, 2))
```

--------------------------------

### @wait_for Decorator (Deferred)

Source: https://context7.com/itamarst/crochet/llms.txt

Runs a function in the Twisted reactor thread and blocks until the result is available. Supports traditional Deferred-based Twisted functions.

```APIDOC
## @wait_for(timeout)

### Description
A decorator that runs a function in the Twisted reactor thread and blocks until the result is available. If the function returns a Deferred, it is automatically resolved. Supports both traditional Twisted functions and async/await syntax. If the timeout (in seconds) is exceeded, raises `TimeoutError` and cancels the underlying Deferred.

### Method
Decorator

### Endpoint
N/A

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from twisted.names import client
from crochet import setup, wait_for, TimeoutError

setup()

@wait_for(timeout=5.0)
def gethostbyname(name):
    """Lookup the IP of a given hostname with a 5-second timeout."""
    d = client.lookupAddress(name)
    d.addCallback(lambda result: result[0][0].payload.dottedQuad())
    return d

# Usage - blocks until result is ready or timeout
try:
    ip = gethostbyname("twistedmatrix.com")
    print(f"twistedmatrix.com -> {ip}")
    # Output: twistedmatrix.com -> 66.35.39.66
except TimeoutError:
    print("DNS lookup timed out")
except Exception as e:
    print(f"DNS lookup failed: {e}")
```

### Response
#### Success Response (200)
- **result** (any) - The resolved value of the Deferred or the return value of the async function.

#### Response Example
```json
{
  "result": "66.35.39.66"
}
```
```

--------------------------------

### Access original failure with EventualResult.original_failure()

Source: https://context7.com/itamarst/crochet/llms.txt

When an operation fails, `original_failure()` can be used to retrieve the Twisted `Failure` object, which contains the traceback of the error.

```python
@run_in_reactor
def failing_operation():
    raise ValueError("Something went wrong")

result = failing_operation()
try:
    result.wait(timeout=1.0)
except Exception:
    failure = result.original_failure()
    if failure:
        print("Original traceback:")
        print(failure.getTraceback())
```

--------------------------------

### Use @run_in_reactor for background operations

Source: https://context7.com/itamarst/crochet/llms.txt

Decorates a function to run in the reactor thread, returning an `EventualResult` immediately for asynchronous operations. Does not block the calling thread.

```python
from crochet import setup, run_in_reactor, TimeoutError

setup()

@run_in_reactor
def download_page(url):
    """Download a page in the background."""
    from twisted.web.client import getPage
    return getPage(url.encode())

# Start download - returns immediately with EventualResult
result = download_page("http://example.com")
print("Download started in background...")

# Do other work while download proceeds
import time
time.sleep(1)
```

--------------------------------

### Limit Twisted log verbosity

Source: https://github.com/itamarst/crochet/blob/master/docs/workarounds.md

Reduces log output by setting the Twisted logger level to ERROR.

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

--------------------------------

### Run Twisted reactor function with @run_in_reactor

Source: https://context7.com/itamarst/crochet/llms.txt

The `@run_in_reactor` decorator allows calling Twisted reactor functions from synchronous code. It returns an `EventualResult` object that can be used to wait for the result or cancel the operation.

```python
from crochet import setup, run_in_reactor, TimeoutError

setup()

@run_in_reactor
def slow_operation(duration):
    from twisted.internet import reactor, defer
    d = defer.Deferred()
    reactor.callLater(duration, d.callback, f"Completed after {duration}s")
    return d

# Get an EventualResult
eventual = slow_operation(2.0)

# wait(timeout) - block until result or timeout
try:
    result = eventual.wait(timeout=3.0)
    print(result)  # "Completed after 2.0s"
except TimeoutError:
    print("Operation timed out")

# cancel() - attempt to cancel the underlying Deferred
eventual2 = slow_operation(10.0)
eventual2.cancel()  # Tries to cancel; may or may not succeed
```

--------------------------------

### Accessing exception tracebacks from EventualResult

Source: https://github.com/itamarst/crochet/blob/master/docs/workarounds.md

Retrieves the string version of a traceback from an EventualResult object when an exception occurs during a reactor-wrapped function call.

```python
from crochet import run_in_reactor, TimeoutError

@run_in_reactor
def download_page(url):
    from twisted.web.client import getPage
    return getPage(url)

result = download_page("https://github.com")
try:
    page = result.wait(timeout=1000)
except TimeoutError:
    # Handle timeout ...
except:
    # Something else happened:
    print(result.original_failure().getTraceback())
```

--------------------------------

### @run_in_reactor Decorator

Source: https://context7.com/itamarst/crochet/llms.txt

Runs a function in the reactor thread without blocking the calling thread, returning an `EventualResult` object for background operations.

```APIDOC
## @run_in_reactor

### Description
A decorator that runs a function in the reactor thread and immediately returns an `EventualResult` object, allowing asynchronous/background operations. Unlike `@wait_for`, it does not block the calling thread. The `EventualResult` can be polled or waited on later.

### Method
Decorator

### Endpoint
N/A

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
```python
from crochet import setup, run_in_reactor, TimeoutError

setup()

@run_in_reactor
def download_page(url):
    """Download a page in the background."""
    from twisted.web.client import getPage
    return getPage(url.encode())

# Start download - returns immediately with EventualResult
result = download_page("http://example.com")
print("Download started in background...")

# Do other work while download proceeds
import time
time.sleep(1)

# To get the result later (example):
try:
    # Wait for the result with a timeout
    page_content = result.wait(timeout=10)
    print(f"Downloaded {len(page_content)} bytes.")
except TimeoutError:
    print("Download timed out.")
except Exception as e:
    print(f"Download failed: {e}")
```

### Response
#### Success Response (200)
- **EventualResult** - An object that can be used to retrieve the result of the background operation later.

#### Response Example
```json
{
  "EventualResult": "<crochet.session.EventualResult object at 0x...>"
}
```
```

--------------------------------

### Cast Deferred results for type checking

Source: https://github.com/itamarst/crochet/blob/master/docs/type-checking.md

Use typing.cast to inform the type checker of the expected return type when a function returns a Deferred object.

```default
@run_in_reactor
def get_time_in_x_seconds(delay: float) -> float:
    def get_time() -> float:
        return reactor.seconds()  # type: ignore

    if delay < 0.001:
        # Close enough; just return the current time.
        return get_time()
    else:
        d = Deferred()

        def complete():
            d.callback(get_time())

        reactor.callLater(delay, complete)  # type: ignore
        return typing.cast(float, d)
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.