Flask与PostgreSQL交互教程

1. 项目结构

首先，让我们了解项目的基本结构：

├── .env                  # 环境变量配置
├── app.py                # Flask应用主文件
├── docker-compose.yml    # Docker配置文件
├── models.py             # 数据库模型定义
└── requirements.txt      # 项目依赖

2. 环境准备

2.1 安装依赖

首先，我们需要安装必要的依赖包：

pip install -r requirements.txt

依赖包包括：

• Flask: Web框架
• psycopg2-binary: PostgreSQL的Python驱动
• flask-sqlalchemy: Flask的ORM扩展
• python-dotenv: 环境变量管理

2.2 使用Docker启动PostgreSQL

我们使用Docker Compose来管理PostgreSQL数据库。以下是docker-compose.yml的配置：

services:
  db:
    image: postgres:13
    volumes:
      - postgres_data:/var/lib/postgresql/data/
    environment:
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_USER=postgres
      - POSTGRES_DB=postgres
    ports:
      - "5432:5432"

volumes:
  postgres_data:

启动PostgreSQL容器：

docker-compose up -d

这将启动一个PostgreSQL 13的实例，并将其端口映射到本地的5432端口。

3. 数据库配置

3.1 环境变量配置

在.env文件中，我们定义了数据库连接和Flask应用的环境变量：

DATABASE_URL=postgresql://postgres:postgres@localhost:5432/postgres

3.2 数据库连接配置

在app.py中，我们使用以下代码配置数据库连接：

from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

app = Flask(__name__)

# 配置数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# 初始化数据库
db.init_app(app)

# 创建数据库表
with app.app_context():
    db.create_all()

4. 定义数据库模型

在models.py中，我们定义了一个简单的User模型：

from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

db = SQLAlchemy()

class User(db.Model):
    __tablename__ = 'users'

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    def __init__(self, username, email):
        self.username = username
        self.email = email

    def __repr__(self):
        return f'<User {self.username}>'
    
    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'email': self.email,
            'created_at': self.created_at.isoformat() if self.created_at else None
        }

这个模型定义了用户表的结构，包括id、用户名、邮箱和创建时间等字段。to_dict()方法用于将用户对象转换为字典，方便JSON序列化。

5. 实现API接口

在app.py中，我们实现了一组RESTful API接口，用于对用户进行CRUD操作：

5.1 创建用户

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    
    if not data or not data.get('username') or not data.get('email'):
        return jsonify({'error': 'Invalid data'}), 400
    
    try:
        user = User(username=data['username'], email=data['email'])
        db.session.add(user)
        db.session.commit()
        return jsonify({'message': 'User created successfully', 'user': user.to_dict()}), 201
    except Exception as e:
        db.session.rollback()
        return jsonify({'error': str(e)}), 500

5.2 获取所有用户

@app.route('/users', methods=['GET'])
def get_users():
    try:
        users = User.query.all()
        return jsonify({'users': [user.to_dict() for user in users]})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

5.3 获取单个用户

@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    try:
        user = User.query.get(user_id)
        if not user:
            return jsonify({'error': 'User not found'}), 404
        return jsonify({'user': user.to_dict()})
    except Exception as e:
        return jsonify({'error': str(e)}), 500

5.4 更新用户

@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
    try:
        user = User.query.get(user_id)
        if not user:
            return jsonify({'error': 'User not found'}), 404
        
        data = request.get_json()
        if data.get('username'):
            user.username = data['username']
        if data.get('email'):
            user.email = data['email']
        
        db.session.commit()
        return jsonify({'message': 'User updated successfully', 'user': user.to_dict()})
    except Exception as e:
        db.session.rollback()
        return jsonify({'error': str(e)}), 500

5.5 删除用户

@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
    try:
        user = User.query.get(user_id)
        if not user:
            return jsonify({'error': 'User not found'}), 404
        
        db.session.delete(user)
        db.session.commit()
        return jsonify({'message': 'User deleted successfully'})
    except Exception as e:
        db.session.rollback()
        return jsonify({'error': str(e)}), 500

6. 运行应用

启动Flask应用：

python app.py

或者使用Flask命令：

flask run --host=0.0.0.0

应用将在http://localhost:5000上运行。
或者用uv:

7. API测试

你可以使用工具如Postman或curl来测试API：

7.1 创建用户

curl -X POST http://localhost:5000/users -H "Content-Type: application/json" -d '{"username":"test_user","email":"test@example.com"}'

7.2 获取所有用户

curl http://localhost:5000/users

7.3 获取单个用户

curl http://localhost:5000/users/1

7.4 更新用户

curl -X PUT http://localhost:5000/users/1 -H "Content-Type: application/json" -d '{"username":"updated_user"}'

7.5 删除用户

curl -X DELETE http://localhost:5000/users/1