gevent-toolbelt

Small utilities for gevent that do not belong in gevent itself: cooperative services, graceful shutdown, optional StatsD health metrics, and test helpers.

Install

From PyPI (after you publish a release):

pip install gevent-toolbelt

With uv:

uv add gevent-toolbelt

Requirements

Python 3.10 or newer.

Table of contents

• Configuration
• Usage
• Development
• Publishing to PyPI
• License

Configuration

GEVENT_MAX_BLOCKING_TIME

Default: 5

If the monitor thread is enabled, this is approximately how long (in seconds) the event loop may block before a warning is issued.

GEVENT_MONITOR_THREAD_ENABLE

Default: True

If true, the monitor thread is created the first time the hub is switched to.

STATSD_HOST

Default: 'localhost'

StatsD host for metrics when using the StatsD-backed health helpers.

STATSD_PORT

Default: 8125

StatsD port.

Usage

gevent_toolbelt.service.Service

import gevent
from gevent_toolbelt.service import Service


class MyService(Service):
    def handle(self):
        while not self.stop_event.is_set():
            gevent.sleep()


my_service = MyService()
my_service.serve_forever()

gevent_toolbelt.service.ServiceFunc

import gevent
from gevent_toolbelt.service import ServiceFunc


def my_service_func(stop_event):
    while not stop_event.is_set():
        gevent.sleep()


my_service = ServiceFunc(my_service_func)
my_service.serve_forever()

gevent_toolbelt.service.ServiceGroup

import gevent
from gevent_toolbelt.service import ServiceFunc, ServiceGroup


def my_service_func(stop_event):
    while not stop_event.is_set():
        gevent.sleep()


my_service_group = ServiceGroup(
    [
        ServiceFunc(my_service_func),
        ServiceFunc(my_service_func),
        ServiceFunc(my_service_func),
    ]
)
my_service_group.serve_forever()

StatsD metrics (ServiceWithMetrics, ServiceFuncWithMetrics, ServiceGroupWithMetrics)

Same roles as Service, ServiceFunc, and ServiceGroup, but emit worker gauges over StatsD when the event loop reports blocking. See gevent_toolbelt.service.healthcheck.statsd.

Test helpers

gevent_toolbelt.test.joinall_gracefully joins greenlets and runs a sidecar that calls stop_func after a timeout.

import gevent
from gevent.event import Event
from gevent_toolbelt.test import joinall_gracefully

stop_event = Event()


def my_service_func(stop_event):
    while not stop_event.is_set():
        gevent.sleep()


joinall_gracefully(gevent.spawn(my_service_func, stop_event), timeout=5)

Graceful shutdown

from gevent_toolbelt.service import Service
from gevent_toolbelt.signals import install_signal_handler

my_service = Service(...)

install_signal_handler(my_service.stop)

my_service.serve_forever()

Development

Clone the repository, then install dependencies with uv (includes dev tools such as pytest and Ruff):

uv sync

Run the linter and formatter:

uv run ruff check gevent_toolbelt tests
uv run ruff format gevent_toolbelt tests

Run tests (pytest is started via gevent’s monkey entry point so socket patches in tests line up with StatsD and the hub):

uv run python -m gevent.monkey "$(uv run which pytest)"

Publishing to PyPI

1. Bump the version in gevent_toolbelt/__init__.py (constant VERSION; __version__ follows it), then create and push an annotated tag vX.Y.Z (for example v1.0.13). You can also use bump2version with .bumpversion.cfg.
2. Build and publish are handled by .github/workflows/release.yml when the tag is pushed: it builds with uv build --no-sources, smoke-tests the wheel and sdist, then runs uv publish using PyPI Trusted Publishing (OIDC). On GitHub, create an environment named pypi (optionally with required reviewers) and register that publisher on PyPI so uploads are approved the same way as in wsgiven.

Alternatively, build locally and upload with a PyPI API token:

uv build --no-sources
UV_PUBLISH_TOKEN=… uv publish

Configure the Homepage / Repository URLs in pyproject.toml if your GitHub organization or repository name differs from the placeholders.

License

MIT — see LICENSE.