Когда запускаешь pytest --cov код выполняется как обычно, но в конце появляется отчёт о покрытии. Как pytest узнаёт, какие строки выполнились? Ответ в sys.settrace, это низкоуровневый хук, который позволяет перехватывать каждый шаг интерпретатора.
На этом механизме построены coverage.py, pdb, PyCharm debugger, hunters, и десятки других инструментов. Разберём, как это работает изнутри и почему трассировка устроена именно так.
Модель выполнения CPython
Прежде чем говорить о трассировке, нужно понять, как CPython выполняет код. Python-код компилируется в байткод — последовательность инструкций для виртуальной машины. Каждая функция имеет объект code, содержащий байткод и метаданные.
Когда функция вызывается, создаётся frame — объект, хранящий состояние выполнения: локальные переменные, текущую позицию в байткоде, ссылку на код. Frames образуют стек вызовов.
Интерпретатор — это цикл, который берёт следующую инструкцию из текущего frame и выполняет её. Трассировка встраивается в этот цикл: перед выполнением определённых инструкций интерпретатор проверяет, установлена ли trace-функция, и если да — вызывает её.
Базовый API: sys.settrace
sys.settrace(func) устанавливает глобальную trace-функцию. Она вызывается при определённых событиях:
import sys def trace_calls(frame, event, arg): """ frame — текущий frame выполнения event — тип события ('call', 'line', 'return', 'exception') arg — дополнительные данные (зависит от event) """ print(f"{event}: {frame.f_code.co_name}, line {frame.f_lineno}") return trace_calls # Возвращаем trace-функцию для локальных событий sys.settrace(trace_calls) def example(): x = 1 y = 2 return x + y example() sys.settrace(None)
Вывод:
call: example, line 1 line: example, line 2 line: example, line 3 line: example, line 4 return: example, line 4
Четыре типа событий:
call — вызов функции,
arg= Noneline — переход на новую строку,
arg= Nonereturn — возврат из функции,
arg= возвращаемое значениеexception — возникло исключение,
arg= (type, value, traceback)
trace-функция должна вернуть функцию для отслеживания локальных событий (line, return, exception). Если вернуть None,то локальные события этого frame игнорируются. Это механизм фильтрации: можно отслеживать вызовы всех функций, но line-события только для интересующих.
Анатомия frame-объекта
Frame — это окно в состояние выполнения. Доступные атрибуты:
def inspect_frame(frame, event, arg): # Информация о коде code = frame.f_code print(f"Function: {code.co_name}") print(f"Filename: {code.co_filename}") print(f"First line: {code.co_firstlineno}") # Текущее состояние print(f"Current line: {frame.f_lineno}") print(f"Locals: {frame.f_locals}") # Стек вызовов print(f"Caller frame: {frame.f_back}") return inspect_frame
f_locals — словарь локальных переменных. Но есть нюанс, это снапшот на момент вызова trace-функции. Изменения в f_locals не отражаются на реальных переменных (кроме специальных случаев с ctypes). Для дебаггеров это проблема,нельзя просто так изменить переменную во время отладки.
f_back — ссылка на frame вызывающей функции. Позволяет ходить по стеку вызовов.
Как работает coverage.py
Coverage.py — стандартный инструмент измерения покрытия кода. Его архитектура: trace-функция отмечает выполненные строки, потом сравниваем с исходным кодом.
Упрощённая версия:
class SimpleCoverage: def __init__(self): self.executed_lines = {} # filename -> set of line numbers def start(self): sys.settrace(self._trace) def stop(self): sys.settrace(None) def _trace(self, frame, event, arg): if event == 'call': # Для каждого вызова возвращаем локальный tracer return self._trace_lines return None def _trace_lines(self, frame, event, arg): if event == 'line': filename = frame.f_code.co_filename lineno = frame.f_lineno if filename not in self.executed_lines: self.executed_lines[filename] = set() self.executed_lines[filename].add(lineno) return self._trace_lines def report(self): for filename, lines in self.executed_lines.items(): print(f"{filename}: {sorted(lines)}")
Реальный coverage.py сложнее, он написан на C для производительности, отслеживает арки (transitions между строками) для branch coverage, интегрируется с multiprocessing.
Кстати, тут еще интересно, что coverage должен знать, какие строки вообще являются исполняемыми. Пустые строки, комментарии, строки с только else: не являются. Для этого coverage.py парсит AST исходного кода и определяет «исполняемые» строки.
Дебаггеры
pdb использует тот же механизм, но с другой логикой:
class SimpleDebugger: def __init__(self): self.breakpoints = set() # (filename, lineno) self.step_mode = False def set_break(self, filename, lineno): self.breakpoints.add((filename, lineno)) def _trace(self, frame, event, arg): if event == 'call': return self._trace_lines return None def _trace_lines(self, frame, event, arg): if event != 'line': return self._trace_lines filename = frame.f_code.co_filename lineno = frame.f_lineno should_stop = ( self.step_mode or (filename, lineno) in self.breakpoints ) if should_stop: self._interact(frame) return self._trace_lines def _interact(self, frame): """REPL для отладки""" while True: cmd = input(f"(Pdb) {frame.f_code.co_name}:{frame.f_lineno}> ") if cmd == 'n': # next self.step_mode = True break elif cmd == 'c': # continue self.step_mode = False break elif cmd == 'p': # print locals print(frame.f_locals) elif cmd == 'q': # quit sys.exit(0)
Команда step (войти в функцию) vs next (перешагнуть) реализуется через анализ событий: при step мы реагируем на следующее событие, при next игнорируем вложенные call/return и ждём line в текущем frame.
Условные breakpoints — просто проверка условия при остановке:
# breakpoint с условием x > 10 if eval(condition, frame.f_globals, frame.f_locals): self._interact(frame)
Почему трассировка медленная
Trace-функция вызывается на каждую строку кода. Все это беда для производительности. Почему так:
Вызов Python-функции — сам по себе дорогой (создание frame, setup arguments)
Переход Python ↔ C — trace-функция на Python, интерпретатор на C
Невозможность оптимизаций — трассировка отключает многие оптимизации CPython
Поэтому coverage.py имеет C-расширение, а профилировщики используют другие механизмы.
sys.setprofile: легковесная альтернатива
sys.setprofile похож на settrace, но события только на границах функций:
call
return
c_call (вызов C-функции)
c_return
c_exception
Нет событий line, огромная экономия. cProfile построен на этом:
import sys call_counts = {} def profiler(frame, event, arg): if event == 'call': name = frame.f_code.co_name call_counts[name] = call_counts.get(name, 0) + 1 return profiler sys.setprofile(profiler) # ... код ... sys.setprofile(None)
Для профилирования по времени можно использовать time.perf_counter() на call/return и считать разницу.
Per-thread трассировка
sys.settrace устанавливает trace-функцию для текущего потока. Для многопоточных приложений нужно устанавливать в каждом потоке:
import threading import sys def install_tracer(): sys.settrace(my_trace_func) # Устанавливаем в главном потоке sys.settrace(my_trace_func) # Для новых потоков threading.settrace(install_tracer) # deprecated в 3.12 # Или вручную original_start = threading.Thread.start def patched_start(self): original_run = self.run def traced_run(): sys.settrace(my_trace_func) original_run() self.run = traced_run original_start(self) threading.Thread.start = patched_start
coverage.py делает подобный monkey-patching для отслеживания всех потоков.
Python 3.12+: sys.monitoring
Python 3.12 добавил sys.monitoring — новый, более эффективный API для инструментации:
import sys def line_handler(code, line_number): print(f"Line {line_number} in {code.co_name}") # Регистрируем callback sys.monitoring.register_callback( sys.monitoring.PROFILER_ID, sys.monitoring.events.LINE, line_handler ) # Включаем события sys.monitoring.set_events( sys.monitoring.PROFILER_ID, sys.monitoring.events.LINE )
Меньше overhead — callbacks вызываются напрямую, без создания frame, плюсом имеем гранулярный контроль, можно включать события для конкретных функций.
API ещё стабилизируется.
Детектор утечек
Пример нетипичного применения — отслеживание незакрытых файлов:
import sys import weakref import traceback open_files = {} # id -> (weakref, traceback) original_open = open def tracked_open(*args, **kwargs): f = original_open(*args, **kwargs) # Сохраняем traceback места открытия tb = ''.join(traceback.format_stack()) open_files[id(f)] = (weakref.ref(f), tb) return f def trace_closes(frame, event, arg): if event == 'return': # Проверяем, не закрылся ли файл for fid, (ref, tb) in list(open_files.items()): if ref() is None: del open_files[fid] return trace_closes # Установка import builtins builtins.open = tracked_open sys.settrace(trace_closes) # В конце программы def report_leaks(): for fid, (ref, tb) in open_files.items(): f = ref() if f and not f.closed: print(f"Unclosed file: {f.name}") print(f"Opened at:\n{tb}")
В проде использовали бы del или context managers.
sys.settrace — низкоуровневый механизм. Для большинства задач достаточно готовых инструментов.
Хотите разбираться в Python не только на уровне «пишу код», но и понимать, что делает интерпретатор, где теряется время и как устроены инструменты вроде coverage и debugger’ов? Специализация Python Developer помогает пройти путь от базы до уверенного middle: практика, код-ревью, проекты и инженерные п��ивычки для продакшена.
Для знакомства с форматом обучения и экспертами приходите на бесплатные демо-уроки:
9 февраля, 20:00. «Telegram-бот за 60 минут: получение данных из любого API и ответ пользователю». Записаться
17 февраля, 20:00. «Запуск Python-приложения в Docker: FastAPI и база данных». Записаться
Больше открытых уроков смотрите в посте.
