Docker环境下Flask应用连接PostgreSQL的配置方法与问题排查

一、搞定数据库连接（解决版本查询失败） #

1. 先把Docker Compose配置捋顺 #

• 别用localhost当数据库主机名，要用Compose里PostgreSQL服务的名称（比如你定义的服务叫db，那Flask里的HOST就是db）
• 两边环境变量要对齐：PostgreSQL的POSTGRES_USER/POSTGRES_PASSWORD/POSTGRES_DB，和Flask里的环境变量必须完全一致
• 加健康检查，让Flask等Postgres完全启动再跑，避免数据库没就绪就连接：

  services:
    db:
      image: postgres:15-alpine
      environment:
        POSTGRES_USER: myuser
        POSTGRES_PASSWORD: mypass
        POSTGRES_DB: mydb
      healthcheck:
        test: ["CMD-SHELL", "pg_isready -U myuser -d mydb"]
        interval: 5s
        timeout: 5s
        retries: 5
    web:
      build: .
      ports:
        - "5000:5000"
      depends_on:
        db:
          condition: service_healthy
      environment:
        DB_HOST: db
        DB_USER: myuser
        DB_PASSWORD: mypass
        DB_NAME: mydb
  

2. 修正Flask的连接代码 #

• 用环境变量读参数，别硬编码，方便后续改配置
• 加重试逻辑，就算有健康检查，偶尔还是会有启动延迟
• 示例用psycopg2的连接代码：

  import os
  import psycopg2
  from psycopg2 import OperationalError
  import time
  
  def get_db_connection():
      conn = None
      try:
          conn = psycopg2.connect(
              host=os.getenv('DB_HOST'),
              database=os.getenv('DB_NAME'),
              user=os.getenv('DB_USER'),
              password=os.getenv('DB_PASSWORD')
          )
          # 测试连接：查版本
          with conn.cursor() as cur:
              cur.execute("SELECT version();")
              version = cur.fetchone()
              print(f"连上PostgreSQL了：{version[0]}")
      except OperationalError as e:
          print(f"连接出错：{e}")
          # 等2秒再重试
          time.sleep(2)
          return get_db_connection()
      return conn
  

  用SQLAlchemy的话更简单：

  from flask import Flask
  from flask_sqlalchemy import SQLAlchemy
  import os
  
  app = Flask(__name__)
  app.config['SQLALCHEMY_DATABASE_URI'] = f"postgresql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}/{os.getenv('DB_NAME')}"
  app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
  db = SQLAlchemy(app)
  
  # 测试接口
  @app.route('/test-db')
  def test_db():
      try:
          result = db.session.execute(db.text("SELECT version();")).fetchone()
          return f"PostgreSQL版本：{result[0]}"
      except Exception as e:
          return f"连接失败：{str(e)}"
  

3. 检查网络和权限 #

• 确保Flask和Postgres在同一个Docker网络（同一个Compose文件默认会创建共享网络，不用额外配置）
• 确认Postgres用户有访问指定数据库的权限（默认用POSTGRES_USER创建的用户会有对应数据库的权限）
• 不需要给Postgres映射端口到宿主机，Flask内部用服务名+默认5432端口就能访问

二、查询真实数据的注意事项&技巧 #

1. 绝对要参数化查询，防SQL注入 #

别手动拼SQL字符串，比如这种是找死：

# 错误示例！
cur.execute(f"SELECT * FROM users WHERE id = {user_id}")

正确姿势用占位符：

# psycopg2用%s当占位符
cur.execute("SELECT * FROM users WHERE id = %s", (user_id,))
# SQLAlchemy ORM自动处理参数化
user = User.query.get(user_id)

2. 用连接池，别每次都新建连接 #

频繁创建销毁连接会拖垮性能，用连接池复用连接：

from psycopg2 import pool

# 初始化连接池：最小1个，最大10个
connection_pool = pool.SimpleConnectionPool(
    1, 10,
    host=os.getenv('DB_HOST'),
    database=os.getenv('DB_NAME'),
    user=os.getenv('DB_USER'),
    password=os.getenv('DB_PASSWORD')
)

def get_conn():
    return connection_pool.getconn()

def release_conn(conn):
    connection_pool.putconn(conn)

3. 写操作必须加事务 #

插入、更新、删除时用事务，保证数据一致：

conn = get_conn()
try:
    with conn.cursor() as cur:
        cur.execute("INSERT INTO users (name) VALUES (%s)", ("Bob",))
    conn.commit()  # 成功再提交
except Exception as e:
    conn.rollback()  # 失败回滚
    print(f"操作失败：{e}")
finally:
    release_conn(conn)

4. 日志和错误处理要到位 #

• 捕获数据库异常，记录详细日志（错误信息、执行的SQL、参数），方便排查
• 别把原始错误返回给前端，给用户看友好提示就行

5. 自动初始化数据库 #

可以在Postgres容器启动时自动跑初始化脚本，比如建表、插初始数据：
在Compose的db服务里加volumes：

volumes:
  - ./init.sql:/docker-entrypoint-initdb.d/init.sql

init.sql示例：

CREATE TABLE IF NOT EXISTS users (
    id SERIAL PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL
);

6. 性能优化小技巧 #

• 给经常查询的字段加索引，比如CREATE INDEX idx_users_email ON users(email);
• 查大量数据用分页，别一次性拉全表：SELECT * FROM users LIMIT 10 OFFSET 20;
• 复杂查询用EXPLAIN ANALYZE看执行计划，优化慢查询

内容的提问来源于stack exchange，提问作者Alon