@@ -2,11 +2,12 @@ __all__ = ["STDOUT", "PIPE", "Wurlitzer", "pipes", "sys_pipes", "sys_pipes_forev
2
2
3
3
import contextlib
4
4
import io
5
+ import logging
5
6
from _typeshed import SupportsWrite
6
7
from contextlib import _GeneratorContextManager
7
8
from types import TracebackType
8
9
from typing import Any , Final , Literal , Protocol , TextIO , TypeVar , overload
9
- from typing_extensions import TypeAlias
10
+ from typing_extensions import Self , TypeAlias
10
11
11
12
STDOUT : Final = 2
12
13
PIPE : Final = 3
@@ -31,8 +32,8 @@ class Wurlitzer:
31
32
32
33
def __init__ (
33
34
self ,
34
- stdout : SupportsWrite [str ] | SupportsWrite [bytes ] | None = None ,
35
- stderr : _STDOUT | SupportsWrite [str ] | SupportsWrite [bytes ] | None = None ,
35
+ stdout : SupportsWrite [str ] | SupportsWrite [bytes ] | logging . Logger | None = None ,
36
+ stderr : _STDOUT | SupportsWrite [str ] | SupportsWrite [bytes ] | logging . Logger | None = None ,
36
37
encoding : str | None = ...,
37
38
bufsize : int | None = ...,
38
39
) -> None : ...
@@ -70,6 +71,30 @@ def pipes(
70
71
stdout : _PIPE , stderr : _StreamErrT , encoding : str = ..., bufsize : int | None = None
71
72
) -> _GeneratorContextManager [tuple [io .StringIO , _StreamErrT ]]: ...
72
73
@overload
74
+ def pipes (
75
+ stdout : _PIPE , stderr : logging .Logger , encoding : str = ..., bufsize : int | None = None
76
+ ) -> _GeneratorContextManager [tuple [io .StringIO , _LogPipe ]]: ...
77
+ @overload
78
+ def pipes (
79
+ stdout : logging .Logger , stderr : _STDOUT , encoding : str | None = ..., bufsize : int | None = None
80
+ ) -> _GeneratorContextManager [tuple [_LogPipe , None ]]: ...
81
+ @overload
82
+ def pipes (
83
+ stdout : logging .Logger , stderr : _PIPE , encoding : None , bufsize : int | None = None
84
+ ) -> _GeneratorContextManager [tuple [_LogPipe , io .BytesIO ]]: ...
85
+ @overload
86
+ def pipes (
87
+ stdout : logging .Logger , stderr : _PIPE , encoding : str = ..., bufsize : int | None = None
88
+ ) -> _GeneratorContextManager [tuple [_LogPipe , io .StringIO ]]: ...
89
+ @overload
90
+ def pipes (
91
+ stdout : logging .Logger , stderr : _StreamErrT , encoding : str | None = ..., bufsize : int | None = None
92
+ ) -> _GeneratorContextManager [tuple [_LogPipe , _StreamErrT ]]: ...
93
+ @overload
94
+ def pipes (
95
+ stdout : logging .Logger , stderr : logging .Logger , encoding : str | None = ..., bufsize : int | None = None
96
+ ) -> _GeneratorContextManager [tuple [_LogPipe , _LogPipe ]]: ...
97
+ @overload
73
98
def pipes (
74
99
stdout : _StreamOutT , stderr : _STDOUT , encoding : str | None = ..., bufsize : int | None = None
75
100
) -> _GeneratorContextManager [tuple [_StreamOutT , None ]]: ...
@@ -87,11 +112,28 @@ def pipes(
87
112
) -> _GeneratorContextManager [tuple [_StreamOutT , _StreamErrT ]]: ...
88
113
@overload
89
114
def pipes (
90
- stdout : _PIPE | _StreamOutT = 3 ,
91
- stderr : _STDOUT | _PIPE | _StreamErrT = 3 ,
115
+ stdout : _StreamOutT , stderr : logging .Logger , encoding : str | None = ..., bufsize : int | None = None
116
+ ) -> _GeneratorContextManager [tuple [_StreamOutT , _LogPipe ]]: ...
117
+ @overload
118
+ def pipes (
119
+ stdout : _PIPE | logging .Logger | _StreamOutT = 3 ,
120
+ stderr : _STDOUT | _PIPE | logging .Logger | _StreamErrT = 3 ,
92
121
encoding : str | None = ...,
93
122
bufsize : int | None = None ,
94
- ) -> _GeneratorContextManager [tuple [io .BytesIO | io .StringIO | _StreamOutT , io .BytesIO | io .StringIO | _StreamErrT | None ]]: ...
123
+ ) -> _GeneratorContextManager [
124
+ tuple [io .BytesIO | io .StringIO | _StreamOutT | _LogPipe , io .BytesIO | io .StringIO | _StreamErrT | _LogPipe | None ]
125
+ ]: ...
126
+
127
+ class _LogPipe (io .BufferedWriter ):
128
+ logger : logging .Logger
129
+ stream_name : str
130
+ level : int
131
+ def __init__ (self , logger : logging .Logger , stream_name : str , level : int = 20 ) -> None : ...
132
+ def write (self , chunk : str ) -> None : ... # type: ignore[override]
133
+ def flush (self ) -> None : ...
134
+ def __enter__ (self ) -> Self : ...
135
+ def __exit__ (self , * exc_info : object ) -> None : ...
136
+
95
137
def sys_pipes_forever (encoding : str = ..., bufsize : int | None = None ) -> None : ...
96
138
def stop_sys_pipes () -> None : ...
97
139
def load_ipython_extension (ip : _InteractiveShell ) -> None : ...
0 commit comments