Сетевое программирование на Python: протоколы от TCP до HTTP

address = ('127.0.0.1', 8000) # Локальный адрес и порт 8000

import socket

# Создаем TCP сокет
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Привязываем сокет к адресу и порту
server_socket.bind(('127.0.0.1', 8000))

# Начинаем прослушивание (максимум 5 ожидающих подключений)
server_socket.listen(5)
print("Сервер запущен на 127.0.0.1:8000")

while True:
# Принимаем входящее подключение
client_socket, client_address = server_socket.accept()
print(f"Подключение от {client_address}")

# Получаем данные от клиента
data = client_socket.recv(1024)
if not data:
break

# Отправляем ответ
response = f"Получено: {data.decode()}"
client_socket.send(response.encode())

# Закрываем соединение с клиентом
client_socket.close()

# Закрываем серверный сокет
server_socket.close()

import socket

# Создаем TCP сокет
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# Подключаемся к серверу
client_socket.connect(('127.0.0.1', 8000))

# Отправляем сообщение
message = "Привет, сервер!"
client_socket.send(message.encode())

# Получаем ответ
response = client_socket.recv(1024)
print(f"Ответ сервера: {response.decode()}")

# Закрываем сокет
client_socket.close()

# UDP-сервер
import socket

server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server_socket.bind(('127.0.0.1', 8000))

while True:
data, client_address = server_socket.recvfrom(1024)
print(f"Сообщение от {client_address}: {data.decode()}")
response = f"UDP-ответ на: {data.decode()}"
server_socket.sendto(response.encode(), client_address)

# UDP-клиент
import socket

client_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
message = "Привет, UDP-сервер!"
client_socket.sendto(message.encode(), ('127.0.0.1', 8000))
response, _ = client_socket.recvfrom(1024)
print(f"Ответ: {response.decode()}")
client_socket.close()

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.connect(('127.0.0.1', 8000))
s.send(b'Hello, world')
data = s.recv(1024)
# Сокет автоматически закроется после выхода из блока with

import requests

# GET-запрос
response = requests.get('https://api.github.com/events')
print(f"Статус: {response.status_code}")
print(f"Тип содержимого: {response.headers['Content-Type']}")
print(f"Содержимое: {response.text[:100]}...") # Печатаем первые 100 символов

# POST-запрос с JSON-данными
data = {'key': 'value', 'another_key': 'another_value'}
response = requests.post('https://httpbin.org/post', json=data)
print(f"Ответ сервера: {response.json()}")

# Загрузка файла
files = {'file': open('example.txt', 'rb')}
response = requests.post('https://httpbin.org/post', files=files)

# Параметры запроса
params = {'q': 'python', 'sort': 'stars'}
response = requests.get('https://api.github.com/search/repositories', params=params)

# Установка заголовков
headers = {'User-Agent': 'Python Requests App', 'Authorization': 'Bearer token123'}
response = requests.get('https://api.example.com/data', headers=headers)

# Работа с куками
cookies = {'session_id': '123456'}
response = requests.get('https://httpbin.org/cookies', cookies=cookies)

with requests.Session() as session:
# Устанавливаем параметры, которые будут использоваться во всех запросах
session.headers.update({'User-Agent': 'Python Requests App'})
session.auth = ('username', 'password')

# Первый запрос – авторизация
session.get('https://example.com/login')

# Последующие запросы будут использовать те же куки и заголовки
response = session.get('https://example.com/dashboard')

from urllib.request import urlopen, Request
from urllib.parse import urlencode

# Простой GET-запрос
with urlopen('https://www.python.org') as response:
html = response.read()
print(f"Получено {len(html)} байт")

# POST-запрос с данными
data = urlencode({'name': 'John', 'age': 30}).encode()
req = Request('https://httpbin.org/post', data=data, method='POST')
with urlopen(req) as response:
print(response.read().decode())

# Добавление заголовков
headers = {'User-Agent': 'Mozilla/5.0'}
req = Request('https://api.github.com/events', headers=headers)
with urlopen(req) as response:
print(response.read(100)) # Первые 100 байт

import requests
import logging

# Настройка логирования
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)

# Запрос с включенным логированием
response = requests.get('https://api.github.com/events')

import asyncio

async def handle_client(reader, writer):
data = await reader.read(100)
message = data.decode()
addr = writer.get_extra_info('peername')

print(f"Получено {message!r} от {addr!r}")

response = f"Обработано: {message!r}"
writer.write(response.encode())
await writer.drain()

writer.close()
await writer.wait_closed()
print(f"Соединение с {addr!r} закрыто")

async def main():
server = await asyncio.start_server(
handle_client, '127.0.0.1', 8888)

addr = server.sockets[0].getsockname()
print(f'Сервер запущен на {addr}')

async with server:
await server.serve_forever()

asyncio.run(main())

import asyncio

async def tcp_client():
reader, writer = await asyncio.open_connection(
'127.0.0.1', 8888)

message = 'Привет, asyncio сервер!'
print(f'Отправка: {message!r}')
writer.write(message.encode())
await writer.drain()

data = await reader.read(100)
print(f'Получено: {data.decode()!r}')

writer.close()
await writer.wait_closed()

asyncio.run(tcp_client())

import asyncio
import aiohttp

async def fetch(session, url):
async with session.get(url) as response:
return await response.text()

async def main():
# Создаем сессию для повторного использования соединений
async with aiohttp.ClientSession() as session:
# Список URL для запросов
urls = [
'https://api.github.com/events',
'https://api.github.com/emojis',
'https://api.github.com/meta'
]

# Создаем список задач
tasks = [fetch(session, url) for url in urls]

# Запускаем все задачи конкурентно
results = await asyncio.gather(*tasks)

for url, result in zip(urls, results):
print(f"{url}: получено {len(result)} байт")

# Запускаем асинхронный код
asyncio.run(main())

import asyncio
import aiohttp
from aiohttp import ClientSession

async def fetch_with_semaphore(semaphore, session, url):
# Семафор ограничивает количество одновременных запросов
async with semaphore:
try:
async with session.get(url, timeout=10) as response:
return await response.text(), url
except Exception as e:
return f"Ошибка: {str(e)}", url

async def main():
# Максимум 5 одновременных запросов
semaphore = asyncio.Semaphore(5)

# Список URL
urls = [f"https://httpbin.org/delay/{i%3}" for i in range(20)]

async with ClientSession() as session:
tasks = [
fetch_with_semaphore(semaphore, session, url) 
for url in urls
]

# Обрабатываем результаты по мере их готовности
for future in asyncio.as_completed(tasks):
result, url = await future
print(f"Завершен запрос к {url}: {len(result) if not isinstance(result, str) or not result.startswith('Ошибка') else result}")

asyncio.run(main())

# chat_server.py
import asyncio
import json
from datetime import datetime

# Список для хранения подключенных клиентов
connected_clients = []

async def handle_client(reader, writer):
# Получаем информацию о клиенте
client_addr = writer.get_extra_info('peername')
client_id = f"{client_addr[0]}:{client_addr[1]}"
print(f"Новое подключение от {client_id}")

# Добавляем клиента в список
connected_clients.append(writer)

# Отправляем приветственное сообщение
welcome_message = {
"type": "system",
"time": datetime.now().strftime("%H:%M:%S"),
"content": f"Добро пожаловать в чат! Онлайн пользователей: {len(connected_clients)}"
}
writer.write(json.dumps(welcome_message).encode() + b'\n')
await writer.drain()

# Оповещаем всех о новом пользователе
broadcast_message = {
"type": "system",
"time": datetime.now().strftime("%H:%M:%S"),
"content": f"Пользователь {client_id} присоединился к чату"
}
await broadcast(broadcast_message, writer)

try:
while True:
# Получаем данные от клиента
data = await reader.readline()
if not data: # Соединение закрыто
break

message = data.decode().strip()
print(f"Получено от {client_id}: {message}")

try:
# Пытаемся разобрать JSON
parsed_message = json.loads(message)
parsed_message["time"] = datetime.now().strftime("%H:%M:%S")
parsed_message["sender"] = client_id

# Рассылаем сообщение всем клиентам
await broadcast(parsed_message, None)
except json.JSONDecodeError:
error_message = {
"type": "error",
"time": datetime.now().strftime("%H:%M:%S"),
"content": "Неверный формат сообщения"
}
writer.write(json.dumps(error_message).encode() + b'\n')
await writer.drain()

except asyncio.CancelledError:
pass
except Exception as e:
print(f"Ошибка при обработке клиента {client_id}: {str(e)}")
finally:
# Удаляем клиента из списка при отключении
if writer in connected_clients:
connected_clients.remove(writer)

# Закрываем соединение
writer.close()
await writer.wait_closed()

# Оповещаем о выходе пользователя
leave_message = {
"type": "system",
"time": datetime.now().strftime("%H:%M:%S"),
"content": f"Пользователь {client_id} покинул чат"
}
await broadcast(leave_message, None)
print(f"Соединение с {client_id} закрыто")

async def broadcast(message, exclude):
"""Отправляет сообщение всем подключенным клиентам, кроме отправителя"""
encoded_message = json.dumps(message).encode() + b'\n'

for client in connected_clients:
if client != exclude:
try:
client.write(encoded_message)
await client.drain()
except:
# Если отправка не удалась, просто пропускаем клиента
pass

async def main():
server = await asyncio.start_server(
handle_client, '0.0.0.0', 8888)

addr = server.sockets[0].getsockname()
print(f'Чат-сервер запущен на {addr}')

async with server:
await server.serve_forever()

if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Сервер остановлен")

# chat_client.py
import asyncio
import json
import sys
import aioconsole

async def receive_messages(reader):
"""Асинхронно получает и выводит сообщения от сервера"""
try:
while True:
data = await reader.readline()
if not data:
print("Соединение с сервером потеряно")
break

message = json.loads(data.decode())

if message["type"] == "system":
print(f"[{message['time']}] СИСТЕМА: {message['content']}")
elif message["type"] == "message":
print(f"[{message['time']}] {message['sender']}: {message['content']}")
elif message["type"] == "error":
print(f"[{message['time']}] ОШИБКА: {message['content']}")
except Exception as e:
print(f"Ошибка при получении сообщений: {str(e)}")
finally:
# Завершаем программу при потере соединения
sys.exit(0)

async def send_messages(writer):
"""Асинхронно отправляет сообщения на сервер"""
try:
print("Введите ваше имя:")
username = await aioconsole.ainput()

print("Введите сообщение (или /exit для выхода):")

while True:
# Асинхронно ждем ввода пользователя
message_text = await aioconsole.ainput()

if message_text.lower() == '/exit':
break

# Формируем JSON-сообщение
message = {
"type": "message",
"username": username,
"content": message_text
}

# Отправляем на сервер
writer.write(json.dumps(message).encode() + b'\n')
await writer.drain()
except Exception as e:
print(f"Ошибка при отправке сообщений: {str(e)}")
finally:
# Закрываем соединение при выходе
writer.close()
await writer.wait_closed()
sys.exit(0)

async def main():
try:
# Подключаемся к серверу
reader, writer = await asyncio.open_connection('127.0.0.1', 8888)
print("Подключено к серверу чата")

# Запускаем две задачи: отправка и прием сообщений
receive_task = asyncio.create_task(receive_messages(reader))
send_task = asyncio.create_task(send_messages(writer))

# Ждем завершения любой из задач
await asyncio.gather(receive_task, send_task, return_exceptions=True)
except ConnectionRefusedError:
print("Не удалось подключиться к серверу. Убедитесь, что сервер запущен.")
except Exception as e:
print(f"Ошибка: {str(e)}")

if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Клиент остановлен")