codekingpro/portable-devtools
115k
1from typing import Union, Any
2from array import array
3from .protocol import HTTPProtocol
4
5class HttpParser:
6 def __init__(self, protocol: Union[HTTPProtocol, Any]) -> None:
7 """
8 protocol -- a Python object with the following methods
9 (all optional):
10
11 - on_message_begin()
12 - on_url(url: bytes)
13 - on_header(name: bytes, value: bytes)
14 - on_headers_complete()
15 - on_body(body: bytes)
16 - on_message_complete()
17 - on_chunk_header()
18 - on_chunk_complete()
19 - on_status(status: bytes)
20 """
21
22 def get_http_version(self) -> str:
23 """Return an HTTP protocol version."""
24 ...
25
26 def should_keep_alive(self) -> bool:
27 """Return ``True`` if keep-alive mode is preferred."""
28 ...
29
30 def should_upgrade(self) -> bool:
31 """Return ``True`` if the parsed request is a valid Upgrade request.
32 The method exposes a flag set just before on_headers_complete.
33 Calling this method earlier will only yield `False`."""
34 ...
35
36 def feed_data(self, data: Union[bytes, bytearray, memoryview, array]) -> None:
37 """Feed data to the parser.
38
39 Will eventually trigger callbacks on the ``protocol``
40 object.
41
42 On HTTP upgrade, this method will raise an
43 ``HttpParserUpgrade`` exception, with its sole argument
44 set to the offset of the non-HTTP data in ``data``.
45 """
46
47class HttpRequestParser(HttpParser):
48 """Used for parsing http requests from the server's side"""
49
50 def get_method(self) -> bytes:
51 """Return HTTP request method (GET, HEAD, etc)"""
52
53class HttpResponseParser(HttpParser):
54 """Used for parsing http requests from the client's side"""
55
56 def get_status_code(self) -> int:
57 """Return the status code of the HTTP response"""
58 