methane · March 13, 2026 11:11 · methane · Mar 13, 2026 · methane · Mar 13, 2026
diff --git a/niquests_block.py b/niquests_block.py
 # /// script
 # requires-python = ">=3.14"
 # dependencies = [
 #     "niquests>=3.18.2",
 # ]
 # ///
 import niquests
 from niquests.adapters import HTTPAdapter
 import threading
 import time
 import sys
 import logging
 import faulthandler

 logging.basicConfig(
    level=logging.DEBUG, format="%(asctime)s %(threadName)s %(levelname)s %(message)s"
 )
 logger = logging.getLogger("niquests_block")

 session = niquests.Session(pool_maxsize=1)


 def get_worker():
    for _ in range(4):
        try:
            response = session.get(f"http://localhost:3001/delay/2")
            response.raise_for_status()
        except Exception as e:
            logger.exception("request failed: %s", e)
        time.sleep(0.2)


 def main():
    threads = []
    for i in range(2):
        t = threading.Thread(target=get_worker, name=f"worker-{i}")
        t.start()
        threads.append(t)

    #for _ in range(7):
    #    faulthandler.dump_traceback()
    #    time.sleep(1)

    for t in threads:
        t.join()


 pools = session.adapters["http://"].poolmanager.pools
 print(f"{pools.maxsize=}, {pools.concurrency=}, {pools.strict_maxsize=}")

 if __name__ == "__main__":
    main()
diff --git a/requests_block.py b/requests_block.py
 # /// script
 # requires-python = ">=3.14"
 # dependencies = [
 #     "requests>=2.32.5",
 # ]
 # ///
 import requests
 from requests.adapters import HTTPAdapter
 import threading
 import time
 import sys
 import logging

 logging.basicConfig(
    level=logging.DEBUG, format="%(asctime)s %(threadName)s %(levelname)s %(message)s"
 )
 logger = logging.getLogger("requests_block")

 session = requests.Session()
 session.mount("http://", HTTPAdapter(pool_maxsize=1, pool_block=False))
 session.mount("https://", HTTPAdapter(pool_maxsize=1, pool_block=False))


 def get_worker():
    for _ in range(4):
        try:
            response = session.get(f"http://localhost:3001/delay/2")
            response.raise_for_status()
        except Exception as e:
            logger.exception("request failed: %s", e)
        time.sleep(0.2)


 def main():
    threads = []
    for i in range(2):
        t = threading.Thread(target=get_worker, name=f"worker-{i}")
        t.start()
        threads.append(t)

    for t in threads:
        t.join()


 if __name__ == "__main__":
    main()
	# /// script
	# requires-python = ">=3.14"
	# dependencies = [
	# "niquests>=3.18.2",
	# ]
	# ///
	import niquests
	from niquests.adapters import HTTPAdapter
	import threading
	import time
	import sys
	import logging
	import faulthandler

	logging.basicConfig(
	level=logging.DEBUG, format="%(asctime)s %(threadName)s %(levelname)s %(message)s"
	)
	logger = logging.getLogger("niquests_block")

	session = niquests.Session(pool_maxsize=1)


	def get_worker():
	for _ in range(4):
	try:
	response = session.get(f"http://localhost:3001/delay/2")
	response.raise_for_status()
	except Exception as e:
	logger.exception("request failed: %s", e)
	time.sleep(0.2)


	def main():
	threads = []
	for i in range(2):
	t = threading.Thread(target=get_worker, name=f"worker-{i}")
	t.start()
	threads.append(t)

	#for _ in range(7):
	# faulthandler.dump_traceback()
	# time.sleep(1)

	for t in threads:
	t.join()


	pools = session.adapters["http://"].poolmanager.pools
	print(f"{pools.maxsize=}, {pools.concurrency=}, {pools.strict_maxsize=}")

	if __name__ == "__main__":
	main()
No results found