Python and TypeScript utilities: async, retries, typing

How it works: asyncio.Semaphore(limit) allows at most limit tasks to run simultaneously. Each coroutine is wrapped in bounded, which waits for a free slot via async with semaphore. asyncio.gather launches everything in parallel, the semaphore controls the flow.

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]:
    """Run coroutines in parallel, at most `limit` at a time."""
    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 — fetch 100 URLs, at most 5 in parallel
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

How it works: pathlib.Path wraps filesystem paths in an object with .is_file()/.is_dir() methods, which is more readable and less error-prone than string-based os.path.isfile()/os.path.isdir() calls.

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')

How it works: for each item, if it's itself a list the function recurses into it and extends the result; otherwise the item is appended as-is.

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]

How it works: csv.DictReader uses the first row as field names automatically, so each row comes back as a dict you can access by column name instead of a fragile numeric index.

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'])

How it works: @wraps(func) preserves the original function's metadata. On each failure the delay doubles (base_delay * 2 ** attempt), capped at max_delay. A 0–20% jitter spreads retries from many clients across time.

import time
import random
from functools import wraps


def retry(max_attempts=3, base_delay=0.2, max_delay=2.0, exceptions=(Exception,)):
    """Retry a call with exponential backoff and jitter."""
    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')

How it works: iterates over the array in steps of size, slicing a chunk at each step. The last chunk may be smaller than size if there aren't enough elements.

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']]

// Batch Telegram sends (30 msg/sec limit)
const userBatches = chunk(userIds, 25)
for (const batch of userBatches) {
  await Promise.all(batch.map((id) => sendMessage(id, text)))
  await sleep(1000)
}

How it works: uses reduce to accumulate items into an object where keys are the grouping field values and values are arrays of matching elements. In modern environments (Node 21+) you can use the native Object.groupBy instead.

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[]>)
}

// Native alternative (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,...}] }

How it works: paginate converts page/perPage into offset/limit for a SQL query. paginationMeta takes the total count and returns all metadata for a consistent API response format.

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 } = 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 }

How it works: every new execute call first cancels the previous request via AbortController.abort(). This prevents race conditions — a stale response won't overwrite a fresh one. AbortError is ignored; all other errors are written to 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 = {}) {
    // Cancel the previous request
    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[]>()

// Switching pages — the old request is cancelled automatically
watch(currentPage, (page) => {
  execute(`/api/users?page=${page}`)
})

How it works: Zod schemas define the expected shape and constraints of incoming data. parse throws on invalid data; safeParse returns a result object so you can handle errors gracefully without 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'),
})

// Use with safeParse for readable errors
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 is fully typed
createUser(result.data)

// Reusable fragments
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 handler example
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))
})