Convenience function which creates a SOCK_STREAM type socket bound to *address* (a 2-tuple (host, port)) and return the socket object. *family* should be either AF_INET or AF_INET6. *backlog* is the queue size passed to socket.listen(). *reuse_port* dictates whether to use
(address, *, family=AF_INET, backlog=None, reuse_port=False,
dualstack_ipv6=False)
| 874 | |
| 875 | |
| 876 | def create_server(address, *, family=AF_INET, backlog=None, reuse_port=False, |
| 877 | dualstack_ipv6=False): |
| 878 | """Convenience function which creates a SOCK_STREAM type socket |
| 879 | bound to *address* (a 2-tuple (host, port)) and return the socket |
| 880 | object. |
| 881 | |
| 882 | *family* should be either AF_INET or AF_INET6. |
| 883 | *backlog* is the queue size passed to socket.listen(). |
| 884 | *reuse_port* dictates whether to use the SO_REUSEPORT socket option. |
| 885 | *dualstack_ipv6*: if true and the platform supports it, it will |
| 886 | create an AF_INET6 socket able to accept both IPv4 or IPv6 |
| 887 | connections. When false it will explicitly disable this option on |
| 888 | platforms that enable it by default (e.g. Linux). |
| 889 | |
| 890 | >>> with create_server(('', 8000)) as server: |
| 891 | ... while True: |
| 892 | ... conn, addr = server.accept() |
| 893 | ... # handle new connection |
| 894 | """ |
| 895 | if reuse_port and not hasattr(_socket, "SO_REUSEPORT"): |
| 896 | raise ValueError("SO_REUSEPORT not supported on this platform") |
| 897 | if dualstack_ipv6: |
| 898 | if not has_dualstack_ipv6(): |
| 899 | raise ValueError("dualstack_ipv6 not supported on this platform") |
| 900 | if family != AF_INET6: |
| 901 | raise ValueError("dualstack_ipv6 requires AF_INET6 family") |
| 902 | sock = socket(family, SOCK_STREAM) |
| 903 | try: |
| 904 | # Note about Windows. We don't set SO_REUSEADDR because: |
| 905 | # 1) It's unnecessary: bind() will succeed even in case of a |
| 906 | # previous closed socket on the same address and still in |
| 907 | # TIME_WAIT state. |
| 908 | # 2) If set, another socket is free to bind() on the same |
| 909 | # address, effectively preventing this one from accepting |
| 910 | # connections. Also, it may set the process in a state where |
| 911 | # it'll no longer respond to any signals or graceful kills. |
| 912 | # See: https://learn.microsoft.com/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse |
| 913 | if os.name not in ('nt', 'cygwin') and \ |
| 914 | hasattr(_socket, 'SO_REUSEADDR'): |
| 915 | try: |
| 916 | sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) |
| 917 | except error: |
| 918 | # Fail later on bind(), for platforms which may not |
| 919 | # support this option. |
| 920 | pass |
| 921 | if reuse_port: |
| 922 | sock.setsockopt(SOL_SOCKET, SO_REUSEPORT, 1) |
| 923 | if has_ipv6 and family == AF_INET6: |
| 924 | if dualstack_ipv6: |
| 925 | sock.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, 0) |
| 926 | elif hasattr(_socket, "IPV6_V6ONLY") and \ |
| 927 | hasattr(_socket, "IPPROTO_IPV6"): |
| 928 | sock.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, 1) |
| 929 | try: |
| 930 | sock.bind(address) |
| 931 | except error as err: |
| 932 | msg = '%s (while attempting to bind on address %r)' % \ |
| 933 | (err.strerror, address) |
nothing calls this directly
no test coverage detected