Работа с Apache Airflow 2.8: новые возможности

from airflow.decorators import dag, task
from airflow.models import Variable
from datetime import datetime
from typing import Dict
@dag(
    schedule="0 0 * * *",
    start_date=datetime(2024, 2, 24),
    catchup=False,
    tags=['data_processing']
)
def process_daily_data():
    
    @task
    def extract_data() -> Dict:
        return {
            'source': 'api',
            'timestamp': datetime.now()
        }
    
    @task
    def transform_data(data: Dict) -> Dict:
        return {
            **data,
            'processed': True
        }
    
    @task
    def load_data(data: Dict):
        print(f"Loading data: {data}")
    
    # TaskFlow API
    data = extract_data()
    processed = transform_data(data)
    load_data(processed)
dag = process_daily_data()

Преимущества:
— TaskFlow API
— Типизация задач
— Автоматическая сериализация
— Встроенный UI
— Простая отладка

Важно:
— Используйте Virtual Env
— Правильно обрабатывайте ошибки
— Следите за зависимостями
— Документируйте DAG'и

Подписывайся на @the_daily_python

Реализация Saga паттерна в Python

from enum import Enum
from typing import List, Callable, Dict, Any
import asyncio
import logging
class StepStatus(Enum):
    SUCCESS = "success"
    FAILED = "failed"
class SagaStep:
    def __init__(
        self,
        action: Callable,
        compensation: Callable = None,
        name: str = None
    ):
        self.action = action
        self.compensation = compensation
        self.name = name or action.__name__
        
    async def execute(self, context: Dict[str, Any]) -> StepStatus:
        try:
            await self.action(context)
            return StepStatus.SUCCESS
        except Exception as e:
            logging.error(f"Step {self.name} failed: {e}")
            return StepStatus.FAILED
            
    async def compensate(self, context: Dict[str, Any]) -> None:
        if self.compensation:
            try:
                await self.compensation(context)
            except Exception as e:
                logging.error(
                    f"Compensation for {self.name} failed: {e}"
                )

Использование паттерна:

class Saga:
    def __init__(self):
        self.steps: List[SagaStep] = []
        
    def add_step(
        self,
        action: Callable,
        compensation: Callable = None
    ) -> 'Saga':
        self.steps.append(SagaStep(action, compensation))
        return self
        
    async def execute(self, context: Dict[str, Any] = None) -> bool:
        if context is None:
            context = {}
            
        executed_steps = []
        
        for step in self.steps:
            status = await step.execute(context)
            executed_steps.append(step)
            
            if status == StepStatus.FAILED:
                # Запускаем компенсацию в обратном порядке
                for completed_step in reversed(executed_steps):
                    await completed_step.compensate(context)
                return False
                
        return True

Преимущества:
— Управление распределенными транзакциями
— Поддержка отмены операций
— Изоляция шагов
— Контекстно-зависимая логика
— Прозрачная обработка ошибок

Важно помнить:
— Компенсации должны быть идемпотентными
— Логируйте все действия
— Обрабатывайте ошибки компенсаций
— Сохраняйте контекст транзакции

Подписывайся на @the_daily_python

Профилирование Django приложений с django-silk

# settings.py
MIDDLEWARE = [
    # ...
    'silk.middleware.SilkyMiddleware',
    # ...
]
INSTALLED_APPS = [
    # ...
    'silk',
    # ...
]
# urls.py
urlpatterns = [
    # ...
    path('silk/', include('silk.urls')),
    # ...
]
# views.py
from django.db.models import Prefetch
from django.shortcuts import render
from silk.profiling.profiler import silk_profile
@silk_profile(name="Dashboard View")
def dashboard(request):
    with silk_profile(name="Data Query"):
        # Оптимизируем запросы
        users = User.objects.prefetch_related(
            Prefetch(
                'orders',
                queryset=Order.objects.filter(status='completed')
            )
        )[:100]
    
    with silk_profile(name="Template Rendering"):
        return render(request, 'dashboard.html', {'users': users})

Преимущества:
— Подробная информация о SQL запросах
— Визуализация узких мест
— Анализ загрузки страниц
— Детализация по HTTP-запросам
— Профилирование участков кода

Важно помнить:
— Используйте только в разработке
— Собирайте статистику на реальных данных
— Настройте фильтры запросов
— Обращайте внимание на N+1 запросы
— Оптимизируйте шаблоны Django

Подписывайся на @the_daily_python

Работа с Apache Cassandra через python-driver

from cassandra.cluster import Cluster
from cassandra.auth import PlainTextAuthProvider
from cassandra.query import SimpleStatement
from cassandra.policies import TokenAwarePolicy, DCAwareRoundRobinPolicy
from typing import List, Dict, Any
class CassandraClient:
    def __init__(self, contact_points: List[str], keyspace: str):
        auth = PlainTextAuthProvider(
            username='cassandra',
            password='cassandra'
        )
        
        # Оптимизируем политику маршрутизации
        load_balancing = TokenAwarePolicy(
            DCAwareRoundRobinPolicy(local_dc='datacenter1')
        )
        
        self.cluster = Cluster(
            contact_points=contact_points,
            auth_provider=auth,
            load_balancing_policy=load_balancing
        )
        self.session = self.cluster.connect(keyspace)
        
    async def get_user_by_id(self, user_id: str) -> Dict[str, Any]:
        query = SimpleStatement(
            "SELECT * FROM users WHERE id = %s",
            consistency_level=ConsistencyLevel.LOCAL_QUORUM
        )
        
        row = self.session.execute(
            query, (user_id,)
        ).one()
        
        if not row:
            return None
            
        return dict(row._asdict())
    
    async def insert_user(self, user: Dict[str, Any]) -> None:
        # Использование батча для транзакционности
        batch = BatchStatement(consistency_level=ConsistencyLevel.QUORUM)
        
        batch.add(
            "INSERT INTO users (id, name, email) VALUES (%s, %s, %s)",
            (user['id'], user['name'], user['email'])
        )
        
        batch.add(
            "INSERT INTO user_by_email (email, user_id) VALUES (%s, %s)",
            (user['email'], user['id'])
        )
        
        self.session.execute(batch)

Преимущества:
— Высокая производительность
— Линейная масштабируемость
— Настраиваемая консистентность
— Отказоустойчивость
— Поддержка асинхронности

Важно помнить:
— Проектируйте схему под запросы
— Правильно выбирайте ключи партиционирования
— Следите за уровнем консистентности
— Используйте prepared statements
— Контролируйте размер данных

Подписывайся на @the_daily_python

Использование Hypothesis для property-based testing

from hypothesis import given, strategies as st
from typing import List, Dict
import pytest
def remove_duplicates(items: List[int]) -> List[int]:
    """Удаляет дубликаты из списка, сохраняя порядок"""
    seen = set()
    result = []
    for item in items:
        if item not in seen:
            seen.add(item)
            result.append(item)
    return result
# Property-based тесты
@given(st.lists(st.integers()))
def test_no_duplicates_in_result(items):
    """Проверяем, что результат не содержит дубликатов"""
    result = remove_duplicates(items)
    assert len(result) == len(set(result))
@given(st.lists(st.integers()))
def test_all_items_preserved(items):
    """Проверяем, что все оригинальные элементы сохраняются"""
    result = remove_duplicates(items)
    assert set(result) == set(items)
@given(st.lists(st.integers()))
def test_order_preserved(items):
    """Проверяем сохранение порядка элементов"""
    result = remove_duplicates(items)
    
    # Строим словарь с индексами первого появления
    original_indices = {}
    for i, item in enumerate(items):
        if item not in original_indices:
            original_indices[item] = i
    
    # Проверяем, что порядок сохранился
    for i in range(len(result) - 1):
        assert original_indices[result[i]] < original_indices[result[i+1]]

Преимущества:
— Автоматическая генерация тестовых данных
— Нахождение граничных случаев
— Проверка свойств, а не конкретных примеров
— Воспроизводимость ошибок
— Минимизация тестовых случаев

Важно помнить:
— Определяйте ясные свойства
— Ограничивайте размер входных данных
— Используйте композицию стратегий
— Сохраняйте failed примеры

Подписывайся на @the_daily_python

Оптимизация GraphQL запросов в Python

import strawberry
from strawberry.types import Info
from typing import List, Optional
from dataclasses import dataclass
import asyncio
@strawberry.type
class Author:
    id: str
    name: str
    
    @strawberry.field
    async def books(self, info: Info) -> List['Book']:
        # Получаем DataLoader из контекста
        loader = info.context['loaders'].book_by_author_loader
        # Батчим запросы к БД
        return await loader.load(self.id)
@strawberry.type
class Book:
    id: str
    title: str
    author_id: str
    
    @strawberry.field
    async def author(self, info: Info) -> Author:
        loader = info.context['loaders'].author_loader
        return await loader.load(self.author_id)
@strawberry.type
class Query:
    @strawberry.field
    async def books(
        self, 
        info: Info,
        limit: Optional[int] = 10
    ) -> List[Book]:
        # Используем префетч для оптимизации
        result = await get_books(limit=limit)
        
        # Подготавливаем DataLoader
        author_ids = [book.author_id for book in result]
        info.context['loaders'].author_loader.prime_many(
            author_ids, 
            await get_authors_by_ids(author_ids)
        )
        
        return result

Преимущества:
— Решает проблему N+1 запросов
— Оптимизирует загрузку данных
— Кэширование связанных объектов
— Батчинг запросов к БД
— Поддержка асинхронных запросов

Важно помнить:
— Используйте DataLoader для связанных объектов
— Применяйте prefetching где возможно
— Оптимизируйте запросы к БД
— Следите за глубиной запросов

Подписывайся на @the_daily_python

Работа с Apache Pulsar в Python

from pulsar import Client, ConsumerType, MessageId
from typing import Callable, Dict, Any
import json
import asyncio
class PulsarManager:
    def __init__(self, service_url: str):
        self.client = Client(service_url)
        
    def create_producer(self, topic: str):
        return self.client.create_producer(
            topic,
            block_if_queue_full=True,
            batching_enabled=True,
            batching_max_publish_delay_ms=10
        )
    
    async def send_message(
        self, 
        topic: str, 
        data: Dict[str, Any], 
        key: str = None
    ):
        producer = self.create_producer(topic)
        
        try:
            # Сериализуем сообщение
            serialized = json.dumps(data).encode('utf-8')
            
            # Добавляем метаданные
            properties = {
                "timestamp": str(int(time.time())),
                "version": "1.0"
            }
            
            # Отправляем сообщение
            producer.send(
                serialized,
                properties=properties,
                partition_key=key
            )
        finally:
            producer.close()
    
    async def consume(
        self,
        topic: str,
        subscription: str,
        handler: Callable,
        consumer_type: ConsumerType = ConsumerType.Shared
    ):
        consumer = self.client.subscribe(
            topic,
            subscription_name=subscription,
            consumer_type=consumer_type,
            receiver_queue_size=1000
        )
        
        try:
            while True:
                msg = consumer.receive()
                data = json.loads(msg.data().decode('utf-8'))
                
                try:
                    await handler(data)
                    consumer.acknowledge(msg)
                except Exception as e:
                    consumer.negative_acknowledge(msg)
        finally:
            consumer.close()

Преимущества:
— Высокая пропускная способность
— Multi-tenancy архитектура
— Географическая репликация
— Гарантированная доставка
— Поддержка различных схем подписки

Важно помнить:
— Правильно настраивайте партиционирование
— Используйте батчинг для производительности
— Обрабатывайте ошибки доставки
— Следите за размером очередей
— Правильно выбирайте тип потребителя

Подписывайся на @the_daily_python

Профилирование и оптимизация asyncio приложений

import asyncio
import time
from functools import wraps
import tracemalloc
from typing import Dict, Callable, Any
def async_profiler(func):
    """Декоратор для профилирования асинхронных функций"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # Запускаем отслеживание памяти
        tracemalloc.start()
        start_memory = tracemalloc.get_traced_memory()[0]
        
        # Замеряем время
        start_time = time.time()
        
        try:
            result = await func(*args, **kwargs)
            return result
        finally:
            # Вычисляем метрики
            end_time = time.time()
            current, peak = tracemalloc.get_traced_memory()
            
            # Останавливаем трассировку
            tracemalloc.stop()
            
            # Выводим статистику
            print(f"Функция: {func.__name__}")
            print(f"  Время выполнения: {end_time - start_time:.4f} сек")
            print(f"  Текущая память: {(current - start_memory) / 1024:.2f} KB")
            print(f"  Пиковая память: {peak / 1024:.2f} KB")
            
            # Выводим top-10 источников утечек
            snapshot = tracemalloc.take_snapshot()
            stats = snapshot.statistics('lineno')
            print("\nТоп 10 источников выделения памяти:")
            for stat in stats[:10]:
                print(f"  {stat}")
    
    return wrapper

Применение профилировщика:

@async_profiler
async def fetch_data(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()
# Оптимизация работы с несколькими тасками
async def process_concurrently(urls: List[str], limit: int = 10):
    # Используем Semaphore для ограничения параллельности
    sem = asyncio.Semaphore(limit)
    
    async def fetch_with_limit(url):
        async with sem:
            return await fetch_data(url)
    
    # Запускаем задачи с ограничением
    tasks = [fetch_with_limit(url) for url in urls]
    return await asyncio.gather(*tasks)

Преимущества:
— Детальный анализ потребления памяти
— Отслеживание времени выполнения
— Контроль параллельности
— Обнаружение утечек памяти
— Оптимизация узких мест

Важно помнить:
— Ограничивайте параллельность
— Следите за блокирующими операциями
— Используйте asyncio.gather для параллельного выполнения
— Контролируйте размер буферов
— Профилируйте в условиях, близких к production


Подписывайся на @the_daily_python

Структурное сопоставление образцов (pattern matching) позволяет элегантно обрабатывать сложные структуры данных в Python с помощью оператора match.

Синтаксис базового сопоставления:

match выражение:
    case образец_1:
        # действия, если выражение соответствует образцу_1
    case образец_2:
        # действия, если выражение соответствует образцу_2
    case _:
        # действия по умолчанию (как else)

Сопоставление со сложными структурами данных:

def process_data(data):
    match data:
        case {'type': 'user', 'id': id_value, 'name': name} if id_value > 100:
            return f"VIP пользователь: {name}"
        case {'type': 'user', 'id': _, 'name': name}:
            return f"Обычный пользователь: {name}"
        case {'type': 'task', 'status': 'completed', **rest}:
            return f"Завершенная задача: {rest.get('title', 'Без названия')}"
        case ['error', code, msg]:
            return f"Ошибка {code}: {msg}"
        case _:
            return "Неизвестный формат данных"
# Примеры использования
print(process_data({'type': 'user', 'id': 101, 'name': 'Анна'}))  # VIP пользователь: Анна
print(process_data({'type': 'user', 'id': 42, 'name': 'Пётр'}))   # Обычный пользователь: Пётр
print(process_data({'type': 'task', 'status': 'completed', 'title': 'Изучить Python 3.10'}))  # Завершенная задача: Изучить Python 3.10
print(process_data(['error', 404, 'Страница не найдена']))  # Ошибка 404: Страница не найдена

Объясняем
Захват значений: Переменные в шаблонах (как id_value, name) захватывают соответствующие значения
Условные шаблоны: Часть if id_value > 100 добавляет дополнительное условие для соответствия
Подстановочный знак: Символ _ игнорирует значение (аналогично id: _)
Распаковка: **rest захватывает оставшиеся ключи словаря

Преимущества:
Читаемость: Упрощает работу со сложными структурами данных
Компактность: Сокращает объем кода по сравнению с цепочками if-elif
Выразительность: Позволяет одновременно проверять структуру и извлекать данные
Гибкость: Можно комбинировать разные типы проверок в одном блоке

Важно помнить:
Доступно только в Python 3.10 и выше
При сопоставлении словарей учитывается только наличие указанных ключей (а не точное соответствие всех ключей)
Порядок case-блоков имеет значение - используется первое совпадение

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

Подписывайся на @the_daily_python

Pydantic v2 позволяет проверять и преобразовывать данные с помощью аннотаций типов Python, обеспечивая производительность и типобезопасность.

from pydantic import BaseModel, Field, EmailStr, ValidationError, field_validator
from typing import List, Optional
from datetime import datetime
class User(BaseModel):
    id: int
    name: str = Field(..., min_length=2, max_length=50)
    email: EmailStr
    tags: List[str] = []
    created_at: datetime = Field(default_factory=datetime.now)
    age: Optional[int] = Field(None, ge=18, lt=100)
    
    @field_validator('name')
    @classmethod
    def normalize_name(cls, v: str) -> str:
        return v.strip().title()
# Создание и валидация модели
try:
    user = User(
        id=1,
        name="  john smith  ",
        email="john@example.com",
        tags=["admin", "user"],
        age=25
    )
    print(user.model_dump_json(indent=2))
    
    # Неверные данные
    invalid_user = User(
        id="not_an_int",  # Строка вместо int
        name="J",  # Слишком короткое имя
        email="not_an_email",  # Неверный email
        age=15  # Меньше минимального возраста
    )
except ValidationError as e:
    print(f"Ошибка валидации: {e}")

Объясняем
— Базовая модель: Наследуемся от BaseModel для определения схемы данных
— Валидация полей: Field обеспечивает ограничения (min_length, ge и т.д.)
— Типизация: Аннотации типов гарантируют правильный формат данных
— Валидаторы: Метод normalize_name автоматически форматирует имя

Преимущества Pydantic v2:
— Скорость: Переписан с использованием Rust, в 30-50 раз быстрее v1
— Типобезопасность: Статическая проверка типов с помощью mypy
— Интеграция: Отлично работает с FastAPI, ORM и другими библиотеками
— JSON схемы: Автоматически генерирует JSON Schema для документации API

Важно помнить:
— В v2 изменены многие API по сравнению с v1 (например, model_dump() вместо dict())
— Для обратной совместимости можно использовать режим пространства имен v1
— EmailStr требует установки pydantic[email]

Pydantic v2 — необходимый инструмент для современной Python-разработки, который помогает избежать ошибок, автоматизировать проверки данных и ускорить разработку приложений, особенно при работе с API.


Подписывайся на @the_daily_python

Найдите проблему в коде для многопоточной обработки больших файлов:

import threading
from queue import Queue
import time
class ChunkProcessor:
    def __init__(self, num_workers=4):
        self.queue = Queue(maxsize=10)
        self.results = []
        self.workers = []
        self.num_workers = num_workers
        
    def process_chunk(self, chunk):
        # Имитация обработки данных
        time.sleep(0.1)
        return sum(chunk)
        
    def worker(self):
        while True:
            chunk = self.queue.get()
            if chunk is None:
                self.queue.task_done()
                break
            result = self.process_chunk(chunk)
            self.results.append(result)
            self.queue.task_done()
    
    def process_data(self, data, chunk_size=1000):
        # Запуск рабочих потоков
        for _ in range(self.num_workers):
            t = threading.Thread(target=self.worker)
            t.start()
            self.workers.append(t)
        
        # Разбиение данных на чанки и добавление в очередь
        for i in range(0, len(data), chunk_size):
            chunk = data[i:i+chunk_size]
            self.queue.put(chunk)
            
        # Ожидание завершения обработки
        for _ in range(self.num_workers):
            self.queue.put(None)
        
        self.queue.join()
        return sum(self.results)
# Пример использования
processor = ChunkProcessor(num_workers=4)
data = list(range(10000))
total = processor.process_data(data, chunk_size=1000)
print(f"Сумма: {total}, Ожидаемая сумма: {sum(data)}")

Пишите свои варианты в комментариях, а завтра мы опубликуем правильный ответ.

Подписывайся на @the_daily_python

Beautiful Soup 4 позволяет эффективно парсить и извлекать данные из HTML с помощью CSS-селекторов, что значительно упрощает веб-скрапинг.

import requests
from bs4 import BeautifulSoup
# Получаем HTML страницы
url = "https://python.org/events/"
response = requests.get(url)
html = response.text
# Создаем объект BeautifulSoup
soup = BeautifulSoup(html, 'html.parser')
# Использование CSS-селекторов для извлечения данных
events = soup.select("ul.list-recent-events li")
# Извлекаем и структурируем информацию
event_list = []
for event in events:
    title = event.select_one("h3.event-title a")
    time_elem = event.select_one("time")
    location = event.select_one("span.event-location")
    
    if title and time_elem:
        event_list.append({
            "title": title.text.strip(),
            "url": title["href"] if title.has_attr("href") else "",
            "date": time_elem["datetime"] if time_elem.has_attr("datetime") else time_elem.text.strip(),
            "location": location.text.strip() if location else "N/A"
        })
# Выводим результаты
for i, event in enumerate(event_list[:3], 1):
    print(f"{i}. {event['title']} ({event['date']})")
    print(f"   Место: {event['location']}")
    print(f"   Ссылка: {event['url']}")

Объясняем
— Селекторы CSS: Используем .select() для получения списка элементов, .select_one() для первого совпадения
— Вложенные селекторы: ul.list-recent-events li находит все элементы списка внутри определенного ul
— Извлечение данных: Получаем текст через .text и атрибуты через ["имя_атрибута"]
— Структурирование: Формируем словари с данными о каждом событии

Преимущества CSS-селекторов:
— Читаемость: Селекторы более лаконичны, чем цепочки методов BeautifulSoup
— Гибкость: Легко выбирать элементы по классам, ID и сложным паттернам
— Знакомый синтаксис: Если вы работали с CSS, синтаксис интуитивно понятен
— Скорость: Поиск через CSS-селекторы часто быстрее, чем через методы find/find_all

Полезные селекторы:
div#main — элемент с ID "main"
div.content p — все параграфы внутри div с классом "content"
a[href^="https"] — ссылки, начинающиеся с https
ul > li:first-child — первый элемент списка

Beautiful Soup с CSS-селекторами — мощное сочетание для веб-скрапинга, значительно упрощающее извлечение данных из HTML-страниц и требующее меньше кода по сравнению с традиционными методами навигации по DOM.

Подписывайся на @the_daily_python

Django ORM отлично работает в экосистеме Django, но иногда SQLAlchemy предлагает больше гибкости. Рассмотрим плюсы и минусы миграции, а также базовый пример перехода.

# Django ORM модель
from django.db import models
class User(models.Model):
    username = models.CharField(max_length=100, unique=True)
    email = models.EmailField(unique=True)
    is_active = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True)
    
# Пример запроса в Django ORM
active_users = User.objects.filter(is_active=True).order_by('-created_at')[:10]
# Эквивалент в SQLAlchemy
from sqlalchemy import Column, String, Boolean, DateTime, func, select
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(100), unique=True)
    email = Column(String, unique=True)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=func.now())
# Пример запроса в SQLAlchemy
from sqlalchemy.orm import Session
with Session(engine) as session:
    stmt = select(User).where(User.is_active == True).order_by(User.created_at.desc()).limit(10)
    active_users = session.execute(stmt).scalars().all()

Объясняем
— Определение моделей: В SQLAlchemy явно указываем таблицу и столбцы
— Выполнение запросов: Django использует менеджер объектов, SQLAlchemy - сессии и выражения
— Синтаксис: SQLAlchemy ближе к SQL, Django более "питонический"

Преимущества SQLAlchemy:
— Гибкость: Больше контроля над SQL-запросами и низкоуровневым взаимодействием с БД
— Производительность: Возможность оптимизации запросов (raw SQL, bulk operations)
— Универсальность: Работает независимо от веб-фреймворка
— Расширенные возможности: Сложные join-запросы, пользовательские выражения, оконные функции

Недостатки миграции:
— Кривая обучения: SQLAlchemy сложнее в освоении
— Потеря экосистемы: Отказ от Django admin, форм и других компонентов
— Совместимость: Дополнительные усилия для интеграции с Django-middleware

Советы по миграции:
— Начните с гибридного подхода: используйте SQLAlchemy для сложных запросов
— Используйте инструменты миграции Alembic вместо Django-миграций
— Рассмотрите SQLModel как промежуточный вариант между Django ORM и SQLAlchemy

SQLAlchemy даёт больше свободы и контроля, но это приходит с ценой более сложного кода. Выбирайте инструмент под конкретные потребности проекта, а не слепо следуйте трендам.

Подписывайся на @the_daily_python

Библиотека pyjq позволяет использовать мощный язык запросов jq прямо в Python для продвинутой фильтрации и трансформации JSON-данных.

import pyjq
import json
import requests
# Получаем данные из API
response = requests.get('https://jsonplaceholder.typicode.com/users')
data = response.json()
# Базовый запрос: извлечь имена всех пользователей
names = pyjq.all('.[].name', data)
print("Имена:", names[:3], "...")
# Сложный запрос: создать словарь {id: email} для пользователей с веб-сайтом .org
org_users = pyjq.all('.[] | select(.website | endswith(".org")) | {id: .id, email: .email}', data)
print("Пользователи с .org сайтами:", org_users)
# Трансформация данных: создать иерархическую структуру по компаниям
companies = pyjq.all('group_by(.company.name) | map({company: .[0].company.name, users: map({id, name, email})})' , data)
print(f"Найдено {len(companies)} компаний")
# Вложенная фильтрация: пользователи с широтой > 0 и городом, начинающимся на "S"
filtered = pyjq.all('.[] | select(.address.geo.lat | tonumber > 0 and .address.city | startswith("S")) | {name, city: .address.city}', data)
print("Отфильтрованные пользователи:", filtered)

Объясняем
jq синтаксис: .[] - итерация по массиву, select() - фильтрация, | - пайплайн операций
pyjq.all(): Выполняет запрос и возвращает результат как Python-объекты
Трансформации: Можно создавать сложные преобразования и агрегации

Преимущества jq в Python:
Выразительность: Компактные запросы для сложных преобразований JSON
Производительность: Реализация на C быстрее чистого Python
Мощная фильтрация: Функции для работы со строками, числами, логическими выражениями
JSON-специфичность: Создан специально для работы с JSON-структурами

Важно помнить:
— Требуется установка: pip install pyjq и системная библиотека jq
— Кривая обучения jq немного крутая, но окупается при работе со сложными JSON
— Документация jq: https://stedolan.github.io/jq/manual/

jq в Python — это мощный инструмент для работы с JSON-данными, позволяющий в одну строку делать то, что потребовало бы десятки строк кода на чистом Python, особенно при сложных преобразованиях и фильтрации вложенных структур.

Подписывайся на @the_daily_python

Pytest делает тестирование Python-кода удобным и эффективным. Рассмотрим продвинутые техники для сокращения дублирования и повышения читаемости тестов.

import pytest
from datetime import datetime, timedelta
# Функция, которую будем тестировать
def is_valid_date_range(start_date, end_date, max_days=30):
    """Проверяет, что диапазон дат не превышает max_days и end_date > start_date"""
    if not isinstance(start_date, datetime) or not isinstance(end_date, datetime):
        raise TypeError("Даты должны быть объектами datetime")
    
    if end_date <= start_date:
        return False
    
    date_diff = (end_date - start_date).days
    return date_diff <= max_days
# Параметризованные тесты
@pytest.mark.parametrize("start,end,max_days,expected", [
    (datetime(2023, 1, 1), datetime(2023, 1, 15), 30, True),
    (datetime(2023, 1, 1), datetime(2023, 1, 1), 30, False),  # равные даты
    (datetime(2023, 1, 15), datetime(2023, 1, 1), 30, False),  # конец раньше начала
    (datetime(2023, 1, 1), datetime(2023, 2, 1), 30, False),  # слишком большой диапазон
    (datetime(2023, 1, 1), datetime(2023, 2, 1), 40, True),  # увеличенный max_days
])
def test_is_valid_date_range(start, end, max_days, expected):
    assert is_valid_date_range(start, end, max_days) == expected
# Фикстуры для подготовки тестового окружения
@pytest.fixture
def date_factory():
    """Фикстура для генерации тестовых дат"""
    base_date = datetime(2023, 3, 1)
    
    def _make_date(days_delta=0):
        return base_date + timedelta(days=days_delta)
    
    return _make_date
def test_with_fixture(date_factory):
    """Использование фикстуры для создания тестовых дат"""
    start = date_factory()
    end = date_factory(days_delta=15)
    assert is_valid_date_range(start, end) is True
# Тестирование исключений
def test_invalid_input():
    """Проверка вызова исключения при неверных типах"""
    with pytest.raises(TypeError, match="Даты должны быть объектами datetime"):
        is_valid_date_range("2023-01-01", "2023-01-15")

Объясняем
— Параметризация: @pytest.mark.parametrize позволяет запускать один тест с разными входными данными
— Фикстуры: @pytest.fixture создает переиспользуемые компоненты для тестов
— Проверка исключений: pytest.raises проверяет, что код правильно вызывает исключения

Преимущества:
— Меньше дублирования: Один тест для множества случаев
— Лучшая организация: Фикстуры помогают структурировать тестовые данные
— Читаемость: Явное определение входных и ожидаемых значений
— Гибкость: Фикстуры можно использовать в нескольких тестах

Дополнительные возможности:
— Scope фикстур: @pytest.fixture(scope="module") для переиспользования состояния
— Маркеры: @pytest.mark.slow для категоризации тестов
— Пропуск тестов: @pytest.mark.skipif при определенных условиях

Грамотное использование параметризации и фикстур в Pytest значительно сокращает объем кода, повышает его поддерживаемость и делает результаты тестов более информативными при возникновении ошибок.

Подписывайся на @the_daily_python

Memcached — легковесное, высокопроизводительное решение для кэширования данных в оперативной памяти, которое может значительно ускорить ваши приложения.

import pymemcache
import time
import json
from functools import wraps
# Настройка клиента Memcached
client = pymemcache.client.base.Client(('localhost', 11211))
# Сериализатор для сложных типов данных
class JSONSerde:
    def serialize(self, key, value):
        if isinstance(value, str):
            return value.encode('utf-8'), 1
        return json.dumps(value).encode('utf-8'), 2
    
    def deserialize(self, key, value, flags):
        if flags == 1:
            return value.decode('utf-8')
        if flags == 2:
            return json.loads(value.decode('utf-8'))
        return value
# Создаем клиент с сериализатором
json_client = pymemcache.client.base.Client(
    ('localhost', 11211),
    serde=JSONSerde()
)
# Функция-декоратор для кэширования
def cache_result(expire_time=300):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Создаем ключ кэша из имени функции и аргументов
            key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            result = client.get(key)
            
            if result is not None:
                print(f"Загружено из кэша: {key}")
                return json.loads(result.decode('utf-8'))
            
            # Вычисляем результат и сохраняем в кэш
            result = func(*args, **kwargs)
            client.set(key, json.dumps(result).encode('utf-8'), expire=expire_time)
            print(f"Сохранено в кэш: {key}")
            return result
        return wrapper
    return decorator
# Пример использования декоратора кэширования
@cache_result(expire_time=60)
def get_user_data(user_id):
    # Имитация дорогостоящей операции
    time.sleep(2)
    return {"id": user_id, "name": f"User {user_id}", "email": f"user{user_id}@example.com"}
# Демонстрация работы
print("Первый вызов (без кэша):")
start = time.time()
user = get_user_data(123)
print(f"Время выполнения: {time.time() - start:.2f} сек")
print("\nВторой вызов (с кэшем):")
start = time.time()
user = get_user_data(123)
print(f"Время выполнения: {time.time() - start:.2f} сек")
# Использование сложных типов данных
json_client.set('user_list', [1, 2, 3, 4, 5])
print("\nПолучение списка из кэша:", json_client.get('user_list'))

Объясняем
— Базовое использование: Методы set и get для сохранения и получения данных
— Декоратор кэширования: Автоматически кэширует результаты функций
— Сериализация: JSONSerde позволяет хранить сложные объекты (списки, словари)
Время жизни: Параметр expire управляет временем хранения данных

Преимущества Memcached:
— Скорость: Хранение в памяти обеспечивает сверхбыстрый доступ
— Масштабируемость: Легко создать кластер из нескольких серверов
— Простота: Минималистичный API без сложных функций
— Снижение нагрузки: Уменьшает запросы к базе данных и вычисления

Когда использовать:
— Кэширование результатов запросов к БД
— Хранение данных сессий пользователей
— Кэширование результатов API-запросов
— Временное хранение часто используемых расчетов

Memcached — отличный выбор, когда нужно быстрое распределенное кэширование с простым API. Для сложных сценариев с поддержкой персистентности или более продвинутых типов данных стоит рассмотреть Redis, но для базового кэширования Memcached часто остается оптимальным решением благодаря своей скорости и простоте.

Подписывайся на @the_daily_python

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

# Пример использования Rust-функции в Python
import time
import numpy as np
from rust_module import fibonacci_rust  # Подключаем Rust-модуль
# Реализация на чистом Python для сравнения
def fibonacci_python(n: int) -> int:
    if n <= 1:
        return n
    return fibonacci_python(n-1) + fibonacci_python(n-2)
# Сравнение производительности
def benchmark(n: int):
    # Python-версия
    start = time.time()
    result_py = fibonacci_python(n)
    py_time = time.time() - start
    
    # Rust-версия
    start = time.time()
    result_rust = fibonacci_rust(n)
    rust_time = time.time() - start
    
    speedup = py_time / rust_time
    
    print(f"Число Фибоначчи({n}):")
    print(f"Python: {result_py}, время: {py_time:.6f} сек")
    print(f"Rust: {result_rust}, время: {rust_time:.6f} сек")
    print(f"Ускорение: {speedup:.1f}x")
# Запускаем бенчмарк
benchmark(30)

Код Rust-модуля (src/lib.rs):

use pyo3::prelude::*;
#[pyfunction]
fn fibonacci_rust(n: u64) -> u64 {
    match n {
        0 => 0,
        1 => 1,
        _ => fibonacci_rust(n-1) + fibonacci_rust(n-2),
    }
}
#[pymodule]
fn rust_module(_py: Python, m: &PyModule) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(fibonacci_rust, m)?)?;
    Ok(())
}

Объясняем
PyO3: Это фреймворк для создания Python-расширений на Rust
Декораторы: #[pyfunction] для экспорта функций, #[pymodule] для модуля
Компиляция: Создается двоичный .so/.pyd файл, который импортируется в Python

Преимущества Rust в Python:
— Производительность: Ускорение в 10-100 раз для CPU-интенсивных операций
— Безопасность памяти: Rust гарантирует отсутствие утечек и гонок данных
— Многопоточность: Безопасное параллельное выполнение без GIL
— Типобезопасность: Строгая типизация на уровне компиляции

Как начать:
1. Создать проект: cargo new --lib rust_module
2. Добавить в Cargo.toml:

tomlCopy[lib]
name = "rust_module"
crate-type = ["cdylib"]
[dependencies]
pyo3 = { version = "0.19", features = ["extension-module"] }

3. Реализовать функции с декораторами PyO3
4. Собрать: maturin develop (или pip install maturin && maturin develop)

PyO3 — идеальный инструмент, когда критические для производительности части кода становятся узким местом приложения. Сочетание Rust и Python даёт лучшее из обоих миров: скорость разработки и высокую производительность, особенно для математических вычислений, обработки данных и низкоуровневых операций.

Подписывайся на @the_daily_python

Uplink позволяет объявлять API-клиенты в декларативном стиле, преобразуя Python-классы и методы в HTTP-запросы. Это значительно упрощает работу с REST API.

import uplink
from uplink import Query, Path, Body, json
import requests
from typing import List, Dict, Any, Optional
# Определение модели данных для типизации
class Post:
    def __init__(self, id: int, title: str, body: str, userId: int):
        self.id = id
        self.title = title
        self.body = body
        self.userId = userId
# Создание API-клиента
@uplink.json
@uplink.headers({"Accept": "application/json"})
class JsonPlaceholderAPI:
    """Клиент для API JSONPlaceholder"""
    
    def __init__(self, base_url="https://jsonplaceholder.typicode.com"):
        self._base_url = base_url
    
    @uplink.get("/posts")
    def get_posts(self) -> List[Post]:
        """Получить все посты"""
    
    @uplink.get("/posts/{post_id}")
    def get_post(self, post_id: Path) -> Post:
        """Получить пост по ID"""
    
    @uplink.get("/posts")
    def get_posts_by_user(self, userId: Query) -> List[Post]:
        """Получить посты по ID пользователя"""
    
    @uplink.post("/posts")
    def create_post(self, **post: Body) -> Post:
        """Создать новый пост"""
    
    @uplink.put("/posts/{post_id}")
    def update_post(self, post_id: Path, **post: Body) -> Post:
        """Обновить существующий пост"""
    
    @uplink.delete("/posts/{post_id}")
    def delete_post(self, post_id: Path) -> Dict[str, Any]:
        """Удалить пост"""
# Использование клиента
def demo_api():
    # Создаем экземпляр клиента
    api = JsonPlaceholderAPI()
    
    # Получаем посты
    posts = api.get_posts()
    print(f"Всего постов: {len(posts)}")
    
    # Получаем определенный пост
    post = api.get_post(1)
    print(f"Пост #1: {post.title}")
    
    # Создаем новый пост
    new_post = api.create_post(
        title="Новый пост",
        body="Содержимое поста",
        userId=1
    )
    print(f"Создан пост с ID: {new_post.id}")
    
    # Обновляем пост
    updated = api.update_post(
        post_id=1,
        title="Обновленный заголовок",
        body=post.body,
        userId=post.userId
    )
    print(f"Обновлен пост: {updated.title}")
if __name__ == "__main__":
    demo_api()

Объясняем
— Декораторы методов: @uplink.get, @uplink.post и др. определяют HTTP-метод и путь
— Аннотации параметров: Path, Query, Body указывают, где использовать параметр
— Аннотации возврата: Типизация возвращаемых значений для автоматической десериализации
— Декораторы класса: Настройка обработки JSON и заголовков для всех методов

Преимущества uplink:
— Декларативность: Чистый и понятный синтаксис для описания API
— Типизация: Поддержка типов Python для валидации и автодокументации
— Расширяемость: Возможность добавления пользовательских конвертеров и обработчиков
— Переиспользование: Легко тестировать и инжектировать в другие компоненты

Дополнительные возможности:
— Настройка пользовательских сессий requests
— Перехват и модификация запросов/ответов
— Асинхронные запросы с aiohttp
— Обработка ошибок и ретраи

Uplink превращает рутинную работу с REST API в элегантный код, уменьшая количество шаблонного кода и повышая его поддерживаемость. Это особенно полезно при интеграции с несколькими внешними сервисами и регулярном использовании API-клиентов в проекте.

Подписывайся на @the_daily_python

Apache Avro — бинарный формат сериализации данных со схемой, обеспечивающий компактность и высокую производительность.

import avro.schema
from avro.datafile import DataFileReader, DataFileWriter
from avro.io import DatumReader, DatumWriter
import json
import io
# Определяем схему Avro
schema_json = '''
{
    "namespace": "example",
    "type": "record",
    "name": "User",
    "fields": [
        {"name": "id", "type": "int"},
        {"name": "name", "type": "string"},
        {"name": "email", "type": ["string", "null"]},
        {"name": "roles", "type": {"type": "array", "items": "string"}},
        {"name": "active", "type": "boolean", "default": true}
    ]
}
'''
# Парсим схему
schema = avro.schema.parse(schema_json)
# Тестовые данные
user = {
    "id": 1,
    "name": "John Doe",
    "email": "john@example.com",
    "roles": ["admin", "user"],
    "active": True
}
# Сериализация в Avro
def serialize_to_avro(data, schema):
    bytes_io = io.BytesIO()
    writer = DataFileWriter(bytes_io, DatumWriter(), schema)
    writer.append(data)
    writer.close()
    return bytes_io.getvalue()
# Десериализация из Avro
def deserialize_from_avro(binary_data):
    bytes_io = io.BytesIO(binary_data)
    reader = DataFileReader(bytes_io, DatumReader())
    data = list(reader)[0]  # Получаем первую запись
    reader.close()
    return data
# Демонстрация использования
avro_data = serialize_to_avro(user, schema)
print(f"Размер в Avro: {len(avro_data)} байт")
print(f"Размер в JSON: {len(json.dumps(user).encode())} байт")
# Проверка десериализации
restored = deserialize_from_avro(avro_data)
print(f"Восстановленные данные: {restored['name']}")

Объясняем
— Схема: Описывает структуру данных в формате JSON
— Сериализация: Преобразует Python-объекты в компактный бинарный формат
— Десериализация: Восстанавливает исходные данные из бинарного представления

Преимущества Avro:
— Компактность: Занимает меньше места, чем JSON или XML
— Скорость: Эффективная сериализация и десериализация
— Схемы: Поддержка эволюции схем и совместимости версий
— Интеграция: Отлично работает с Hadoop, Kafka и другими системами

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


Подписывайся на @the_daily_python

RestrictedPython позволяет безопасно выполнять недоверенный код, ограничивая доступ к системным функциям и предотвращая потенциальные атаки.

from RestrictedPython import compile_restricted
from RestrictedPython import safe_globals
import math
# Подготовка безопасных глобальных переменных
def get_safe_globals():
    safe_env = safe_globals.copy()
    # Добавляем безопасные математические функции
    safe_env.update({
        'abs': abs,
        'round': round,
        'max': max,
        'min': min,
        'sum': sum,
        'math': math,
    })
    return safe_env
# Функция для выполнения кода в безопасной среде
def execute_safe(code_string, input_data=None):
    try:
        # Компилируем код с ограничениями
        byte_code = compile_restricted(
            code_string, 
            filename='<string>',
            mode='exec'
        )
        
        # Создаем ограниченное пространство имен
        restricted_globals = get_safe_globals()
        local_vars = {'result': None, 'input_data': input_data}
        
        # Выполняем код
        exec(byte_code, restricted_globals, local_vars)
        return local_vars['result']
    except Exception as e:
        return f"Ошибка: {str(e)}"
# Безопасный код
safe_code = """
# Можем использовать математические функции
numbers = input_data['numbers']
result = {
    'sum': sum(numbers),
    'average': sum(numbers) / len(numbers) if numbers else 0,
    'max': max(numbers) if numbers else None,
    'min': min(numbers) if numbers else None,
    'range': max(numbers) - min(numbers) if numbers else 0
}
"""
# Небезопасный код с попыткой доступа к файловой системе
unsafe_code = """
import os  # Попытка импорта системного модуля
result = os.listdir('/')
"""
# Примеры использования
data = {'numbers': [10, 5, 8, 15, 3, 20]}
print("Безопасный код:")
print(execute_safe(safe_code, data))
print("\nНебезопасный код:")
print(execute_safe(unsafe_code))

Объясняем
compile_restricted: Компилирует код с безопасными ограничениями
safe_globals: Предоставляет ограниченный набор функций и объектов
Песочница: Блокирует доступ к системным модулям и опасным функциям

Применения:
Веб-приложения с пользовательскими скриптами
Плагины и расширения для безопасного выполнения
Учебные платформы для запуска кода студентов
Инструменты оценки для тестирования кода

RestrictedPython — мощный инструмент для создания безопасных песочниц, позволяющий выполнять пользовательский код без риска для системы. Особенно полезен для создания расширяемых приложений и образовательных платформ.

Подписывайся на @the_daily_python