Server Class — requests Architecture
Architecture documentation for the Server class in server.py from the requests codebase.
Entity Profile
Dependency Diagram
graph TD b9ecf8db_ec54_4dfb_626c_b3769674f676["Server"] 1299f040_d82e_36bb_76ee_892d203f2f6a["server.py"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|defined in| 1299f040_d82e_36bb_76ee_892d203f2f6a 06351a62_57f4_3faf_2007_0444d8671498["__init__()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| 06351a62_57f4_3faf_2007_0444d8671498 507b0f59_b4bd_1c2b_f879_58d253933ec3["text_response_server()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| 507b0f59_b4bd_1c2b_f879_58d253933ec3 77b5091f_26f5_443a_fd8b_425cdd5ca984["basic_response_server()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| 77b5091f_26f5_443a_fd8b_425cdd5ca984 fab2c5f1_8672_849d_f0c5_e11fb76db997["run()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| fab2c5f1_8672_849d_f0c5_e11fb76db997 eee1803d_ff9d_a5bb_fc93_93a47f6b787c["_create_socket_and_bind()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| eee1803d_ff9d_a5bb_fc93_93a47f6b787c e3f7deb6_7bf5_840b_f66f_68302df65e04["_close_server_sock_ignore_errors()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| e3f7deb6_7bf5_840b_f66f_68302df65e04 30970975_142a_647d_c257_2b44a920d005["_handle_requests()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| 30970975_142a_647d_c257_2b44a920d005 597ead60_fb07_4fe0_6952_0c013fcb435d["_accept_connection()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| 597ead60_fb07_4fe0_6952_0c013fcb435d c91007f7_3064_604b_1b08_a6a1fbee84b7["__enter__()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| c91007f7_3064_604b_1b08_a6a1fbee84b7 8c1b292f_223c_8e02_9ce7_34e7a5581360["__exit__()"] b9ecf8db_ec54_4dfb_626c_b3769674f676 -->|method| 8c1b292f_223c_8e02_9ce7_34e7a5581360
Relationship Graph
Source Code
tests/testserver/server.py lines 25–135
class Server(threading.Thread):
"""Dummy server using for unit testing"""
WAIT_EVENT_TIMEOUT = 5
def __init__(
self,
handler=None,
host="localhost",
port=0,
requests_to_handle=1,
wait_to_close_event=None,
):
super().__init__()
self.handler = handler or consume_socket_content
self.handler_results = []
self.host = host
self.port = port
self.requests_to_handle = requests_to_handle
self.wait_to_close_event = wait_to_close_event
self.ready_event = threading.Event()
self.stop_event = threading.Event()
@classmethod
def text_response_server(cls, text, request_timeout=0.5, **kwargs):
def text_response_handler(sock):
request_content = consume_socket_content(sock, timeout=request_timeout)
sock.send(text.encode("utf-8"))
return request_content
return Server(text_response_handler, **kwargs)
@classmethod
def basic_response_server(cls, **kwargs):
return cls.text_response_server(
"HTTP/1.1 200 OK\r\n" + "Content-Length: 0\r\n\r\n", **kwargs
)
def run(self):
try:
self.server_sock = self._create_socket_and_bind()
# in case self.port = 0
self.port = self.server_sock.getsockname()[1]
self.ready_event.set()
self._handle_requests()
if self.wait_to_close_event:
self.wait_to_close_event.wait(self.WAIT_EVENT_TIMEOUT)
finally:
self.ready_event.set() # just in case of exception
self._close_server_sock_ignore_errors()
self.stop_event.set()
def _create_socket_and_bind(self):
sock = socket.socket()
sock.bind((self.host, self.port))
sock.listen()
return sock
def _close_server_sock_ignore_errors(self):
try:
self.server_sock.close()
except OSError:
pass
def _handle_requests(self):
for _ in range(self.requests_to_handle):
sock = self._accept_connection()
if not sock:
break
handler_result = self.handler(sock)
self.handler_results.append(handler_result)
sock.close()
def _accept_connection(self):Domain
Defined In
Source
Frequently Asked Questions
What is the Server class?
Server is a class in the requests codebase, defined in tests/testserver/server.py.
Where is Server defined?
Server is defined in tests/testserver/server.py at line 25.
Analyze Your Own Codebase
Get architecture documentation, dependency graphs, and domain analysis for your codebase in minutes.
Try Supermodel Free