### Build a High-Performance HTTP Server with aiohttp and winloop
Source: https://context7.com/vizonex/winloop/llms.txt
Demonstrates setting up an aiohttp application using winloop as the underlying event loop. It includes handlers for GET and POST requests and manages the application lifecycle with a runner.
```python
import aiohttp
import aiohttp.web
import asyncio
import winloop
async def handle_request(request):
"""Handle HTTP GET requests"""
name = request.match_info.get('name', 'World')
return aiohttp.web.Response(text=f"Hello, {name}!")
async def handle_json(request):
"""Handle JSON POST requests"""
try:
data = await request.json()
response_data = {
'status': 'success',
'received': data,
'echo': data.get('message', '')
}
return aiohttp.web.json_response(response_data)
except Exception as e:
return aiohttp.web.json_response(
{'status': 'error', 'message': str(e)},
status=400
)
async def main():
# Create application
app = aiohttp.web.Application()
# Setup routes
app.router.add_get('/', handle_request)
app.router.add_get('/{name}', handle_request)
app.router.add_post('/api/echo', handle_json)
# Create and start server
runner = aiohttp.web.AppRunner(app)
await runner.setup()
site = aiohttp.web.TCPSite(runner, '127.0.0.1', 8080)
await site.start()
print("Server started at http://127.0.0.1:8080")
# Keep server running
try:
await asyncio.Event().wait()
except KeyboardInterrupt:
pass
finally:
await runner.cleanup()
if __name__ == "__main__":
winloop.run(main())
```
--------------------------------
### Test aiohttp basic functionality with Winloop
Source: https://github.com/vizonex/winloop/blob/main/README.md
This Python code snippet demonstrates how to test basic aiohttp functionality using the Winloop event loop. It sets up an aiohttp web application, runs it, and then uses a client to make requests to verify the server's response. It includes setup for the event loop and application, and cleanup of the runner.
```python
try:
import aiohttp
import aiohttp.web
except ImportError:
skip_tests = True
else:
skip_tests = False
import asyncio
import unittest
import weakref
import winloop
import sys
class TestAioHTTP(unittest.TestCase):
def __init__(self, methodName: str = "test_aiohttp_basic_1") -> None:
super().__init__(methodName)
def setUp(self):
self.loop = asyncio.get_event_loop()
def test_aiohttp_basic_1(self):
PAYLOAD = 'It Works!
' * 10000
async def on_request(request):
return aiohttp.web.Response(text=PAYLOAD)
asyncio.set_event_loop(self.loop)
app = aiohttp.web.Application()
app.router.add_get('/', on_request)
runner = aiohttp.web.AppRunner(app)
self.loop.run_until_complete(runner.setup())
site = aiohttp.web.TCPSite(runner, '0.0.0.0', '10000')
self.loop.run_until_complete(site.start())
port = site._server.sockets[0].getsockname()[1]
async def test():
# Make sure we're using the correct event loop.
self.assertIs(asyncio.get_event_loop(), self.loop)
for addr in (('localhost', port),
('127.0.0.1', port)):
async with aiohttp.ClientSession() as client:
async with client.get('http://{}:{}'.format(*addr)) as r:
self.assertEqual(r.status, 200)
result = await r.text()
self.assertEqual(result, PAYLOAD)
self.loop.run_until_complete(test())
self.loop.run_until_complete(runner.cleanup())
def test_aiohttp_graceful_shutdown(self):
async def websocket_handler(request):
ws = aiohttp.web.WebSocketResponse()
await ws.prepare(request)
request.app['websockets'].add(ws)
try:
async for msg in ws:
await ws.send_str(msg.data)
finally:
request.app['websockets'].discard(ws)
return ws
async def on_shutdown(app):
for ws in set(app['websockets']):
await ws.close(
code=aiohttp.WSCloseCode.GOING_AWAY,
message='Server shutdown')
asyncio.set_event_loop(self.loop)
app = aiohttp.web.Application()
app.router.add_get('/', websocket_handler)
app.on_shutdown.append(on_shutdown)
app['websockets'] = weakref.WeakSet()
runner = aiohttp.web.AppRunner(app)
self.loop.run_until_complete(runner.setup())
site = aiohttp.web.TCPSite(runner, '0.0.0.0', '10000')
self.loop.run_until_complete(site.start())
port = site._server.sockets[0].getsockname()[1]
async def client():
async with aiohttp.ClientSession() as client:
async with client.ws_connect(
'http://127.0.0.1:{}'.format(port)) as ws:
await ws.send_str("hello")
async for msg in ws:
assert msg.data == "hello"
client_task = asyncio.ensure_future(client())
async def stop():
await asyncio.sleep(0.1)
try:
await asyncio.wait_for(runner.cleanup(), timeout=0.1)
except Exception as e:
print(e)
finally:
try:
client_task.cancel()
await client_task
except asyncio.CancelledError:
pass
self.loop.run_until_complete(stop())
if __name__ == "__main__":
# print("tesing without winloop")
# asyncio.DefaultEventLoopPolicy = asyncio.WindowsSelectorEventLoopPolicy
# asyncio.DefaultEventLoopPolicy = asyncio.WindowsProactorEventLoopPolicy
unittest.main()
# Looks like winloop might be 3x faster than the Proctor Event Loop , THAT's A HUGE IMPROVEMENT!
print("testing again but with winloop enabled")
winloop.install()
unittest.main()
```
--------------------------------
### Configure Production-Ready FastAPI with winloop
Source: https://context7.com/vizonex/winloop/llms.txt
Integrates winloop with FastAPI and Uvicorn to optimize request handling. The script explicitly installs the winloop policy and configures Uvicorn to run within the optimized event loop.
```python
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse, JSONResponse
import winloop
import uvicorn
import asyncio
import datetime
from typing import Dict, Any
app = FastAPI(title="Winloop FastAPI Demo")
@app.on_event("startup")
def verify_winloop():
"""Verify that winloop event loop policy is active"""
policy = asyncio.get_event_loop_policy()
assert isinstance(policy, winloop.EventLoopPolicy)
print("✓ Winloop event loop policy is active")
@app.get("/", response_class=HTMLResponse)
async def root():
"""Root endpoint returning HTML"""
return """
FastAPI + Winloop
FastAPI with Winloop is Running!
High-performance async event loop for Windows
"""
@app.get("/api/time")
async def get_time() -> Dict[str, str]:
"""Get current server timestamp"""
return {
"timestamp": datetime.datetime.now().isoformat(),
"timezone": "UTC" if datetime.datetime.now().tzinfo else "Local"
}
@app.get("/api/health")
async def health_check() -> Dict[str, Any]:
"""Health check endpoint"""
return {
"status": "healthy",
"event_loop": "winloop",
"timestamp": datetime.datetime.now().isoformat()
}
@app.post("/api/echo")
async def echo_data(data: Dict[str, Any]) -> JSONResponse:
"""Echo JSON data back to client"""
return JSONResponse({
"received": data,
"message": "Data echoed successfully"
})
if __name__ == "__main__":
# Install winloop before starting uvicorn
winloop.install()
# Get the current event loop
loop = asyncio.get_event_loop()
# Configure uvicorn with the winloop event loop
config = uvicorn.Config(
app=app,
host="0.0.0.0",
port=8000,
loop=loop,
log_level="info"
)
server = uvicorn.Server(config)
asyncio.run(server.serve())
```
--------------------------------
### Install Winloop as global policy
Source: https://context7.com/vizonex/winloop/llms.txt
Sets winloop as the default asyncio event loop policy globally. This method is deprecated in favor of winloop.run() for newer Python versions and should be used primarily for backward compatibility.
```python
import winloop
import asyncio
winloop.install()
async def handle_request():
await asyncio.sleep(0.1)
return "Request handled"
loop = asyncio.get_event_loop()
assert isinstance(asyncio.get_event_loop_policy(), winloop.EventLoopPolicy)
result = asyncio.run(handle_request())
print(result)
```
--------------------------------
### Implement Cross-Platform Event Loop Execution
Source: https://github.com/vizonex/winloop/blob/main/README.md
Shows a pattern for selecting the appropriate high-performance event loop based on the operating system, defaulting to Winloop on Windows and uvloop on other platforms.
```python
import sys
import aiohttp
import asyncio
async def main():
async with aiohttp.ClientSession("https://httpbin.org") as client:
async with client.get("/ip") as resp:
print(await resp.json())
if __name__ == "__main__":
if sys.platform in ('win32', 'cygwin', 'cli'):
from winloop import run
else:
from uvloop import run
run(main())
```
--------------------------------
### Integrate Winloop with FastAPI
Source: https://github.com/vizonex/winloop/blob/main/README.md
Demonstrates how to configure a FastAPI application to utilize the WinLoopPolicy. It includes an event loop assertion check and manual server orchestration using uvicorn.
```python
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
import winloop
import uvicorn
import asyncio
import datetime
app = FastAPI()
@app.on_event("startup")
def make_assertion():
assert isinstance(asyncio.get_event_loop_policy(), winloop.WinLoopPolicy)
@app.get("/test")
async def test_get_request():
return HTMLResponse("FAST API WORKS WITH WINLOOP!
")
@app.get("/date")
def test_dynamic_response():
return str(datetime.datetime.now())
if __name__ == "__main__":
winloop.install()
loop = asyncio.get_event_loop()
config = uvicorn.Config(app=app,port=10000,loop=loop)
server = uvicorn.Server(config)
asyncio.run(server.serve())
```
--------------------------------
### Implement SSL/TLS TCP server with winloop
Source: https://context7.com/vizonex/winloop/llms.txt
Demonstrates how to create an asynchronous TCP server that supports SSL/TLS encryption. It utilizes the winloop event loop for enhanced performance on Windows and includes standard certificate chain loading and connection echo functionality.
```python
import asyncio
import ssl
import winloop
from pathlib import Path
async def handle_secure_client(reader, writer):
"""Handle SSL/TLS client connections"""
peername = writer.get_extra_info('peername')
cipher = writer.get_extra_info('cipher')
print(f"Secure connection from {peername}")
print(f"Cipher: {cipher}")
try:
# Send welcome message
writer.write(b"Welcome to secure server\n")
await writer.drain()
# Echo loop
while True:
data = await reader.read(4096)
if not data:
break
print(f"Received {len(data)} bytes from {peername}")
writer.write(data)
await writer.drain()
except Exception as e:
print(f"Error handling {peername}: {e}")
finally:
writer.close()
await writer.wait_closed()
print(f"Closed connection from {peername}")
async def run_ssl_server():
"""Start SSL/TLS server"""
# Create SSL context
ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
# Load certificate and key
try:
ssl_context.load_cert_chain(
certfile='server_cert.pem',
keyfile='server_key.pem'
)
except FileNotFoundError:
print("Warning: SSL certificates not found")
return
# Optional: Configure SSL settings
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE
# Start server
server = await asyncio.start_server(
handle_secure_client,
'127.0.0.1',
8443,
ssl=ssl_context
)
addr = server.sockets[0].getsockname()
print(f"Secure server listening on {addr}")
async with server:
await server.serve_forever()
if __name__ == "__main__":
try:
winloop.run(run_ssl_server())
except KeyboardInterrupt:
print("\nServer stopped")
```
--------------------------------
### Implement a TCP echo server with Winloop
Source: https://context7.com/vizonex/winloop/llms.txt
Showcases a high-performance TCP server implementation utilizing winloop for efficient network I/O handling on Windows.
```python
import asyncio
import winloop
async def handle_echo_client(reader, writer):
addr = writer.get_extra_info('peername')
print(f"Connected: {addr}")
try:
while True:
data = await reader.read(1000000)
if not data:
break
writer.write(data)
await writer.drain()
except Exception as e:
print(f"Error: {e}")
finally:
writer.close()
await writer.wait_closed()
print(f"Disconnected: {addr}")
async def start_server():
server = await asyncio.start_server(handle_echo_client, '127.0.0.1', 8888)
async with server:
await server.serve_forever()
if __name__ == "__main__":
winloop.run(start_server())
```
--------------------------------
### Create a custom Winloop event loop
Source: https://context7.com/vizonex/winloop/llms.txt
Provides a factory method to instantiate a new winloop event loop. This is useful for manual control over the loop lifecycle within specific application scopes.
```python
import winloop
import asyncio
loop = winloop.new_event_loop()
asyncio.set_event_loop(loop)
async def fetch_data():
await asyncio.sleep(0.1)
return "Data fetched"
try:
result = loop.run_until_complete(fetch_data())
print(result)
finally:
loop.close()
```
--------------------------------
### Execute coroutines with Winloop
Source: https://context7.com/vizonex/winloop/llms.txt
Demonstrates the recommended method to run an asyncio coroutine using the winloop event loop. This approach is compatible with Python 3.8+ and replaces the standard asyncio execution flow.
```python
import winloop
import aiohttp
import asyncio
async def main():
async with aiohttp.ClientSession("https://httpbin.org") as client:
async with client.get("/ip") as resp:
data = await resp.json()
print(f"Your IP: {data['origin']}")
return data
if __name__ == "__main__":
result = winloop.run(main())
```
--------------------------------
### Cross-Platform Event Loop Selection (Python)
Source: https://context7.com/vizonex/winloop/llms.txt
Selects the best available event loop, prioritizing winloop on Windows and uvloop on Unix-like systems. Falls back to the default asyncio loop if neither is available. This pattern ensures optimal performance across different operating systems.
```python
import sys
import asyncio
import aiohttp
from typing import Any, Coroutine
async def fetch_url(url: str) -> dict:
"""Fetch data from URL asynchronously"""
async with aiohttp.ClientSession(url) as client:
async with client.get("/json") as resp:
if resp.status != 200:
raise Exception(f"HTTP {resp.status}")
return await resp.json()
async def main():
"""Main async function"""
try:
data = await fetch_url("https://httpbin.org")
print(f"Fetched data: {data}")
return data
except Exception as e:
print(f"Error: {e}")
return None
def run_with_best_loop(coro: Coroutine) -> Any:
"""Run coroutine with the best available event loop"""
if sys.platform in ('win32', 'cygwin', 'cli'):
# Use winloop on Windows
try:
from winloop import run
print("Using winloop (Windows)")
return run(coro)
except ImportError:
print("Winloop not available, using default asyncio")
return asyncio.run(coro)
else:
# Use uvloop on Unix/Linux/macOS
try:
from uvloop import run
print("Using uvloop (Unix)")
return run(coro)
except ImportError:
print("Uvloop not available, using default asyncio")
return asyncio.run(coro)
if __name__ == "__main__":
result = run_with_best_loop(main())
```
--------------------------------
### Asynchronous DNS Resolution with Winloop (Python)
Source: https://context7.com/vizonex/winloop/llms.txt
Performs asynchronous DNS lookups for a list of hostnames using winloop's `getaddrinfo` implementation. It handles potential `socket.gaierror` exceptions and concurrently resolves multiple hosts. This is useful for network applications requiring efficient name resolution.
```python
import winloop
import socket
import asyncio
async def resolve_hosts(hostnames: list[str]):
"""Resolve multiple hostnames concurrently"""
loop = asyncio.get_running_loop()
async def resolve_one(hostname: str):
try:
# Asynchronous getaddrinfo using winloop
addrs = await loop.getaddrinfo(
hostname,
80,
family=socket.AF_INET,
type=socket.SOCK_STREAM,
flags=socket.AI_CANONNAME
)
results = []
for family, socktype, proto, canonname, sockaddr in addrs:
results.append({
'hostname': hostname,
'canonical_name': canonname,
'ip': sockaddr[0],
'port': sockaddr[1]
})
return results
except socket.gaierror as e:
return [{'hostname': hostname, 'error': str(e)}]
# Resolve all hosts concurrently
tasks = [resolve_one(host) for host in hostnames]
results = await asyncio.gather(*tasks)
# Flatten results
all_results = []
for result_list in results:
all_results.extend(result_list)
return all_results
async def main():
hosts = ['google.com', 'github.com', 'python.org', 'invalid-host-xyz.com']
print(f"Resolving {len(hosts)} hostnames...")
results = await resolve_hosts(hosts)
for result in results:
if 'error' in result:
print(f"❌ {result['hostname']}: {result['error']}")
else:
print(f"✓ {result['hostname']} -> {result['ip']} (canonical: {result['canonical_name']})")
if __name__ == "__main__":
winloop.run(main())
```
=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.