rkhullar

import csv
from collections.abc import Iterator
from pathlib import Path
from tempfile import TemporaryDirectory


def _build_reverse_mapping(fields: list[str], mapping: dict[str, str] = None) -> dict[str, str]:
    mapping = mapping or dict()
    return {mapping.get(field, field): field for field in fields}

rkhullar

We couldn’t find that file to show.

rkhullar

We couldn’t find that file to show.

rkhullar

import time
from dataclasses import dataclass, field
from typing import Generic, TypeVar
from collections.abc import AsyncIterable


T = TypeVar('T')


async def async_list(stream: AsyncIterable[T]) -> list[T]:

rkhullar

We couldn’t find that file to show.

rkhullar

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

T = TypeVar('T')


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

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