Wrap a given `file`-like object's `read()` or `write()` to report lengths to the given `callback`
(self, callback, stream, method="read")
| 225 | |
| 226 | class CallbackIOWrapper(ObjectWrapper): |
| 227 | def __init__(self, callback, stream, method="read"): |
| 228 | """ |
| 229 | Wrap a given `file`-like object's `read()` or `write()` to report |
| 230 | lengths to the given `callback` |
| 231 | """ |
| 232 | super().__init__(stream) |
| 233 | func = getattr(stream, method) |
| 234 | if method == "write": |
| 235 | @wraps(func) |
| 236 | def write(data, *args, **kwargs): |
| 237 | res = func(data, *args, **kwargs) |
| 238 | callback(len(data)) |
| 239 | return res |
| 240 | self.wrapper_setattr('write', write) |
| 241 | elif method == "read": |
| 242 | @wraps(func) |
| 243 | def read(*args, **kwargs): |
| 244 | data = func(*args, **kwargs) |
| 245 | callback(len(data)) |
| 246 | return data |
| 247 | self.wrapper_setattr('read', read) |
| 248 | else: |
| 249 | raise KeyError("Can only wrap read/write methods") |
| 250 | |
| 251 | |
| 252 | def _is_utf(encoding): |
nothing calls this directly
no test coverage detected