RU
EN
RU
EN
Стандартные потоки ввода-вывода в Python
Работа с потоками ввода-вывода (I/O) в Python является фундаментальным аспектом программирования, который включает обработку стандартных потоков ввода, вывода и ошибок, а также работу с файлами, сетевыми соединениями и другими источниками данных. Понимание этих механизмов критически важно для создания эффективных и надежных приложений.
Основные стандартные потоки
В Python стандартные потоки ввода-вывода представлены тремя основными объектами из модуля sys:
- sys.stdin - поток для чтения данных из стандартного ввода (обычно клавиатура)
- sys.stdout - поток для вывода данных в стандартный вывод (обычно экран терминала)
- sys.stderr - поток для вывода сообщений об ошибках в стандартный поток ошибок
Чтение данных из стандартного ввода
import sys
# Чтение одной строки
data = sys.stdin.readline()
print("Вы ввели:", data.strip())
# Чтение всех данных
all_data = sys.stdin.read()
print("Все данные:", all_data)
# Построчное чтение
for line in sys.stdin:
print("Строка:", line.strip())
Вывод данных в стандартные потоки
import sys
# Вывод в стандартный поток вывода
sys.stdout.write("Привет, мир!\n")
sys.stdout.flush() # Принудительная отправка буфера
# Вывод сообщения об ошибке
sys.stderr.write("Это сообщение об ошибке!\n")
sys.stderr.flush()
# Использование print() с указанием потока
print("Обычное сообщение", file=sys.stdout)
print("Сообщение об ошибке", file=sys.stderr)
Работа с файлами
Основные режимы открытия файлов
Python предоставляет функцию open() для работы с файлами в различных режимах:
- 'r' - чтение (по умолчанию)
- 'w' - запись (перезаписывает существующий файл)
- 'a' - добавление в конец файла
- 'x' - эксклюзивное создание (файл не должен существовать)
- 'b' - бинарный режим (например, 'rb', 'wb')
- 't' - текстовый режим (по умолчанию)
Примеры работы с файлами
# Чтение файла
with open("example.txt", "r", encoding="utf-8") as file:
content = file.read()
print(content)
# Построчное чтение
with open("example.txt", "r", encoding="utf-8") as file:
for line in file:
print(line.strip())
# Запись в файл
with open("output.txt", "w", encoding="utf-8") as file:
file.write("Hello, world!\n")
file.write("Вторая строка\n")
# Добавление в файл
with open("output.txt", "a", encoding="utf-8") as file:
file.write("Добавленная строка\n")
# Работа с бинарными файлами
with open("image.jpg", "rb") as binary_file:
data = binary_file.read()
print(f"Размер файла: {len(data)} байт")
Обработка ошибок при работе с файлами
import os
try:
with open("nonexistent.txt", "r") as file:
content = file.read()
except FileNotFoundError:
print("Файл не найден")
except PermissionError:
print("Нет прав доступа к файлу")
except IOError as e:
print(f"Ошибка ввода-вывода: {e}")
# Проверка существования файла
if os.path.exists("example.txt"):
with open("example.txt", "r") as file:
content = file.read()
Работа с сетевыми потоками
TCP-сервер
import socket
import threading
def handle_client(conn, addr):
"""Обработчик подключения клиента"""
print(f'Подключен клиент: {addr}')
try:
while True:
data = conn.recv(1024)
if not data:
break
print(f'Получено от {addr}: {data.decode()}')
conn.sendall(data) # Эхо-сервер
except Exception as e:
print(f'Ошибка при обработке клиента {addr}: {e}')
finally:
conn.close()
print(f'Соединение с {addr} закрыто')
def start_server():
HOST = '127.0.0.1'
PORT = 12345
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind((HOST, PORT))
server.listen(5)
print(f'Сервер запущен на {HOST}:{PORT}')
while True:
conn, addr = server.accept()
client_thread = threading.Thread(
target=handle_client,
args=(conn, addr)
)
client_thread.start()
if __name__ == "__main__":
start_server()
TCP-клиент
import socket
def tcp_client():
HOST = '127.0.0.1'
PORT = 12345
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as client:
client.connect((HOST, PORT))
# Отправка данных
message = b'Hello, world!'
client.sendall(message)
# Получение ответа
data = client.recv(1024)
print(f'Получено: {data.decode()}')
except ConnectionRefusedError:
print("Не удалось подключиться к серверу")
except Exception as e:
print(f"Ошибка: {e}")
if __name__ == "__main__":
tcp_client()
Буферизация и производительность
Управление буферизацией
import sys
# Отключение буферизации для stdout
sys.stdout = open(sys.stdout.fileno(), 'w', buffering=0)
# Работа с буферизацией при записи файлов
with open("large_file.txt", "w", buffering=8192) as file:
for i in range(1000000):
file.write(f"Строка {i}\n")
if i % 1000 == 0:
file.flush() # Принудительная запись буфера
Контекстные менеджеры для потоков
import contextlib
import sys
@contextlib.contextmanager
def redirect_stdout(new_target):
"""Контекстный менеджер для перенаправления stdout"""
old_target = sys.stdout
sys.stdout = new_target
try:
yield new_target
finally:
sys.stdout = old_target
# Использование
with open("output.txt", "w") as f:
with redirect_stdout(f):
print("Это будет записано в файл")
print("И это тоже")
Асинхронный ввод-вывод
Использование asyncio для асинхронного I/O
import asyncio
import aiofiles
async def read_file_async(filename):
"""Асинхронное чтение файла"""
async with aiofiles.open(filename, 'r') as file:
content = await file.read()
return content
async def write_file_async(filename, data):
"""Асинхронная запись в файл"""
async with aiofiles.open(filename, 'w') as file:
await file.write(data)
async def main():
# Параллельное чтение нескольких файлов
tasks = [
read_file_async("file1.txt"),
read_file_async("file2.txt"),
read_file_async("file3.txt")
]
results = await asyncio.gather(*tasks)
for i, content in enumerate(results):
print(f"Файл {i+1}: {len(content)} символов")
# Запуск асинхронного кода
asyncio.run(main())
Лучшие практики
Рекомендации по работе с потоками I/O
- Всегда используйте контекстные менеджеры (
withstatement) для автоматического закрытия ресурсов - Указывайте кодировку при работе с текстовыми файлами
- Обрабатывайте исключения для предотвращения аварийного завершения программы
- Используйте буферизацию для улучшения производительности при работе с большими объемами данных
- Применяйте асинхронный I/O для операций с высокой задержкой
Пример комплексного использования
import sys
import logging
from pathlib import Path
# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('app.log'),
logging.StreamHandler(sys.stdout)
]
)
def process_files(input_dir, output_dir):
"""Обработка файлов с логированием"""
input_path = Path(input_dir)
output_path = Path(output_dir)
if not input_path.exists():
logging.error(f"Входная директория не существует: {input_dir}")
return
output_path.mkdir(exist_ok=True)
for file_path in input_path.glob("*.txt"):
try:
with open(file_path, 'r', encoding='utf-8') as infile:
content = infile.read()
processed_content = content.upper()
output_file = output_path / f"processed_{file_path.name}"
with open(output_file, 'w', encoding='utf-8') as outfile:
outfile.write(processed_content)
logging.info(f"Обработан файл: {file_path.name}")
except Exception as e:
logging.error(f"Ошибка при обработке {file_path.name}: {e}")
if __name__ == "__main__":
process_files("input", "output")
Работа с потоками ввода-вывода в Python предоставляет мощные возможности для создания эффективных приложений, способных обрабатывать данные из различных источников. Правильное понимание и использование этих механизмов является ключом к созданию надежного и производительного кода.