Public Git Hosting - python-iview.git/blob (original) (raw)

1 import zlib

2 from io import BufferedIOBase, BytesIO, RawIOBase

3 from urllib.parse import quote_plus

4 from io import SEEK_CUR, SEEK_END

5 import urllib.request

6 import http.client

7 from errno import EPIPE, ESHUTDOWN, ENOTCONN, ECONNRESET

8 import selectors

9 from socketserver import StreamRequestHandler, BaseServer

10 import sys

12 try: # Python 3.3

13 ConnectionError

14 except NameError: # Python < 3.3

15 ConnectionError = ()

17 DISCONNECTION_ERRNOS = {EPIPE, ESHUTDOWN, ENOTCONN, ECONNRESET}

19 def xml_text_elements(parent, namespace=""):

20 """Extracts text from Element Tree into a dict()

22 Each key is the tag name of a child of the given parent element, and

23 the value is the text of that child. Only tags with no attributes are

24 included. If the "namespace" parameter is given, it should specify an

25 XML namespace enclosed in curly brackets {. . .}, and only tags in

26 that namespace are included."""

28 d = dict()

29 for child in parent:

30 if child.tag.startswith(namespace) and not child.keys():

31 tag = child.tag[len(namespace):]

32 d[tag] = child.text or ""

33 return d

35 def read_int(stream, size):

36 bytes = read_strict(stream, size)

37 return int.from_bytes(bytes, "big")

39 def read_string(stream):

40 buf = bytearray()

41 while True:

42 b = read_strict(stream, 1)

43 if not ord(b):

44 return buf

45 buf.extend(b)

47 def read_strict(stream, size):

48 data = stream.read(size)

49 if len(data) != size:

50 raise EOFError()

51 return data

53 value_unsafe = '%+&;#'

54 VALUE_SAFE = ''.join(chr(c) for c in range(33, 127)

55 if chr(c) not in value_unsafe)

56 def urlencode_param(value):

57 """Minimal URL encoding for query parameter"""

58 return quote_plus(value, safe=VALUE_SAFE)

60 class CounterWriter(BufferedIOBase):

61 def __init__(self, output):

62 self.length = 0

63 self.output = output

64 def write(self, b):

65 self.length += len(b)

66 return self.output.write(b)

67 def tell(self):

68 return self.length

70 class ZlibDecompressorWriter(BufferedIOBase):

71 def __init__(self, output, *pos, buffer_size=0x10000, **kw):

72 self.output = output

73 self.buffer_size = buffer_size

74 self.decompressor = zlib.decompressobj(*pos, **kw)

75 def write(self, b):

76 while b:

77 data = self.decompressor.decompress(b, self.buffer_size)

78 self.output.write(data)

79 b = self.decompressor.unconsumed_tail

80 def close(self):

81 self.decompressor.flush()

83 class TeeWriter(BufferedIOBase):

84 def __init__(self, *outputs):

85 self.outputs = outputs

86 def write(self, b):

87 for output in self.outputs:

88 output.write(b)

90 def streamcopy(input, output, length):

91 assert length >= 0

92 while length:

93 chunk = read_strict(input, min(length, 0x10000))

94 output.write(chunk)

95 length -= len(chunk)

97 def fastforward(stream, offset):

98 assert offset >= 0

99 if stream.seekable():

100 pos = stream.seek(offset, SEEK_CUR)

101 if pos > stream.seek(0, SEEK_END):

102 raise EOFError()

103 stream.seek(pos)

105 while offset:

106 chunk = read_strict(stream, min(offset, 0x10000))

107 offset -= len(chunk)

109 class WritingReader(BufferedIOBase):

110 """Filter for a reader stream that writes the data read to another stream

112 def __init__(self, reader, writer):

113 self.reader = reader

114 self.writer = writer

115 def read(self, n):

116 data = self.reader.read(n)

117 self.writer.write(data)

118 return data

120 class GeneratorReader(RawIOBase):

121 def __init__(self, gen, initial=b""):

122 self._gen = gen

123 self._pending = initial

125 def read(self, size):

126 if not self._pending:

127 self._pending = next(self._gen, b"")

128 result = self._pending[:size]

129 self._pending = self._pending[size:]

130 return result

132 def close(self):

133 self._gen.close()

135 def setitem(dict, key):

136 """Decorator that adds the definition to a dictionary with a given key"""

137 def decorator(func):

138 dict[key] = func

139 return func

140 return decorator

142 class PersistentConnectionHandler(urllib.request.BaseHandler):

143 """URL handler for HTTP persistent connections

145 connection = PersistentConnectionHandler()

146 session = urllib.request.build_opener(connection)

148 # First request opens connection

149 with session.open("http://localhost/one") as response:

150 response.read()

152 # Subsequent requests reuse the existing connection, unless it got closed

153 with session.open("http://localhost/two") as response:

154 response.read()

156 # Closes old connection when new host specified

157 with session.open("http://example/three") as response:

158 response.read()

160 connection.close() # Frees socket

162 Currently does not reuse an existing connection if

163 two host names happen to resolve to the same Internet address.

166 def __init__(self, *pos, **kw):

167 self._type = None

168 self._host = None

169 self._pos = pos

170 self._kw = kw

171 self._connection = None

173 def default_open(self, req):

174 if req.type != "http":

175 return None

177 if req.type != self._type or req.host != self._host:

178 if self._connection:

179 self._connection.close()

180 self._connection = http.client.HTTPConnection(req.host,

181 *self._pos, **self._kw)

182 self._type = req.type

183 self._host = req.host

185 headers = dict(req.header_items())

186 self._attempt_request(req, headers)

189 response = self._connection.getresponse()

190 except EnvironmentError as err: # Python < 3.3 compatibility

191 if err.errno not in DISCONNECTION_ERRNOS:

193 raise http.client.BadStatusLine(err) from err

194 except (ConnectionError, http.client.BadStatusLine):

195 idempotents = {

196 "GET", "HEAD", "PUT", "DELETE", "TRACE", "OPTIONS"}

197 if req.get_method() not in idempotents:

199 # Retry requests whose method indicates they are idempotent

200 self._connection.close()

201 response = None

203 if response.status == http.client.REQUEST_TIMEOUT:

204 # Server indicated it did not handle request

205 response = None

206 if not response:

207 # Retry request

208 self._attempt_request(req, headers)

209 response = self._connection.getresponse()

211 # Odd impedance mismatch between "http.client" and "urllib.request"

212 response.msg = response.reason

213 return response

215 def _attempt_request(self, req, headers):

216 """Send HTTP request, ignoring broken pipe and similar errors"""

218 self._connection.request(req.get_method(), req.selector,

219 req.data, headers)

220 except (ConnectionRefusedError, ConnectionAbortedError):

221 raise # Assume connection was not established

222 except ConnectionError:

223 pass # Continue and read server response if available

224 except EnvironmentError as err: # Python < 3.3 compatibility

225 if err.errno not in DISCONNECTION_ERRNOS:

228 def close(self):

229 if self._connection:

230 self._connection.close()

232 def __enter__(self):

233 return self

234 def __exit__(self, *exc):

235 self.close()

237 def http_get(session, url, types=None, *, headers=dict(), **kw):

238 headers = dict(headers)

239 if types is not None:

240 headers["Accept"] = ", ".join(types)

241 req = urllib.request.Request(url, headers=headers, **kw)

242 response = session.open(req)

244 headers = response.info()

245 headers.set_default_type(None)

246 type = headers.get_content_type()

247 if types is not None and type not in types:

248 msg = "Unexpected content type {}"

249 raise TypeError(msg.format(type))

250 return response

252 response.close()

255 def format_addr(address):

256 [address, port] = address

257 if not frozenset("[]:").isdisjoint(address):

258 address = "[{}]".format(address)

259 if port is not None:

260 address = "{}:{}".format(address, port)

261 return address

263 def header_list(message, header):

264 for header in message.get_all(header, ()):

265 yield from header_split(header, ",")

267 def header_split(header, delim):

268 while header:

269 [elem, header] = header_partition(header, delim)

271 yield elem

273 def header_partition(header, sep):

274 sentinelled = header + sep + '"\\'

276 while True: # For each quoted segment

277 end = sentinelled.index(sep, pos)

278 quote = sentinelled.index('"', pos)

279 if end < quote:

281 pos = quote + 1

282 while True: # For each backslash escape in quote

283 quote = sentinelled.index('"', pos)

284 backslash = sentinelled.index("\\", pos)

285 if quote < backslash:

287 pos = min(backslash + 2, len(header))

288 pos = min(quote + 1, len(header))

290 return (header[:end].strip(), header[end + 1:].strip())

292 class RollbackReader(BufferedIOBase):

293 def __init__(self, wrapped):

294 self.wrapped = wrapped

295 self.readbuffer = BytesIO()

297 def fileno(self, *pos, **kw):

298 return self.wrapped.fileno(*pos, **kw)

300 def start_capture(self):

301 self.writebuffer = BytesIO()

302 def drop_capture(self):

303 self.writebuffer = None

304 def roll_back(self):

305 self.readbuffer = self.writebuffer

306 self.readbuffer.seek(0)

307 self.writebuffer = None

309 def read(self, size=None):

310 data = self.readbuffer.read(size)

311 if size is not None and size >= 0:

312 size -= len(data)

313 data += self.wrapped.read(size)

314 if self.writebuffer:

315 self.writebuffer.write(data)

316 return data

318 class SelectableServer(BaseServer):

319 def handle_error(self, *pos, **kw):

320 '''Inhibit reporting of remote connection dropouts'''

321 [_, exc, *_] = sys.exc_info()

322 if not isinstance(exc, ConnectionError) and (

323 not isinstance(exc, EnvironmentError) or

324 exc.errno not in DISCONNECTION_ERRNOS):

325 super().handle_error(*pos, **kw)

327 def __init__(self, *pos, **kw):

328 super().__init__(*pos, **kw)

329 self.selector = None

330 self.selected = False

331 self.handlers = set()

333 def register(self, selector):

334 self.selector = selector

335 self.selector.register(self.fileno(), selectors.EVENT_READ,

336 self.handle_select)

338 def handle_select(self):

339 self.selected = True

340 self.handle_request()

341 self.selected = False

343 def process_request(self, *pos, **kw):

344 if not self.selected:

345 return super().process_request(*pos, **kw)

346 self.finish_request(*pos, **kw)

348 def server_close(self):

349 while self.handlers:

350 next(iter(self.handlers)).close()

351 if self.selector:

352 self.selector.unregister(self.fileno())

353 return super().server_close()

355 class SelectableHandler(StreamRequestHandler):

356 def handle(self):

357 if not self.server.selected:

358 return super().handle()

359 self.server.selector.register(self.rfile, selectors.EVENT_READ,

360 self.handle_select)

361 self.server.handlers.add(self)

363 def handle_select(self):

364 self.close_connection = True

366 self.handle_one_request()

368 if self.close_connection:

369 self.close()

371 def close(self):

372 self.server.handlers.remove(self)

373 self.server.selector.unregister(self.rfile)

374 self.finish()

375 self.server.shutdown_request(self.request)

377 def finish(self):

378 if not self.server.selected:

379 return super().finish()

381 def select_callbacks(selector):

382 '''Invokes the selector and then the relevant callbacks'''

383 ready = selector.select()

384 for [ready, _] in ready:

385 if ready.fileobj not in selector.get_map():

386 continue # File unregistered since select() returned

388 ready.data()

389 except ConnectionError:

391 except Exception:

392 sys.excepthook(*sys.exc_info())