FM / asyncio

/snap/core24/1643/usr/lib/python3.12/asyncio/

/snap/core24/1643/usr/lib/python3.12/asyncio

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
base_events.py	77971	0644	edit dl rm
base_futures.py	1974	0644	edit dl rm
base_subprocess.py	8869	0644	edit dl rm
base_tasks.py	2672	0644	edit dl rm
constants.py	1413	0644	edit dl rm
coroutines.py	3342	0644	edit dl rm
events.py	29339	0644	edit dl rm
exceptions.py	1752	0644	edit dl rm
format_helpers.py	2404	0644	edit dl rm
futures.py	14210	0644	edit dl rm
locks.py	18994	0644	edit dl rm
log.py	124	0644	edit dl rm
mixins.py	481	0644	edit dl rm
proactor_events.py	33385	0644	edit dl rm
protocols.py	6957	0644	edit dl rm
queues.py	7974	0644	edit dl rm
runners.py	7159	0644	edit dl rm
selector_events.py	48282	0644	edit dl rm
sslproto.py	31739	0644	edit dl rm
staggered.py	5992	0644	edit dl rm
streams.py	27619	0644	edit dl rm
subprocess.py	7737	0644	edit dl rm
taskgroups.py	8747	0644	edit dl rm
tasks.py	37362	0644	edit dl rm
threads.py	790	0644	edit dl rm
timeouts.py	5321	0644	edit dl rm
transports.py	10722	0644	edit dl rm
trsock.py	2475	0644	edit dl rm
unix_events.py	53124	0644	edit dl rm
windows_events.py	32587	0644	edit dl rm
windows_utils.py	5060	0644	edit dl rm
__init__.py	1220	0644	edit dl rm
__main__.py	3343	0644	edit dl rm

Edit: /snap/core24/1643/usr/lib/python3.12/asyncio/trsock.py (2475B)

import socket

class TransportSocket:

"""A socket-like wrapper for exposing real transport sockets.

These objects can be safely returned by APIs like
    `transport.get_extra_info('socket')`.  All potentially disruptive
    operations (like "socket.close()") are banned.
    """

__slots__ = ('_sock',)

def __init__(self, sock: socket.socket):
        self._sock = sock

@property
    def family(self):
        return self._sock.family

@property
    def type(self):
        return self._sock.type

@property
    def proto(self):
        return self._sock.proto

def __repr__(self):
        s = (
            f"<asyncio.TransportSocket fd={self.fileno()}, "
            f"family={self.family!s}, type={self.type!s}, "
            f"proto={self.proto}"
        )

if self.fileno() != -1:
            try:
                laddr = self.getsockname()
                if laddr:
                    s = f"{s}, laddr={laddr}"
            except socket.error:
                pass
            try:
                raddr = self.getpeername()
                if raddr:
                    s = f"{s}, raddr={raddr}"
            except socket.error:
                pass

return f"{s}>"

def __getstate__(self):
        raise TypeError("Cannot serialize asyncio.TransportSocket object")

def fileno(self):
        return self._sock.fileno()

def dup(self):
        return self._sock.dup()

def get_inheritable(self):
        return self._sock.get_inheritable()

def shutdown(self, how):
        # asyncio doesn't currently provide a high-level transport API
        # to shutdown the connection.
        self._sock.shutdown(how)

def getsockopt(self, *args, **kwargs):
        return self._sock.getsockopt(*args, **kwargs)

def setsockopt(self, *args, **kwargs):
        self._sock.setsockopt(*args, **kwargs)

def getpeername(self):
        return self._sock.getpeername()

def getsockname(self):
        return self._sock.getsockname()

def getsockbyname(self):
        return self._sock.getsockbyname()

def settimeout(self, value):
        if value == 0:
            return
        raise ValueError(
            'settimeout(): only 0 timeout is allowed on transport sockets')

def gettimeout(self):
        return 0

def setblocking(self, flag):
        if not flag:
            return
        raise ValueError(
            'setblocking(): transport sockets cannot be blocking')

import socket class TransportSocket: """A socket-like wrapper for exposing real transport sockets. These objects can be safely returned by APIs like `transport.get_extra_info('socket')`. All potentially disruptive operations (like "socket.close()") are banned. """ __slots__ = ('_sock',) def __init__(self, sock: socket.socket): self._sock = sock @property def family(self): return self._sock.family @property def type(self): return self._sock.type @property def proto(self): return self._sock.proto def __repr__(self): s = ( f"" def __getstate__(self): raise TypeError("Cannot serialize asyncio.TransportSocket object") def fileno(self): return self._sock.fileno() def dup(self): return self._sock.dup() def get_inheritable(self): return self._sock.get_inheritable() def shutdown(self, how): # asyncio doesn't currently provide a high-level transport API # to shutdown the connection. self._sock.shutdown(how) def getsockopt(self, *args, **kwargs): return self._sock.getsockopt(*args, **kwargs) def setsockopt(self, *args, **kwargs): self._sock.setsockopt(*args, **kwargs) def getpeername(self): return self._sock.getpeername() def getsockname(self): return self._sock.getsockname() def getsockbyname(self): return self._sock.getsockbyname() def settimeout(self, value): if value == 0: return raise ValueError( 'settimeout(): only 0 timeout is allowed on transport sockets') def gettimeout(self): return 0 def setblocking(self, flag): if not flag: return raise ValueError( 'setblocking(): transport sockets cannot be blocking')