Wenbo Zhao zhaowb

zhaowb / threadpool_imap.py

Last active September 26, 2022 01:48

multithreading.pool.ThreadPool.imap with limited memory footprint

	from multiprocessing.pool import ThreadPool
	import queue
	from typing import Any, Callable, Iterator


	def threadpool_imap(func: Callable[[Any], Any],
	iterable: Iterator[Any],
	*,
	num_thread: int = 4
	) -> Iterator[Any]:

zhaowb / concurrent_imap.py

Last active September 26, 2022 01:47

Use concurrent.futures.ThreadPoolExecutor do similar as multiprocessing.ThreadPool.imap, with limited memory footprint

	import queue
	from concurrent.futures import ThreadPoolExecutor
	from contextlib import contextmanager
	from typing import Any, Callable, Iterator


	@contextmanager
	def concurrent_imap(func: Callable[[Any], Any],
	iterable: Iterator[Any],
	*,

zhaowb / gist:7d5eac4fc6ef20ee29fa6907e558438d

Last active June 5, 2019 22:53

stripe pagination using auto_paging_iter()

	In [2]: inv100 = stripe.Invoice.list(limit=100)
	In [8]: from itertools import islice
	In [9]: [(i, inv.id, inv.created) for i, inv in islice(enumerate(inv100.data), 20)]
	Out[9]:
	[(0, 'in_1EhxnEGObuJZja1rsNGjJcNe', 1559736000),
	(1, 'in_1EhxnFGObuJZja1rT295LqNl', 1559736000),
	(2, 'in_1EhxnFGObuJZja1r9xZutCpP', 1559736000),
	(3, 'in_1EhxnGGObuJZja1riY3HzPDl', 15597360

zhaowb / gist:5c148a0a6f6d337f3259905b0f13e4ea

Created June 4, 2019 19:21

stripe list paging incorrect with starting_after

	In [2]: pg1 = stripe.Invoice.list()

	In [3]: pg2 = stripe.Invoice.list(starting_after=pg1.data[-1].id)

	In [4]: pg2.data[0].id
	Out[4]: 'in_1EgJwwDhOpaK71Z9NO4eU8Jz'

	In [5]: invs100 = stripe.Invoice.list(limit=100)

	In [6]: [i.id for i in invs100].index(pg2.data[0].id)