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

Overview

Eval results

Files

Servers

Name: tessl/pypi-gevent
Author: tessl

High-performance server implementations including generic TCP/UDP servers and a complete WSGI HTTP server for web applications. All servers are built on gevent's cooperative concurrency model for handling thousands of simultaneous connections.

Capabilities

Stream Server

Generic TCP server for handling streaming connections.

class StreamServer:
    """
    Generic TCP server that spawns greenlets to handle connections.
    """
    
    def __init__(self, listener, handle=None, backlog=None, spawn='default', **ssl_args):
        """
        Create a stream server.
        
        Parameters:
        - listener: tuple (host, port) or socket object for listening
        - handle: callable, function to handle each connection
        - backlog: int, listen backlog size
        - spawn: str or callable, greenlet spawning strategy
        - **ssl_args: SSL configuration arguments
        """
    
    def serve_forever(self, stop_timeout=None):
        """
        Start server and handle connections forever.
        
        Parameters:
        - stop_timeout: float, timeout for stopping server
        
        Returns:
        None
        """
    
    def start(self):
        """
        Start accepting connections in background.
        
        Returns:
        None
        """
    
    def stop(self, timeout=None):
        """
        Stop accepting new connections.
        
        Parameters:
        - timeout: float, timeout for graceful shutdown
        
        Returns:
        None
        """
    
    def close(self):
        """
        Close server socket.
        
        Returns:
        None
        """
    
    def restart(self):
        """
        Restart the server.
        
        Returns:
        None
        """
    
    @property
    def started(self) -> bool:
        """Whether server has been started."""
    
    @property
    def closed(self) -> bool:
        """Whether server socket is closed."""

Datagram Server

UDP server for handling datagram packets.

class DatagramServer:
    """
    UDP server for handling datagram packets.
    """
    
    def __init__(self, listener, handle=None, spawn='default'):
        """
        Create a datagram server.
        
        Parameters:
        - listener: tuple (host, port) or socket object
        - handle: callable, function to handle each packet
        - spawn: str or callable, greenlet spawning strategy
        """
    
    def serve_forever(self, stop_timeout=None):
        """
        Start server and handle packets forever.
        
        Parameters:
        - stop_timeout: float, timeout for stopping
        
        Returns:
        None
        """
    
    def start(self):
        """
        Start server in background.
        
        Returns:
        None
        """
    
    def stop(self, timeout=None):
        """
        Stop server.
        
        Parameters:
        - timeout: float, timeout for shutdown
        
        Returns:
        None
        """

Base Server

Abstract base class for all servers.

class BaseServer:
    """
    Abstract base class for gevent servers.
    """
    
    def serve_forever(self):
        """
        Abstract method to run server forever.
        
        Returns:
        None
        """
    
    def start(self):
        """
        Start server.
        
        Returns:
        None
        """
    
    def stop(self):
        """
        Stop server.
        
        Returns:
        None
        """
    
    def close(self):
        """
        Close server resources.
        
        Returns:
        None
        """

WSGI Server

Complete HTTP server implementing WSGI interface for web applications.

class WSGIServer(StreamServer):
    """
    HTTP server implementing WSGI interface.
    """
    
    def __init__(self, listener, application, **kwargs):
        """
        Create WSGI server.
        
        Parameters:
        - listener: tuple (host, port) or socket object
        - application: WSGI application callable
        - **kwargs: additional server configuration
        """
    
    def serve_forever(self):
        """
        Run WSGI server forever.
        
        Returns:
        None
        """

class WSGIHandler:
    """
    Handler for individual WSGI requests.
    """
    
    def handle(self):
        """
        Handle a single WSGI request.
        
        Returns:
        None
        """

class LoggingLogAdapter:
    """
    Log adapter for WSGI request logging.
    """

class Environ(dict):
    """
    WSGI environ dictionary with additional functionality.
    """

class SecureEnviron(Environ):
    """
    Secure WSGI environ that redacts sensitive information.
    """

class WSGISecureEnviron(SecureEnviron):
    """
    WSGI-specific secure environ implementation.
    """

Usage Examples

Basic TCP Echo Server

import gevent
from gevent import server

def echo_handler(socket, address):
    print(f"Connection from {address}")
    
    try:
        while True:
            data = socket.recv(1024)
            if not data:
                break
            socket.send(data)  # Echo back
    except Exception as e:
        print(f"Error: {e}")
    finally:
        socket.close()

# Create and start server
echo_server = server.StreamServer(('localhost', 8080), echo_handler)
print("Echo server starting on port 8080")
echo_server.serve_forever()

HTTP Server with WSGI

from gevent import pywsgi

def simple_app(environ, start_response):
    """Simple WSGI application."""
    status = '200 OK'
    headers = [('Content-Type', 'text/plain')]
    start_response(status, headers)
    
    path = environ['PATH_INFO']
    method = environ['REQUEST_METHOD']
    
    return [f"Hello! You requested {method} {path}\n".encode()]

# Create WSGI server
wsgi_server = pywsgi.WSGIServer(('localhost', 8000), simple_app)
print("WSGI server starting on port 8000")
wsgi_server.serve_forever()

UDP Echo Server

import gevent
from gevent import server

def udp_echo_handler(socket, address):
    data, client_addr = socket.recvfrom(1024)
    print(f"UDP packet from {client_addr}: {data}")
    socket.sendto(data, client_addr)  # Echo back

# Create UDP server
udp_server = server.DatagramServer(('localhost', 9090), udp_echo_handler)
print("UDP echo server starting on port 9090")
udp_server.serve_forever()

Server with SSL

import gevent
from gevent import server, ssl

def secure_handler(socket, address):
    print(f"Secure connection from {address}")
    
    try:
        # Read request
        data = socket.recv(1024)
        
        # Send HTTP response
        response = b"HTTP/1.1 200 OK\r\nContent-Length: 12\r\n\r\nHello HTTPS!"
        socket.send(response)
    finally:
        socket.close()

# Create SSL server
ssl_server = server.StreamServer(
    ('localhost', 8443), 
    secure_handler,
    keyfile='server.key',
    certfile='server.crt'
)

print("HTTPS server starting on port 8443")
ssl_server.serve_forever()

Server Lifecycle Management

import gevent
from gevent import server, signal as gevent_signal
import signal

def connection_handler(socket, address):
    print(f"Handling connection from {address}")
    
    try:
        socket.send(b"Welcome to the server!\n")
        
        while True:
            data = socket.recv(1024)
            if not data or data.strip().lower() == b'quit':
                break
            socket.send(b"Echo: " + data)
            
    finally:
        socket.close()

# Create server
tcp_server = server.StreamServer(('localhost', 8080), connection_handler)

# Set up graceful shutdown
def shutdown_handler():
    print("Shutting down server...")
    tcp_server.stop(timeout=5)
    print("Server stopped")

# Handle SIGTERM and SIGINT
gevent_signal.signal(signal.SIGTERM, shutdown_handler)
gevent_signal.signal(signal.SIGINT, shutdown_handler)

# Start server
print("Server starting... Press Ctrl+C to stop")
try:
    tcp_server.serve_forever()
except KeyboardInterrupt:
    shutdown_handler()

Multi-Protocol Server

import gevent
from gevent import server

def http_handler(socket, address):
    """Simple HTTP handler."""
    try:
        request = socket.recv(1024).decode()
        
        response = (
            "HTTP/1.1 200 OK\r\n"
            "Content-Type: text/html\r\n"
            "Content-Length: 13\r\n"
            "\r\n"
            "Hello, HTTP!"
        )
        
        socket.send(response.encode())
    finally:
        socket.close()

def raw_handler(socket, address):
    """Raw TCP handler."""
    try:
        socket.send(b"Raw TCP Server\n")
        data = socket.recv(1024)
        socket.send(b"You sent: " + data)
    finally:
        socket.close()

# Start multiple servers
servers = [
    server.StreamServer(('localhost', 8080), http_handler),
    server.StreamServer(('localhost', 8081), raw_handler),
]

# Start all servers
for s in servers:
    s.start()

print("HTTP server on port 8080, Raw TCP on port 8081")
print("Press Ctrl+C to stop")

try:
    # Wait forever
    gevent.wait()
except KeyboardInterrupt:
    print("Stopping servers...")
    for s in servers:
        s.stop()

Install with Tessl CLI