MCPcopy Index your code
hub / github.com/python/cpython / __init__

Method __init__

Lib/asyncio/windows_utils.py:137–180  ·  view source on GitHub ↗
(self, args, stdin=None, stdout=None, stderr=None, **kwds)

Source from the content-addressed store, hash-verified

135 The stdin, stdout, stderr are None or instances of PipeHandle.
136 """
137 def __init__(self, args, stdin=None, stdout=None, stderr=None, **kwds):
138 assert not kwds.get('universal_newlines')
139 assert kwds.get('bufsize', 0) == 0
140 stdin_rfd = stdout_wfd = stderr_wfd = None
141 stdin_wh = stdout_rh = stderr_rh = None
142 if stdin == PIPE:
143 stdin_rh, stdin_wh = pipe(overlapped=(False, True), duplex=True)
144 stdin_rfd = msvcrt.open_osfhandle(stdin_rh, os.O_RDONLY)
145 else:
146 stdin_rfd = stdin
147 if stdout == PIPE:
148 stdout_rh, stdout_wh = pipe(overlapped=(True, False))
149 stdout_wfd = msvcrt.open_osfhandle(stdout_wh, 0)
150 else:
151 stdout_wfd = stdout
152 if stderr == PIPE:
153 stderr_rh, stderr_wh = pipe(overlapped=(True, False))
154 stderr_wfd = msvcrt.open_osfhandle(stderr_wh, 0)
155 elif stderr == STDOUT:
156 stderr_wfd = stdout_wfd
157 else:
158 stderr_wfd = stderr
159 try:
160 super().__init__(args, stdin=stdin_rfd, stdout=stdout_wfd,
161 stderr=stderr_wfd, **kwds)
162 except:
163 for h in (stdin_wh, stdout_rh, stderr_rh):
164 if h is not None:
165 _winapi.CloseHandle(h)
166 raise
167 else:
168 if stdin_wh is not None:
169 self.stdin = PipeHandle(stdin_wh)
170 if stdout_rh is not None:
171 self.stdout = PipeHandle(stdout_rh)
172 if stderr_rh is not None:
173 self.stderr = PipeHandle(stderr_rh)
174 finally:
175 if stdin == PIPE:
176 os.close(stdin_rfd)
177 if stdout == PIPE:
178 os.close(stdout_wfd)
179 if stderr == PIPE:
180 os.close(stderr_wfd)

Callers

nothing calls this directly

Calls 5

pipeFunction · 0.85
superClass · 0.85
PipeHandleClass · 0.85
getMethod · 0.45
closeMethod · 0.45

Tested by

no test coverage detected