Серверное программирование в Python: создание и настройка серверов

Расширенные примеры создания серверов на Python

Многопоточный TCP сервер с использованием threading

Этот пример демонстрирует сервер, обрабатывающий каждое соединение в отдельном потоке, что позволяет параллельно обслуживать нескольких клиентов.

import socket
import threading

def handle_client(client_socket, addr):
    print(f'Обработка клиента: {addr}')
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            client_socket.sendall(data.upper())
    except Exception as e:
        print(f'Ошибка: {e}')
    finally:
        client_socket.close()
        print(f'Клиент {addr} отключен')

def main():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(('0.0.0.0', 9999))
    server.listen(10)
    print('Многопоточный сервер запущен на порту 9999...')

    while True:
        client, addr = server.accept()
        thread = threading.Thread(target=handle_client, args=(client, addr))
        thread.start()

if __name__ == '__main__':
    main()

Результат: после запуска сервера можно подключиться несколькими клиентами (например, через telnet). Каждое соединение обрабатывается в своем потоке. Пример вывода в консоль сервера:

Многопоточный сервер запущен на порту 9999...
Обработка клиента: ('127.0.0.1', 54321)
Обработка клиента: ('127.0.0.1', 54322)
Клиент ('127.0.0.1', 54321) отключен

Асинхронный echo сервер с asyncio и обработкой таймаутов

Расширенный вариант асинхронного сервера с таймаутом на чтение и логированием.

import asyncio
import logging

logging.basicConfig(level=logging.INFO)

async def echo_handler(reader, writer):
    addr = writer.get_extra_info('peername')
    logging.info(f'Новое соединение: {addr}')
    try:
        while True:
            data = await asyncio.wait_for(reader.read(100), timeout=30)
            if not data:
                break
            writer.write(data)
            await writer.drain()
    except asyncio.TimeoutError:
        logging.warning(f'Таймаут соединения с {addr}')
    except Exception as e:
        logging.error(f'Ошибка: {e}')
    finally:
        writer.close()
        logging.info(f'Соединение с {addr} закрыто')

async def main():
    server = await asyncio.start_server(echo_handler, 'localhost', 7777)
    addr = server.sockets[0].getsockname()
    logging.info(f'Сервер запущен на {addr}')
    async with server:
        await server.serve_forever()

if __name__ == '__main__':
    asyncio.run(main())

Результат: сервер принимает соединения, отвечает эхом и завершает при неактивности более 30 секунд. В лог выводится информация.

INFO:root:Сервер запущен на ('127.0.0.1', 7777)
INFO:root:Новое соединение: ('127.0.0.1', 60001)
INFO:root:Соединение с ('127.0.0.1', 60001) закрыто

Сервер на FastAPI с обработкой JSON запросов и документацией

FastAPI - современный асинхронный веб-фреймворк с автоматической генерацией документации OpenAPI.

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post('/items/')
async def create_item(item: Item):
    # Имитация сохранения в БД
    return {'id': 123, 'name': item.name, 'price': item.price}

@app.get('/')
async def root():
    return {'message': 'Сервер работает!'}

if __name__ == '__main__':
    uvicorn.run(app, host='0.0.0.0', port=8000)

Результат: сервер запускается на порту 8000. Документация доступна по адресу /docs. Пример запроса через curl:

$ curl -X POST http://localhost:8000/items/ -H "Content-Type: application/json" -d '{"name":"Шляпа", "price": 100}'
{"id":123,"name":"Шляпа","price":100.0}

Автоматически генерируется интерактивная документация Swagger.