Продвинутый Python

Маршрут: Устройство среды выполнения · тема 2 из 3

Навыки: S1, S2, S5

До урока: oop, iterators_generators, closures_decorators, context_managers, type_system.

Результат: безопасное исследование модели данных, AST и границ процессов.

Основа: расширенные протоколы и интроспекция. Работа в эксплуатации: инструменты AST, многопроцессность и контролируемые динамические возможности. Углубление: перехват поиска атрибутов, преобразование кода и границы нативного кода/среды выполнения.

Подтверждение: проба атрибутов/AST/процесса с ограничениями и технической запиской.

Урок связывает атрибутный протокол, AST и процессы с публичными механизмами Python. Внутренние детали CPython рассматриваются только там, где явно указана версия и граница переносимости.

#1. Атрибутный протокол — вся цепочка

#`getattr` vs `getattribute`

# __getattribute__ вызывается ВСЕГДА при доступе к атрибуту
# __getattr__ вызывается ТОЛЬКО если атрибут не найден обычным путём

class DebugAccess:
    def __init__(self, data):
        # Используем super().__setattr__ чтобы избежать рекурсии
        super().__setattr__("_data", data)
        super().__setattr__("_access_log", [])

    def __getattribute__(self, name):
        # Вызывается для ЛЮБОГО доступа к атрибуту
        if not name.startswith("_"):
            log = super().__getattribute__("_access_log")
            log.append(f"GET {name}")
        return super().__getattribute__(name)

    def __getattr__(self, name):
        # Вызывается только если атрибут не найден через __getattribute__
        return f"<missing: {name}>"

obj = DebugAccess({"x": 1})
print(obj.x)        # "<missing: x>" + залогировано "GET x"
print(obj._data)    # {"x": 1} — через __getattribute__
print(obj.missing)  # "<missing: missing>"

#`setattr` и `delattr`

class ValidatedObject:
    _VALIDATORS = {}   # имя поля → функция валидации

    def __setattr__(self, name, value):
        if name in self._VALIDATORS:
            self._VALIDATORS[name](value)
        super().__setattr__(name, value)

    def __delattr__(self, name):
        if name in {"id", "created_at"}:
            raise AttributeError(f"Cannot delete '{name}'")
        super().__delattr__(name)

class User(ValidatedObject):
    _VALIDATORS = {
        "age": lambda v: None if 0 <= v <= 150 else (_ for _ in ()).throw(ValueError(f"Invalid age: {v}")),
        "email": lambda v: None if "@" in v else (_ for _ in ()).throw(ValueError("Invalid email")),
    }

u = User()
u.name = "Alice"
u.age = 25          # OK
u.age = -5          # ValueError

#Порядок поиска атрибутов

# Полный порядок разрешения obj.attr:
# 1. type(obj).__mro__ — ищем data descriptor (имеет __set__)
# 2. obj.__dict__       — ищем в instance dict
# 3. type(obj).__mro__ — ищем non-data descriptor или class var
# 4. __getattr__        — если ничего не нашли

class DataDesc:
    def __get__(self, obj, t): return "data-desc"
    def __set__(self, obj, v): pass   # наличие __set__ делает data descriptor

class NonDataDesc:
    def __get__(self, obj, t): return "non-data-desc"

class MyClass:
    x = DataDesc()
    y = NonDataDesc()

obj = MyClass()
obj.__dict__["x"] = "instance"    # не поможет — data desc перехватит
obj.__dict__["y"] = "instance"    # поможет — non-data desc проигрывает instance dict

print(obj.x)   # "data-desc"    — data descriptor победил
print(obj.y)   # "instance"     — instance dict победил

#2. Кастомные контейнеры

Python позволяет реализовать любую коллекцию через набор dunder-методов.

from typing import Iterator, Any

class SortedList:
    """Список, всегда остающийся отсортированным."""

    def __init__(self, items=None):
        self._data = sorted(items) if items else []

    # Длина
    def __len__(self) -> int:
        return len(self._data)

    # Доступ по индексу
    def __getitem__(self, index):
        return self._data[index]

    # Нельзя напрямую устанавливать — нарушит порядок
    def __setitem__(self, index, value):
        raise TypeError("Use add() method to maintain sort order")

    # Удаление
    def __delitem__(self, index):
        del self._data[index]

    # Итерация
    def __iter__(self) -> Iterator:
        return iter(self._data)

    # Обратная итерация
    def __reversed__(self) -> Iterator:
        return reversed(self._data)

    # Проверка вхождения
    def __contains__(self, item) -> bool:
        import bisect
        idx = bisect.bisect_left(self._data, item)
        return idx < len(self._data) and self._data[idx] == item

    # Добавление
    def add(self, item) -> None:
        import bisect
        bisect.insort(self._data, item)

    # Строковое представление
    def __repr__(self) -> str:
        return f"SortedList({self._data!r})"

sl = SortedList([3, 1, 4, 1, 5, 9, 2, 6])
print(sl)           # SortedList([1, 1, 2, 3, 4, 5, 6, 9])
print(5 in sl)      # True (O(log n) через bisect)
sl.add(7)
print(sl[4])        # 5
for x in sl:
    print(x)

#Полная таблица dunder-методов для контейнеров

Метод	Назначение
`__len__`	`len(obj)`
`__getitem__`	`obj[key]`
`__setitem__`	`obj[key] = value`
`__delitem__`	`del obj[key]`
`__contains__`	`item in obj`
`__iter__`	`for x in obj`
`__reversed__`	`reversed(obj)`
`__next__`	`next(obj)` — для итераторов
`__missing__`	`dict.__getitem__` при отсутствии ключа
`__length_hint__`	Подсказка о размере для оптимизаций

#3. Магические методы для операций

#Арифметика

class Vector:
    def __init__(self, x: float, y: float):
        self.x, self.y = x, y

    def __add__(self, other: "Vector") -> "Vector":
        return Vector(self.x + other.x, self.y + other.y)

    def __sub__(self, other: "Vector") -> "Vector":
        return Vector(self.x - other.x, self.y - other.y)

    def __mul__(self, scalar: float) -> "Vector":
        return Vector(self.x * scalar, self.y * scalar)

    def __rmul__(self, scalar: float) -> "Vector":
        return self.__mul__(scalar)   # scalar * Vector работает через __rmul__

    def __neg__(self) -> "Vector":
        return Vector(-self.x, -self.y)

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

    def __bool__(self) -> bool:
        return abs(self) != 0.0

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Vector):
            return NotImplemented
        return self.x == other.x and self.y == other.y

    def __repr__(self) -> str:
        return f"Vector({self.x}, {self.y})"

v1 = Vector(1, 2)
v2 = Vector(3, 4)
print(v1 + v2)    # Vector(4, 6)
print(3 * v1)     # Vector(3, 6) — __rmul__
print(abs(v2))    # 5.0
print(bool(Vector(0, 0)))  # False

#Операторы in-place

# __iadd__ — +=
# __isub__ — -=
# __imul__ — *=

class Counter:
    def __init__(self, value=0):
        self.value = value

    def __iadd__(self, n):
        self.value += n
        return self   # ВАЖНО: возвращать self!

c = Counter()
c += 5    # вызывает __iadd__, не создаёт новый объект

#4. `inspect` — интроспекция кода

import inspect

def greet(name: str, times: int = 1, *, prefix: str = "Hello") -> str:
    """Приветствует пользователя."""
    return f"{prefix}, {name}! " * times

# Информация о функции
print(inspect.signature(greet))          # (name: str, times: int = 1, *, prefix: str = 'Hello') -> str
print(inspect.getdoc(greet))             # "Приветствует пользователя."
print(inspect.getsource(greet))          # исходный код

# Параметры
sig = inspect.signature(greet)
for name, param in sig.parameters.items():
    print(f"{name}: kind={param.kind.name}, default={param.default}")

# Тип параметра:
# POSITIONAL_ONLY      — только позиционный (/)
# POSITIONAL_OR_KEYWORD — обычный
# VAR_POSITIONAL        — *args
# KEYWORD_ONLY          — после *
# VAR_KEYWORD           — **kwargs

# Проверки
inspect.isfunction(greet)    # True
inspect.ismethod(greet)      # False
inspect.isclass(int)         # True
inspect.ismodule(os)         # True
inspect.iscoroutinefunction(async_func)  # True

# Стек вызовов
def get_caller_info():
    frame = inspect.currentframe().f_back
    return {
        "function": frame.f_code.co_name,
        "file": frame.f_code.co_filename,
        "line": frame.f_lineno,
    }

#5. AST — абстрактное синтаксическое дерево

import ast

# Парсинг кода
code = """
def add(x: int, y: int) -> int:
    return x + y

result = add(1, 2)
"""
tree = ast.parse(code)
print(ast.dump(tree, indent=2))

# Обход дерева
class FunctionCollector(ast.NodeVisitor):
    def __init__(self):
        self.functions = []

    def visit_FunctionDef(self, node):
        self.functions.append({
            "name": node.name,
            "lineno": node.lineno,
            "args": [a.arg for a in node.args.args],
        })
        self.generic_visit(node)   # обойти дочерние узлы

collector = FunctionCollector()
collector.visit(tree)
print(collector.functions)   # [{"name": "add", "lineno": 2, "args": ["x", "y"]}]

# Трансформация: заменить все умножения на сложения
class MultToAdd(ast.NodeTransformer):
    def visit_BinOp(self, node):
        self.generic_visit(node)  # сначала обработать дочерние
        if isinstance(node.op, ast.Mult):
            node.op = ast.Add()
        return node

new_tree = MultToAdd().visit(ast.parse("x = 2 * 3"))
ast.fix_missing_locations(new_tree)  # заполнить позиции
code = compile(new_tree, "<string>", "exec")
exec(code)
# x = 2 + 3 = 5

#Безопасное вычисление выражений через AST

import ast
import operator

SAFE_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
    ast.USub: operator.neg,
}

def safe_eval(expr: str) -> float:
    """Безопасно вычисляет математическое выражение."""
    tree = ast.parse(expr, mode="eval")

    def eval_node(node):
        if isinstance(node, ast.Constant):
            if isinstance(node.value, (int, float)):
                return node.value
            raise ValueError(f"Unsupported constant: {node.value}")
        elif isinstance(node, ast.BinOp):
            left = eval_node(node.left)
            right = eval_node(node.right)
            op_func = SAFE_OPS.get(type(node.op))
            if op_func is None:
                raise ValueError(f"Unsupported operation: {type(node.op)}")
            return op_func(left, right)
        elif isinstance(node, ast.UnaryOp):
            operand = eval_node(node.operand)
            op_func = SAFE_OPS.get(type(node.op))
            return op_func(operand)
        else:
            raise ValueError(f"Unsupported node: {type(node)}")

    return eval_node(tree.body)

print(safe_eval("2 + 3 * 4"))    # 14.0
print(safe_eval("2 ** 10"))      # 1024.0
safe_eval("__import__('os')")    # ValueError!

#6. `multiprocessing` и shared memory

import multiprocessing
from multiprocessing import Process, Pool, Queue, Pipe, Value, Array
from multiprocessing import shared_memory
import ctypes

# Pool — пул рабочих процессов
def square(n):
    return n ** 2

if __name__ == "__main__":
    with Pool(processes=4) as pool:
        results = pool.map(square, range(10))
        # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

        # imap — ленивый map (не загружает всё в память)
        for result in pool.imap(square, range(1_000_000), chunksize=1000):
            pass

        # starmap — для функций с несколькими аргументами
        results = pool.starmap(pow, [(2, n) for n in range(5)])
        # [1, 2, 4, 8, 16]

# Value и Array — разделяемая память
def increment_shared(shared_val, n):
    for _ in range(n):
        with shared_val.get_lock():
            shared_val.value += 1

if __name__ == "__main__":
    counter = Value(ctypes.c_int, 0)
    processes = [Process(target=increment_shared, args=(counter, 100000))
                 for _ in range(4)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    print(counter.value)  # 400000 (с блокировкой)

# shared_memory (Python 3.8+) — эффективнее для больших данных
import numpy as np

shm = shared_memory.SharedMemory(create=True, size=10 * 4)  # 10 int32
arr = np.ndarray((10,), dtype=np.int32, buffer=shm.buf)
arr[:] = range(10)

# В другом процессе:
# shm2 = shared_memory.SharedMemory(name=shm.name)
# arr2 = np.ndarray((10,), dtype=np.int32, buffer=shm2.buf)

shm.close()
shm.unlink()   # обязательно освободить!

#7. `exec` и `eval` — динамическое выполнение

# eval — вычисляет выражение, возвращает значение
x = eval("2 + 3")         # 5
y = eval("[i**2 for i in range(5)]")  # [0, 1, 4, 9, 16]

# exec — выполняет произвольный код, возвращает None
exec("a = 42")
exec("def foo(): return 42")
exec("""
class DynamicClass:
    def method(self):
        return 'dynamic!'
""")

# Ограниченный контекст
safe_globals = {"__builtins__": {}}  # убираем встроенные функции!
exec("result = 2 + 2", safe_globals)
print(safe_globals["result"])  # 4

# Проблема: НИКОГДА с пользовательским вводом!
# exec(user_input)   # катастрофическая уязвимость
# eval(user_input)   # тоже опасно

# Динамическое создание функций (редко нужно, но бывает)
code = compile("def f(x): return x * 2", "<string>", "exec")
namespace = {}
exec(code, namespace)
f = namespace["f"]
print(f(21))   # 42

#8. Паттерны функционального программирования

from functools import reduce
import operator

# Composition — составные функции
def compose(*functions):
    """f(g(h(x))) — применяет функции справа налево."""
    def composed(x):
        for f in reversed(functions):
            x = f(x)
        return x
    return composed

double = lambda x: x * 2
add_one = lambda x: x + 1
square = lambda x: x ** 2

transform = compose(double, add_one, square)  # double(add_one(square(x)))
print(transform(3))  # double(add_one(9)) = double(10) = 20

# Pipeline — цепочка преобразований (слева направо)
def pipe(*functions):
    def piped(x):
        for f in functions:
            x = f(x)
        return x
    return piped

process = pipe(square, add_one, double)   # double(add_one(square(x)))

# Memoize (ручная реализация lru_cache)
def memoize(func):
    cache = {}
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    wrapper.cache = cache
    return wrapper

# Currying
def curry(func):
    """Превращает f(a, b, c) в f(a)(b)(c)."""
    import inspect
    n = len(inspect.signature(func).parameters)

    def curried(*args):
        if len(args) >= n:
            return func(*args[:n])
        return lambda *more: curried(*args, *more)

    return curried

@curry
def add3(a, b, c):
    return a + b + c

add_10 = add3(10)        # функция ожидающая b, c
add_10_20 = add_10(20)   # функция ожидающая c
print(add_10_20(5))      # 35

functools.partial() создаёт вызываемый объект с заранее зафиксированной частью аргументов. Модуль operator предоставляет функции add, mul, itemgetter и другие операции, когда callable нужно передать как значение.

from functools import partial
from operator import itemgetter

parse_binary = partial(int, base=2)
assert parse_binary("1010") == 10

rows = [{"name": "Ada", "score": 10}, {"name": "Lin", "score": 8}]
rows.sort(key=itemgetter("score"), reverse=True)

#Контекстный менеджер без отдельного класса

contextlib.contextmanager превращает генератор с одним yield в контекстный менеджер. Код до yield соответствует входу, код в finally — гарантированной очистке при обычном выходе и исключении.

from contextlib import contextmanager


@contextmanager
def transaction(connection):
    connection.begin()
    try:
        yield connection
    except Exception:
        connection.rollback()
        raise
    else:
        connection.commit()

#9. `__class_getitem__` — поддержка дженерик-синтаксиса

# Позволяет писать MyClass[int] как синтаксис

class TypedContainer:
    def __class_getitem__(cls, item):
        # item — тип в скобках (MyContainer[int] → item=int)
        return type(f"{cls.__name__}[{item.__name__}]", (cls,), {
            "_item_type": item
        })

    def validate(self, value):
        if hasattr(self, "_item_type") and not isinstance(value, self._item_type):
            raise TypeError(f"Expected {self._item_type}, got {type(value)}")

class Stack(TypedContainer):
    def __init__(self):
        self._items = []

    def push(self, item):
        self.validate(item)
        self._items.append(item)

IntStack = Stack[int]   # создаёт подкласс Stack с _item_type=int
s = IntStack()
s.push(42)       # OK
s.push("hello")  # TypeError

#10. Продвинутые паттерны

#Proxy-объект

class Proxy:
    """Прозрачный прокси для любого объекта."""

    __slots__ = ("_target",)

    def __init__(self, target):
        object.__setattr__(self, "_target", target)

    def __getattr__(self, name):
        return getattr(object.__getattribute__(self, "_target"), name)

    def __setattr__(self, name, value):
        setattr(object.__getattribute__(self, "_target"), name, value)

    def __repr__(self):
        return f"Proxy({object.__getattribute__(self, '_target')!r})"

proxy = Proxy([1, 2, 3])
proxy.append(4)   # вызывает list.append!
print(proxy)      # Proxy([1, 2, 3, 4])

#Lazy loading через `getattr`

class LazyLoader:
    """Загружает атрибуты только при первом обращении."""

    def __getattr__(self, name):
        # Вызывается только если атрибут не найден
        print(f"Loading {name}...")
        value = self._load(name)
        setattr(self, name, value)  # кэшируем — следующий раз через __dict__
        return value

    def _load(self, name):
        # Симуляция дорогой загрузки
        import time
        time.sleep(0.1)
        return f"loaded_{name}"

obj = LazyLoader()
print(obj.data)   # Loading data... → "loaded_data"
print(obj.data)   # "loaded_data" (из кэша, без загрузки)

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

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

Интроспекция и преобразование кода требуют узкого контракта и безопасного отказа. Результат основы — анализатор AST, который принимает разрешённую конструкцию и явно отвергает запрещённую.

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

import ast


def imported_modules(source: str) -> set[str]:
    tree = ast.parse(source)
    return {
        alias.name
        for node in ast.walk(tree)
        if isinstance(node, ast.Import)
        for alias in node.names
    }


assert imported_modules("import json\nimport pathlib as p") == {"json", "pathlib"}
assert imported_modules("result = 2 + 2") == set()

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

Практика с подсказками: напишите проверку небольшого правила исходного кода, добавьте номер строки и безопасное сообщение. Проверьте синтаксическую ошибку, alias, вложенную функцию и конструкцию, которую анализатор пока не понимает.

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

Проследите __getattribute__, дескриптор и __getattr__ на одном объекте, затем сравните AST-преобразование с явным декоратором. Для процесса-работника зафиксируйте границы сериализации, запуска и остановки.

#Упражнения

Реализуйте ImmutableDict — словарь запрещающий изменения через __setattr__, __setitem__, __delitem__. При попытке изменить должен возникать TypeError.
Напишите TracingProxy(obj) который логирует каждый вызов метода с аргументами и результатом.
Используя ast, напишите функцию find_undefined_names(code) — ищет имена, которые используются но не объявлены в коде.
Реализуйте curry декоратор для функций с произвольным числом аргументов.
Напишите Pipeline класс поддерживающий оператор |: Pipeline(data) | filter_func | transform_func | output_func.
Создайте дескриптор LazyProperty вычисляющий значение при первом обращении и кэширующий его. Убедитесь что у разных экземпляров независимые кэши.

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

ImmutableDict закрывает все публичные пути изменения, но инициализация всё ещё возможна; сравните с MappingProxyType и обычной композицией над dict.
Proxy корректно связывает методы, не логирует секретные аргументы и сохраняет исключение исходного объекта.
AST-анализатор тестируется на локальной области, импортированном имени, параметре, comprehension и вложенной функции; простого множества всех Name недостаточно для полной модели областей видимости.
Curry сохраняет ошибку исходной сигнатуры для лишних и неизвестных аргументов.
Pipeline фиксирует, возвращает ли операция новое значение или завершает конвейер; ошибка этапа содержит имя этапа без потери __cause__.
LazyProperty хранит кэш в экземпляре и документирует способ его сброса.

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

Закрепите тему в лаборатории «AST-аудит границы процесса без исполнения кода». Вы замените поиск опасных слов структурным анализом, соберёте каноническое сообщение исполнителю и объясните, почему такой аудит не является песочницей.

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 python_advanced/tests

Метод

Назначение

__len__

len(obj)

__getitem__

obj[key]

__setitem__

obj[key] = value

__delitem__

del obj[key]

__contains__

item in obj

__iter__

for x in obj

__reversed__

reversed(obj)

__next__

next(obj) — для итераторов

__missing__

dict.__getitem__ при отсутствии ключа

__length_hint__

Подсказка о размере для оптимизаций

Продвинутый Python