Skip to content

Instantly share code, notes, and snippets.

@methane

methane/niquests_block.py

Created March 13, 2026 11:11
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save methane/8bc735c373747fe9c5d57c5f12eadf69 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save methane/8bc735c373747fe9c5d57c5f12eadf69 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
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()
Raw
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()
@methane
Copy link
Author

methane commented Mar 13, 2026 

$ uv run niquests_block.py
2026-03-13 20:12:42,085 MainThread DEBUG Converted retries value: 0 -> Retry(total=0, connect=None, read=None, redirect=None, status=None)
2026-03-13 20:12:42,085 MainThread DEBUG Converted retries value: 0 -> Retry(total=0, connect=None, read=None, redirect=None, status=None)
2026-03-13 20:12:42,085 MainThread DEBUG Converted retries value: 0 -> Retry(total=0, connect=None, read=None, redirect=None, status=None)
pools.maxsize=10, pools.concurrency=True, pools.strict_maxsize=True
2026-03-13 20:12:42,089 worker-1 DEBUG Starting new HTTP connection (1): localhost:3001
2026-03-13 20:12:44,138 worker-1 DEBUG http://localhost:3001 "GET /delay/2 HTTP/1.1" 200 317
2026-03-13 20:12:46,185 worker-0 DEBUG http://localhost:3001 "GET /delay/2 HTTP/1.1" 200 317
2026-03-13 20:12:48,445 worker-1 DEBUG http://localhost:3001 "GET /delay/2 HTTP/1.1" 200 317
2026-03-13 20:12:50,476 worker-0 DEBUG http://localhost:3001 "GET /delay/2 HTTP/1.1" 200 317
2026-03-13 20:12:52,729 worker-1 DEBUG http://localhost:3001 "GET /delay/2 HTTP/1.1" 200 317
2026-03-13 20:12:54,768 worker-0 DEBUG http://localhost:3001 "GET /delay/2 HTTP/1.1" 200 317
2026-03-13 20:12:57,015 worker-1 DEBUG http://localhost:3001 "GET /delay/2 HTTP/1.1" 200 317
2026-03-13 20:12:59,048 worker-0 DEBUG http://localhost:3001 "GET /delay/2 HTTP/1.1" 200 317

@methane
Copy link
Author

methane commented Mar 13, 2026 •
edited
Loading

Before run these scripts, you need to run:
docker run -p 127.0.0.1:3001:80 kennethreitz/httpbin

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment