Python Timeouts

An unresponsive service can be worse than a down one. It can tie up your entire system if not handled properly. All network requests should have a timeout.

Here’s how to add timeouts for popular Python packages. All have been tested. The default is no timeout, unless otherwise specified. Enjoy!

Also available for Ruby, Node, Go, PHP, and Rust

Packages

Standard Library

• ftplib
• http
• imaplib
• nntplib
• poplib
• smtplib
• socket
• subprocess
• telnetlib
• urllib

Data Stores

• asyncpg
• cassandra-driver
• elasticsearch
• influxdb
• mongoengine
• mysqlclient
• opensearch-py
• pg8000
• psycopg
• psycopg2
• pymemcache
• pymongo
• redis
• SQLAlchemy
• trino
• typesense
• valkey

HTTP Clients

• aiohttp
• httpx
• requests
• urllib3

Solvers

• osqp
• scs

3rd Party Services

• boto3
• PyGithub
• stripe

Standard Library

ftplib

FTP(host, timeout=1)

Raises socket.timeout

http

HTTPConnection(host, port, timeout=1)

Raises socket.timeout

imaplib

IMAP4(host, timeout=1)

Raises socket.timeout

Note: Requires Python 3.9+

nntplib

NNTP(host, timeout=1)

Raises socket.timeout

poplib

POP3(host, timeout=1)

Raises socket.timeout

smtplib

SMTP(host, timeout=1)

Raises

• socket.timeout on connect timeout
• smtplib.SMTPServerDisconnected on read timeout

socket

sock.settimeout(1)

Raises socket.timeout

subprocess

subprocess.run(cmd, timeout=1)

Raises subprocess.TimeoutExpired

telnetlib

Telnet(host, timeout=1)

Raises socket.timeout

urllib

urlopen(url, timeout=1)

Raises

• urllib.error.URLError on connect timeout
• socket.timeout on read timeout

Data Stores

asyncpg

asyncpg.connect(timeout=1)

Default: 60s

Raises asyncio.exceptions.TimeoutError

cassandra-driver

Cluster([host], connect_timeout=1)

Raises cassandra.cluster.NoHostAvailable on connect timeout

elasticsearch

Elasticsearch(request_timeout=1)

Raises elastic_transport.ConnectionTimeout

influxdb

InfluxDBClient(timeout=1)

Raises

• requests.exceptions.ConnectTimeout on connect timeout
• requests.exceptions.ReadTimeout on read timeout

mongoengine

connect(connectTimeoutMS=1000, socketTimeoutMS=1000, serverSelectionTimeoutMS=1000)

Raises pymongo.errors.ServerSelectionTimeoutError

mysqlclient

MySQLdb.connect(connect_timeout=1)

Raises MySQLdb._exceptions.OperationalError

opensearch-py

OpenSearch(timeout=1)

Raises opensearchpy.exceptions.ConnectionError

pg8000

pg8000.connect(timeout=1)

Raises

• pg8000.exceptions.InterfaceError on connect timeout
• socket.timeout on read timeout

psycopg

psycopg.connect(connect_timeout=1)

Raises psycopg.OperationalError

psycopg2

psycopg2.connect(connect_timeout=1)

Raises psycopg2.OperationalError

pymemcache

Client(host, connect_timeout=1, timeout=1)

Raises socket.timeout

pymongo

MongoClient(connectTimeoutMS=1000, socketTimeoutMS=1000, serverSelectionTimeoutMS=1000)

Default: 20s connect timeout, 30s server selection timeout

Raises pymongo.errors.ServerSelectionTimeoutError

redis

Redis(socket_connect_timeout=1, socket_timeout=1)

Raises redis.exceptions.TimeoutError

SQLAlchemy

create_engine(url, connect_args={'connect_timeout': 1})

Raises sqlalchemy.exc.OperationalError

trino

trino.dbapi.connect(request_timeout=1)
# or
trino.dbapi.connect(request_timeout=(1, 1)) # (connect, read)

Raises trino.exceptions.TrinoConnectionError

typesense

Client({'connection_timeout_seconds': 1})

Raises

• requests.exceptions.ConnectTimeout on connect timeout
• requests.exceptions.ReadTimeout on read timeout

valkey

Valkey(socket_connect_timeout=1, socket_timeout=1)

Raises valkey.exceptions.TimeoutError

HTTP Clients

aiohttp

timeout = aiohttp.ClientTimeout(total=1)
async with aiohttp.ClientSession(timeout=timeout) as session:
    # ...

Raises asyncio.exceptions.TimeoutError

httpx

httpx.get(url, timeout=1)
# or
httpx.Client(timeout=1)

Raises

• httpx.ConnectTimeout on connect timeout
• httpx.ReadTimeout on read timeout

requests

requests.get(url, timeout=1)

Raises

• requests.exceptions.ConnectTimeout on connect timeout
• requests.exceptions.ReadTimeout on read timeout

urllib3

http = urllib3.PoolManager(timeout=urllib3.Timeout(connect=1, read=1))
# or
http.request('GET', url, timeout=urllib3.Timeout(connect=1, read=1))

Raises urllib3.exceptions.MaxRetryError

Solvers

osqp

m = osqp.OSQP()
m.setup(..., time_limit=1)
results = m.solve()

Check for a results.info.status of run time limit reached for a timeout

scs

sol = scs.solve(data, cone, time_limit_secs=1)

Check for a sol['info']['status'] of solved (inaccurate - reached time_limit_secs) for a timeout

3rd Party Services

boto3

boto3.client('s3', config=Config(connect_timeout=1, read_timeout=1))

Raises

• botocore.exceptions.ConnectTimeoutError on connect timeout
• botocore.exceptions.ReadTimeoutError on read timeout

PyGithub

Github(timeout=1)

Raises

• requests.exceptions.ConnectTimeout on connect timeout
• requests.exceptions.ReadTimeout on read timeout

stripe

StripeClient(api_key, http_client=stripe.RequestsClient(timeout=1))

Raises stripe.APIConnectionError

Don’t see a library you use?

Let us know. Even better, create a pull request for it.

Running the Tests

git clone https://github.com/ankane/python-timeouts.git
cd python-timeouts
pip install --group dev

To run all tests, use:

pytest

To run individual tests, use:

pytest tests/test_redis.py