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
networking.mddocs/
Networking
Complete networking stack with cooperative sockets, SSL support, and gevent-aware I/O operations. All network operations work seamlessly with greenlets, yielding control during I/O to enable high concurrency.
Capabilities
Socket Operations
Cooperative socket interface that mirrors Python's standard socket API.
def create_connection(address, timeout=None, source_address=None) -> socket:
"""
Create a socket connection to address.
Parameters:
- address: tuple (host, port) or string path for Unix sockets
- timeout: float, connection timeout in seconds
- source_address: tuple, source address to bind to
Returns:
Connected socket object
"""
class socket:
"""
Cooperative socket that yields during I/O operations.
"""
def __init__(self, family=AF_INET, type=SOCK_STREAM, proto=0, fileno=None):
"""
Create a socket.
Parameters:
- family: int, address family (AF_INET, AF_INET6, AF_UNIX)
- type: int, socket type (SOCK_STREAM, SOCK_DGRAM)
- proto: int, protocol number
- fileno: int, existing file descriptor
"""
def connect(self, address):
"""
Connect to address.
Parameters:
- address: tuple (host, port) or string path
Returns:
None
"""
def bind(self, address):
"""
Bind socket to address.
Parameters:
- address: tuple (host, port) or string path
Returns:
None
"""
def listen(self, backlog=None):
"""
Listen for connections.
Parameters:
- backlog: int, maximum number of pending connections
Returns:
None
"""
def accept(self) -> tuple[socket, tuple]:
"""
Accept a connection.
Returns:
tuple of (socket, address) for new connection
"""
def recv(self, bufsize, flags=0) -> bytes:
"""
Receive data from socket.
Parameters:
- bufsize: int, maximum bytes to receive
- flags: int, optional flags
Returns:
bytes received
"""
def send(self, data, flags=0) -> int:
"""
Send data on socket.
Parameters:
- data: bytes, data to send
- flags: int, optional flags
Returns:
int, number of bytes sent
"""
def sendall(self, data, flags=0):
"""
Send all data on socket.
Parameters:
- data: bytes, data to send
- flags: int, optional flags
Returns:
None
"""
def close(self):
"""
Close the socket.
Returns:
None
"""
def wait_read(self, timeout=None):
"""
Wait for socket to become readable.
Parameters:
- timeout: float, maximum time to wait
Returns:
None
"""
def wait_write(self, timeout=None):
"""
Wait for socket to become writable.
Parameters:
- timeout: float, maximum time to wait
Returns:
None
"""DNS Resolution
Cooperative DNS resolution functions.
def getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
"""
Get address info for host and port.
Parameters:
- host: str, hostname or IP address
- port: int or str, port number or service name
- family: int, address family filter
- type: int, socket type filter
- proto: int, protocol filter
- flags: int, optional flags
Returns:
list of (family, type, proto, canonname, sockaddr) tuples
"""
def gethostbyname(hostname) -> str:
"""
Get IP address for hostname.
Parameters:
- hostname: str, hostname to resolve
Returns:
str, IP address
"""
def gethostbyaddr(ip_address) -> tuple:
"""
Get hostname for IP address.
Parameters:
- ip_address: str, IP address to reverse lookup
Returns:
tuple of (hostname, aliaslist, ipaddrlist)
"""SSL Support
Secure socket layer support for encrypted connections.
class SSLSocket:
"""
SSL-wrapped socket for encrypted communication.
"""
def read(self, len=1024) -> bytes:
"""
Read data from SSL socket.
Parameters:
- len: int, maximum bytes to read
Returns:
bytes received
"""
def write(self, data) -> int:
"""
Write data to SSL socket.
Parameters:
- data: bytes, data to write
Returns:
int, number of bytes written
"""
def do_handshake(self):
"""
Perform SSL handshake.
Returns:
None
"""
def unwrap(self) -> socket:
"""
Unwrap SSL socket to plain socket.
Returns:
Plain socket object
"""
def wrap_socket(sock, keyfile=None, certfile=None, server_side=False, cert_reqs=CERT_NONE, ssl_version=PROTOCOL_TLS, ca_certs=None, do_handshake_on_connect=True, suppress_ragged_eofs=True, ciphers=None) -> SSLSocket:
"""
Wrap socket with SSL.
Parameters:
- sock: socket object to wrap
- keyfile: str, path to private key file
- certfile: str, path to certificate file
- server_side: bool, whether this is server side
- cert_reqs: int, certificate requirements
- ssl_version: int, SSL version to use
- ca_certs: str, path to CA certificates
- do_handshake_on_connect: bool, perform handshake immediately
- suppress_ragged_eofs: bool, suppress SSL EOF errors
- ciphers: str, cipher suite specification
Returns:
SSLSocket object
"""
class SSLContext:
"""
SSL context for configuring SSL connections.
"""
def wrap_socket(self, sock, server_side=False, do_handshake_on_connect=True, suppress_ragged_eofs=True, server_hostname=None) -> SSLSocket:
"""
Wrap socket with this SSL context.
Parameters:
- sock: socket to wrap
- server_side: bool, server mode
- do_handshake_on_connect: bool, immediate handshake
- suppress_ragged_eofs: bool, suppress EOF errors
- server_hostname: str, expected server hostname
Returns:
SSLSocket object
"""File Objects
File-like objects for socket I/O.
class FileObject:
"""
File-like object wrapper for sockets.
"""
def read(self, size=-1) -> bytes:
"""
Read data from file object.
Parameters:
- size: int, bytes to read (-1 for all)
Returns:
bytes read
"""
def readline(self, size=-1) -> bytes:
"""
Read one line from file object.
Parameters:
- size: int, maximum bytes to read
Returns:
bytes for one line
"""
def write(self, data) -> int:
"""
Write data to file object.
Parameters:
- data: bytes, data to write
Returns:
int, bytes written
"""
def flush(self):
"""
Flush write buffers.
Returns:
None
"""
def close(self):
"""
Close file object.
Returns:
None
"""Usage Examples
TCP Client
import gevent
from gevent import socket
def tcp_client():
# Create and connect socket
sock = socket.socket()
sock.connect(('example.com', 80))
# Send HTTP request
request = b'GET / HTTP/1.0\r\nHost: example.com\r\n\r\n'
sock.sendall(request)
# Receive response
response = b''
while True:
data = sock.recv(4096)
if not data:
break
response += data
sock.close()
return response
response = tcp_client()
print(response[:200]) # Print first 200 bytesTCP Server
import gevent
from gevent import socket
def echo_server():
server = socket.socket()
server.bind(('localhost', 8080))
server.listen(1000)
print("Echo server listening on port 8080")
while True:
conn, addr = server.accept()
gevent.spawn(handle_client, conn, addr)
def handle_client(conn, addr):
print(f"Connection from {addr}")
try:
while True:
data = conn.recv(1024)
if not data:
break
conn.sendall(data) # Echo back
except Exception as e:
print(f"Error handling client {addr}: {e}")
finally:
conn.close()
# Start server
echo_server()HTTPS Client
import gevent
from gevent import socket, ssl
def https_client():
# Create SSL context
context = ssl.create_default_context()
# Create socket and wrap with SSL
sock = socket.socket()
ssl_sock = context.wrap_socket(sock, server_hostname='httpbin.org')
# Connect and send request
ssl_sock.connect(('httpbin.org', 443))
request = b'GET /get HTTP/1.1\r\nHost: httpbin.org\r\n\r\n'
ssl_sock.sendall(request)
# Read response
response = ssl_sock.recv(4096)
ssl_sock.close()
return response
response = https_client()
print(response.decode())Install with Tessl CLI
npx tessl i tessl/pypi-gevent