Mercurial > code > home > repos > rdferry
annotate examples/_run_server_child.py @ 10:52e1bb1532f2
serve_inline_graph
| author | drewp@bigasterisk.com |
|---|---|
| date | Sat, 16 Mar 2024 16:02:23 -0700 |
| parents | 25538e3ee531 |
| children | 0bc06da6bf74 |
| rev | line source |
|---|---|
| 3 | 1 import asyncio |
| 2 from dataclasses import dataclass | |
| 3 from pathlib import Path | |
| 4 | |
| 5 import aiohttp | |
| 6 | |
| 7 | |
| 8 @dataclass | |
| 8 | 9 class RunHttpServerChildProcess: |
| 3 | 10 server_path: Path |
| 11 | |
| 12 async def __aenter__(self): | |
| 13 self.subprocess = await asyncio.create_subprocess_exec( | |
| 14 'pdm', 'run', 'python', self.server_path) | |
| 15 self._session = await aiohttp.ClientSession().__aenter__() | |
| 16 return self | |
| 17 | |
| 18 async def __aexit__(self, exc_type, exc, tb): | |
|
6
d885fce5e4e7
try to reduce warnings about leaked sessions
drewp@bigasterisk.com
parents:
5
diff
changeset
|
19 await self._session.close() |
| 3 | 20 self.subprocess.terminate() |
|
5
b34cd6619316
new test. server has to be able to exit between tests.
drewp@bigasterisk.com
parents:
3
diff
changeset
|
21 await self.subprocess.wait() |
|
6
d885fce5e4e7
try to reduce warnings about leaked sessions
drewp@bigasterisk.com
parents:
5
diff
changeset
|
22 |
| 10 | 23 async def get(self, url: str, headers=None) -> aiohttp.ClientResponse: |
| 3 | 24 while True: |
| 25 try: | |
| 10 | 26 return await self._session.get(url, headers=headers) |
| 3 | 27 except aiohttp.ClientConnectorError: |
| 28 await asyncio.sleep(0.05) |