cengal.io.serve_free_ports.versions.v_0.serve_free_ports

Module Docstring Docstrings: http://www.python.org/dev/peps/pep-0257/

  1#!/usr/bin/env python
  2# coding=utf-8
  3
  4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>
  5# 
  6# Licensed under the Apache License, Version 2.0 (the "License");
  7# you may not use this file except in compliance with the License.
  8# You may obtain a copy of the License at
  9# 
 10#     http://www.apache.org/licenses/LICENSE-2.0
 11# 
 12# Unless required by applicable law or agreed to in writing, software
 13# distributed under the License is distributed on an "AS IS" BASIS,
 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 15# See the License for the specific language governing permissions and
 16# limitations under the License.
 17
 18"""
 19Module Docstring
 20Docstrings: http://www.python.org/dev/peps/pep-0257/
 21"""
 22
 23__author__ = "ButenkoMS <gtalk@butenkoms.space>"
 24__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>"
 25__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ]
 26__license__ = "Apache License, Version 2.0"
 27__version__ = "4.4.1"
 28__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>"
 29__email__ = "gtalk@butenkoms.space"
 30# __status__ = "Prototype"
 31__status__ = "Development"
 32# __status__ = "Production"
 33
 34
 35__all__ = ['find_free_port', 'simple_port_search', 'asimple_port_search', 'find_free_ports_range', 'simple_ports_range_search', 'asimple_ports_range_search']
 36
 37
 38import asyncio
 39from cengal.io.used_ports import used_ports, PortsIterator, Protocol, PortStatus, UsedPorts
 40from cengal.math.numbers import RationalNumber
 41from typing import Optional, Union, Tuple, Set
 42from gc import collect
 43
 44
 45def client_connected_cb(*args, **kwargs):
 46    pass
 47
 48
 49async def find_free_port(host, ports_iterartor: PortsIterator, timeout: RationalNumber = 0) -> Optional[int]:
 50    """_summary_
 51    Example:
 52
 53    from cengal.io.serve_free_ports import find_free_port
 54    from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports
 55    import asyncio
 56
 57    async def main():
 58        ports_rages_iterator: PortsIterator = used_ports().range(Protocol.tcp, slice(18000, 18100), {PortStatus.na, PortStatus.no})
 59        port: int = await find_free_port('0.0.0.0', ports_rages_iterator, 5.0)
 60        if port is not None:
 61            print(port)  # Output: 18000
 62
 63    Args:
 64        host (_type_): _description_
 65        ports_iterartor (PortsIterator): _description_
 66        timeout (RationalNumber, optional): _description_. Defaults to 0.
 67
 68    Returns:
 69        Optional[int]: _description_
 70    """    
 71    for ports_range in ports_iterartor:
 72        for port in range(ports_range.start, ports_range.stop):
 73            try:
 74                if timeout:
 75                    server = await asyncio.wait_for(asyncio.start_server(client_connected_cb, host, port), timeout)
 76                else:
 77                    server = await asyncio.start_server(client_connected_cb, host, port)
 78                
 79                try:
 80                    server.close()
 81                except Exception:
 82                    pass
 83                finally:
 84                    del server
 85                    collect()
 86
 87                return port
 88            except TimeoutError as err:
 89                pass
 90            except (ConnectionRefusedError, ConnectionError, OSError) as err:
 91                pass
 92
 93
 94def simple_port_search(host: str, port_or_range: Union[int, slice, Tuple[int, int]], protocol: Protocol = Protocol.tcp, statuses: Optional[Union[PortStatus, Set[PortStatus]]] = None, timeout: RationalNumber = 0, used_ports_instance: Optional[UsedPorts] = None) -> Optional[int]:
 95    """_summary_
 96
 97    Example:
 98
 99    from cengal.io.serve_free_ports import simple_port_search
100    from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports
101
102    def main():
103        port: int = simple_port_search('0.0.0.0', slice(18000, 18100), Protocol.tcp, {PortStatus.na, PortStatus.no}, 5.0)
104        if port is not None:
105            print(port)  # Output: 18000
106
107    Args:
108        host (str): _description_
109        port_or_range (Union[int, slice, Tuple[int, int]]): _description_
110        protocol (Protocol, optional): _description_. Defaults to Protocol.tcp.
111        statuses (Optional[Union[PortStatus, Set[PortStatus]]], optional): _description_. Defaults to None.
112        desired_number_of_ports (int, optional): _description_. Defaults to 1.
113        timeout (int, optional): _description_. Defaults to 0.
114        used_ports_instance (Optional[UsedPorts], optional): .
115
116    Returns:
117        _type_: _description_
118    """    
119    if statuses is None: statuses = {PortStatus.na, PortStatus.no}
120    if used_ports_instance is None: used_ports_instance = used_ports()
121    return asyncio.run(find_free_port(host, used_ports_instance.port(protocol, port_or_range, statuses), timeout))
122
123
124async def asimple_port_search(host: str, port_or_range: Union[int, slice, Tuple[int, int]], protocol: Protocol = Protocol.tcp, statuses: Optional[Union[PortStatus, Set[PortStatus]]] = None, timeout: RationalNumber = 0, used_ports_instance: Optional[UsedPorts] = None) -> Optional[int]:
125    """_summary_
126
127    Example:
128
129    from cengal.io.serve_free_ports import asimple_port_search
130    from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports
131    import asyncio
132
133    async def main():
134        port: int = await asimple_port_search('0.0.0.0', slice(18000, 18100), Protocol.tcp, {PortStatus.na, PortStatus.no}, 5.0)
135        if port is not None:
136            print(port)  # Output: 18000
137
138    Args:
139        host (str): _description_
140        port_or_range (Union[int, slice, Tuple[int, int]]): _description_
141        protocol (Protocol, optional): _description_. Defaults to Protocol.tcp.
142        statuses (Optional[Union[PortStatus, Set[PortStatus]]], optional): _description_. Defaults to None.
143        desired_number_of_ports (int, optional): _description_. Defaults to 1.
144        timeout (int, optional): _description_. Defaults to 0.
145        used_ports_instance (Optional[UsedPorts], optional): .
146
147    Returns:
148        _type_: _description_
149    """    
150    if statuses is None: statuses = {PortStatus.na, PortStatus.no}
151    if used_ports_instance is None: used_ports_instance = used_ports()
152    return await find_free_port(host, used_ports_instance.port(protocol, port_or_range, statuses), timeout)
153
154
155async def find_free_ports_range(host, ports_iterartor: PortsIterator, timeout: RationalNumber = 0) -> Optional[slice]:
156    """
157    Example:
158
159    from cengal.io.serve_free_ports import find_free_ports_range
160    from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports
161    import asyncio
162
163    async def main():
164        ports_rages_iterator: PortsIterator = used_ports().range(Protocol.tcp, slice(18000, 18100), {PortStatus.na, PortStatus.no}, 3)
165        ports_range = await find_free_ports_range('0.0.0.0', ports_rages_iterator, 5.0)
166        if ports_range is not None:
167            first_port = ports_range.start  # Output: 18000
168            last_port = ports_range.stop - 1  # Output: 18002
169            first_port, last_port = unify_ports(ports_range)  # Output: (18000, 18002)
170            for port in range(ports_range.start, ports_range.stop):
171                print(port)
172            
173            print(list(range(ports_range.start, ports_range.stop)))  # Output: [18000, 18001, 18002]
174            if isinstance(purify_ports(ports_range), slice):
175                # type is `slice`
176                print('Ports range')
177            else:
178                # type is `int`
179                print('Single port')
180
181    Args:
182        host (_type_): _description_
183        ports_iterartor (PortsIterator): _description_
184        timeout (int, optional): _description_. Defaults to 0.
185
186    Returns:
187        Optional[slice]: _description_
188    """
189    for ports_range in ports_iterartor:
190        try:
191            for port in range(ports_range.start, ports_range.stop):
192                if timeout:
193                    server = await asyncio.wait_for(asyncio.start_server(client_connected_cb, host, port), timeout)
194                else:
195                    server = await asyncio.start_server(client_connected_cb, host, port)
196                
197                try:
198                    server.close()
199                except Exception:
200                    pass
201                finally:
202                    del server
203                    collect()
204
205            return ports_range
206        except TimeoutError as err:
207            pass
208        except (ConnectionRefusedError, ConnectionError, OSError) as err:
209            pass
210
211
212def simple_ports_range_search(host: str, port_or_range: Union[int, slice, Tuple[int, int]], protocol: Protocol = Protocol.tcp, statuses: Optional[Union[PortStatus, Set[PortStatus]]] = None, desired_number_of_ports: int = 1, timeout: RationalNumber = 0, used_ports_instance: Optional[UsedPorts] = None) -> Optional[slice]:
213    """_summary_
214
215    Example:
216
217    from cengal.io.serve_free_ports import simple_ports_range_search
218    from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports
219
220    def main():
221        ports_range = simple_ports_range_search('0.0.0.0', slice(18000, 18100), Protocol.tcp, {PortStatus.na, PortStatus.no}, 3, 5.0)
222        if ports_range is not None:
223            first_port = ports_range.start  # Output: 18000
224            last_port = ports_range.stop - 1  # Output: 18002
225            first_port, last_port = unify_ports(ports_range)  # Output: (18000, 18002)
226            for port in range(ports_range.start, ports_range.stop):
227                print(port)
228            
229            print(list(range(ports_range.start, ports_range.stop)))  # Output: [18000, 18001, 18002]
230            if isinstance(purify_ports(ports_range), slice):
231                # type is `slice`
232                print('Ports range')
233            else:
234                # type is `int`
235                print('Single port')
236
237    Args:
238        host (str): _description_
239        port_or_range (Union[int, slice, Tuple[int, int]]): _description_
240        protocol (Protocol, optional): _description_. Defaults to Protocol.tcp.
241        statuses (Optional[Union[PortStatus, Set[PortStatus]]], optional): _description_. Defaults to None.
242        desired_number_of_ports (int, optional): _description_. Defaults to 1.
243        timeout (int, optional): _description_. Defaults to 0.
244        used_ports_instance (Optional[UsedPorts], optional): .
245
246    Returns:
247        _type_: _description_
248    """    
249    if statuses is None: statuses = {PortStatus.na, PortStatus.no}
250    if used_ports_instance is None: used_ports_instance = used_ports()
251    return asyncio.run(find_free_ports_range(host, used_ports_instance.range(protocol, port_or_range, statuses, desired_number_of_ports), timeout))
252
253
254async def asimple_ports_range_search(host: str, port_or_range: Union[int, slice, Tuple[int, int]], protocol: Protocol = Protocol.tcp, statuses: Optional[Union[PortStatus, Set[PortStatus]]] = None, desired_number_of_ports: int = 1, timeout: RationalNumber = 0, used_ports_instance: Optional[UsedPorts] = None) -> Optional[slice]:
255    """_summary_
256
257    Example:
258
259    from cengal.io.serve_free_ports import asimple_ports_range_search
260    from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports
261    import asyncio
262
263    async def main():
264        ports_range = await asimple_ports_range_search('0.0.0.0', slice(18000, 18100), Protocol.tcp, {PortStatus.na, PortStatus.no}, 3, 5.0)
265        if ports_range is not None:
266            first_port = ports_range.start  # Output: 18000
267            last_port = ports_range.stop - 1  # Output: 18002
268            first_port, last_port = unify_ports(ports_range)  # Output: (18000, 18002)
269            for port in range(ports_range.start, ports_range.stop):
270                print(port)
271            
272            print(list(range(ports_range.start, ports_range.stop)))  # Output: [18000, 18001, 18002]
273            if isinstance(purify_ports(ports_range), slice):
274                # type is `slice`
275                print('Ports range')
276            else:
277                # type is `int`
278                print('Single port')
279
280    Args:
281        host (str): _description_
282        port_or_range (Union[int, slice, Tuple[int, int]]): _description_
283        protocol (Protocol, optional): _description_. Defaults to Protocol.tcp.
284        statuses (Optional[Union[PortStatus, Set[PortStatus]]], optional): _description_. Defaults to None.
285        desired_number_of_ports (int, optional): _description_. Defaults to 1.
286        timeout (int, optional): _description_. Defaults to 0.
287        used_ports_instance (Optional[UsedPorts], optional): .
288
289    Returns:
290        _type_: _description_
291    """    
292    if statuses is None: statuses = {PortStatus.na, PortStatus.no}
293    if used_ports_instance is None: used_ports_instance = used_ports()
294    return await find_free_ports_range(host, used_ports_instance.range(protocol, port_or_range, statuses, desired_number_of_ports), timeout)
async def find_free_port( host, ports_iterartor: cengal.io.used_ports.versions.v_0.used_ports.PortsIterator, timeout: Union[int, float] = 0) -> Union[int, NoneType]: 
50async def find_free_port(host, ports_iterartor: PortsIterator, timeout: RationalNumber = 0) -> Optional[int]:
51    """_summary_
52    Example:
53
54    from cengal.io.serve_free_ports import find_free_port
55    from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports
56    import asyncio
57
58    async def main():
59        ports_rages_iterator: PortsIterator = used_ports().range(Protocol.tcp, slice(18000, 18100), {PortStatus.na, PortStatus.no})
60        port: int = await find_free_port('0.0.0.0', ports_rages_iterator, 5.0)
61        if port is not None:
62            print(port)  # Output: 18000
63
64    Args:
65        host (_type_): _description_
66        ports_iterartor (PortsIterator): _description_
67        timeout (RationalNumber, optional): _description_. Defaults to 0.
68
69    Returns:
70        Optional[int]: _description_
71    """    
72    for ports_range in ports_iterartor:
73        for port in range(ports_range.start, ports_range.stop):
74            try:
75                if timeout:
76                    server = await asyncio.wait_for(asyncio.start_server(client_connected_cb, host, port), timeout)
77                else:
78                    server = await asyncio.start_server(client_connected_cb, host, port)
79                
80                try:
81                    server.close()
82                except Exception:
83                    pass
84                finally:
85                    del server
86                    collect()
87
88                return port
89            except TimeoutError as err:
90                pass
91            except (ConnectionRefusedError, ConnectionError, OSError) as err:
92                pass

_summary_ Example:

from cengal.io.serve_free_ports import find_free_port from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports import asyncio

async def main(): ports_rages_iterator: PortsIterator = used_ports().range(Protocol.tcp, slice(18000, 18100), {PortStatus.na, PortStatus.no}) port: int = await find_free_port('0.0.0.0', ports_rages_iterator, 5.0) if port is not None: print(port) # Output: 18000

Args: host (_type_): _description_ ports_iterartor (PortsIterator): _description_ timeout (RationalNumber, optional): _description_. Defaults to 0.

Returns: Optional[int]: _description_

async def find_free_ports_range( host, ports_iterartor: cengal.io.used_ports.versions.v_0.used_ports.PortsIterator, timeout: Union[int, float] = 0) -> Union[slice, NoneType]: 
156async def find_free_ports_range(host, ports_iterartor: PortsIterator, timeout: RationalNumber = 0) -> Optional[slice]:
157    """
158    Example:
159
160    from cengal.io.serve_free_ports import find_free_ports_range
161    from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports
162    import asyncio
163
164    async def main():
165        ports_rages_iterator: PortsIterator = used_ports().range(Protocol.tcp, slice(18000, 18100), {PortStatus.na, PortStatus.no}, 3)
166        ports_range = await find_free_ports_range('0.0.0.0', ports_rages_iterator, 5.0)
167        if ports_range is not None:
168            first_port = ports_range.start  # Output: 18000
169            last_port = ports_range.stop - 1  # Output: 18002
170            first_port, last_port = unify_ports(ports_range)  # Output: (18000, 18002)
171            for port in range(ports_range.start, ports_range.stop):
172                print(port)
173            
174            print(list(range(ports_range.start, ports_range.stop)))  # Output: [18000, 18001, 18002]
175            if isinstance(purify_ports(ports_range), slice):
176                # type is `slice`
177                print('Ports range')
178            else:
179                # type is `int`
180                print('Single port')
181
182    Args:
183        host (_type_): _description_
184        ports_iterartor (PortsIterator): _description_
185        timeout (int, optional): _description_. Defaults to 0.
186
187    Returns:
188        Optional[slice]: _description_
189    """
190    for ports_range in ports_iterartor:
191        try:
192            for port in range(ports_range.start, ports_range.stop):
193                if timeout:
194                    server = await asyncio.wait_for(asyncio.start_server(client_connected_cb, host, port), timeout)
195                else:
196                    server = await asyncio.start_server(client_connected_cb, host, port)
197                
198                try:
199                    server.close()
200                except Exception:
201                    pass
202                finally:
203                    del server
204                    collect()
205
206            return ports_range
207        except TimeoutError as err:
208            pass
209        except (ConnectionRefusedError, ConnectionError, OSError) as err:
210            pass

Example:

from cengal.io.serve_free_ports import find_free_ports_range from cengal.io.used_ports import PortsIterator, used_ports, Protocol, PortStatus, unify_ports, purify_ports import asyncio

async def main(): ports_rages_iterator: PortsIterator = used_ports().range(Protocol.tcp, slice(18000, 18100), {PortStatus.na, PortStatus.no}, 3) ports_range = await find_free_ports_range('0.0.0.0', ports_rages_iterator, 5.0) if ports_range is not None: first_port = ports_range.start # Output: 18000 last_port = ports_range.stop - 1 # Output: 18002 first_port, last_port = unify_ports(ports_range) # Output: (18000, 18002) for port in range(ports_range.start, ports_range.stop): print(port)

    print(list(range(ports_range.start, ports_range.stop)))  # Output: [18000, 18001, 18002]
    if isinstance(purify_ports(ports_range), slice):
        # type is `slice`
        print('Ports range')
    else:
        # type is `int`
        print('Single port')

Args: host (_type_): _description_ ports_iterartor (PortsIterator): _description_ timeout (int, optional): _description_. Defaults to 0.

Returns: Optional[slice]: _description_