Wsgiserver 0.2 -

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

async def handle_client(self, reader, writer): data = await reader.read(4096) # Process request writer.write(response) await writer.drain() writer.close() 6.1 Hello World ( examples/hello_world.py ) from wsgiserver.server import WSGIServer def hello_world_app(environ, start_response): status = '200 OK' headers = [('Content-Type', 'text/html; charset=utf-8')] start_response(status, headers) wsgiserver 0.2

def __init__(self, host, port, application, workers=4): self.host = host self.port = port self.application = application self.workers = workers def run(self): for _ in range(self.workers): pid = os.fork() if pid == 0: # Child process server = WSGIServer(self.host, self.port, self.application) server.serve_forever() exit() # Parent waits os.wait() # /etc/systemd/system/wsgiserver.service [Unit] Description=WSGIServer 0.2 Application After=network.target [Service] Type=simple User=www-data Group=www-data WorkingDirectory=/var/www/myapp Environment="PYTHONPATH=/var/www/myapp" ExecStart=/var/www/myapp/venv/bin/python /var/www/myapp/app.py Restart=always # Give server time to start import time time

def __init__(self, app): self.app = app def __call__(self, environ, start_response): import time start = time.time() response = self.app(environ, start_response) duration = time.time() - start print(f"Request took: duration:.4fs") return response app = hello_world_app app = LoggingMiddleware(app) app = TimingMiddleware(app) headers) def __init__(self

class MultiProcessServer: """Fork multiple worker processes"""

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

Related Apps