Atomically publish the assigned port after the listening socket exists.
(port_file: pathlib.Path, port: int)
| 32 | |
| 33 | |
| 34 | def publish_port(port_file: pathlib.Path, port: int) -> None: |
| 35 | """Atomically publish the assigned port after the listening socket exists.""" |
| 36 | port_file.parent.mkdir(parents=True, exist_ok=True) |
| 37 | with tempfile.NamedTemporaryFile( |
| 38 | mode="w", |
| 39 | encoding="ascii", |
| 40 | dir=port_file.parent, |
| 41 | prefix=f".{port_file.name}.", |
| 42 | delete=False, |
| 43 | ) as temporary: |
| 44 | temporary.write(f"{port}\n") |
| 45 | temporary.flush() |
| 46 | temporary_path = pathlib.Path(temporary.name) |
| 47 | try: |
| 48 | os.replace(temporary_path, port_file) |
| 49 | except BaseException: |
| 50 | temporary_path.unlink(missing_ok=True) |
| 51 | raise |
| 52 | |
| 53 | |
| 54 | def main() -> None: |