server.py — requests Source File

Architecture documentation for server.py, a python file in the requests codebase. 4 imports, 2 dependents.

File python CoreAPI VerbHandlers 4 imports 2 dependents 1 functions 2 classes

Entity Profile

CoreAPI→ VerbHandlers→ server.py — requests Source File

Dependency Diagram

graph LR
  1299f040_d82e_36bb_76ee_892d203f2f6a["server.py"]
  b006b4a1_ec19_c51f_ab37_63777bd33b0c["select"]
  1299f040_d82e_36bb_76ee_892d203f2f6a --> b006b4a1_ec19_c51f_ab37_63777bd33b0c
  81807cb8_22b9_a4ae_c8b0_91ee147c8a3a["socket"]
  1299f040_d82e_36bb_76ee_892d203f2f6a --> 81807cb8_22b9_a4ae_c8b0_91ee147c8a3a
  d4c9b75a_57ea_9b07_1986_5bd0fc134387["ssl"]
  1299f040_d82e_36bb_76ee_892d203f2f6a --> d4c9b75a_57ea_9b07_1986_5bd0fc134387
  d04f5b22_a751_5438_e937_aa8b0666d15e["threading"]
  1299f040_d82e_36bb_76ee_892d203f2f6a --> d04f5b22_a751_5438_e937_aa8b0666d15e
  48561d17_8bd1_bf20_5710_ec3053a534d2["test_lowlevel.py"]
  48561d17_8bd1_bf20_5710_ec3053a534d2 --> 1299f040_d82e_36bb_76ee_892d203f2f6a
  6cf98a23_4f9c_9f92_66f0_e3af57898d30["test_testserver.py"]
  6cf98a23_4f9c_9f92_66f0_e3af57898d30 --> 1299f040_d82e_36bb_76ee_892d203f2f6a
  style 1299f040_d82e_36bb_76ee_892d203f2f6a fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

import select
import socket
import ssl
import threading


def consume_socket_content(sock, timeout=0.5):
    chunks = 65536
    content = b""

    while True:
        more_to_read = select.select([sock], [], [], timeout)[0]
        if not more_to_read:
            break

        new_content = sock.recv(chunks)
        if not new_content:
            break

        content += new_content

    return content


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)

// ... (117 more lines)