跳转到内容

Flask框架入门和RESTful API开发

1. Flask框架介绍

Flask是一个轻量级的Python Web框架,以其简洁、灵活的设计而闻名。它被广泛应用于构建API、Web应用和微服务。

1.1 Flask的特点

  • 轻量级:核心功能简单,易于理解和使用
  • 灵活:不强制使用特定的工具或库
  • 可扩展:通过扩展插件实现更多功能
  • WSGI兼容:支持标准的Web服务器网关接口
  • 内置开发服务器:便于快速开发和测试

1.2 Flask与其他框架的比较

框架特点适用场景
Flask轻量级、灵活小型应用、API开发
Django全功能、电池包含大型应用、企业级项目
FastAPI类型提示、自动文档API开发、高性能需求
Tornado异步IO、长连接实时应用、WebSocket

2. Flask环境搭建

2.1 安装Python

Flask需要Python 3.6或更高版本。首先确保你的系统已经安装了Python。

2.2 安装Flask

使用pip安装Flask:

bash
# 创建虚拟环境
python3 -m venv venv

# 激活虚拟环境
# Windows
venv\Scripts\activate
# Linux/macOS
source venv/bin/activate

# 安装Flask
pip install Flask

2.3 验证安装

创建一个简单的Flask应用来验证安装:

python
# app.py
from flask import Flask

app = Flask(__name__)

@app.route('/')
def hello():
    return 'Hello, Flask!'

if __name__ == '__main__':
    app.run(debug=True)

运行应用:

bash
python app.py

访问 http://localhost:5000 看到 "Hello, Flask!" 即表示安装成功。

3. Flask基础语法

3.1 路由定义

Flask使用装饰器定义路由:

python
@app.route('/')
def index():
    return '首页'

@app.route('/about')
def about():
    return '关于我们'

3.2 HTTP方法

指定HTTP方法:

python
@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # 处理表单提交
        return '登录成功'
    else:
        # 显示登录表单
        return '显示登录表单'

3.3 路径参数

获取URL路径中的参数:

python
@app.route('/user/<username>')
def user_profile(username):
    return f'用户: {username}'

@app.route('/post/<int:post_id>')
def show_post(post_id):
    return f'帖子ID: {post_id}'

3.4 请求处理

获取请求数据:

python
from flask import request

@app.route('/submit', methods=['POST'])
def submit():
    # 获取表单数据
    name = request.form.get('name')
    # 获取查询参数
    age = request.args.get('age')
    # 获取JSON数据
    data = request.get_json()
    return f'姓名: {name}, 年龄: {age}'

3.5 响应处理

返回不同类型的响应:

python
from flask import jsonify, make_response

@app.route('/json')
def json_response():
    # 返回JSON响应
    return jsonify({'name': '张三', 'age': 25})

@app.route('/custom')
def custom_response():
    # 自定义响应
    response = make_response('自定义响应')
    response.status_code = 201
    response.headers['X-Custom-Header'] = 'value'
    return response

4. Flask扩展

4.1 常用扩展

扩展功能安装命令
Flask-RESTfulRESTful API开发pip install Flask-RESTful
Flask-SQLAlchemyORM数据库操作pip install Flask-SQLAlchemy
Flask-Migrate数据库迁移pip install Flask-Migrate
Flask-JWT-ExtendedJWT认证pip install Flask-JWT-Extended
Flask-CORS跨域资源共享pip install Flask-CORS

4.2 安装和使用扩展

以Flask-RESTful为例:

bash
pip install Flask-RESTful

使用示例:

python
from flask import Flask
from flask_restful import Api, Resource

app = Flask(__name__)
api = Api(app)

class HelloWorld(Resource):
    def get(self):
        return {'hello': 'world'}

api.add_resource(HelloWorld, '/')

if __name__ == '__main__':
    app.run(debug=True)

5. RESTful API开发

5.1 RESTful API设计原则

  • 资源导向:使用URL表示资源
  • HTTP方法:使用GET、POST、PUT、DELETE等HTTP方法
  • 无状态:每个请求都是独立的
  • 统一接口:使用标准的HTTP状态码和错误处理
  • 缓存:支持缓存机制

5.2 实现RESTful API

使用Flask-RESTful实现RESTful API:

python
from flask import Flask
from flask_restful import Api, Resource, reqparse, abort

app = Flask(__name__)
api = Api(app)

# 模拟数据库
todos = {
    '1': {'task': '学习Flask', 'done': False},
    '2': {'task': '开发API', 'done': False}
}

# 请求参数解析
parser = reqparse.RequestParser()
parser.add_argument('task', required=True, help='任务内容不能为空')
parser.add_argument('done', type=bool, default=False)

class TodoList(Resource):
    def get(self):
        return todos
    
    def post(self):
        args = parser.parse_args()
        todo_id = str(len(todos) + 1)
        todos[todo_id] = {'task': args['task'], 'done': args['done']}
        return todos[todo_id], 201

class Todo(Resource):
    def get(self, todo_id):
        if todo_id not in todos:
            abort(404, message=f'任务 {todo_id} 不存在')
        return todos[todo_id]
    
    def put(self, todo_id):
        if todo_id not in todos:
            abort(404, message=f'任务 {todo_id} 不存在')
        args = parser.parse_args()
        todos[todo_id] = {'task': args['task'], 'done': args['done']}
        return todos[todo_id]
    
    def delete(self, todo_id):
        if todo_id not in todos:
            abort(404, message=f'任务 {todo_id} 不存在')
        del todos[todo_id]
        return '', 204

api.add_resource(TodoList, '/todos')
api.add_resource(Todo, '/todos/<todo_id>')

if __name__ == '__main__':
    app.run(debug=True)

5.3 API测试

使用curl测试API:

bash
# 获取所有任务
curl http://localhost:5000/todos

# 添加新任务
curl -X POST http://localhost:5000/todos -H "Content-Type: application/json" -d '{"task": "测试任务", "done": false}'

# 获取单个任务
curl http://localhost:5000/todos/1

# 更新任务
curl -X PUT http://localhost:5000/todos/1 -H "Content-Type: application/json" -d '{"task": "更新的任务", "done": true}'

# 删除任务
curl -X DELETE http://localhost:5000/todos/1

6. Flask项目结构

6.1 推荐的项目结构

myapp/
├── app/
│   ├── __init__.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── resources/
│   │   └── routes.py
│   ├── models/
│   ├── schemas/
│   ├── services/
│   └── utils/
├── config.py
├── requirements.txt
├── run.py
└── tests/

6.2 项目结构说明

  • app/:应用主目录
    • api/:API相关代码
    • models/:数据库模型
    • schemas/:数据验证和序列化
    • services/:业务逻辑
    • utils/:工具函数
  • config.py:配置文件
  • requirements.txt:依赖项
  • run.py:应用入口
  • tests/:测试代码

7. 配置管理

7.1 配置文件

创建config.py文件:

python
class Config:
    SECRET_KEY = 'your-secret-key'
    DEBUG = False
    TESTING = False

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///dev.db'

class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///test.db'

class ProductionConfig(Config):
    SQLALCHEMY_DATABASE_URI = 'postgresql://user:password@localhost/dbname'

config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,
    'default': DevelopmentConfig
}

7.2 应用配置

在app/init.py中配置应用:

python
from flask import Flask
from config import config

def create_app(config_name='default'):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    
    # 注册蓝图、初始化扩展等
    
    return app

8. 错误处理

8.1 自定义错误处理

python
from flask import jsonify

@app.errorhandler(404)
def not_found_error(error):
    return jsonify({'error': '资源不存在'}), 404

@app.errorhandler(400)
def bad_request_error(error):
    return jsonify({'error': '请求参数错误'}), 400

@app.errorhandler(500)
def internal_error(error):
    return jsonify({'error': '服务器内部错误'}), 500

8.2 异常处理

python
from flask import jsonify

@app.route('/error')
def error_example():
    try:
        # 可能抛出异常的代码
        result = 1 / 0
    except ZeroDivisionError as e:
        return jsonify({'error': '除数不能为零'}), 400
    except Exception as e:
        return jsonify({'error': str(e)}), 500
    return jsonify({'result': result})

9. 部署Flask应用

9.1 使用Gunicorn

安装Gunicorn:

bash
pip install gunicorn

运行应用:

bash
gunicorn -w 4 -b 0.0.0.0:8000 run:app

9.2 使用Docker

创建Dockerfile:

dockerfile
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 5000

CMD ["gunicorn", "-w", "4", "-b",

评论区

专业的Linux技术学习平台,从入门到精通的完整学习路径