import dis
def optimized_comprehension():
squares = [x ** 2 for x in range(1000)]
return squares
dis.dis(optimized_comprehension)
import timeit
import statistics
setup_old = """
def old_style():
return sum([x * 2 for x in range(10000)])
"""
setup_new = """
def new_style():
return sum([x * 2 for x in range(10000)])
"""
old_time = timeit.timeit("old_style()", setup=setup_old, number=10000)
new_time = timeit.timeit("new_style()", setup=setup_new, number=10000)
print(f"Python 3.11: {old_time:.3f} сек")
print(f"Python 3.12: {new_time:.3f} сек")
print(f"Ускорение: {(old_time/new_time - 1)*100:.1f}%")
Практическое применение:
def process_matrix(matrix):
"""Обработка матрицы с оптимизированными генераторами."""
flattened = [
cell * 2
for row in matrix
for cell in row
if cell is not None
]
config = {
key: value.upper()
for key, value in config_raw.items()
if value and key.startswith('API_')
}
unique_values = {
item.strip().lower()
for item in raw_data
if len(item) > 0
}
return flattened, config, unique_values
def analyze_large_dataset(data_stream):
"""Анализ потока данных с оптимизированными генераторами."""
processed = [
{
'id': item['id'],
'value': transform(item['value']),
'timestamp': item['ts']
}
for item in data_stream
if is_valid(item) and within_timeframe(item['ts'])
]
grouped = {
category: [item for item in processed if item['category'] == category]
for category in set(item['category'] for item in processed)
}
return grouped
1.2. PEP 684: Per-Interpreter GIL — GIL на уровень интерпретатора
Самое революционное изменение Python 3.12 — внедрение Per-Interpreter GIL, которое открывает путь к настоящему параллелизму.
import sys
import threading
import time
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor
import numpy as np
def cpu_intensive_work_gil(data_chunk):
"""Вычислительно сложная задача с общим GIL."""
result = 0
for x in data_chunk:
result += x ** 0.5 * np.log1p(x)
return result
def cpu_intensive_work_isolated(data_chunk, interpreter_id):
"""Вычислительно сложная задача в изолированном интерпретаторе."""
result = 0
for x in data_chunk:
result += x ** 0.5 * np.log1p(x)
shared_results[interpreter_id] = result
return result
def benchmark_gil_vs_isolated():
"""Сравнение производительности с общим и изолированным GIL."""
data = list(range(1, 1000000))
chunk_size = len(data) // 4
chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
print("Тестирование с общим GIL...")
start = time.time()
threads = []
results_gil = []
for chunk in chunks:
thread = threading.Thread(
target=lambda c: results_gil.append(cpu_intensive_work_gil(c)),
args=(chunk,)
)
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
gil_time = time.time() - start
print(f"Общий GIL: {gil_time:.3f} сек, результат: {sum(results_gil):.2f}")
print("\nТестирование с изолированными интерпретаторами...")
start = time.time()
with ProcessPoolExecutor(max_workers=4) as executor:
futures = [
executor.submit(cpu_intensive_work_gil, chunk)
for chunk in chunks
]
results_isolated = [f.result() for f in futures]
isolated_time = time.time() - start
print(f"Изолированные интерпретаторы: {isolated_time:.3f} сек, результат: {sum(results_isolated):.2f}")
if gil_time > 0:
speedup = gil_time / isolated_time
print(f"\nУскорение: {speedup:.2f}x")
print(f"Эффективность использования CPU: {speedup/4*100:.1f}%")
return gil_time, isolated_time
try:
import _xxsubinterpreters as interpreters
def run_in_isolated_interpreter(code_str, shared_data=None):
"""Запуск кода в изолированном интерпретаторе."""
interp_id = interpreters.create()
config = {
'gil': 'isolated',
'memory': 'shared',
'resources': 'dedicated'
}
if shared_data:
interpreters.set_shared(interp_id, 'data', shared_data)
result = interpreters.run_string(interp_id, code_str)
output = interpreters.get_shared(interp_id, 'result')
interpreters.destroy(interp_id)
return output
except ImportError:
print("Расширенная поддержка изолированных интерпретаторов доступна в Python 3.12+")
def parallel_image_processing(images):
"""Параллельная обработка изображений с использованием изолированных интерпретаторов."""
from PIL import Image
import concurrent.futures
def process_single_image(image_data):
"""Обработка одного изображения."""
img = Image.frombytes('RGB', image_data['size'], image_data['pixels'])
img = img.filter(ImageFilter.GaussianBlur(radius=2))
img = img.rotate(45, expand=True)
img = img.resize((800, 600))
enhancer = ImageEnhance.Contrast(img)
img = enhancer.enhance(1.5)
return img.tobytes()
with concurrent.futures.ProcessPoolExecutor(
max_workers=mp.cpu_count(),
mp_context=mp.get_context('spawn')
) as executor:
futures = [
executor.submit(process_single_image, img)
for img in images
]
processed_images = []
for future in concurrent.futures.as_completed(futures):
try:
result = future.result(timeout=30)
processed_images.append(result)
except Exception as e:
print(f"Ошибка обработки изображения: {e}")
return processed_images
Новый синтаксис и возможности языка
2.1. PEP 701: Улучшенный синтаксис f-строк
Python 3.12 значительно расширил возможности f-строк, сделав их более гибкими и мощными.
def format_complex_query(table_name, filters, columns):
"""Форматирование сложного SQL-запроса."""
query = f"""
SELECT
{', '.join(columns)}
FROM
{table_name}
WHERE
{' AND '.join(f"{k} = {repr(v)}" for k, v in filters.items())}
ORDER BY
id DESC
LIMIT 100
"""
return query
def format_json_response(data, status):
"""Форматирование JSON-ответа."""
response = f"""
{{
"status": "{status}",
"data": {data},
"timestamp": "{datetime.now().isoformat()}"
}}
"""
return response
def format_windows_path(user, folder):
"""Форматирование путей Windows."""
path = fr"C:\Users\{user}\Documents\{folder}\file.txt"
return path
def generate_dynamic_template(template_name, context):
"""Генерация динамических шаблонов."""
html = f"""
<!DOCTYPE html>
<html>
<head>
<title>{template_name.replace('_', ' ').title()}</title>
<style>
body {{ font-family: Arial, sans-serif; }}
.container {{ max-width: {context.get('width', 1200)}px; }}
</style>
</head>
<body>
<div class="container">
<h1>{context['title'] if 'title' in context else 'Default Title'}</h1>
{"".join(f'<p>{paragraph}</p>' for paragraph in context.get('paragraphs', []))}
</div>
</body>
</html>
"""
return html
def calculate_invoice(items, tax_rate=0.20):
"""Расчет счета с автоматическим форматированием."""
subtotal = sum(item['price'] * item['quantity'] for item in items)
tax = subtotal * tax_rate
total = subtotal + tax
invoice = f"""
==============================
ИНВОЙС
==============================
{"\n".join(
f"{item['name'][:30]:30} {item['quantity']:3} × ${item['price']:7.2f} = ${item['price'] * item['quantity']:8.2f}"
for item in items
)}
------------------------------
Промежуточный итог: ${subtotal:>10.2f}
Налог ({tax_rate*100:.1f}%): ${tax:>10.2f}
==============================
ИТОГО: ${total:>10.2f}
==============================
"""
return invoice
def debug_complex_operation(data, config):
"""Расширенная отладка с f-строками."""
print(f"{data.shape=}, {data.dtype=}")
print(f"{config.get('threshold', 0.5)=}")
result = process_data(data, config)
print(f"""
Результат обработки:
- Статус: {result.status}
- Время выполнения: {result.execution_time:.3f} сек
- Потребление памяти: {result.memory_usage / 1024 / 1024:.2f} MB
- Успешные операции: {sum(1 for r in result.operations if r.success)}
""")
return result
def safe_user_greeting(user_input):
"""Безопасное использование пользовательского ввода в f-строках."""
name = user_input[:50]
greeting = f"""
Добро пожаловать, {name}!
Ваш профиль:
- Имя пользователя: {name}
- Дата регистрации: {datetime.now().strftime('%Y-%m-%d')}
- Статус: {'активен' if is_user_active(name) else 'неактивен'}
{'⚠️ Внимание: ваш аккаунт требует верификации' if needs_verification(name) else ''}
"""
safe_greeting = greeting.replace('<', '<').replace('>', '>')
return safe_greeting
import timeit
import dis
def benchmark_fstrings():
"""Бенчмарк различных методов форматирования строк."""
name = "John"
age = 30
salary = 50000.50
concat_time = timeit.timeit(
'"Name: " + name + ", Age: " + str(age) + ", Salary: $" + str(salary)',
globals=globals(),
number=1000000
)
format_time = timeit.timeit(
'"Name: {}, Age: {}, Salary: ${}".format(name, age, salary)',
globals=globals(),
number=1000000
)
fstring_time = timeit.timeit(
'f"Name: {name}, Age: {age}, Salary: ${salary}"',
globals=globals(),
number=1000000
)
print("Результаты бенчмарка (1,000,000 итераций):")
print(f"Конкатенация: {concat_time:.3f} сек")
print(f"Метод format: {format_time:.3f} сек")
print(f"f-строка 3.12: {fstring_time:.3f} сек")
print(f"\nУскорение f-строк vs format: {(format_time/fstring_time - 1)*100:.1f}%")
print("\nБайт-код f-строки в Python 3.12:")
dis.dis(compile('f"Name: {name}, Age: {age}"', '<string>', 'exec'))
def analyze_fstring_bytecode():
"""Анализ оптимизаций байт-кода для f-строк."""
code_obj = compile(
'f"Результат: {calculate(42)}, Путь: {path}, Дата: {date}"',
'<fstring>',
'eval'
)
print("Оптимизации f-строк в Python 3.12:")
print("1. Более эффективная константная складка")
print("2. Прямое встраивание выражений")
print("3. Минимизация операций форматирования")
print("4. Улучшенное кэширование строк")
return code_obj
2.2. PEP 695: Улучшенный синтаксис для типизации и дженериков
Python 3.12 вводит новый, более выразительный синтаксис для типизации.
from typing import TypeVar, Generic, Optional, Union, Callable
from typing_extensions import TypeAlias
T = TypeVar('T')
U = TypeVar('U')
class Container(Generic[T]):
def __init__(self, value: T) -> None:
self.value = value
def map(self, func: Callable[[T], U]) -> 'Container[U]':
return Container(func(self.value))
Result: TypeAlias = Union[dict[str, str], list[str]]
OptionalResult = Optional[Result]
class Container[T]:
"""Обобщенный контейнер с новым синтаксисом."""
def __init__(self, value: T) -> None:
self.value = value
def map[U](self, func: Callable[[T], U]) -> 'Container[U]':
"""Преобразование значения с сохранением типа."""
return Container[U](func(self.value))
def flat_map[U](self, func: Callable[[T], 'Container[U]']) -> 'Container[U]':
"""Монадическое связывание."""
return func(self.value)
def zip[U](self, other: 'Container[U]') -> 'Container[tuple[T, U]]':
"""Объединение двух контейнеров."""
return Container[tuple[T, U]]((self.value, other.value))
type Result = dict[str, str] | list[str]
type OptionalResult[T] = T | None
type AsyncResult[T] = Coroutine[Any, Any, T]
type Matrix = list[list[float]]
type JsonValue = str | int | float | bool | None | list['JsonValue'] | dict[str, 'JsonValue']
def process_data[T: (int, float, str), U](data: list[T]) -> dict[T, U]:
"""Обработка данных с ограничениями типов."""
return {item: transform(item) for item in data}
type Response[T = dict] = {
"status": str,
"data": T,
"metadata": dict[str, Any]
}
class PaginatedResponse[T, U = dict]:
"""Пагинированный ответ с типами по умолчанию."""
def __init__(
self,
items: list[T],
page: int,
total_pages: int,
metadata: U | None = None
) -> None:
self.items = items
self.page = page
self.total_pages = total_pages
self.metadata = metadata or {}
def to_dict(self) -> dict[str, Any]:
return {
"items": self.items,
"page": self.page,
"total_pages": self.total_pages,
"metadata": self.metadata
}
from typing import Protocol
class Addable(Protocol):
"""Протокол для типов, поддерживающих сложение."""
def __add__(self, other: Any) -> Any: ...
def sum_values[T: Addable](values: list[T]) -> T:
"""Суммирование значений с проверкой типа."""
if not values:
raise ValueError("Список значений пуст")
result = values[0]
for value in values[1:]:
result += value
return result
def cache[T](max_size: int = 100) -> Callable[[Callable[..., T]], Callable[..., T]]:
"""Декоратор кэширования с поддержкой типов."""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
cache_dict: dict[str, T] = {}
cache_keys: list[str] = []
@wraps(func)
def wrapper(*args, **kwargs) -> T:
key = f"{func.__name__}:{args}:{kwargs}"
if key in cache_dict:
return cache_dict[key]
result = func(*args, **kwargs)
if len(cache_keys) >= max_size:
oldest_key = cache_keys.pop(0)
del cache_dict[oldest_key]
cache_dict[key] = result
cache_keys.append(key)
return result
return wrapper
return decorator
type UserID = NewType('UserID', int)
type Email = NewType('Email', str)
class UserRepository[T]:
"""Обобщенный репозиторий пользователей."""
def __init__(self, model_type: type[T]) -> None:
self.model_type = model_type
self._storage: dict[UserID, T] = {}
self._next_id: UserID = UserID(1)
def create(self, data: dict[str, Any]) -> T:
"""Создание нового пользователя."""
user = self.model_type(**data, id=self._next_id)
self._storage[self._next_id] = user
self._next_id = UserID(self._next_id + 1)
return user
def get(self, user_id: UserID) -> T | None:
"""Получение пользователя по ID."""
return self._storage.get(user_id)
def find_by_email(self, email: Email) -> T | None:
"""Поиск пользователя по email."""
for user in self._storage.values():
if getattr(user, 'email', None) == email:
return user
return None
def update(self, user_id: UserID, **updates) -> T | None:
"""Обновление данных пользователя."""
if user_id not in self._storage:
return None
user = self._storage[user_id]
for key, value in updates.items():
if hasattr(user, key):
setattr(user, key, value)
return user
from pydantic import BaseModel, Field
from pydantic.generics import GenericModel
class ApiResponse[T](BaseModel, GenericModel):
"""Типизированный API-ответ."""
success: bool
data: T | None = None
error: str | None = None
meta: dict[str, Any] = Field(default_factory=dict)
class Config:
validate_assignment = True
frozen = False
extra = 'forbid'
import typing
from typing import TYPE_CHECKING
if TYPE_CHECKING:
type DatabaseConnection = Connection[DatabaseConfig]
type QueryResult[T] = tuple[list[T], int]
class ServiceProtocol[T](Protocol):
async def execute(self, query: str) -> T: ...
def close(self) -> None: ...
def benchmark_type_syntax():
"""Бенчмарк нового синтаксиса типизации."""
import timeit
old_code = """
from typing import TypeVar, Generic
T = TypeVar('T')
class OldContainer(Generic[T]):
def __init__(self, value: T):
self.value = value
"""
new_code = """
class NewContainer[T]:
def __init__(self, value: T):
self.value = value
"""
old_time = timeit.timeit(
"OldContainer[int](42)",
setup=old_code,
number=1000000
)
new_time = timeit.timeit(
"NewContainer[int](42)",
setup=new_code,
number=1000000
)
print(f"Старый синтаксис: {old_time:.3f} сек")
print(f"Новый синтаксис: {new_time:.3f} сек")
print(f"Ускорение: {(old_time/new_time - 1)*100:.1f}%")
Новые API и улучшения стандартной библиотеки
3.1. Улучшенный модуль itertools
import itertools
import collections
from typing import Iterable, TypeVar, Generator
T = TypeVar('T')
def process_in_batches(
items: Iterable[T],
batch_size: int
) -> Generator[list[T], None, None]:
"""Обработка элементов батчами с использованием нового batched."""
for batch in itertools.batched(items, batch_size):
yield process_batch(batch)
def analyze_large_file(file_path: str, chunk_size: int = 1000):
"""Анализ большого файла по частям."""
with open(file_path, 'r', encoding='utf-8') as f:
for batch in itertools.batched(f, chunk_size):
parsed_lines = [parse_line(line) for line in batch]
yield aggregate_data(parsed_lines)
def calculate_differences(measurements: list[float]) -> list[float]:
"""Вычисление разниц между последовательными измерениями."""
return [b - a for a, b in itertools.pairwise(measurements)]
def find_patterns(sequence: list[int], window_size: int = 3) -> list[tuple]:
"""Поиск паттернов в последовательности с использованием sliding_window."""
windows = list(itertools.sliding_window(sequence, window_size))
patterns = collections.Counter(windows)
return patterns.most_common(5)
def running_statistics(data: list[float]) -> dict[str, list]:
"""Вычисление текущей статистики по мере поступления данных."""
sums = list(itertools.accumulate(data, initial=0))
counts = list(range(1, len(data) + 1))
moving_averages = [
sums[i] / counts[i-1] if i > 0 else data[0]
for i in range(1, len(sums))
]
return {
'sums': sums[1:],
'counts': counts,
'averages': moving_averages
}
class DataStreamProcessor:
"""Обработчик потоковых данных с использованием новых itertools."""
def __init__(self, window_size: int = 10):
self.window_size = window_size
self._buffer = collections.deque(maxlen=window_size * 2)
def process_stream(self, data_stream: Iterable[dict]) -> Generator[dict, None, None]:
"""Обработка потока данных с сохранением контекста."""
for window in itertools.batched(data_stream, self.window_size):
self._buffer.extend(window)
window_stats = self._analyze_window(list(window))
if len(self._buffer) >= self.window_size:
context_stats = self._analyze_context()
window_stats.update(context_stats)
yield window_stats
def _analyze_window(self, window: list[dict]) -> dict:
"""Анализ текущего окна данных."""
if not window:
return {}
timestamps = [item['timestamp'] for item in window]
values = [item['value'] for item in window]
time_diffs = [b - a for a, b in itertools.pairwise(timestamps)]
value_diffs = [b - a for a, b in itertools.pairwise(values)]
return {
'window_size': len(window),
'avg_time_diff': sum(time_diffs) / len(time_diffs) if time_diffs else 0,
'avg_value_diff': sum(value_diffs) / len(value_diffs) if value_diffs else 0,
'min_value': min(values),
'max_value': max(values),
'avg_value': sum(values) / len(values)
}
def _analyze_context(self) -> dict:
"""Анализ контекста на основе предыдущих данных."""
recent_data = list(self._buffer)[-self.window_size*2:-self.window_size]
if not recent_data:
return {}
values = [item['value'] for item in recent_data]
return {
'context_avg': sum(values) / len(values),
'trend': self._calculate_trend(values)
}
def _calculate_trend(self, values: list[float]) -> str:
"""Вычисление тренда на основе sliding window."""
if len(values) < 3:
return 'stable'
windows = list(itertools.sliding_window(values, 3))
trends = []
for window in windows:
if window[2] > window[0]:
trends.append('up')
elif window[2] < window[0]:
trends.append('down')
else:
trends.append('stable')
trend_counts = collections.Counter(trends)
return trend_counts.most_common(1)[0][0]
def benchmark_itertools():
"""Сравнение производительности новых функций itertools."""
import timeit
data = list(range(1000000))
old_batch_code = """
def batch_old(iterable, n):
iterator = iter(iterable)
while True:
chunk = list(itertools.islice(iterator, n))
if not chunk:
break
yield chunk
"""
new_batch_code = """
from itertools import batched
"""
old_time = timeit.timeit(
"list(batch_old(data, 1000))",
setup=old_batch_code,
globals={'data': data, 'itertools': itertools},
number=100
)
new_time = timeit.timeit(
"list(itertools.batched(data, 1000))",
setup=new_batch_code,
globals={'data': data, 'itertools': itertools},
number=100
)
print(f"Старая реализация: {old_time:.3f} сек")
print(f"Новая функция batched: {new_time:.3f} сек")
print(f"Ускорение: {(old_time/new_time - 1)*100:.1f}%")
3.2. Новый модуль tomllib для работы с TOML
import tomllib
import tomli_w
from pathlib import Path
from typing import TypedDict, NotRequired
import json
def load_project_config(config_path: str | Path) -> dict:
"""Загрузка конфигурации проекта из TOML файла."""
with open(config_path, 'rb') as f:
config = tomllib.load(f)
required_keys = ['project', 'dependencies']
for key in required_keys:
if key not in config:
raise ValueError(f"Отсутствует обязательный ключ: {key}")
return config
class DatabaseConfig(TypedDict):
"""Конфигурация базы данных."""
host: str
port: int
database: str
username: str
password: str
pool_size: NotRequired[int]
timeout: NotRequired[float]
class ProjectConfig(TypedDict):
"""Конфигурация проекта."""
name: str
version: str
description: NotRequired[str]
dependencies: dict[str, str]
database: DatabaseConfig
api: dict[str, Any]
def load_typed_config(config_path: Path) -> ProjectConfig:
"""Загрузка типизированной конфигурации."""
with open(config_path, 'rb') as f:
raw_config = tomllib.load(f)
config: ProjectConfig = {
'name': raw_config['project']['name'],
'version': raw_config['project']['version'],
'dependencies': raw_config.get('dependencies', {}),
'database': {
'host': raw_config['database']['host'],
'port': raw_config['database']['port'],
'database': raw_config['database']['name'],
'username': raw_config['database']['user'],
'password': raw_config['database']['password'],
'pool_size': raw_config['database'].get('pool_size', 10),
'timeout': raw_config['database'].get('timeout', 30.0)
},
'api': raw_config.get('api', {})
}
if 'description' in raw_config['project']:
config['description'] = raw_config['project']['description']
return config
def create_default_config(output_path: Path) -> None:
"""Создание конфигурационного файла по умолчанию."""
default_config = {
'project': {
'name': 'My Awesome Project',
'version': '1.0.0',
'description': 'A project using Python 3.12 features',
'readme': 'README.md',
'requires-python': '>=3.12'
},
'build-system': {
'requires': ['setuptools>=61.0', 'wheel'],
'build-backend': 'setuptools.build_meta'
},
'dependencies': {
'fastapi': '>=0.100.0',
'pydantic': '>=2.0.0',
'sqlalchemy': '>=2.0.0',
'redis': '>=4.5.0'
},
'database': {
'host': 'localhost',
'port': 5432,
'name': 'app_db',
'user': 'app_user',
'password': 'secret_password',
'pool_size': 20,
'timeout': 30.0,
'ssl': True
},
'api': {
'host': '0.0.0.0',
'port': 8000,
'debug': False,
'cors_origins': [
'http://localhost:3000',
'https://example.com'
],
'rate_limit': {
'requests': 100,
'per_seconds': 60
}
},
'logging': {
'level': 'INFO',
'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
'file': 'logs/app.log',
'rotation': '1 day',
'retention': '30 days'
},
'monitoring': {
'enabled': True,
'prometheus_port': 9090,
'health_check_interval': 30,
'metrics': ['cpu', 'memory', 'requests', 'errors']
}
}
with open(output_path, 'wb') as f:
tomli_w.dump(default_config, f)
print(f"Конфигурационный файл создан: {output_path}")
def merge_configs(base_path: Path, override_path: Path) -> dict:
"""Объединение базовой и переопределяющей конфигураций."""
with open(base_path, 'rb') as f:
base_config = tomllib.load(f)
with open(override_path, 'rb') as f:
override_config = tomllib.load(f)
def deep_merge(base: dict, override: dict) -> dict:
"""Рекурсивное объединение словарей."""
result = base.copy()
for key, value in override.items():
if key in result and isinstance(result[key], dict) and isinstance(value, dict):
result[key] = deep_merge(result[key], value)
else:
result[key] = value
return result
return deep_merge(base_config, override_config)
from typing import Any
import re
class TOMLValidator:
"""Валидатор TOML конфигураций."""
def __init__(self, schema: dict):
self.schema = schema
self.rules = self._compile_rules(schema)
def _compile_rules(self, schema: dict) -> dict:
"""Компиляция правил валидации."""
rules = {}
for key, spec in schema.items():
if isinstance(spec, dict):
if 'type' in spec:
rules[key] = {
'type': self._parse_type(spec['type']),
'required': spec.get('required', True),
'default': spec.get('default'),
'regex': spec.get('regex'),
'min': spec.get('min'),
'max': spec.get('max'),
'choices': spec.get('choices')
}
else:
rules[key] = self._compile_rules(spec)
else:
rules[key] = {'type': self._parse_type(spec), 'required': True}
return rules
def _parse_type(self, type_str: str) -> type:
"""Парсинг строки типа."""
type_map = {
'string': str,
'integer': int,
'float': float,
'boolean': bool,
'array': list,
'object': dict
}
return type_map.get(type_str.lower(), Any)
def validate(self, config: dict) -> list[str]:
"""Валидация конфигурации."""
errors = []
self._validate_section(config, self.rules, [], errors)
return errors
def _validate_section(self, data: dict, rules: dict, path: list[str], errors: list[str]):
"""Валидация секции конфигурации."""
for key, rule in rules.items():
current_path = path + [key]
path_str = '.'.join(current_path)
if isinstance(rule, dict) and 'type' in rule:
if rule['required'] and key not in data:
errors.append(f"Обязательное поле отсутствует: {path_str}")
continue
if key not in data:
if 'default' in rule:
data[key] = rule['default']
continue
value = data[key]
expected_type = rule['type']
if not isinstance(value, expected_type):
errors.append(
f"Неверный тип для {path_str}: "
f"ожидался {expected_type.__name__}, "
f"получен {type(value).__name__}"
)
continue
if rule.get('regex') and isinstance(value, str):
if not re.match(rule['regex'], value):
errors.append(
f"Значение {path_str} не соответствует паттерну: "
f"{rule['regex']}"
)
if rule.get('min') is not None:
if isinstance(value, (int, float)) and value < rule['min']:
errors.append(
f"Значение {path_str} должно быть >= {rule['min']}"
)
if rule.get('max') is not None:
if isinstance(value, (int, float)) and value > rule['max']:
errors.append(
f"Значение {path_str} должно быть <= {rule['max']}"
)
if rule.get('choices') and value not in rule['choices']:
errors.append(
f"Значение {path_str} должно быть одним из: "
f"{rule['choices']}"
)
elif isinstance(rule, dict):
if key in data:
if not isinstance(data[key], dict):
errors.append(
f"Поле {path_str} должно быть объектом, "
f"получен {type(data[key]).__name__}"
)
else:
self._validate_section(
data[key], rule, current_path, errors
)
CONFIG_SCHEMA = {
'project': {
'name': {'type': 'string', 'required': True},
'version': {'type': 'string', 'required': True, 'regex': r'^\d+\.\d+\.\d+
Заключение
Python 3.12 представляет собой значительный шаг вперед в эволюции языка, предлагая не просто новые функции, а фундаментальные улучшения, которые меняют подход к разработке на Python в 2026 году.
Python 3.12 доказывает, что язык продолжает развиваться в сторону высокой производительности и выразительности, оставаясь при этом доступным для начинающих и мощным для экспертов. Это обновление — не просто еще одна версия, а качественный скачок, который делает Python еще более конкурентоспособным выбором для современных high-load систем 2026 года.