Flask与PostgreSQL交互教程

本教程将指导你如何使用Flask框架与PostgreSQL数据库进行交互，实现一个简单的用户管理API。

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:13volumes:- postgres_data:/var/lib/postgresql/data/environment:- POSTGRES_PASSWORD=postgres- POSTGRES_USER=postgres- POSTGRES_DB=postgresports:- "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 datetimedb = 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 = usernameself.email = emaildef __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'}), 400try: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()}), 201except 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'}), 404return 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'}), 404data = 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'}), 404db.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":"[e-mail]"}'

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