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