Docker Compose环境下Python应用无法连接PostgreSQL求助
Docker中Python应用连接PostgreSQL失败的解决方法
问题1:数据库主机名配置错误
- 原因:Docker Compose默认网络中,容器间通过服务名通信,而非
localhost。你的.env中DB_HOST=localhost会让Python容器尝试连接自身的localhost,而非PostgreSQL容器。 - 解决:修改
.env中的DB_HOST为db(对应docker-compose里的db服务名):
DB_NAME=database_dev DB_USER=postgres DB_PASSWORD=secret DB_HOST=db DB_PORT=5432
问题2:Python代码中URL格式化未生效
- 原因:
str.format()返回新字符串,但你未将结果赋值回self.db_url,导致原始带占位符的字符串被传给create_engine,触发"{hostname}"无法解析的错误。同时URL中的\\@是多余转义,需删除。 - 解决:修正代码逻辑:
class DatabaseManager: def __init__(self): self.db_url = 'postgresql+psycopg2://{user}:{password}@{hostname}/{database_name}' # 将格式化结果赋值回变量 self.db_url = self.db_url.format( user=os.environ['DB_USER'], password=os.environ['DB_PASSWORD'], hostname=os.environ['DB_HOST'], database_name=os.environ['DB_NAME'] ) self.engine = create_engine(self.db_url) def make_session(self): self.Session = sessionmaker(bind=self.engine) self.session = self.Session() def add_data(self, data): self.session.add(data) self.session.commit() def close(self): self.session.close()
问题3:数据库未就绪时应用尝试连接
- 原因:
depends_on: - db仅保证db容器先启动,但PostgreSQL服务可能未完成初始化,此时应用连接会失败。 - 解决:
- 在docker-compose的db服务中添加健康检查,确保数据库就绪后再启动app:
version: '3' services: db: image: postgres:13.4-alpine environment: POSTGRES_PASSWORD: secret POSTGRES_HOST_AUTH_METHOD: trust env_file: - .env ports: - "5432:5432" volumes: - ./database/init.sql:/docker-entrypoint-initdb.d/init.sql healthcheck: test: ["CMD-SHELL", "pg_isready -U postgres -d database_dev"] interval: 5s timeout: 5s retries: 5 app: build: . restart: always depends_on: db: condition: service_healthy stdin_open: true tty: true env_file: - .env
- 或在Python代码中添加连接重试逻辑:
from sqlalchemy import create_engine from sqlalchemy.exc import OperationalError import time import os class DatabaseManager: def __init__(self): self.db_url = 'postgresql+psycopg2://{user}:{password}@{hostname}/{database_name}'.format( user=os.environ['DB_USER'], password=os.environ['DB_PASSWORD'], hostname=os.environ['DB_HOST'], database_name=os.environ['DB_NAME'] ) max_retries = 5 retry_count = 0 while retry_count < max_retries: try: self.engine = create_engine(self.db_url) with self.engine.connect(): break except OperationalError: retry_count += 1 time.sleep(2) if retry_count == max_retries: raise Exception("无法连接到数据库") # 其他方法保持不变
内容的提问来源于stack exchange,提问作者Sebastián Martínez
相关产品推荐
相关产品推荐

