import socket sock = socket.create_server(address, *, family=AF_INET, backlog=None, reuse_port=False, dualstack_ipv6=False)
address - адресу сервера в виде (host, port),family=AF_INET - семейства адресов,backlog=None - размер очереди из клиентов,reuse_port=False - следует ли устанавливать socket.SO_REUSEPORT,dualstack_ipv6=False - поддержка IPv6.Функция create_server() модуля socket создает сокет, слушающий TCP/IP подключения, привязанный к адресу address в виде 2-кортеж (host, port) и возвращает объект сокета.
Аргумент семейства адресов family должен быть socket.AF_INET или socket.AF_INET6.
Аргумент backlog - это размер очереди, переданной в метод объекта сокета Socket.listen(). Когда это значение равно 0, то выбирается разумное значение по умолчанию.
Аргумент reuse_port указывает, следует ли устанавливать параметр сокета socket.SO_REUSEPORT.
Если аргумент dualstack_ipv6=True и платформа его поддерживает, то сокет сможет принимать соединения как IPv4, так и IPv6, в противном случае возникнет ошибка ValueError.
Предполагается, что большинство платформ POSIX и Windows поддерживают эту функцию. Когда эта функция включена, то адрес, возвращаемый методом объекта сокета Socket.getpeername() при подключении IPv4, будет IPv6-адресом, представленным как IPv4-сопоставленный адрес.
Если для dualstack_ipv6=False, то метод объекта сокета Socket.getpeername() будет явно отключен на платформах, которые включают ее по умолчанию (например, Linux). Этот параметр можно использовать вместе с функцией socket.has_dualstack_ipv6():
import socket # слушать все IP на порту 8080 addr = ('', 8080) if socket.has_dualstack_ipv6(): s = socket.create_server(addr, family=socket.AF_INET6, dualstack_ipv6=True) else: s = socket.create_server(addr)
Примечание. На платформах POSIX параметр сокета socket.SO_REUSEADDR установлен для немедленного повторного использования предыдущих сокетов, которые были привязаны к тому же адресу и оставались в состоянии TIME_WAIT.
Новое в Python 3.8.
Так как мы имеем дело с сокетами, то для тестирования кода сервера, необходим клиент. Код клиента представлен ниже.
Примечание. Для успешного тестирования примера, код клиента и сервера необходимо запускать в разных окнах терминала. Код сервера запускается первым.
# 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) # Ожидание соединения...
# 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)