"""detach() -> file descriptor
Close the socket object without closing the underlying file descriptor.
The object cannot be used after this call, but the file descriptor
can be reused for other purposes. The file descriptor is returned.
"""Read-only access to the address family for this socket.
return _intenum_converter(super().family, AddressFamily)
"""Read-only access to the socket type.
return _intenum_converter(super().type, SocketKind)
def get_inheritable(self):
return os.get_handle_inheritable(self.fileno())
def set_inheritable(self, inheritable):
os.set_handle_inheritable(self.fileno(), inheritable)
def get_inheritable(self):
return os.get_inheritable(self.fileno())
def set_inheritable(self, inheritable):
os.set_inheritable(self.fileno(), inheritable)
get_inheritable.__doc__ = "Get the inheritable flag of the socket"
set_inheritable.__doc__ = "Set the inheritable flag of the socket"
def fromfd(fd, family, type, proto=0):
""" fromfd(fd, family, type[, proto]) -> socket object
Create a socket object from a duplicate of the given file
descriptor. The remaining arguments are the same as for socket().
return socket(family, type, proto, nfd)
if hasattr(_socket.socket, "share"):
""" fromshare(info) -> socket object
Create a socket object from the bytes object returned by
return socket(0, 0, 0, info)
__all__.append("fromshare")
if hasattr(_socket, "socketpair"):
def socketpair(family=None, type=SOCK_STREAM, proto=0):
"""socketpair([family[, type[, proto]]]) -> (socket object, socket object)
Create a pair of socket objects from the sockets returned by the platform
The arguments are the same as for socket() except the default family is
AF_UNIX if defined on the platform; otherwise, the default is AF_INET.
a, b = _socket.socketpair(family, type, proto)
a = socket(family, type, proto, a.detach())
b = socket(family, type, proto, b.detach())
# Origin: https://gist.github.com/4325783, by Geert Jansen. Public domain.
def socketpair(family=AF_INET, type=SOCK_STREAM, proto=0):
raise ValueError("Only AF_INET and AF_INET6 socket address families "
raise ValueError("Only SOCK_STREAM socket type is supported")
raise ValueError("Only protocol zero is supported")
# We create a connected TCP socket. Note the trick with
# setblocking(False) that prevents us from having to create a thread.
lsock = socket(family, type, proto)
# On IPv6, ignore flow_info and scope_id
addr, port = lsock.getsockname()[:2]
csock = socket(family, type, proto)
csock.connect((addr, port))
except (BlockingIOError, InterruptedError):
ssock, _ = lsock.accept()
__all__.append("socketpair")
socketpair.__doc__ = """socketpair([family[, type[, proto]]]) -> (socket object, socket object)
Create a pair of socket objects from the sockets returned by the platform
The arguments are the same as for socket() except the default family is AF_UNIX
if defined on the platform; otherwise, the default is AF_INET.
_blocking_errnos = { EAGAIN, EWOULDBLOCK }
class SocketIO(io.RawIOBase):
"""Raw I/O implementation for stream sockets.
This class supports the makefile() method on sockets. It provides
the raw I/O interface on top of a socket object.
# One might wonder why not let FileIO do the job instead. There are two
# main reasons why FileIO is not adapted:
# - it wouldn't work under Windows (where you can't used read() and
# write() on a socket handle)
# - it wouldn't work with socket timeouts (FileIO would ignore the
# timeout and consider the socket non-blocking)
def __init__(self, sock, mode):
if mode not in ("r", "w", "rw", "rb", "wb", "rwb"):
raise ValueError("invalid mode: %r" % mode)
io.RawIOBase.__init__(self)
self._reading = "r" in mode
self._writing = "w" in mode
self._timeout_occurred = False
"""Read up to len(b) bytes into the writable buffer *b* and return
the number of bytes read. If the socket is non-blocking and no bytes
are available, None is returned.
If *b* is non-empty, a 0 return value indicates that the connection
was shutdown at the other end.
if self._timeout_occurred:
raise OSError("cannot read from timed out object")
return self._sock.recv_into(b)
self._timeout_occurred = True
if e.args[0] in _blocking_errnos:
"""Write the given bytes or bytearray object *b* to the socket
and return the number of bytes written. This can be less than
len(b) if not all data could be written. If the socket is
non-blocking and no bytes could be written None is returned.
return self._sock.send(b)
if e.args[0] in _blocking_errnos:
"""True if the SocketIO is open for reading.
raise ValueError("I/O operation on closed socket.")
"""True if the SocketIO is open for writing.
raise ValueError("I/O operation on closed socket.")
"""True if the SocketIO is open for seeking.
raise ValueError("I/O operation on closed socket.")
return super().seekable()
"""Return the file descriptor of the underlying socket.
return self._sock.fileno()
"""Close the SocketIO object. This doesn't close the underlying
socket, except if all references to it have disappeared.
self._sock._decref_socketios()
"""Get fully qualified domain name from name.
An empty argument is interpreted as meaning the local host.
First the hostname returned by gethostbyaddr() is checked, then
possibly existing aliases. In case no FQDN is available, hostname
from gethostname() is returned.
if not name or name == '0.0.0.0':
hostname, aliases, ipaddrs = gethostbyaddr(name)
aliases.insert(0, hostname)
_GLOBAL_DEFAULT_TIMEOUT = object()
def create_connection(address, timeout=_GLOBAL_DEFAULT_TIMEOUT,
"""Connect to *address* and return the socket object.
Convenience function. Connect to *address* (a 2-tuple ``(host,
port)``) and return the socket object. Passing the optional
*timeout* parameter will set the timeout on the socket instance
before attempting to connect. If no *timeout* is supplied, the
global default timeout setting returned by :func:`getdefaulttimeout`
is used. If *source_address* is set it must be a tuple of (host, port)
for the socket to bind as a source address before making the connection.
A host of '' or port 0 tells the OS to use the default.
for res in getaddrinfo(host, port, 0, SOCK_STREAM):
af, socktype, proto, canonname, sa = res
sock = socket(af, socktype, proto)
if timeout is not _GLOBAL_DEFAULT_TIMEOUT:
sock.bind(source_address)
# Break explicitly a reference cycle
# Break explicitly a reference cycle
raise error("getaddrinfo returns an empty list")
def has_dualstack_ipv6():
"""Return True if the platform supports creating a SOCK_STREAM socket
which can handle both AF_INET and AF_INET6 (IPv4 / IPv6) connections.
or not hasattr(_socket, 'IPPROTO_IPV6') \
or not hasattr(_socket, 'IPV6_V6ONLY'):
with socket(AF_INET6, SOCK_STREAM) as sock:
sock.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, 0)
def create_server(address, *, family=AF_INET, backlog=None, reuse_port=False,
"""Convenience function which creates a SOCK_STREAM type socket
bound to *address* (a 2-tuple (host, port)) and return the socket
*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 SO_REUSEPORT socket option.
*dualstack_ipv6*: if true and the platform supports it, it will
create an AF_INET6 socket able to accept both IPv4 or IPv6
connections. When false it will explicitly disable this option on
platforms that enable it by default (e.g. Linux).
>>> with create_server(('', 8000)) as server:
... conn, addr = server.accept()
... # handle new connection
if reuse_port and not hasattr(_socket, "SO_REUSEPORT"):
raise ValueError("SO_REUSEPORT not supported on this platform")
if not has_dualstack_ipv6():
raise ValueError("dualstack_ipv6 not supported on this platform")
raise ValueError("dualstack_ipv6 requires AF_INET6 family")
sock = socket(family, SOCK_STREAM)
# Note about Windows. We don't set SO_REUSEADDR because:
# 1) It's unnecessary: bind() will succeed even in case of a
# previous closed socket on the same address and still in
# 2) If set, another socket is free to bind() on the same
# address, effectively preventing this one from accepting
# connections. Also, it may set the process in a state where
# it'll no longer respond to any signals or graceful kills.
# See: msdn2.microsoft.com/en-us/library/ms740621(VS.85).aspx
if os.name not in ('nt', 'cygwin') and \
hasattr(_socket, 'SO_REUSEADDR'):
sock.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
# Fail later on bind(), for platforms which may not
sock.setsockopt(SOL_SOCKET, SO_REUSEPORT, 1)
if has_ipv6 and family == AF_INET6:
sock.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, 0)
elif hasattr(_socket, "IPV6_V6ONLY") and \
hasattr(_socket, "IPPROTO_IPV6"):
sock.setsockopt(IPPROTO_IPV6, IPV6_V6ONLY, 1)
msg = '%s (while attempting to bind on address %r)' % \
raise error(err.errno, msg) from None
def getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
"""Resolve host and port into list of address info entries.
Translate the host/port argument into a sequence of 5-tuples that contain
all the necessary arguments for creating a socket connected to that service.
host is a domain name, a string representation of an IPv4/v6 address or
None. port is a string service name such as 'http', a numeric port number or
None. By passing None as the value of host and port, you can pass NULL to
The family, type and proto arguments can be optionally specified in order to
narrow the list of addresses returned. Passing zero as a value for each of
these arguments selects the full range of results.
# We override this function since we want to translate the numeric family
# and socket type values to enum constants.
for res in _socket.getaddrinfo(host, port, family, type, proto, flags):
af, socktype, proto, canonname, sa = res
addrlist.append((_intenum_converter(af, AddressFamily),
_intenum_converter(socktype, SocketKind),
It is recommended that you Edit text format, this type of Fix handles quite a lot in one request
