Sockets Python

Введение

Этот модуль обеспечивает доступ к интерфейсу сокета BSD. Он доступен во всех современных системах Unix , Windows , macOS и, возможно, на дополнительных платформах.

Про то, что такое TCP сокет и про сети в целом вы можете прочитать в статье «Компьютерные сети»

Некоторое поведение может зависеть от платформы, поскольку выполняются вызовы API-интерфейсов сокетов операционной системы.

Этот модуль не работает или недоступен на платформах WebAssembly wasm32-emscripten и wasm32-wasi. Дополнительные сведения см. в разделе Платформы веб-сборки .

Интерфейс Python представляет собой простую транслитерацию системного вызова Unix и библиотечного интерфейса для сокетов в объектно-ориентированный стиль Python: функция socket() возвращает объект socket, методы которого реализуют различные системные вызовы сокетов. Типы параметров несколько более высокоуровневые, чем в интерфейсе C: как и в случае операций чтения() и записи() в файлах Python, распределение буфера при операциях приема происходит автоматически, а длина буфера неявно указывается при операциях отправки.

Официальная документация

Пример

Простейший пример - открываем сокет, ждём подключение, фиксируем, что кто-то подключился и больше ничего не делаем.

Клиент от такого сервера ничего полезного не получит.

import socket import sys sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_address = ('', 10000) print(sys.stderr, 'starting up on %s port %s' % server_address) sock.bind(server_address) sock.listen(1) while True: print(sys.stderr, 'waiting for connection') connection, client_address = sock.accept() try: print(sys.stderr, 'connection from', client_address) while True: data = connection.recv(16) print(sys.stderr, 'received "%s"' % data) if data: print(sys.stderr, 'sending data back to the client') connection.sendall(data) else: print(sys.stderr, 'no more data from', client_address) break finally: connection.close()

Для этого сервера подойдёт самый простой клиент

import socket import sys # Create a TCP/IP socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # Connect the socket to the port where the server is listening # server_address = ('localhost', 10000) server_address = ('192.168.56.113', 10000) print(sys.stderr, 'connecting to %s port %s' % server_address) sock.connect(server_address) try: # Send data message = 'This is the message. It will be repeated.' print(sys.stderr, 'sending "%s"' % message) sock.sendall(message) # Look for the response amount_received = 0 amount_expected = len(message) while amount_received < amount_expected: data = sock.recv(16) amount_received += len(data) print(sys.stderr, 'received "%s"' % data) finally: print(sys.stderr, 'closing socket') sock.close()

Пример с обработкой данных

В этом примере открывается сокет, который слушает входящие данные и обрабатывает их.

Подробности в комментариях к коду.

# Нужно принять сообщение типа # name=Sergey&last_name=Ivanov&birthday=1990-10-05 # и вытащить имя, фамилию и ДР # Используем модуль socket: import socket # Создаём сокет sock = socket.socket() # Выбираем хост и порт. # Хост сделаем пустым для большей доступности sock.bind(('', 9090)) # Запускаем режим прослушивания и определяем размер очереди sock.listen(1) # Принимаем подключение с помощью метода accept # accept возвращает кортеж (tuple) с двумя элементами # новый сокет и адрес клента # адрес клиент это тоже кортеж из двух элементов # IP это элемент с индексом 0 (class 'str') conn, addr = sock.accept() print('connected:', addr) client_ip = addr[0] # В my_list мы будем собирать символы по 4 чтобы понять # было два перевода строки или нет my_list = [] # my_decoded_list сделан для подстаховки, чтобы убедиться # что мы не приняли за перевод строки что-то другое # это ещё не реализовано my_decoded_list = [] # i просто считает символы i = 0 while True: symbol = conn.recv(1) # сюда нужно встроить проверку декодируется ли этот символ в принципе или нет # хотя бы try поставить decoded_symbol = symbol.decode('utf-8') my_list += symbol my_decoded_list += decoded_symbol # пока что мы не ожидаем сообщение # нужно дождаться двух переводов строки message_coming = False # поймали два перевода строки if i > 2 and my_list[i] == 10 and my_list[i - 1] == 13 and my_list[i - 2] == 10 and my_list[i - 3] == 13: # теперь мы ждём сообщение message_coming = not message_coming while message_coming == True: # о размере нужно договориться заранее # мы ждём до 2-х килобайт message = conn.recv(2048) # Декодируем из 'b' в 'str' decoded_message = message.decode('utf-8') # Проверяем что сообщение начинается с name, иначе нет смысла с ним # работать дальше if decoded_message[:5] == 'name=': # Разбиваем по разделительному символу & new_list = decoded_message.split('&') new_list_len = len(new_list) # Проверяем, что элемента три if new_list_len == 3: name = new_list[0][5:] last_name_check = new_list[1][:9] if last_name_check == 'last_name': last_name = new_list[1][10:] else: print("Error with last_name name") message_coming = False birthday = new_list[2][span class="lightblue">9:span class="lightblue">-2] print(name) print(last_name) print(birthday) if birthday: birthday_len = len(birthday) if birthday_len != 10: print(f"Error in birthday format. Expected len = 10." f"Actual len = {birthday_len}") # Здесь я шлю данные обратно, # но вообще можно вызывать какую-то функцию и передавать аргументы conn.send(bytes(name, 'utf-8')) conn.send(bytes(last_name, 'utf-8')) conn.send(bytes(birthday, 'utf-8')) conn.send(bytes(client_ip, 'utf-8')) print("OK") message_coming = False else: print(f"message is not full. Expected len = 3." f"Actual len = {new_list_len}") message_coming = False else: print("message is not of correct format") message_coming = False i += 1 if not symbol: break # Закрываем подключение # Как именно закрывется подключение я ещё не разобрался conn.close()

gethostbyname()

import socket def main(): print(resolve) ip = resolve("www.eth1.ru") print(ip) def resolve(host): return socket.gethostbyname(host) if __name__ == "__main__": main()

Пример с кэшированием gethostbyname() вы можете изучить здесь

РЕКЛАМА хостинга Beget, которым я пользуюсь более десяти лет

Конец рекламы хостинга Beget, который я всем рекомендую.