Skip to content

Instantly share code, notes, and snippets.

@mesaglio
Created February 19, 2021 23:21
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save mesaglio/057b22cee97d705d20171399f97fafa5 to your computer and use it in GitHub Desktop.
Save mesaglio/057b22cee97d705d20171399f97fafa5 to your computer and use it in GitHub Desktop.
Python http socket server
import socket
import threading
import json
import datetime
class Server:
def __init__(self, port: int, multi_threading=True):
self.port = port
self.threads = multi_threading
@staticmethod
def _request_data(headers_string: str, _socket: socket.socket):
request_obj = {}
headers_list = headers_string.splitlines()
request_data = headers_list.pop(0).split(' ')
request_obj['Method'] = request_data[0]
request_obj['Path'] = request_data[1]
request_obj['Protocol'] = request_data[2]
for elem in headers_list:
if ': ' in elem:
header_splited = elem.split(': ')
request_obj[f'{header_splited[0]}'] = header_splited[1]
if request_obj.get('Content-Length'):
body = _socket.recv(int(request_obj['Content-Length']))
request_obj['Body'] = body.decode('utf-8')
return request_obj
@staticmethod
def _response(body: dict, _socket: socket.socket, request):
real_body = json.dumps(body, indent=4)
body_length = len(real_body)
_date = datetime.datetime.now()
date = _date.strftime("%a, %d %b %Y %H:%M:%S")
headers = f'''{request['Protocol']} 200\nDate: {date}\nContent-Type: application/json\nContent-Length: {body_length}\r\n\r\n{real_body}'''
_socket.sendall(bytes(headers, 'utf-8'))
_socket.close()
def _get_socket_date(self, _socket: socket.socket):
header_data = ""
while True:
data = _socket.recv(1)
header_data += data.decode('utf-8')
if data == b'\r':
data = _socket.recv(1)
header_data += data.decode('utf-8')
if data == b'\n':
data = _socket.recv(1)
header_data += data.decode('utf-8')
if data == b'\r':
data = _socket.recv(1)
header_data += data.decode('utf-8')
if data == b'\n':
header_data += data.decode('utf-8')
break
return self._request_data(header_data, _socket)
def _consume_request(self, _socket: socket.socket):
data = self._get_socket_date(_socket)
print(f"{datetime.datetime.now()} - {data.get('Method')} - {data.get('Path')}")
self._response(data, _socket, data)
def run(self):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
s.bind(('localhost', self.port))
s.listen(1)
print('Server up, waiting for requests...')
if self.threads:
while True:
connection, client_address = s.accept()
threading.Thread(target=self._consume_request, args=(connection,)).start()
while True:
connection, client_address = s.accept()
self._consume_request(connection)
if __name__ == '__main__':
server = Server(8080)
server.run()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment