tessl/pypi-gevent
A coroutine-based Python networking library that uses greenlet to provide a high-level synchronous API on top of libev event loop
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
core-greenlets.mddocs/
Core Greenlets
Fundamental greenlet operations that form the foundation of gevent's cooperative concurrency model. Greenlets are lightweight cooperative threads that yield control during I/O operations, enabling high concurrency without traditional threading complexity.
Capabilities
Greenlet Creation and Management
Create and manage individual greenlets for concurrent execution of functions.
def spawn(function, *args, **kwargs) -> Greenlet:
"""
Spawn a new greenlet to run function with given arguments.
Parameters:
- function: callable to execute in greenlet
- *args: positional arguments for function
- **kwargs: keyword arguments for function
Returns:
Greenlet object that will execute the function
"""
def spawn_later(seconds, function, *args, **kwargs) -> Greenlet:
"""
Spawn a greenlet after a delay.
Parameters:
- seconds: float, delay in seconds before starting
- function: callable to execute
- *args: positional arguments for function
- **kwargs: keyword arguments for function
Returns:
Greenlet object scheduled to start after delay
"""
def spawn_raw(function, *args, **kwargs) -> Greenlet:
"""
Spawn a greenlet without any of the spawn wrapper functionality.
Parameters:
- function: callable to execute
- *args: positional arguments
- **kwargs: keyword arguments
Returns:
Raw greenlet object
"""Greenlet Coordination
Coordinate multiple greenlets for synchronization and resource management.
def joinall(greenlets, timeout=None, raise_error=False):
"""
Wait for all greenlets to finish.
Parameters:
- greenlets: iterable of Greenlet objects
- timeout: float, maximum time to wait in seconds
- raise_error: bool, whether to raise exception if any greenlet failed
Returns:
None
"""
def killall(greenlets, exception=GreenletExit, block=True, timeout=None):
"""
Kill all greenlets in the list.
Parameters:
- greenlets: iterable of Greenlet objects
- exception: exception to raise in greenlets
- block: bool, whether to wait for greenlets to die
- timeout: float, maximum time to wait if blocking
Returns:
None
"""
def kill(greenlet, exception=GreenletExit, block=True, timeout=None):
"""
Kill a single greenlet.
Parameters:
- greenlet: Greenlet object to kill
- exception: exception to raise in greenlet
- block: bool, whether to wait for greenlet to die
- timeout: float, maximum time to wait if blocking
Returns:
None
"""Control Flow
Control execution flow and timing in cooperative environment.
def sleep(seconds=0):
"""
Suspend current greenlet for at least the given time.
Parameters:
- seconds: float, time to sleep in seconds (0 yields to other greenlets)
Returns:
None
"""
def idle(priority=0):
"""
Block until the event loop is idle.
Parameters:
- priority: int, priority level for the idle callback
Returns:
None
"""
def get_hub() -> Hub:
"""
Get the hub for the current thread.
Returns:
Hub object for current thread
"""
def getcurrent() -> Greenlet:
"""
Get the currently executing greenlet.
Returns:
Current Greenlet object
"""Hub Management
Manage the central event loop coordinator.
def reinit():
"""
Reinitialize gevent after forking.
Returns:
None
"""
def getswitchinterval() -> float:
"""
Get the greenlet switch interval.
Returns:
Current switch interval in seconds
"""
def setswitchinterval(interval):
"""
Set the greenlet switch interval.
Parameters:
- interval: float, switch interval in seconds
Returns:
None
"""Signal Handling
Handle Unix signals cooperatively.
def signal_handler(signalnum, handler, *args, **kwargs):
"""
Install a signal handler.
Parameters:
- signalnum: int, signal number
- handler: callable, handler function
- *args: arguments for handler
- **kwargs: keyword arguments for handler
Returns:
signal object
"""Usage Examples
Basic Greenlet Usage
import gevent
def worker(name, delay):
print(f"Worker {name} starting")
gevent.sleep(delay)
print(f"Worker {name} finished")
return f"Result from {name}"
# Spawn multiple workers
workers = [
gevent.spawn(worker, 'A', 1),
gevent.spawn(worker, 'B', 2),
gevent.spawn(worker, 'C', 0.5)
]
# Wait for all to complete
gevent.joinall(workers)
# Get results
for w in workers:
if w.successful():
print(f"Result: {w.value}")Delayed Execution
import gevent
def delayed_task():
print("Task executed after delay")
# Start task after 2 seconds
greenlet = gevent.spawn_later(2, delayed_task)
greenlet.join()Error Handling
import gevent
def failing_task():
raise ValueError("Something went wrong")
def safe_task():
return "Success"
greenlets = [
gevent.spawn(failing_task),
gevent.spawn(safe_task)
]
gevent.joinall(greenlets)
for g in greenlets:
if g.successful():
print(f"Success: {g.value}")
else:
print(f"Failed: {g.exception}")Install with Tessl CLI
npx tessl i tessl/pypi-gevent