Rajan Khullar rkhullar

rkhullar / csv-util.py

Last active November 27, 2023 22:03

csv util with flexible headers

	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 / !terraform-project-structure

Last active September 10, 2023 23:39

tutorials/terraform-project-structure

We couldn’t find that file to show.

rkhullar / !terraform-example-module

Last active September 8, 2023 23:39

tutorials/terraform-example-module

We couldn’t find that file to show.

rkhullar / dataclass-with-cached-field.py

Last active June 11, 2023 18:01

dataclass with cached field

	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 / !fastapi-aws-mongodb-okta

Last active August 1, 2023 01:37

tutorials/fastapi-aws-mongodb-okta

We couldn’t find that file to show.

rkhullar / frozen-dataclass-with-mutable-field.py

Created April 25, 2023 22:16

frozen dataclass with mutable field

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

	T = TypeVar('T')


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

rkhullar / python-iter-batch-v1.py

Created April 4, 2023 14:33

python iter batch v1

	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 / python-iter-batch-v2.py

Created April 4, 2023 14:32

python iter batch v2

	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 / httpx_util.py

Created April 4, 2023 14:31

httpx util

	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 / k8s_client.py

Last active April 4, 2023 14:30

custom python kubernetes client

	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)