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 the S
(address, *, family=AF_INET, backlog=None, reuse_port=False,
dualstack_ipv6=False)
| 902 | |
| 903 | |
| 904 | def create_server(address, *, family=AF_INET, backlog=None, reuse_port=False, |
| 905 | dualstack_ipv6=False): |
| 906 | """Convenience function which creates a SOCK_STREAM type socket |
| 907 | bound to *address* (a 2-tuple (host, port)) and return the socket |
| 908 | object. |
| 909 | |
| 910 | *family* should be either AF_INET or AF_INET6. |
| 911 | *backlog* is the queue size passed to socket.listen(). |
| 912 | *reuse_port* dictates whether to use the SO_REUSEPORT socket option. |
| 913 | *dualstack_ipv6*: if true and the platform supports it, it will |
| 914 | create an AF_INET6 socket able to accept both IPv4 or IPv6 |
| 915 | connections. When false it will explicitly disable this option on |
| 916 | platforms that enable it by default (e.g. Linux). |
| 917 | |
| 918 | >>> with create_server(('', 8000)) as server: |
| 919 | ... while True: |
| 920 | ... conn, addr = server.accept() |
| 921 | ... # handle new connection |
| 922 | """ |
| 923 | if reuse_port and not hasattr(_socket, "SO_REUSEPORT"): |
| 924 | raise ValueError("SO_REUSEPORT not supported on this platform") |
| 925 | if dualstack_ipv6: |
| 926 | if not has_dualstack_ipv6(): |
| 927 | raise ValueError("dualstack_ipv6 not supported on this platform") |
| 928 | if family != AF_INET6: |
| 929 | raise ValueError("dualstack_ipv6 requires AF_INET6 family") |
| 930 | sock = socket(family, SOCK_STREAM) |
| 931 | try: |
| 932 | # Note about Windows. We don't set SO_REUSEADDR because: |
| 933 | # 1) It's unnecessary: bind() will succeed even in case of a |
| 934 | # previous closed socket on the same address and still in |
| 935 | # TIME_WAIT state. |
| 936 | # 2) If set, another socket is free to bind() on the same |
| 937 | # address, effectively preventing this one from accepting |
| 938 | # connections. Also, it may set the process in a state where |
| 939 | # it'll no longer respond to any signals or graceful kills. |
| 940 | # See: https://learn.microsoft.com/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse |
| 941 | if os.name not in ('nt', 'cygwin') and \ |
| 942 | hasattr(_socket, 'SO_REUSEADDR'): |
| 943 | try: |
| 944 | sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) |
| 945 | except error: |
| 946 | # Fail later on bind(), for platforms which may not |
| 947 | # support this option. |
| 948 | pass |
| 949 | # Since Linux 6.12.9, SO_REUSEPORT is not allowed |
| 950 | # on other address families than AF_INET/AF_INET6. |
| 951 | if reuse_port and family in (AF_INET, AF_INET6): |
| 952 | sock.setsockopt(SOL_SOCKET, SO_REUSEPORT, 1) |
| 953 | if has_ipv6 and family == AF_INET6: |
| 954 | if dualstack_ipv6: |
| 955 | sock.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, 0) |
| 956 | elif hasattr(_socket, "IPV6_V6ONLY") and \ |
| 957 | hasattr(_socket, "IPPROTO_IPV6"): |
| 958 | sock.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, 1) |
| 959 | try: |
| 960 | sock.bind(address) |
| 961 | except error as err: |
nothing calls this directly
no test coverage detected
searching dependent graphs…