rkhullar

import functools
import time

class timeit:
  """decorator for benchmarking"""
  def __init__(self, fmt='completed {:s} in {:.3f} seconds'):
    # there is no need to make a class for a decorator if there are no parameters
    self.fmt = fmt
  
  def __call__(self, fn):

rkhullar

function pipe(data, fn) {
    if (typeof data === 'object' && data != null) {
        return Object.fromEntries(
            Object.keys(data).map(key => [key, fn(key, data[key])])
        );
    }
}

const example = {
    hello: 1,

rkhullar

#!/usr/bin/env python3

import os
import re
import subprocess
from argparse import ArgumentParser, Namespace
from dataclasses import dataclass, field


def build_parser() -> ArgumentParser:

rkhullar

from collections.abc import Iterator
from dataclasses import dataclass, field
from typing import Generic, Optional, TypeVar

T = TypeVar('T')
V = TypeVar('V')


@dataclass
class CapturedIteration(Generic[T, V]):

rkhullar

from collections.abc import Iterable, Iterator
from typing import TypeVar

T = TypeVar('T')


def _scan(items: Iterable[T]) -> Iterator[tuple[T, T]]:

    """iterator to traverse previous and current items from iterable"""
    prev, curr, stream = None, None, iter(items)

rkhullar

from collections.abc import AsyncIterator
from dataclasses import dataclass, field

from httpx_util import BearerAuth, async_httpx

# https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/CoreV1Api.md
# https://github.com/kubernetes-client/python/blob/master/kubernetes/docs/BatchV1Api.md


@dataclass(frozen=True)

rkhullar

from dataclasses import dataclass

import httpx


async def async_httpx(method: str, *args, verify: bool = True, **kwargs):
    async with httpx.AsyncClient(verify=verify) as client:
        fn = getattr(client, method)
        return await fn(*args, **kwargs)

rkhullar

from typing import Iterator, TypeVar, List

T = TypeVar('T')


def iter_batch(stream: Iterator[T], batch_size: int) -> Iterator[List[T]]:
    assert batch_size > 0
    buffer, item_count = [None] * batch_size, 0
    for item in stream:
        buffer_index = item_count % batch_size

rkhullar

from typing import Iterator, List, TypeVar

T = TypeVar('T')


def iter_batch(sequence: Iterator[T], size: int) -> Iterator[List[T]]:
    count, buffer = 0, [None] * size
    for item in sequence:
        idx = count % size
        buffer[idx] = item

rkhullar

from dataclasses import dataclass, field
from typing import Generic, TypeVar

T = TypeVar('T')


@dataclass
class Proxy(Generic[T]):
    _data: T