Утилиты Python и TypeScript: async, ретраи, типизация

Как это работает: asyncio.Semaphore(limit) позволяет не более limit задач работать одновременно. Каждая корутина оборачивается в bounded, которая ждёт свободного слота через async with semaphore. asyncio.gather запускает всё параллельно, семафор регулирует поток.

import asyncio
from typing import TypeVar, Coroutine, Any

T = TypeVar('T')


async def gather_limited(coros: list[Coroutine[Any, Any, T]], limit: int = 10) -> list[T]:
    """Запускает корутины параллельно, не более limit одновременно."""
    semaphore = asyncio.Semaphore(limit)

    async def bounded(coro: Coroutine[Any, Any, T]) -> T:
        async with semaphore:
            return await coro

    return await asyncio.gather(*(bounded(c) for c in coros))


# Example usage — скачать 100 URL, не более 5 параллельно
import aiohttp

async def fetch(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as resp:
        return await resp.json()

async def main():
    urls = [f'https://api.example.com/items/{i}' for i in range(100)]
    async with aiohttp.ClientSession() as session:
        results = await gather_limited(
            [fetch(session, url) for url in urls],
            limit=5,
        )
    return results

Как это работает: pathlib.Path оборачивает путь файловой системы в объект с методами .is_file()/.is_dir(), что читабельнее и менее подвержено ошибкам, чем строковые вызовы os.path.isfile()/os.path.isdir().

from pathlib import Path

def file_exists(path):
    return Path(path).is_file()

def dir_exists(path):
    return Path(path).is_dir()

# Example usage
if file_exists('config.json'):
    print('Config found')

Как это работает: для каждого элемента — если это список, функция рекурсивно разворачивает его и добавляет результат целиком; иначе элемент добавляется как есть.

def flatten(items):
    result = []
    for item in items:
        if isinstance(item, list):
            result.extend(flatten(item))
        else:
            result.append(item)
    return result

# Example usage
flatten([1, [2, 3, [4, 5]], 6])  # -> [1, 2, 3, 4, 5, 6]

Как это работает: csv.DictReader автоматически использует первую строку как имена полей, поэтому каждая строка возвращается как словарь, доступный по имени столбца, а не по хрупкому числовому индексу.

import csv

def read_csv(path):
    with open(path, newline='', encoding='utf-8') as f:
        return list(csv.DictReader(f))

# Example usage
rows = read_csv('users.csv')
for row in rows:
    print(row['name'], row['email'])

Как это работает: @wraps(func) сохраняет метаданные оригинальной функции. При каждой неудаче задержка удваивается (base_delay * 2 ** attempt), но ограничена max_delay. Джиттер 0–20% разбрасывает повторы многих клиентов во времени.

import time
import random
from functools import wraps


def retry(max_attempts=3, base_delay=0.2, max_delay=2.0, exceptions=(Exception,)):
    """Повторяет вызов с экспоненциальной задержкой и джиттером."""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as exc:
                    if attempt >= max_attempts:
                        raise
                    delay = min(max_delay, base_delay * 2 ** (attempt - 1))
                    jitter = random.uniform(0, delay * 0.2)
                    time.sleep(delay + jitter)
        return wrapper
    return decorator


# Example usage
import requests

@retry(max_attempts=4, base_delay=0.5, exceptions=(requests.RequestException,))
def fetch_data(url: str) -> dict:
    response = requests.get(url, timeout=5)
    response.raise_for_status()
    return response.json()


data = fetch_data('https://api.example.com/items')

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

function chunk<T>(array: T[], size: number): T[][] {
  if (size <= 0) throw new RangeError('chunk size must be > 0')
  const chunks: T[][] = []
  for (let i = 0; i < array.length; i += size) {
    chunks.push(array.slice(i, i + size))
  }
  return chunks
}

// Example usage
chunk([1, 2, 3, 4, 5], 2)  // [[1,2],[3,4],[5]]
chunk(['a','b','c','d'], 3) // [['a','b','c'],['d']]

// Батчевая отправка в Telegram (лимит 30 сообщений/сек)
const userBatches = chunk(userIds, 25)
for (const batch of userBatches) {
  await Promise.all(batch.map((id) => sendMessage(id, text)))
  await sleep(1000)
}

Как это работает: используется reduce, чтобы накопить элементы в объект, где ключи — значения группирующего поля, а значения — массивы подходящих элементов. В современных средах (Node 21+) можно использовать нативный Object.groupBy.

function groupBy<T>(array: T[], key: keyof T): Record<string, T[]> {
  return array.reduce((groups, item) => {
    const groupKey = String(item[key])
    groups[groupKey] = [...(groups[groupKey] ?? []), item]
    return groups
  }, {} as Record<string, T[]>)
}

// Нативная альтернатива (Node 21+ / Chrome 117+)
// const grouped = Object.groupBy(array, (item) => item.status)

// Example usage
const orders = [
  { id: 1, status: 'paid' },
  { id: 2, status: 'pending' },
  { id: 3, status: 'paid' },
]
groupBy(orders, 'status')
// { paid: [{id:1,...},{id:3,...}], pending: [{id:2,...}] }

Как это работает: paginate переводит page/perPage в offset/limit для SQL-запроса. paginationMeta принимает total и возвращает все метаданные для ответа API — удобно для единообразного формата пагинации.

interface PaginationInput {
  page: number
  perPage?: number
}

interface PaginationMeta {
  total: number
  page: number
  perPage: number
  totalPages: number
  hasNext: boolean
  hasPrev: boolean
}

function paginate({ page, perPage = 20 }: PaginationInput) {
  const safePage = Math.max(1, Math.floor(page))
  const limit = Math.max(1, Math.min(perPage, 100))
  const offset = (safePage - 1) * limit
  return { limit, offset, page: safePage }
}

function paginationMeta(total: number, input: PaginationInput): PaginationMeta {
  const { page, limit } = paginate(input)
  const perPage = input.perPage ?? 20
  const totalPages = Math.ceil(total / perPage) || 1
  return { total, page, perPage, totalPages, hasNext: page < totalPages, hasPrev: page > 1 }
}

// Example usage — SQL query
const { limit, offset } = paginate({ page: 3, perPage: 10 })
// SELECT * FROM orders LIMIT 10 OFFSET 20

const meta = paginationMeta(135, { page: 3, perPage: 10 })
// { total: 135, page: 3, perPage: 10, totalPages: 14, hasNext: true, hasPrev: true }

Как это работает: каждый новый вызов execute сначала отменяет предыдущий запрос через AbortController.abort(). Это предотвращает race condition — устаревший ответ не перезапишет свежий. AbortError игнорируется, остальные ошибки пишутся в error.

import { ref } from 'vue'

export function useFetch<T>() {
  const data = ref<T | null>(null)
  const loading = ref(false)
  const error = ref<Error | null>(null)

  let controller: AbortController | null = null

  async function execute(url: string, options: RequestInit = {}) {
    // Отменяем предыдущий запрос
    controller?.abort()
    controller = new AbortController()

    loading.value = true
    error.value = null

    try {
      const res = await fetch(url, { ...options, signal: controller.signal })
      if (!res.ok) throw new Error(`HTTP ${res.status}: ${res.statusText}`)
      data.value = await res.json()
    } catch (e) {
      if ((e as Error).name !== 'AbortError') {
        error.value = e as Error
      }
    } finally {
      loading.value = false
    }
  }

  function abort() {
    controller?.abort()
  }

  return { data, loading, error, execute, abort }
}

// Example usage
const { data: users, loading, error, execute } = useFetch<User[]>()

// Переключение страниц — старый запрос отменяется автоматически
watch(currentPage, (page) => {
  execute(`/api/users?page=${page}`)
})

Как это работает: Zod-схемы определяют ожидаемую структуру и ограничения входящих данных. parse выбрасывает ошибку при невалидных данных; safeParse возвращает объект-результат для обработки ошибок без try/catch.

import { z } from 'zod'

const CreateUserSchema = z.object({
  email: z.string().email(),
  name: z.string().min(2).max(100),
  age: z.number().int().positive().optional(),
  role: z.enum(['user', 'admin']).default('user'),
})

// Использование с safeParse для читаемых ошибок
const result = CreateUserSchema.safeParse(body)

if (!result.success) {
  const messages = result.error.issues.map(
    issue => `${issue.path.join('.')}: ${issue.message}`
  )
  return { ok: false, errors: messages }
}

// result.data полностью типизирован
createUser(result.data)

// Переиспользуемые фрагменты
const PaginationSchema = z.object({
  page: z.coerce.number().int().positive().default(1),
  limit: z.coerce.number().int().min(1).max(100).default(20),
})

const UuidParam = z.object({
  id: z.string().uuid(),
})

// Пример с Express
app.post('/api/users', (req, res) => {
  const parsed = CreateUserSchema.safeParse(req.body)
  if (!parsed.success) return res.status(400).json(parsed.error.flatten())
  res.json(createUser(parsed.data))
})