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
monkey-patching.mddocs/
Monkey Patching
Transparent replacement of standard library modules with gevent-aware equivalents, enabling existing code to work cooperatively without modification. Monkey patching is gevent's mechanism for making synchronous code asynchronous.
Capabilities
Core Patching Functions
Main functions for applying monkey patches to standard library modules.
def patch_all(socket=True, dns=True, time=True, select=True, thread=True, os=True, ssl=True, subprocess=True, sys=False, aggressive=True, Event=True, builtins=True, signal=True, queue=True, contextvars=True, **kwargs):
"""
Patch all supported modules with gevent equivalents.
Parameters:
- socket: bool, patch socket module
- dns: bool, patch DNS functions
- time: bool, patch time.sleep
- select: bool, patch select module
- thread: bool, patch thread module
- os: bool, patch os module
- ssl: bool, patch ssl module
- subprocess: bool, patch subprocess module
- sys: bool, patch sys module
- aggressive: bool, more comprehensive patching
- Event: bool, patch threading.Event
- builtins: bool, patch builtin functions
- signal: bool, patch signal module
- queue: bool, patch queue module
- contextvars: bool, patch contextvars module
- **kwargs: additional patching options
Returns:
None
"""
def patch_socket():
"""
Patch socket module with gevent socket.
Returns:
None
"""
def patch_ssl():
"""
Patch ssl module with gevent ssl.
Returns:
None
"""
def patch_thread():
"""
Patch thread module with gevent equivalents.
Returns:
None
"""
def patch_threading():
"""
Patch threading module with gevent equivalents.
Returns:
None
"""
def patch_select():
"""
Patch select module with gevent select.
Returns:
None
"""
def patch_subprocess():
"""
Patch subprocess module with gevent subprocess.
Returns:
None
"""
def patch_os():
"""
Patch os module functions with cooperative versions.
Returns:
None
"""
def patch_time():
"""
Patch time.sleep with gevent.sleep.
Returns:
None
"""
def patch_sys():
"""
Patch sys module functions.
Returns:
None
"""
def patch_builtins():
"""
Patch builtin functions with cooperative versions.
Returns:
None
"""
def patch_signal():
"""
Patch signal module with gevent signal handling.
Returns:
None
"""
def patch_queue():
"""
Patch queue module with gevent queue.
Returns:
None
"""
def patch_dns():
"""
Patch DNS resolution functions.
Returns:
None
"""Utility Functions
Functions for inspecting and managing monkey patches.
def get_original(module_name, item_name):
"""
Get original (unpatched) function or class.
Parameters:
- module_name: str, name of module
- item_name: str, name of item in module
Returns:
Original unpatched object
"""
def is_module_patched(module) -> bool:
"""
Check if module has been monkey patched.
Parameters:
- module: module object or string name
Returns:
bool, True if module is patched
"""
def is_object_patched(obj) -> bool:
"""
Check if specific object has been patched.
Parameters:
- obj: object to check
Returns:
bool, True if object is patched
"""Exceptions
class MonkeyPatchWarning(Warning):
"""Warning raised for monkey patching issues."""Usage Examples
Basic Monkey Patching
# This should be done at the very start of your program
from gevent import monkey
monkey.patch_all()
# Now all standard library I/O is cooperative
import socket
import time
import threading
def blocking_function():
# This appears to block but actually yields to other greenlets
time.sleep(2)
# Socket operations are now cooperative
s = socket.socket()
s.connect(('google.com', 80))
s.send(b'GET / HTTP/1.1\r\nHost: google.com\r\n\r\n')
response = s.recv(1024)
s.close()
return response[:100]
# Can be used with gevent as normal
import gevent
greenlets = [gevent.spawn(blocking_function) for _ in range(10)]
results = [g.get() for g in greenlets]
print(f"Got {len(results)} responses")Selective Patching
from gevent import monkey
# Only patch specific modules
monkey.patch_socket()
monkey.patch_ssl()
monkey.patch_time()
import socket
import ssl
import time
def selective_example():
# Socket and SSL are cooperative
context = ssl.create_default_context()
with socket.create_connection(('httpbin.org', 443)) as sock:
with context.wrap_socket(sock, server_hostname='httpbin.org') as ssock:
ssock.send(b'GET /get HTTP/1.1\r\nHost: httpbin.org\r\n\r\n')
# time.sleep is cooperative
time.sleep(0.1)
response = ssock.recv(1024)
return response
result = selective_example()
print("Response received")Checking Patch Status
from gevent import monkey
import socket
import threading
def check_patches():
print("Before patching:")
print(f"Socket module patched: {monkey.is_module_patched('socket')}")
print(f"Threading module patched: {monkey.is_module_patched('threading')}")
print(f"Socket.socket patched: {monkey.is_object_patched(socket.socket)}")
# Apply patches
monkey.patch_all()
print("\nAfter patching:")
print(f"Socket module patched: {monkey.is_module_patched('socket')}")
print(f"Threading module patched: {monkey.is_module_patched('threading')}")
print(f"Socket.socket patched: {monkey.is_object_patched(socket.socket)}")
check_patches()Getting Original Functions
from gevent import monkey
# Patch everything
monkey.patch_all()
import time
import socket
# Get original functions
original_sleep = monkey.get_original('time', 'sleep')
original_socket = monkey.get_original('socket', 'socket')
def mixed_example():
print("Using gevent sleep (cooperative)")
time.sleep(1) # This is now gevent.sleep
print("Using original sleep (blocking)")
original_sleep(1) # This is the original time.sleep
print("Using gevent socket (cooperative)")
gevent_sock = socket.socket() # This is gevent socket
print("Using original socket (blocking)")
original_sock = original_socket() # This is original socket
mixed_example()Patching for Existing Code
# At the top of your main script
from gevent import monkey
monkey.patch_all()
# Now existing libraries work cooperatively
import requests # Uses patched socket/ssl
import urllib3 # Uses patched socket/ssl
import time # sleep is cooperative
def fetch_urls(urls):
"""This function can now handle many concurrent requests."""
import gevent
def fetch_one(url):
try:
# requests will use gevent sockets automatically
response = requests.get(url, timeout=5)
return f"{url}: {response.status_code}"
except Exception as e:
return f"{url}: Error - {e}"
# Spawn greenlets for concurrent fetching
greenlets = [gevent.spawn(fetch_one, url) for url in urls]
# Wait for all to complete
gevent.joinall(greenlets)
# Collect results
return [g.value for g in greenlets]
# Test with multiple URLs
urls = [
'http://httpbin.org/delay/1',
'http://httpbin.org/delay/2',
'http://httpbin.org/delay/1',
'http://httpbin.org/status/200',
'http://httpbin.org/status/404'
]
results = fetch_urls(urls)
for result in results:
print(result)Database Connection Pooling
from gevent import monkey
monkey.patch_all()
import gevent
import time
import sqlite3
# Simulate a database connection pool
class DatabasePool:
def __init__(self, size=5):
self.connections = []
for i in range(size):
# Each connection would be a real DB connection
self.connections.append(f"connection_{i}")
self.available = list(self.connections)
self.in_use = []
def get_connection(self):
while not self.available:
time.sleep(0.1) # Cooperative wait
conn = self.available.pop()
self.in_use.append(conn)
return conn
def return_connection(self, conn):
if conn in self.in_use:
self.in_use.remove(conn)
self.available.append(conn)
def database_worker(worker_id, pool):
conn = pool.get_connection()
print(f"Worker {worker_id} got {conn}")
# Simulate database work
time.sleep(2) # This is now cooperative
print(f"Worker {worker_id} finished with {conn}")
pool.return_connection(conn)
return f"Worker {worker_id} completed"
# Create pool and workers
pool = DatabasePool(3) # Only 3 connections available
workers = [gevent.spawn(database_worker, i, pool) for i in range(10)]
# Wait for all workers
results = [w.get() for w in workers]
print(f"All workers completed: {len(results)}")Testing Patch Compatibility
from gevent import monkey
import warnings
def test_patching():
# Capture warnings
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("always")
# Apply patches
monkey.patch_all()
# Check for warnings
if w:
print("Monkey patching warnings:")
for warning in w:
print(f" {warning.message}")
else:
print("No monkey patching warnings")
# Test basic functionality
import socket
import time
import threading
print("Testing patched modules:")
# Test socket
try:
s = socket.socket()
s.settimeout(1)
s.connect(('8.8.8.8', 53))
s.close()
print(" Socket: OK")
except Exception as e:
print(f" Socket: Error - {e}")
# Test time
start = time.time()
time.sleep(0.01)
elapsed = time.time() - start
print(f" Time.sleep: {elapsed:.3f}s (expected ~0.01s)")
# Test threading
try:
event = threading.Event()
event.set()
print(f" Threading.Event: {'OK' if event.is_set() else 'Error'}")
except Exception as e:
print(f" Threading: Error - {e}")
test_patching()Install with Tessl CLI
npx tessl i tessl/pypi-gevent