Dunder методы глубже, метапрограммирование, AST, Cython, multiprocessing
Этот уровень — для понимания внутренней механики Python: атрибутный протокол, метапрограммирование, AST, мультипроцессорность. Знания необходимы для работы с фреймворками и написания библиотек.
__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 победил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)| Метод | Назначение |
|---|---|
__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__ | Подсказка о размере для оптимизаций |
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# __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__, не создаёт новый объект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,
}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 = 5import 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!multiprocessing и shared memoryimport 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() # обязательно освободить!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)) # 42from 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__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") # TypeErrorclass 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])__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" (из кэша, без загрузки)ImmutableDict — словарь запрещающий изменения через __setattr__, __setitem__, __delitem__. При попытке изменить должен возникать TypeError.TracingProxy(obj) который логирует каждый вызов метода с аргументами и результатом.ast, напишите функцию find_undefined_names(code) — ищет имена, которые используются но не объявлены в коде.curry декоратор для функций с произвольным числом аргументов.Pipeline класс поддерживающий оператор |: Pipeline(data) | filter_func | transform_func | output_func.LazyProperty вычисляющий значение при первом обращении и кэширующий его. Убедитесь что у разных экземпляров независимые кэши.Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.