-
Notifications
You must be signed in to change notification settings - Fork 35
/
Copy pathport_manager.py
41 lines (30 loc) · 1.18 KB
/
port_manager.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
import socket
import random
from typing import Set, Iterable, Optional
class PortForException(Exception):
pass
class PortManager:
def __init__(self, ports_range=(1024, 65535)):
self.ports_range = ports_range
@staticmethod
def is_port_free(port: int) -> bool:
"""Check if a port is free to use."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
try:
s.bind(("", port))
return True
except OSError:
return False
def find_free_port(self, ports: Optional[Set[int]] = None, exclude_ports: Optional[Iterable[int]] = None) -> int:
"""Return a random unused port number."""
if ports is None:
ports = set(range(1024, 65535))
assert type(ports) == set # noqa: E721
if exclude_ports is not None:
assert isinstance(exclude_ports, Iterable)
ports.difference_update(exclude_ports)
sampled_ports = random.sample(tuple(ports), min(len(ports), 100))
for port in sampled_ports:
if self.is_port_free(port):
return port
raise PortForException("Can't select a port")