|
import ssl |
|
from starlette.applications import Starlette |
|
from starlette.routing import Route |
|
from starlette.responses import PlainTextResponse |
|
import httpx |
|
import aiohttp |
|
import multiprocessing |
|
|
|
|
|
URL = "https://localhost:8000" |
|
|
|
ssl_context = ssl.create_default_context(cafile="client.pem") |
|
session = None |
|
client = httpx.AsyncClient(verify=ssl_context) |
|
|
|
|
|
async def startup(): |
|
global session |
|
session = aiohttp.ClientSession() |
|
|
|
|
|
async def index(request): |
|
return PlainTextResponse("world") |
|
|
|
|
|
async def aiohttp_single(request): |
|
async with aiohttp.ClientSession() as session: |
|
async with session.get(URL, ssl=ssl_context) as r: |
|
return _response(await r.text()) |
|
|
|
|
|
async def aiohttp_session(request): |
|
async with session.get(URL, ssl=ssl_context) as r: |
|
return _response(await r.text()) |
|
|
|
|
|
async def httpx_single(request): |
|
async with httpx.AsyncClient(verify=ssl_context) as client: |
|
r = await client.get(URL) |
|
return _response(r.text) |
|
|
|
|
|
async def httpx_client(request): |
|
r = await client.get(URL) |
|
return _response(r.text) |
|
|
|
|
|
def _response(name): |
|
return PlainTextResponse("Hello, " + name) |
|
|
|
|
|
routes = [ |
|
Route("/", endpoint=index), |
|
Route("/aiohttp/single", endpoint=aiohttp_single), |
|
Route("/aiohttp/session", endpoint=aiohttp_session), |
|
Route("/httpx/single", endpoint=httpx_single), |
|
Route("/httpx/client", endpoint=httpx_client), |
|
] |
|
|
|
|
|
app = Starlette(routes=routes, on_startup=[startup]) |
|
|
|
|
|
if __name__ == "__main__": |
|
import uvicorn |
|
|
|
def http(): |
|
uvicorn.run( |
|
app, |
|
host="localhost", |
|
port=4000, |
|
log_level="error", |
|
) |
|
|
|
p = multiprocessing.Process(target=http) |
|
p.start() |
|
|
|
uvicorn.run( |
|
app, |
|
host="localhost", |
|
port=8000, |
|
ssl_keyfile="server.key", |
|
ssl_certfile="server.pem", |
|
log_level="error", |
|
) |