Стандартная библиотека

Стандартная библиотека Python

Маршрут: Инженерная основа · тема 6 из 8

Навыки: J5, M3

До урока: basic_types, functions, exceptions, iterators_generators.

Результат: выбор устойчивого решения стандартной библиотеки вместо лишней зависимости или самописного велосипеда.

Основа: ключевые модули и типовые приёмы. Работа в эксплуатации: жизненный цикл, переносимость и разбор альтернатив. Углубление: subprocess, потоки и правила конкретной платформы/версии.

Подтверждение: одна задача средствами стандартной библиотеки и письменное сравнение вариантов.

Стандартная библиотека закрывает многие повседневные задачи без внешних зависимостей. Важно не запомнить все модули, а уметь выбрать подходящий инструмент и проверить его контракт на нужной версии Python.

#1. `collections` — специализированные контейнеры

#`defaultdict` — словарь с дефолтным значением

from collections import defaultdict

# Без defaultdict: KeyError или громоздкий setdefault
d = {}
d.setdefault("a", []).append(1)

# С defaultdict: чисто
d = defaultdict(list)
d["a"].append(1)
d["a"].append(2)
d["b"].append(3)
print(dict(d))   # {"a": [1, 2], "b": [3]}

# Граф (список смежности)
graph = defaultdict(set)
graph["Moscow"].add("Saint Petersburg")
graph["Moscow"].add("Kazan")

# Подсчёт
word_count = defaultdict(int)
for word in text.split():
    word_count[word] += 1

# defaultdict(lambda: "unknown") — произвольный дефолт
d = defaultdict(lambda: "N/A")
print(d["missing"])   # "N/A"

#`Counter` — подсчёт элементов

from collections import Counter

# Создание
c = Counter("abracadabra")
c = Counter(["a", "b", "a", "c", "a"])
c = Counter({"a": 5, "b": 2})

print(c["a"])       # 5
print(c["z"])       # 0 (не KeyError!)

# Топ N элементов
print(c.most_common(3))   # [("a", 5), ("b", 2), ("r", 2)]

# Арифметика
c1 = Counter(a=3, b=2)
c2 = Counter(a=1, b=4, c=1)
print(c1 + c2)     # Counter(b=6, a=4, c=1)
print(c1 - c2)     # Counter(a=2) — только положительные
print(c1 & c2)     # Counter(a=1, b=2) — минимум
print(c1 | c2)     # Counter(b=4, a=3, c=1) — максимум

# total() — Python 3.10+
print(c.total())   # сумма всех значений

# elements() — итерация с повторами
list(Counter("aab").elements())   # ["a", "a", "b"]

#`deque` — двусторонняя очередь

from collections import deque

d = deque([1, 2, 3], maxlen=5)

# Добавление
d.append(4)         # добавить справа
d.appendleft(0)     # добавить слева
d.extend([5, 6])    # добавить несколько справа
d.extendleft([-1])  # добавить слева (порядок обратный!)

# Удаление
d.pop()             # O(1) — удалить справа
d.popleft()         # O(1) — удалить слева

# Список: pop(0) это O(n), deque.popleft() это O(1)!

# rotate
d = deque([1, 2, 3, 4, 5])
d.rotate(2)    # [4, 5, 1, 2, 3] — вправо
d.rotate(-2)   # [1, 2, 3, 4, 5] — влево

# maxlen — автоматически выбрасывает старые элементы
history = deque(maxlen=100)
for event in events:
    history.append(event)   # старые события удаляются автоматически

#`namedtuple` — именованный кортеж

from collections import namedtuple
from typing import NamedTuple

# Старый стиль
Point = namedtuple("Point", ["x", "y", "z"])
p = Point(1.0, 2.0, z=3.0)
print(p.x, p[0])   # 1.0 1.0 — и атрибут, и индекс
x, y, z = p        # распаковка как tuple
print(p._asdict())  # OrderedDict([('x', 1.0), ...])

# Новый стиль (предпочтительный — типизированный)
class Coordinate(NamedTuple):
    x: float
    y: float
    label: str = ""

c = Coordinate(1.0, 2.0, "origin")
print(c._replace(x=5.0))   # создать копию с изменёнными полями

# Наследование от NamedTuple
class Point3D(NamedTuple):
    x: float
    y: float
    z: float = 0.0

    def distance_to_origin(self) -> float:
        return (self.x**2 + self.y**2 + self.z**2) ** 0.5

#`ChainMap` — цепочка словарей

from collections import ChainMap

# Поиск идёт по очереди в каждом словаре
defaults = {"color": "blue", "size": "medium"}
config = {"color": "red"}
env = {"debug": True}

merged = ChainMap(env, config, defaults)
print(merged["color"])   # "red" — из config (первый найденный)
print(merged["size"])    # "medium" — из defaults
print(merged["debug"])   # True — из env

# Запись идёт только в первый словарь
merged["new_key"] = "value"
print(env)   # {"debug": True, "new_key": "value"}

# Полезно для конфигурации: env vars > config file > defaults

#2. `itertools` — эффективные итераторы

import itertools

#Бесконечные итераторы

# count — бесконечный счётчик
for n in itertools.count(1, 2):   # 1, 3, 5, 7...
    if n > 10:
        break

# cycle — бесконечный цикл по итерируемому
colors = itertools.cycle(["red", "green", "blue"])
for i, color in zip(range(9), colors):
    print(f"{i}: {color}")   # 0:red, 1:green, 2:blue, 3:red...

# repeat — повторение элемента
list(itertools.repeat(0, 5))   # [0, 0, 0, 0, 0]
# Полезно с map: map(lambda x, y: x**y, values, repeat(2))

#Комбинаторные итераторы

# product — декартово произведение
list(itertools.product([1, 2], ["a", "b"]))
# [(1,'a'), (1,'b'), (2,'a'), (2,'b')]

# permutations — перестановки (порядок важен)
list(itertools.permutations("ABC", 2))
# [('A','B'), ('A','C'), ('B','A'), ('B','C'), ('C','A'), ('C','B')]

# combinations — комбинации (порядок не важен, без повторов)
list(itertools.combinations([1, 2, 3, 4], 2))
# [(1,2), (1,3), (1,4), (2,3), (2,4), (3,4)]

# combinations_with_replacement — с повторами
list(itertools.combinations_with_replacement("AB", 2))
# [('A','A'), ('A','B'), ('B','B')]

#Инструменты цепочек

# chain — объединение итераторов
list(itertools.chain([1, 2], [3, 4], [5]))   # [1, 2, 3, 4, 5]

# chain.from_iterable — "flatten" на один уровень
nested = [[1, 2], [3, 4], [5, 6]]
list(itertools.chain.from_iterable(nested))  # [1, 2, 3, 4, 5, 6]

# islice — срез итератора (не загружает всё в память)
gen = (x**2 for x in itertools.count(1))
list(itertools.islice(gen, 5))   # [1, 4, 9, 16, 25]
list(itertools.islice(gen, 2, 7, 2))  # start=2, stop=7, step=2

# takewhile / dropwhile
nums = [1, 2, 3, 4, 5, 1, 2]
list(itertools.takewhile(lambda x: x < 4, nums))  # [1, 2, 3]
list(itertools.dropwhile(lambda x: x < 4, nums))  # [4, 5, 1, 2]

# groupby — группировка по ключу (входные данные должны быть отсортированы!)
data = sorted([{"name": "Alice", "dept": "IT"},
               {"name": "Bob", "dept": "HR"},
               {"name": "Charlie", "dept": "IT"}], key=lambda x: x["dept"])

for dept, group in itertools.groupby(data, key=lambda x: x["dept"]):
    print(dept, list(group))
# IT   [Alice, Charlie]
# HR   [Bob]

#`accumulate` — накопленные значения

import itertools, operator

# Накопленная сумма
list(itertools.accumulate([1, 2, 3, 4, 5]))
# [1, 3, 6, 10, 15]

# Накопленное произведение
list(itertools.accumulate([1, 2, 3, 4, 5], operator.mul))
# [1, 2, 6, 24, 120]

# Скользящий максимум
list(itertools.accumulate([3, 1, 4, 1, 5, 9, 2, 6], max))
# [3, 3, 4, 4, 5, 9, 9, 9]

#3. `functools`

from functools import (
    lru_cache, cache, cached_property,
    partial, partialmethod,
    reduce, wraps,
    singledispatch, total_ordering
)

#`total_ordering` — автогенерация операторов сравнения

from functools import total_ordering

@total_ordering
class Version:
    def __init__(self, major, minor, patch):
        self.major, self.minor, self.patch = major, minor, patch

    def __eq__(self, other):
        return (self.major, self.minor, self.patch) == \
               (other.major, other.minor, other.patch)

    def __lt__(self, other):
        return (self.major, self.minor, self.patch) < \
               (other.major, other.minor, other.patch)
    # Автоматически генерируются: __le__, __gt__, __ge__

v1 = Version(1, 0, 0)
v2 = Version(2, 0, 0)
print(v1 < v2)   # True
print(v2 > v1)   # True
print(v1 <= v1)  # True

#`lru_cache`, `cache` и `partial`

lru_cache хранит результаты вызовов по ключу из аргументов. Поэтому каждый аргумент, входящий в ключ, должен быть хэшируемым. cache делает то же без ограничения размера; у lru_cache(maxsize=N) старые записи вытесняются.

from functools import lru_cache, partial

@lru_cache(maxsize=256)
def load_user(user_id: int) -> dict:
    return read_user_from_database(user_id)

def greet(name: str, greeting: str) -> str:
    return f"{greeting}, {name}"

hello = partial(greet, greeting="Hello")
assert hello("Ada") == "Hello, Ada"

Объект partial вызываем, но это не обычная функция. Например, у него нет автоматического атрибута __name__; обращение к hello.__name__ поднимет AttributeError. Если метаданные нужны инструментам, перенесите их явно через functools.update_wrapper.

#`reduce`

from functools import reduce
import operator

# Свёртка: (((1+2)+3)+4) = 10
result = reduce(operator.add, [1, 2, 3, 4], 0)

# flatten через reduce + operator.add
nested = [[1, 2], [3, 4], [5]]
flat = reduce(operator.add, nested, [])   # [1, 2, 3, 4, 5]
# Но лучше: list(itertools.chain.from_iterable(nested))

# Вычисление факториала
factorial = reduce(operator.mul, range(1, 6), 1)   # 120

#4. `datetime` — дата и время

from datetime import datetime, date, time, timedelta, timezone

# Создание
now = datetime.now()                          # локальное время
utc_now = datetime.now(timezone.utc)          # UTC время (правильно для хранения!)
d = date(2024, 12, 31)
t = time(23, 59, 59)
dt = datetime(2024, 12, 31, 23, 59, 59, tzinfo=timezone.utc)

# Форматирование
print(now.strftime("%Y-%m-%d %H:%M:%S"))      # 2024-03-01 15:30:45
print(now.isoformat())                         # 2024-03-01T15:30:45.123456

# Парсинг
dt = datetime.strptime("2024-03-01 15:30", "%Y-%m-%d %H:%M")
dt = datetime.fromisoformat("2024-03-01T15:30:45")

# Арифметика
tomorrow = date.today() + timedelta(days=1)
next_week = datetime.now() + timedelta(weeks=1)
diff = datetime(2024, 12, 31) - datetime.now()
print(diff.days, diff.seconds)

# Unix timestamp
ts = datetime.now().timestamp()
dt = datetime.fromtimestamp(ts)
dt_utc = datetime.fromtimestamp(ts, tz=timezone.utc)

# Сравнение (только одинаковые tzinfo!)
aware = datetime.now(timezone.utc)
naive = datetime.now()
# aware > naive  # TypeError! Нельзя сравнивать aware и naive

Для часовых поясов из базы IANA используйте zoneinfo.ZoneInfo. Для измерения длительности подходит time.monotonic() или time.perf_counter(), а не календарное datetime.now(): системные часы могут быть переведены.

import time
from datetime import datetime
from zoneinfo import ZoneInfo

moscow_now = datetime.now(ZoneInfo("Europe/Moscow"))
started = time.perf_counter()
run_operation()
elapsed = time.perf_counter() - started

#5. `re` — регулярные выражения

import re

text = "Hello, my email is alice@example.com and phone is +7(495)123-45-67"

# Поиск первого совпадения
match = re.search(r"\d+", text)
if match:
    print(match.group())  # "7"
    print(match.start(), match.end())  # позиции

# Все совпадения
emails = re.findall(r"[\w.+-]+@[\w-]+\.[a-zA-Z]{2,}", text)
# ['alice@example.com']

# Замена
clean = re.sub(r"\s+", " ", "  many   spaces  ")   # "many spaces"
clean = re.sub(r"(\w+)@(\w+)", r"\2@\1", text)     # группы в замене

# Компиляция (для многократного использования)
EMAIL_RE = re.compile(r"^[\w.+-]+@[\w-]+\.[a-zA-Z]{2,}$", re.IGNORECASE)

def is_valid_email(email: str) -> bool:
    return bool(EMAIL_RE.match(email))

# Именованные группы
DATE_RE = re.compile(r"(?P<year>\d{4})-(?P<month>\d{2})-(?P<day>\d{2})")
m = DATE_RE.match("2024-03-15")
if m:
    print(m.group("year"), m.group("month"))   # 2024 03
    print(m.groupdict())   # {'year': '2024', 'month': '03', 'day': '15'}

# split по регулярке
parts = re.split(r"[,;\s]+", "one, two; three  four")
# ['one', 'two', 'three', 'four']

# Флаги
re.IGNORECASE   # (re.I) — без учёта регистра
re.MULTILINE    # (re.M) — ^ и $ для каждой строки
re.DOTALL       # (re.S) — . совпадает с \n
re.VERBOSE      # (re.X) — игнорирует пробелы, разрешает комментарии

VERBOSE_RE = re.compile(r"""
    (\d{1,3})   # первая группа цифр
    \.          # точка
    (\d{1,3})   # вторая группа
""", re.VERBOSE)

#6. `heapq` и `bisect` — эффективные структуры

import heapq

# Min-heap (по умолчанию)
heap = [5, 3, 1, 4, 2]
heapq.heapify(heap)       # преобразовать список в heap inplace: [1, 2, ...]

heapq.heappush(heap, 0)   # добавить элемент
smallest = heapq.heappop(heap)  # O(log n) — извлечь минимум

# N наименьших/наибольших
data = [10, 5, 2, 8, 1, 9, 3]
print(heapq.nsmallest(3, data))   # [1, 2, 3]
print(heapq.nlargest(3, data))    # [10, 9, 8]

# Объединение отсортированных итераторов
sorted_lists = [[1, 4, 7], [2, 5, 8], [3, 6, 9]]
merged = list(heapq.merge(*sorted_lists))   # [1, 2, 3, 4, 5, 6, 7, 8, 9]

import bisect

# Бинарный поиск в отсортированном списке
sorted_list = [1, 3, 5, 7, 9]
print(bisect.bisect_left(sorted_list, 5))   # 2 — позиция для вставки слева
print(bisect.bisect_right(sorted_list, 5))  # 3 — позиция для вставки справа

bisect.insort(sorted_list, 4)   # O(n) — вставить с сохранением порядка
print(sorted_list)   # [1, 3, 4, 5, 7, 9]

#7. `os` и `sys` — взаимодействие с системой

import os, sys

# Переменные окружения
api_key = os.environ.get("API_KEY", "default")
os.environ["NEW_VAR"] = "value"

# Файловая система
# Для новых проектов предпочитайте pathlib (см. раздел 8 ниже):
#   Path("a") / "b" / "c" вместо os.path.join("a", "b", "c")
#   Path(...).read_text() вместо open(...).read()
os.getcwd()                   # текущая директория
os.chdir("/tmp")              # изменить
os.makedirs("/tmp/a/b/c", exist_ok=True)  # создать дерево
os.listdir(".")               # список файлов
os.path.join("a", "b", "c")  # соединить путь

# Информация о системе
os.getpid()    # PID процесса
os.getppid()   # PID родительского процесса
os.cpu_count() # количество ядер

# sys — интерпретатор
sys.version                  # "3.12.0 (default, ...)"
sys.platform                 # "linux", "darwin", "win32"
sys.argv                     # аргументы командной строки
sys.path                     # пути поиска модулей
sys.exit(0)                  # завершить программу

sys.stdin, sys.stdout, sys.stderr  # стандартные потоки

# Перехват stdout
import io
old_stdout = sys.stdout
sys.stdout = io.StringIO()
print("captured")
output = sys.stdout.getvalue()
sys.stdout = old_stdout

#8. Пути, сериализация, CLI и журналирование

Эти модули часто встречаются вместе в небольших служебных программах: pathlib отвечает за пути, json — за переносимые данные, argparse — за интерфейс команды, а logging — за диагностические события.

#`pathlib.Path`

Path хранит путь как объект и предоставляет методы exists, read_text, write_text и оператор /. В отличие от os.path, здесь не нужно вручную передавать строку из одной функции в другую.

from pathlib import Path

report = Path("output") / "report.json"
report.parent.mkdir(parents=True, exist_ok=True)
report.write_text("{}\n", encoding="utf-8")

#`json` и `pickle`

JSON — текстовый переносимый формат с ограниченным набором типов. pickle сохраняет граф Python-объектов, но привязан к Python и может выполнить код при загрузке. Никогда не передавайте данные от недоверенного источника в pickle.loads() или pickle.load().

import json

payload = {"status": "ok", "count": 3}
encoded = json.dumps(payload, ensure_ascii=False)
decoded = json.loads(encoded)
assert decoded == payload

#`argparse`

import argparse
from pathlib import Path

parser = argparse.ArgumentParser(description="Подсчитать события")
parser.add_argument("input", type=Path)
parser.add_argument("--limit", type=int, default=100)
args = parser.parse_args()

#`logging`

Logger создаёт LogRecord, handler выбирает место вывода, formatter задаёт вид записи, а filter может пропустить или отклонить событие. Уровень logger и handler вместе определяют, попадёт ли запись в вывод.

import logging

handler = logging.StreamHandler()
handler.setLevel(logging.INFO)
handler.setFormatter(
    logging.Formatter("%(asctime)s %(levelname)s %(name)s %(message)s")
)

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
logger.addHandler(handler)
logger.info("Обработано записей: %s", 3)

#9. `subprocess` — запуск процессов

import subprocess

# Простой запуск (Python 3.7+)
result = subprocess.run(
    ["ls", "-la"],
    capture_output=True,    # захватить stdout и stderr
    text=True,              # декодировать как строку
    check=True,             # поднять ошибку если returncode != 0
)
print(result.stdout)
print(result.returncode)

# Передача данных через stdin
result = subprocess.run(
    ["grep", "error"],
    input="line1\nerror found\nline3\n",
    capture_output=True,
    text=True,
)
print(result.stdout)   # "error found\n"

# shell=True (осторожно — риск инъекции!)
# subprocess.run(f"cat {filename}", shell=True)  # Проблема: если filename от пользователя
result = subprocess.run(["cat", filename], ...)  # Вариант: список аргументов безопасен

# Popen для асинхронного запуска
proc = subprocess.Popen(
    ["long_running_program"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
)
# Делаем другую работу...
stdout, stderr = proc.communicate(timeout=30)  # ждём завершения

#10. `threading` и `multiprocessing`

import threading
import multiprocessing
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

# threading — для I/O-bound задач
def io_task(url):
    import requests
    return requests.get(url).status_code

urls = ["https://example.com"] * 10

# ThreadPoolExecutor (рекомендуется вместо Thread напрямую)
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(io_task, url) for url in urls]
    results = [f.result() for f in futures]
    # Или через map:
    results = list(executor.map(io_task, urls))

# multiprocessing — для CPU-bound задач
def cpu_task(n):
    return sum(i**2 for i in range(n))

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(cpu_task, [10**6] * 4))

# Lock для синхронизации
lock = threading.Lock()
data = []

def safe_append(item):
    with lock:
        data.append(item)

# Event для сигнализации
ready = threading.Event()

def producer():
    time.sleep(1)
    ready.set()

def consumer():
    ready.wait()   # блокируется до set()
    process()

#11. `copy` — копирование объектов

import copy

original = [[1, 2], [3, 4], [5, 6]]

# Поверхностная копия: новый список, но элементы — те же объекты
shallow = copy.copy(original)
shallow.append([7, 8])
shallow[0].append(99)   # Изменяет original[0]!
print(original)   # [[1, 2, 99], [3, 4], [5, 6]]

# Глубокая копия: рекурсивно копирует всё
deep = copy.deepcopy(original)
deep[0].append(100)   # Не изменяет original
print(original)   # [[1, 2, 99], [3, 4], [5, 6]] — не изменилось

# Кастомизация deepcopy через __deepcopy__
class MyClass:
    def __deepcopy__(self, memo):
        # memo — словарь уже скопированных объектов (избегает бесконечной рекурсии)
        new = self.__class__.__new__(self.__class__)
        memo[id(self)] = new
        for k, v in self.__dict__.items():
            setattr(new, k, copy.deepcopy(v, memo))
        return new

#12. `operator` — функциональные операции

import operator

# Вместо lambda: немного быстрее, более читаемо
operator.add(2, 3)      # 5
operator.mul(4, 5)      # 20
operator.neg(3)         # -3
operator.gt(5, 3)       # True
operator.not_(False)    # True
operator.contains([1,2,3], 2)  # True

# itemgetter — извлечение по индексу/ключу
from operator import itemgetter, attrgetter, methodcaller

data = [("Alice", 30, "IT"), ("Bob", 25, "HR"), ("Charlie", 35, "IT")]
by_age = sorted(data, key=itemgetter(1))
first_and_third = itemgetter(0, 2)(("Alice", 30, "IT"))  # ("Alice", "IT")

# attrgetter — извлечение по атрибуту
class Employee:
    def __init__(self, name, dept):
        self.name = name
        self.dept = dept

employees = [Employee("Alice", "IT"), Employee("Bob", "HR")]
by_dept = sorted(employees, key=attrgetter("dept"))

# Вложенные атрибуты
# attrgetter("dept.manager.name")(emp) == emp.dept.manager.name

# methodcaller — вызов метода
transform = methodcaller("upper")
print(list(map(transform, ["hello", "world"])))  # ["HELLO", "WORLD"]

with_args = methodcaller("replace", " ", "_")
print(with_args("hello world"))  # "hello_world"

#Практика: от пробы к рабочему решению

#Шаг 1. Соберите минимальный пример

Перед добавлением зависимости проверьте стандартную библиотеку и её контракт. Результат основы — решение задачи средствами stdlib с тестом на граничный случай и письменным сравнением альтернатив.

#Шаг 2. Проверьте поведение

from collections import Counter, deque

events = deque(["ok", "retry", "ok", "failed"], maxlen=4)
counts = Counter(events)
assert counts == {"ok": 2, "retry": 1, "failed": 1}

events.append("ok")
assert list(events) == ["retry", "ok", "failed", "ok"]
assert events.maxlen == 4

#Шаг 3. Доведите решение до рабочего сценария

Практика с подсказками: соберите ограниченный журнал последних событий и сводные счётчики. Проверьте переполнение, сериализацию и неизвестный статус; сравните deque/Counter с самописными структурами по контракту, а не по числу строк.

#Шаг 4. Объясните внутренний механизм

Исследуйте subprocess на безопасной фиксированной команде без shell=True. Проверьте срок ожидания, код завершения, кодировку и ограничение объёма вывода; зафиксируйте различия платформ, которые нельзя считать гарантией Python.

#Упражнения

Используя itertools.groupby, сгруппируйте список транзакций по дате (строка вида "2024-01-15"). Убедитесь в правильной сортировке.
Реализуйте top_words(text, n=10) через Counter.most_common() — топ N слов текста.
Напишите генератор sliding_window(iterable, size) возвращающий скользящее окно через collections.deque.
Используя heapq.nsmallest и heapq.nlargest, найдите k ближайших точек к началу координат из списка (x, y) пар.
Напишите функцию deep_merge(d1, d2) рекурсивно объединяющую два dict (значения d2 перезаписывают d1, вложенные dict объединяются, не перезаписываются).
Используя re, напишите парсер лог-строк формата: "2024-01-15 10:30:45 ERROR module: message".

#Самопроверка

groupby получает данные, отсортированные по тому же ключу; две транзакции одной даты не должны оказаться в разных группах.
top_words заранее определяет регистр, пунктуацию и правило равенства частот.
sliding_window явно обрабатывает size <= 0, короткий и бесконечный вход и возвращает независимые снимки окна.
Для heapq сравните результат с полной сортировкой и определите порядок при одинаковом расстоянии.
deep_merge не меняет входы, а политика для list и конфликта разных типов записана тестами.
Парсер лога отклоняет лишний текст, проверяет дату через datetime и сохраняет двоеточия внутри сообщения.

#Практическая лаборатория

Закрепите тему в лаборатории «Атомарная публикация JSONL-снимка». Вы свяжете tempfile, json, hashlib и os.replace в потоковый контракт, который сохраняет предыдущий файл после ошибки данных или файловой системы.

git clone --branch v0.6.2 --depth 1 https://gitlab.potapov.me/courses/python-labs.git
cd python-labs
uv sync --group test
uv run --group test pytest stdlib/tests