import socket sock = socket.create_connection(address, timeout=GLOBAL_DEFAULT, source_address=None, *, all_errors=False)
address - адрес сервера - 2-кортеж (host, port),timeout=GLOBAL_DEFAULT - тайм-аут подключения,source_address=None - исходный адрес - 2-кортеж (host, port),all_errors - если True, то исключения всех попыток подключения попадают в группу исключений ExceptionGroup (добавлено в Python 3.11). Функция create_connection() модуля socket подключается к службе TCP, прослушивающей Интернет- address в виде (host, port) и возвращает объект сокета.
Эта функция более высокого уровня, чем метод объекта сокета Socket.connect(): если host - нечисловое имя хоста, то функция socket.create_connection() попытается разрешить его как для socket.AF_INET, так и для socket.AF_INET6, а затем попытается подключиться ко всем возможным адресам по очереди, пока соединение не установится. Это упрощает создание клиентов, совместимых как с IPv4, так и с IPv6.
Передача необязательного параметра timeout установит тайм-аут для экземпляра сокета перед попыткой подключения. Если тайм-аут не указан, то используется глобальная настройка тайм-аута по умолчанию, возвращаемая функцией socket.getdefaulttimeout().
Если указан source_address, то это должен быть 2-кортеж (host, port), к которому сокет будет привязан в качестве исходного адреса перед подключением. Если host или port имеют значение '' или 0 соответственно, то будет использоваться поведение ОС по умолчанию.
Когда соединение не может быть создано, возникает исключение. По умолчанию это исключение из последнего адреса в списке. Если аргумент all_errors имеет значение True, то это группа исключений ExceptionGroup, содержащая ошибки всех попыток.
Изменено в версии 3.11: добавлен аргумент all_errors.
Так как мы имеем дело с сокетами, то для тестирования кода клиента, необходим сервер. Код сервера представлен ниже.
Примечание. Для успешного тестирования примера, код клиента и сервера необходимо запускать в разных окнах терминала. Код сервера запускается первым.
# test-client.py import socket # Подключаем сокет к порту, через # который прослушивается сервер server_address = ('localhost', 8080) with socket.create_connection(server_address) as sock: print('Подключено к:', server_address) # Отправка данных mess = 'Привет!' print(f'Отправка: {mess}') message = mess.encode() sock.sendall(message) # Смотрим ответ amount_received = 0 amount_expected = len(message) while amount_received < amount_expected: data = sock.recv(16) amount_received += len(data) mess = data.decode() print(f'Получено: {data.decode()}') print('Отключено от:', server_address) # Подключено к: ('localhost', 8080) # Отправка: Привет! # Получено: ПРИВЕТ! # Отключено от: ('localhost', 8080)
# test-server.py import socket server_address = ('localhost', 8080) with socket.create_server(server_address) as sock: while True: # ждем соединения print('Ожидание соединения...') connect, client_address = sock.accept() print('Подключено к:', client_address) # Принимаем данные порциями и ретранслируем их while True: data = connect.recv(16) print(f'Получено: {data.decode()}') if data: print('Обработка данных...') temp_data = data.decode('utf-8').upper() data = temp_data.encode('utf-8') print('Данные обработаны, отправка клиенту...') connect.sendall(data) else: print('Нет данных от:', client_address) break # Ожидание соединения... # Подключено к: ('127.0.0.1', 32800) # Получено: Привет! # Обработка данных... # Данные обработаны, отправка клиенту... # Получено: # Нет данных от: ('127.0.0.1', 32800) # Ожидание соединения...