Как передать данные по сокету Python

Работа с потоками происходит примерно так: Пока 1й поток ждёт новой информации от клиента, 2й - работает (Одновременно выполнять код они не могут из-за GIL). Выглядит это примерно так:

    def _run(self) -> None:
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.bind((HOST, PORT))
        self.server_socket.listen(8)

        while True:  # В бесконечном цикле ждём новых клиентов
            client, address = self.server_socket.accept()  # Принимаем подключение клиента
            thread = threading.Thread(target=self.handle_client, args=(client, address), daemon=True)  # создаём отдельный поток для него
            thread.start()  # Запускаем поток

    def handle_client(self, client_socket: socket.socket, address: tuple) -> None:
        while True:  # В бесконечном цикле ждём сообщений и обрабатываем их как нам нужно
            data = client_socket.recv(1024)
            data = data.decode(encoding=ENCODING)
            if not data:
                break

А как связать 2 клиента через сервер уже дело фантазии. Как пример: через создание пользователей и поиску по id:

import json
import socket
from dataclasses import dataclass
import threading


HOST = 'localhost'
PORT = 8088
ENCODING = 'utf-8'


# можно вынести в отдельный файл
@dataclass
class Client:  # Объект пользователя
    id: int
    client_socket: socket.socket
    name: str


# можно вынести в отдельный файл
def register(client_socket: socket.socket, data: dict, server: 'Server') -> dict:
    client = server.create_user(client_socket, data.get("name", "user"))  # Создаём пользователя и даём ему id
    return {
        "name": client.name,
        "id": client.id
    }


# можно вынести в отдельный файл
def send_message(client_socket, data: dict, server: 'Server') -> dict:
    client_1 = server.connected_clients.get(client_socket)

    if not client_1:  # Проверяем наличие всей нужной нам информации для отправки сообщения (Кто отправил, кому отправил, что отправил)
        return {"error": "user not found"}
    if not (client_2_id := data.get('client_2_id')):
        return {"error": "client_2_id not found"}
    if not (message := data.get("message")):
        return {"error": "message not found"}

    try:
        client_2 = list(filter(lambda x: x.id == int(client_2_id), server.connected_clients.values()))[0]  # Ищем 2го пользователя по id
    except IndexError:
        return {"error": "User to send not found"}

    server.send(client_2.client_socket, {  # Если пользователь найден - отправляем ему сообщение
        "function": "new_message",
        "data": {
            "message": message,
            "sender": {
                "id": client_1.id,
                "name": client_1.name
            },
        }
    })
    return {"success": "ok"}


class Server:
    last_client_id = 0
    functions = {  # Функции, которые может вызвать пользователь
        "register": register,
        "send_message": send_message
    }

    def __init__(self):
        self.connected_clients = {}
        self._run()

    # можно вынести в отдельный файл
    def create_user(self, client_socket: socket.socket, name: str) -> Client:
        client = Client(self.last_client_id, client_socket, name)
        self.connected_clients[client_socket] = client
        self.last_client_id += 1
        return client

    # можно вынести в отдельный файл
    def delete_user(self, client_socket: socket.socket) -> None:
        if self.connected_clients.get(client_socket):
            del self.connected_clients[client_socket]

    # можно вынести в отдельный файл
    def send(self, client_socket: socket.socket, data: dict) -> None:
        client_socket.send(json.dumps(data, indent=4).encode(ENCODING))  # Подготовка данных и их отправка клиенту

    def _run(self) -> None:
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.bind((HOST, PORT))
        self.server_socket.listen(8)

        while True:
            client, address = self.server_socket.accept()  # Принимаем подключение клиента
            thread = threading.Thread(target=self.handle_client, args=(client, address), daemon=True)  # создаём отдельный поток для него
            thread.start()  # Запускаем поток

    def handle_client(self, client_socket: socket.socket, address: tuple) -> None:
        while True:  # В бесконечном цикле ждём сообщений и обрабатываем их как нам нужно
            data = client_socket.recv(1024)
            data = data.decode(encoding=ENCODING)

            try:
                data = json.loads(data)
            except json.JSONDecodeError:
                self.send(client_socket, {"error": "json decode error"})
                return

            if not data:
                self.delete_user(client_socket)  # Пользователь отключился
                break

            if func := self.functions.get(data.get("function")):  # Проверяем есть ли нужная пользователю функция в списке доступных функций сервера
                response = func(client_socket, data.get("data", {}), self)  # Если есть, то вызываем её и отдаём результат
                self.send(client_socket, response)
            else:
                self.send(client_socket, {"error": "function not found"})


if __name__ == '__main__':
    app = Server()

P.S: Код достаточно кустарный и недоделанный, но алгоритм, думаю, понятен.

Или можно воспользоваться не потоками, а асинхронностью, что, на мой взгляд, лучше (т.к потоки в больших количествах могут есть много памяти):

import json
import socket
import asyncio
from dataclasses import dataclass


HOST = 'localhost'
PORT = 8088
ENCODING = 'utf-8'


@dataclass
class Client:
    id: int
    client_socket: socket.socket
    name: str


async def register(client_socket: socket.socket, data: dict, server: 'Server') -> dict:
    client = server.create_user(client_socket, data.get("name", "user"))
    return {
        "name": client.name,
        "id": client.id
    }


async def send_message(client_socket, data: dict, server: 'Server') -> dict:
    client_1 = server.connected_clients.get(client_socket)

    if not client_1:
        return {"error": "user not found"}
    if not (client_2_id := data.get('client_2_id')):
        return {"error": "client_2_id not found"}
    if not (message := data.get("message")):
        return {"error": "message not found"}

    try:
        client_2 = list(filter(lambda x: x.id == int(client_2_id), server.connected_clients.values()))[0]
    except IndexError:
        return {"error": "User to send not found"}

    server.send(client_2.client_socket, {
        "function": "new_message",
        "data": {
            "message": message,
            "sender": {
                "id": client_1.id,
                "name": client_1.name
            },
        }
    })
    return {"success": "ok"}


class Server:
    last_client_id = 0
    functions = {
        "register": register,
        "send_message": send_message
    }

    def __init__(self):
        loop = asyncio.get_event_loop()
        self.connected_clients = {}
        loop.run_until_complete(self._run())  # Запускаем сервер

    def create_user(self, client_socket: socket.socket, name: str) -> Client:
        client = Client(self.last_client_id, client_socket, name)
        self.connected_clients[client_socket] = client
        self.last_client_id += 1
        return client

    def delete_user(self, client_socket: socket.socket) -> None:
        if self.connected_clients.get(client_socket):
            del self.connected_clients[client_socket]

    def send(self, client_socket: socket.socket, data: dict) -> None:
        client_socket.send(json.dumps(data, indent=4).encode(ENCODING))

    async def _run(self) -> None:
        self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.server_socket.bind((HOST, PORT))
        self.server_socket.listen(8)
        self.server_socket.setblocking(False)  # Убираем блокировку

        loop = asyncio.get_event_loop()

        while True:
            client, address = await loop.sock_accept(self.server_socket)  # Асинхронно ждём пользователей
            loop.create_task(self.handle_client(client, address))   # Создаём задачу под пользователя

    async def handle_client(self, client_socket: socket.socket, address: tuple) -> None:
        loop = asyncio.get_event_loop()
        while True:
            data = await loop.sock_recv(client_socket, 1024)  # Асинхронно ждём данные
            data = data.decode(encoding=ENCODING)

            try:
                data = json.loads(data)
            except json.JSONDecodeError:
                self.send(client_socket, {"error": "json decode error"})
                return

            if not data:
                self.delete_user(client_socket)
                break

            if func := self.functions.get(data.get("function")):
                response = await func(client_socket, data.get("data", {}), self)
                self.send(client_socket, response)
            else:
                self.send(client_socket, {"error": "function not found"})

Клиент:

import json
import socket
import threading

client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.connect(('localhost', 8088))


class Listener(threading.Thread):  # Отдельный поток для поиска и вывода новых сообщений
    def run(self) -> None:
        while True:
            from_server = client.recv(1024)
            print(from_server)


def send(request: dict) -> None:
    client.send(json.dumps(request).encode(encoding='utf-8'))


listener = Listener(daemon=True)
listener.start()

send({  # Регистрация
    "function": "register",
    "data": {
        "name": input("Name: ")
    }
})

while True:  # Отправка
    message = input("message: ")

    send({
        "function": "send_message",
        "data": {
            "client_2_id": input("to: "),
            "message": message
        }
    })