# How to Fix Python ConnectionRefusedError
The ConnectionRefusedError occurs when a network connection attempt is actively rejected by the remote host. This typically means the target server is not listening on the specified port, or a firewall is blocking the connection.
Introduction
This article covers troubleshooting steps and solutions for How to Fix Python ConnectionRefusedError. The error typically occurs in production environments and can cause service disruptions if not addressed promptly.
Symptoms
Basic ConnectionRefusedError
Traceback (most recent call last):
File "app.py", line 5, in <module>
socket.connect(('localhost', 8080))
ConnectionRefusedError: [Errno 111] Connection refusedHTTP Request Connection Refused
Traceback (most recent call last):
File "app.py", line 10, in <module>
requests.get('http://localhost:8080/api')
ConnectionRefusedError: [Errno 111] Connection refusedDatabase Connection Refused
Traceback (most recent call last):
File "app.py", line 15, in <module>
psycopg2.connect(host='localhost', port=5432)
ConnectionRefusedError: [Errno 111] Connection refusedWebSocket Connection Refused
Traceback (most recent call last):
File "app.py", line 20, in <module>
websocket.connect('ws://localhost:8080/ws')
ConnectionRefusedError: [Errno 111] Connection refusedCommon Causes
- 1.Server not running - Target service is not started
- 2.Wrong port - Connecting to different port than server listens on
- 3.Wrong host - Server on different IP address
- 4.Firewall blocking - Network firewall rejecting connections
- 5.Service crashed - Server process crashed or stopped
- 6.Port not bound - Server not configured to listen on port
- 7.IPv4/IPv6 mismatch - Connecting with wrong protocol version
- 8.Local firewall - Host firewall blocking inbound connections
Step-by-Step Fix
Step 1: Check if Server is Running
```bash # Linux/Mac ps aux | grep nginx # Check process ps aux | grep python netstat -tlnp # Listening ports ss -tlnp # Alternative to netstat
# Windows tasklist | findstr python netstat -an # All connections ```
Step 2: Check Port is Listening
```bash # Check specific port netstat -tlnp | grep 8080 ss -tlnp | grep 8080 lsof -i :8080
# Or test with nc nc -zv localhost 8080 # Output: localhost 8080 open (if listening) # Output: localhost 8080 refused (if not) ```
Step 3: Check Firewall Rules
```bash # Linux (iptables) iptables -L -n
# Linux (firewalld) firewall-cmd --list-all
# Linux (ufw) ufw status
# Windows netsh advfirewall show allprofiles ```
Step 4: Test from Python
```python import socket
def test_connection(host, port): """Test if connection is possible.""" sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(2)
try: result = sock.connect_ex((host, port)) if result == 0: print(f"Connection to {host}:{port} successful") return True else: print(f"Connection failed: error code {result}") return False finally: sock.close()
# Test common ports test_connection('localhost', 80) # HTTP test_connection('localhost', 443) # HTTPS test_connection('localhost', 5432) # PostgreSQL test_connection('localhost', 3306) # MySQL ```
Solution 1: Start the Server
```bash # If server is not running, start it
# HTTP server python -m http.server 8080 # Simple HTTP server
# Flask app python app.py # Or flask run
# Django python manage.py runserver 0.0.0.0:8080
# PostgreSQL sudo systemctl start postgresql
# MySQL sudo systemctl start mysql
# Redis redis-server
# Custom server python server_script.py ```
```python # Quick HTTP server in Python from http.server import HTTPServer, SimpleHTTPRequestHandler
server = HTTPServer(('localhost', 8080), SimpleHTTPRequestHandler) print("Starting server on port 8080...") server.serve_forever() ```
Solution 2: Verify Correct Port
```python import socket
# Problem: Wrong port sock = socket.socket() sock.connect(('localhost', 8000)) # Server on 8080
# Fix: Use correct port sock.connect(('localhost', 8080)) # Correct port
# Discover correct port def find_service_port(service_name): """Get standard port for service.""" ports = { 'http': 80, 'https': 443, 'ssh': 22, 'ftp': 21, 'postgresql': 5432, 'mysql': 3306, 'redis': 6379, 'mongodb': 27017, 'rabbitmq': 5672, } return ports.get(service_name.lower())
# Check config files for actual port import configparser config = configparser.ConfigParser() config.read('server.conf') port = config.getint('server', 'port', fallback=8080) ```
Solution 3: Check Host Address
```python import socket
# Problem: Wrong host sock.connect(('127.0.0.1', 8080)) # Server on different IP
# Fix: Check available addresses def find_server_addresses(port): """Find addresses listening on port.""" import subprocess result = subprocess.run(['ss', '-tlnp'], capture_output=True, text=True)
addresses = [] for line in result.stdout.split('\n'): if str(port) in line: parts = line.split() if len(parts) >= 4: addr = parts[3].split(':')[0] addresses.append(addr)
return addresses
# Or try all common addresses addresses = ['localhost', '127.0.0.1', '0.0.0.0', '::1'] for addr in addresses: try: sock = socket.socket() sock.settimeout(1) sock.connect((addr, 8080)) print(f"Connected to {addr}:8080") break except ConnectionRefusedError: continue ```
Solution 4: Handle Firewall Issues
```bash # Linux - Allow port sudo iptables -A INPUT -p tcp --dport 8080 -j ACCEPT
# Or using firewalld sudo firewall-cmd --add-port=8080/tcp --permanent sudo firewall-cmd --reload
# Or using ufw sudo ufw allow 8080/tcp
# Windows netsh advfirewall firewall add rule name="Allow 8080" dir=in action=allow protocol=tcp localport=8080 ```
```python # Python code to handle firewall gracefully def connect_with_retry(host, port, retries=5, delay=1): """Connect with retry logic.""" import time
for attempt in range(retries): try: sock = socket.socket() sock.settimeout(5) sock.connect((host, port)) print(f"Connected after {attempt + 1} attempts") return sock except ConnectionRefusedError: if attempt < retries - 1: print(f"Connection refused, retrying in {delay}s...") time.sleep(delay) else: print("Connection refused after all retries") raise ```
Solution 5: Bind Server to Correct Address
```python # Server code - bind to accessible address
# Problem: Server bound to 127.0.0.1 only server = HTTPServer(('127.0.0.1', 8080), handler) # Only localhost can connect
# Fix: Bind to all interfaces server = HTTPServer(('0.0.0.0', 8080), handler) # Any address can connect
# Flask example # app.run(host='0.0.0.0', port=8080) # Accessible from all
# Socket server example import socket
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_sock.bind(('0.0.0.0', 8080)) # Bind to all interfaces server_sock.listen(5)
print("Server listening on 0.0.0.0:8080") ```
Solution 6: Handle IPv4/IPv6 Mismatch
```python import socket
# Problem: IPv4 connection to IPv6 server sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect(('localhost', 8080)) # Fails if server is IPv6
# Fix: Use IPv6 socket sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) sock.connect(('::1', 8080)) # IPv6 localhost
# Fix: Use dual-stack (works for both) sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) # Dual stack sock.connect(('localhost', 8080))
# Best: Use socket.create_connection (auto-selects) sock = socket.create_connection(('localhost', 8080)) ```
Solution 7: Implement Connection Pooling
```python import socket import time from queue import Queue
class ConnectionPool: """Pool of reusable connections."""
def __init__(self, host, port, pool_size=5, timeout=30): self.host = host self.port = port self.pool_size = pool_size self.timeout = timeout self.pool = Queue(maxsize=pool_size) self._initialize_pool()
def _initialize_pool(self): for _ in range(self.pool_size): try: sock = self._create_connection() self.pool.put(sock) except ConnectionRefusedError: print("Warning: Could not initialize full pool")
def _create_connection(self): sock = socket.create_connection((self.host, self.port), timeout=self.timeout) return sock
def get_connection(self): try: sock = self.pool.get(timeout=5) # Verify connection is still valid try: sock.send(b'') # Test return sock except: return self._create_connection() except: return self._create_connection()
def return_connection(self, sock): try: self.pool.put(sock, timeout=1) except: sock.close()
# Usage pool = ConnectionPool('localhost', 8080) conn = pool.get_connection() try: conn.send(data) finally: pool.return_connection(conn) ```
Solution 8: Use Connection Timeout Gracefully
```python import socket
def safe_connect(host, port, timeout=5, message=None): """Connect with comprehensive error handling.""" if message: print(f"Connecting to {host}:{port}...")
try: sock = socket.create_connection((host, port), timeout=timeout) if message: print("Connection successful") return sock
except ConnectionRefusedError: print(f"Connection refused by {host}:{port}") print("Possible causes:") print(" - Server not running") print(" - Wrong port number") print(" - Firewall blocking connection") return None
except socket.timeout: print(f"Connection timed out after {timeout}s") print("Possible causes:") print(" - Server overloaded") print(" - Network latency") return None
except socket.gaierror: print(f"Could not resolve hostname: {host}") return None
except OSError as e: print(f"OS error: {e}") return None
# Usage sock = safe_connect('localhost', 8080, message=True) if sock: # Use connection sock.close() ```
Service-Specific Troubleshooting
PostgreSQL
```python import psycopg2
def connect_postgresql(host='localhost', port=5432, dbname='postgres'): """Connect to PostgreSQL with error handling.""" try: conn = psycopg2.connect( host=host, port=port, dbname=dbname ) return conn except ConnectionRefusedError: print("PostgreSQL connection refused") print("Check if PostgreSQL is running:") print(" sudo systemctl status postgresql") print("Check PostgreSQL port:") print(" cat /etc/postgresql/*/main/postgresql.conf | grep port") raise ```
Redis
```python import redis
def connect_redis(host='localhost', port=6379): """Connect to Redis with error handling.""" try: client = redis.Redis(host=host, port=port) client.ping() # Test connection return client except redis.exceptions.ConnectionError: print("Redis connection refused") print("Check if Redis is running:") print(" redis-cli ping") print("Start Redis:") print(" redis-server") raise ```
HTTP/Web APIs
```python import requests from requests.exceptions import ConnectionError
def safe_request(url, timeout=5, retries=3): """Make HTTP request with error handling.""" for attempt in range(retries): try: response = requests.get(url, timeout=timeout) return response except ConnectionError as e: if 'Connection refused' in str(e): print(f"Server refused connection: {url}") if attempt < retries - 1: print("Retrying...") continue raise return None ```
Verification
After applying the fixes, verify that the ConnectionRefusedError is resolved by testing the connection:
```python import socket
def verify_connection(host, port): """Verify that connection works correctly.""" try: sock = socket.create_connection((host, port), timeout=5) print(f"Successfully connected to {host}:{port}") sock.close() return True except ConnectionRefusedError: print(f"Connection still refused to {host}:{port}") return False except Exception as e: print(f"Connection error: {e}") return False
# Run verification verify_connection('localhost', 8080) ```
Prevention
- 1.Check server status before connecting
- 2.Verify port number in configuration
- 3.Use connection retry logic for transient failures
- 4.Implement health checks in your clients
- 5.Use connection pools for repeated connections
```python # Good pattern: Connection with health check class ConnectionManager: def __init__(self, host, port): self.host = host self.port = port self._sock = None
def connect(self): """Establish connection.""" self._sock = socket.create_connection((self.host, self.port)) return self._sock
def is_connected(self): """Check if connection is valid.""" if not self._sock: return False try: self._sock.send(b'') return True except: return False
def reconnect(self): """Reconnect if needed.""" if not self.is_connected(): return self.connect() return self._sock
def close(self): """Close connection.""" if self._sock: self._sock.close() self._sock = None ```
Related Errors
ConnectionResetError- Connection closed by remotesocket.timeout- Connection attempt timed outsocket.gaierror- Hostname resolution failedTimeoutError- Operation timed out
Related Articles
- [WordPress troubleshooting: Fix Django TypeError - Complete Troubles](fix-django-typeerror)
- [WordPress troubleshooting: Fix async task exception not awaited Iss](async-task-exception-not-awaited)
- [WordPress troubleshooting: Fix FastAPI AttributeError - Complete Tr](fix-fastapi-attributeerror)
- [WordPress troubleshooting: Fix Flask AttributeError - Complete Trou](fix-flask-attributeerror)
- [WordPress troubleshooting: Fix asyncio event loop closed rerun Issu](asyncio-event-loop-closed-rerun)
<script type="application/ld+json"> { "@context": "https://schema.org", "@type": "TechArticle", "headline": "How to Fix Python ConnectionRefusedError", "description": "Solve Python ConnectionRefusedError when connections are rejected. Fix server availability, port binding, firewall issues, and network configuration.", "url": "https://www.fixwikihub.com/fix-python-connectionrefusederror", "publisher": { "@type": "Organization", "name": "FixWikiHub", "url": "https://www.fixwikihub.com" }, "author": { "@type": "Person", "name": "FixWikiHub Editorial Team" }, "datePublished": "2025-11-27T11:04:46.320Z", "dateModified": "2025-11-27T11:04:46.320Z" } </script>