Skip to content

Instantly share code, notes, and snippets.

@agagniere
Created October 31, 2023 08:59
Show Gist options
  • Save agagniere/fa76ded32665d09a839a07857b5a465f to your computer and use it in GitHub Desktop.
Save agagniere/fa76ded32665d09a839a07857b5a465f to your computer and use it in GitHub Desktop.
SImple TCP Client in Python
import sys
import json
import socket
import time
def is_json_complete(message: str) -> bool:
level = 0
for c in message:
level += {'{': 1, '}': -1}.get(c, 0)
if level < 0:
return False
return level == 0
class TcpClient:
def __init__(self, address: str, port: int):
self.address = address
self.port = port
self.socket = None
def connect(self):
# print('/' + '-' * 30, 'Connecting', '-' * 30 + '\\')
self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.socket.connect((self.address, self.port))
def close(self):
# print('\\' + '_' * 30, ' Closing ', '_' * 30 + '/' + '\n')
self.socket.close()
self.socket = None
def send(self, message: str):
self.socket.sendall(message.encode())
def receive_json(self) -> str:
accumulator = ''
received = 0
while True:
chunk = self.socket.recv(2048)
if not chunk:
raise RuntimeError('Connection broken')
accumulator += chunk.decode()
if is_json_complete(accumulator):
return accumulator
def call(self, message: str, log: bool = True):
self.send(message)
reply = self.receive_json()
if log:
print(f'{message:50}: {reply}')
return json.loads(reply)
class ShortTermTcpClient(TcpClient):
''' Allows to make several calls with the same connection '''
def __enter__(self):
self.connect()
return self
def __exit__(self, exception_type, exception_value, traceback):
self.close()
class LongTermTcpClient(ShortTermTcpClient):
''' Creates and closes a new connection for each call. '''
def call(self, query: str, log: bool = True):
with self as client:
return super().call(query, log)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment