Wsgiserver 0.2 -

response = requests.post('http://127.0.0.1:8889/', data=b'Test data') assert response.text == 'Test data'

Overview WSGIServer 0.2 is a lightweight, pure-Python WSGI HTTP server compliant with PEP 3333. This guide covers development setup, architecture, implementation details, testing, and deployment. 1. Project Structure wsgiserver/ ├── wsgiserver/ │ ├── __init__.py │ ├── server.py # Main server class │ ├── handler.py # Request handler │ ├── connection.py # Connection management │ ├── request.py # Request parsing │ ├── response.py # Response building │ └── utils.py # Helper functions ├── tests/ │ ├── test_server.py │ ├── test_handler.py │ └── fixtures/ ├── examples/ │ ├── hello_world.py │ └── middleware_demo.py ├── docs/ ├── setup.py ├── README.md └── requirements.txt 2. Core Implementation 2.1 Main Server Class ( server.py ) import socket import select import threading from wsgiref.handlers import BaseHandler class WSGIServer: """WSGI-compliant HTTP server"""

def __init__(self, host='127.0.0.1', port=8000, application=None): self.host = host self.port = port self.application = application self.server_socket = None self.running = False self.connections = [] def bind_and_listen(self): """Create and bind server socket""" self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_socket.bind((self.host, self.port)) self.server_socket.listen(5) self.server_socket.setblocking(False) def serve_forever(self): """Main server loop""" self.bind_and_listen() self.running = True print(f"Serving on http://self.host:self.port") while self.running: try: readable, _, _ = select.select([self.server_socket], [], [], 1) if readable: client_socket, addr = self.server_socket.accept() self.handle_client(client_socket, addr) except Exception as e: print(f"Error: e") def handle_client(self, client_socket, addr): """Handle incoming client connection""" handler = WSGIHandler(client_socket, addr, self.application) handler.handle_request() def shutdown(self): """Gracefully shutdown server""" self.running = False if self.server_socket: self.server_socket.close() import os import sys from io import BytesIO class WSGIHandler: """Handle individual HTTP requests""" wsgiserver 0.2

server = WSGIServer('0.0.0.0', 8000, app) server.serve_forever() 7.1 Production Configuration # production_server.py import os from wsgiserver.server import WSGIServer Production settings HOST = os.getenv('WSGI_HOST', '0.0.0.0') PORT = int(os.getenv('WSGI_PORT', 8000)) WORKERS = int(os.getenv('WSGI_WORKERS', 4))

start_time = time.time() with ThreadPoolExecutor(max_workers=50) as executor: results = list(executor.map(lambda _: make_request(), range(num_requests))) response = requests

server = WSGIServer('127.0.0.1', 8889, app) thread = threading.Thread(target=server.serve_forever) thread.daemon = True thread.start()

def __init__(self, client_socket, client_address, application): self.client_socket = client_socket self.client_address = client_address self.application = application self.request_data = b'' self.headers = {} self.method = '' self.path = '' self.http_version = '' def parse_request(self): """Parse raw HTTP request""" # Read request line data = self.client_socket.recv(4096) self.request_data = data # Parse request line lines = data.split(b'\r\n') if lines: request_line = lines[0].decode('utf-8') parts = request_line.split() if len(parts) >= 3: self.method = parts[0] self.path = parts[1] self.http_version = parts[2] # Parse headers for line in lines[1:]: if line == b'': break header_line = line.decode('utf-8') if ': ' in header_line: key, value = header_line.split(': ', 1) self.headers[key] = value def build_environ(self): """Build WSGI environ dictionary""" environ = 'REQUEST_METHOD': self.method, 'SCRIPT_NAME': '', 'PATH_INFO': self.path, 'QUERY_STRING': self.path.split('?', 1)[1] if '?' in self.path else '', 'CONTENT_TYPE': self.headers.get('Content-Type', ''), 'CONTENT_LENGTH': self.headers.get('Content-Length', ''), 'SERVER_NAME': self.client_socket.getsockname()[0], 'SERVER_PORT': str(self.client_socket.getsockname()[1]), 'SERVER_PROTOCOL': self.http_version, 'wsgi.version': (1, 0), 'wsgi.url_scheme': 'http', 'wsgi.input': BytesIO(self.request_data), 'wsgi.errors': sys.stderr, 'wsgi.multithread': True, 'wsgi.multiprocess': False, 'wsgi.run_once': False, # Add headers as HTTP_ prefixed variables for key, value in self.headers.items(): env_key = f"HTTP_key.replace('-', '_').upper()" environ[env_key] = value return environ def send_response(self, response_data, status, headers): """Send HTTP response to client""" response_line = f"HTTP/1.1 status\r\n" response = response_line.encode() for key, value in headers: response += f"key: value\r\n".encode() response += b"\r\n" if isinstance(response_data, bytes): response += response_data elif isinstance(response_data, str): response += response_data.encode() self.client_socket.sendall(response) def handle_request(self): """Process the request through WSGI application""" try: self.parse_request() environ = self.build_environ() # Call WSGI application result = self.application(environ, self.start_response) # Send response response_data = b''.join(result) self.send_response(response_data, self.status, self.headers_to_send) except Exception as e: self.send_error_response(str(e)) finally: self.client_socket.close() def start_response(self, status, headers, exc_info=None): """WSGI start_response callback""" self.status = status self.headers_to_send = headers return self.write def write(self, data): """Write response data (for older WSGI)""" # Handle chunked writing if needed pass def send_error_response(self, error_msg): """Send error response to client""" error_body = f"<html><body><h1>500 Internal Server Error</h1><p>error_msg</p></body></html>" self.send_response( error_body, "500 Internal Server Error", [('Content-Type', 'text/html')] ) 3.1 Virtual Environment # Create virtual environment python -m venv venv Activate it source venv/bin/activate # Linux/Mac or venv\Scripts\activate # Windows Upgrade pip pip install --upgrade pip 3.2 Requirements ( requirements.txt ) # Core dependencies (none - pure Python) # Development dependencies pytest>=7.0.0 pytest-cov>=4.0.0 black>=23.0.0 flake8>=6.0.0 mypy>=1.0.0 requests>=2.28.0 gunicorn>=20.1.0 # For comparison/testing 3.3 Installation # Install in development mode pip install -e . Install dev dependencies pip install -r requirements.txt 4. Testing Strategy 4.1 Unit Tests ( tests/test_server.py ) import pytest import threading import requests from wsgiserver.server import WSGIServer def test_basic_request(): """Test basic HTTP request/response""" def app(environ, start_response): start_response('200 OK', [('Content-Type', 'text/plain')]) return [b'Hello, World!'] Testing Strategy 4

# Give server time to start import time time.sleep(0.5)