flask, SQLAlchemy, sqlite3 实现 RESTful API 的 todo list, 同时支持form操作

简介: flask, SQLAlchemy, sqlite3 实现 RESTful API, 同时支持form操作。前端与后台的交互都采用json数据格式,原生javascript实现的ajax。其技术要点如下1.

flask, SQLAlchemy, sqlite3 实现 RESTful API, 同时支持form操作。

前端与后台的交互都采用json数据格式,原生javascript实现的ajax。其技术要点如下

1.前端转换:

1).json数据(来自后台) --> object对象(前端应用)

JSON.parse(xhr2.responseText)

2).FormData对象(前端生成) --> json数据(传入后台)

// 辅助函数:FormData转化为json
var convert_FormData_to_json = function (formData) {
    var objData = {};
    for (var entry of formData.entries()){
        objData[entry[0]] = entry[1];
    }
    return JSON.stringify(objData);
};

2.后端转换:

1).json数据(来自前端) --> 类似dict对象(后台应用)

request.json.get('title', task['title'])

2).dict对象(后台生成) --> json数据(传入前端)

jsonify({'task': replace_id_to_uri(task)})

3.项目结构

/myapp
|----app.py
|----/templates
| |----index.html

4.两个文件

1).app.py

from flask import Flask, jsonify, render_template
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite://'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)


# 定义ORM
class Todo(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(80), unique=True)
    description = db.Column(db.String(120), unique=True)
    done = db.Column(db.Boolean)

    def __init__(self, title, description, done):
        self.title = title
        self.description = description
        self.done = done

    def __repr__(self):
        return '<Todo %r>' % self.title
        

# 创建表格、插入数据
@app.before_first_request
def create_db():
    # Recreate database each time for demo
    #db.drop_all()
    db.create_all()

    tasks = [Todo('Buy groceries', 'Milk, Cheese, Pizza, Fruit, Tylenol', False),
             Todo('Learn Python', 'Need to find a good Python tutorial on the web', False),
             Todo('Mow the lawn', 'Find out some tools', False)]
    db.session.add_all(tasks)
    db.session.commit()


# ==================================
# 下面路由至页面
# ==================================

@app.route('/')
def index():
    return render_template('index.html')

        
# ==================================
# 下面是RESTful api
# ==================================

# 辅助函数
from flask import url_for
def replace_id_to_uri(task):
    return dict(uri = url_for('get_task', task_id=task.id, _external=True),
                title = task.title,
                description = task.description,
                done = task.done)


# 查询全部
@app.route('/todo/api/v1.0/tasks/', methods=['GET'])
def get_tasks():
    tasks = Todo.query.all()
    return jsonify({'tasks': list(map(replace_id_to_uri, tasks))})


# 查询一个
from flask import abort
@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['GET'])
def get_task(task_id):
    task = Todo.query.filter_by(id=task_id).first()
    if task is None:
        abort(404)
    
    return jsonify({'task': replace_id_to_uri(task)})
    
    
# 添加
from flask import request
@app.route('/todo/api/v1.0/tasks/', methods=['POST'])
def create_task():
    # 没有数据,或者数据缺少 title 项,返回 400,表示请求无效
    if not request.json or not 'title' in request.json:
        abort(400)
    
    task = Todo(request.json['title'], request.json.get('description', ""), False)
    
    db.session.add(task)
    db.session.commit()
    return jsonify({'task': replace_id_to_uri(task)}), 201


# 更新
@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['PUT'])
def update_task(task_id):
    task = Todo.query.filter_by(id=task_id).first()
    if task is None:
        abort(404)
    
    if not request.json:
        abort(400)
    if 'title' in request.json and type(request.json['title']) != unicode:
        abort(400)
    if 'description' in request.json and type(request.json['description']) is not unicode:
        abort(400)
    if 'done' in request.json and type(request.json['done']) is not bool:
        abort(400)
    
    task['title'] = request.json.get('title', task['title'])
    task['description'] = request.json.get('description', task['description'])
    task['done'] = request.json.get('done', task['done'])
    
    #db.session.update(task)
    db.session.commit()
    return jsonify({'task': replace_id_to_uri(task)})


# 删除
@app.route('/todo/api/v1.0/tasks/<int:task_id>', methods=['DELETE'])
def delete_task(task_id):
    task = Todo.query.filter_by(id=task_id).first()
    if task is None:
        abort(404)
    
    db.session.delete(task)
    db.session.commit()
    return jsonify({'result': True})
    
    
# 定制404出错页面
@app.errorhandler(404)
def not_found(error):
    return jsonify({'error': 'Not found'}), 404


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

2).index.html

<meta charset="utf-8"/>

<form enctype='application/json' method="post">
    <label>任务标题:</label>
    <input type="text" name="title"></br>
    <label>任务描述:</label>
    <input type="texteara" name="description"></br>
    <button type="submit">添加任务</button>
</form>

<!-- 任务列表 -->
<ol id="container"></ol>

<script>
var container = document.getElementById('container');

// 辅助函数:将一条任务(task),放入form的下方容器的尾部
var add_child = function (task) {
    var li = document.createElement("li");
    li.innerText = task["title"] + "  " + task["description"];
    container.appendChild(li);
};

// 辅助函数:FormData转化为json
var convert_FormData_to_json = function (formData) {
    var objData = {};
    for (var entry of formData.entries()){
        objData[entry[0]] = entry[1];
    }
    return JSON.stringify(objData);
};

// 最初,ajax取全部任务,放入form的下方容器
window.onload = function () {
    var xhr = new XMLHttpRequest();

    xhr.open('GET', '/todo/api/v1.0/tasks/');
    xhr.setRequestHeader('Content-Type', 'application/json');
    xhr.send();

    xhr.addEventListener('loadend', function() {
        if(xhr.status == 200){
            var res = JSON.parse(xhr.responseText);
            //console.log(res);
            res["tasks"].map(add_child)
        }
    }, false);
};
</script>


<script>
var formobj = document.querySelector('form');
//var formobj = document.getElementByTag('form');

formobj.addEventListener('submit', function(event){
    event.preventDefault();
    var xhr2 = new XMLHttpRequest();

    xhr2.open('POST', '/todo/api/v1.0/tasks/');
    xhr2.setRequestHeader('Content-Type', 'application/json');
    xhr2.send(convert_FormData_to_json(new FormData(formobj)));// 发送json数据!

    xhr2.addEventListener('loadend', function() {
        if(xhr2.status == 201){ // 201,去看app.py!!
            var res2 = JSON.parse(xhr2.responseText);
            [res2["task"]].map(add_child); // 这里也用map,没别的目的,只想与前面的形式一致
        }
    }, false);
}, false);
</script>

5.效果图

1). POST数据

app_todo_sqlite3.bmp

2). Response数据

app_todo_sqlite32.bmp

6.未完成的功能

1).更新任务(页面操作)
2).删除任务(页面操作)

7.参考:

1).www.pythondoc.com/flask-restful/first.html

目录
相关文章
|
8月前
|
存储 自然语言处理 关系型数据库
Elasticsearch之Restful操作1
Elasticsearch之Restful操作
100 0
|
3月前
|
XML JSON 数据库
SpringMVC之RESTful(含实际代码操作)
SpringMVC之RESTful(含实际代码操作)
|
9月前
|
Python
【Flask实战】sqlalchemy和Flask-SQLAlchemy使用对比
【Flask实战】sqlalchemy和Flask-SQLAlchemy使用对比
|
9月前
|
数据库 Python
【Flask】Flask-SQLAlchemy的增删改查(CRUD)操作
Flask-SQLAlchemy的增删改查(CRUD)操作
|
7月前
|
Web App开发 JavaScript 前端开发
体验 Orbeon form PE 版本提供的 JavaScript Embedding API
体验 Orbeon form PE 版本提供的 JavaScript Embedding API
41 0
|
7月前
|
JavaScript 前端开发 API
关于 Orbeon form PE 版本使用 JavaScript Embedding API 的一个例子
关于 Orbeon form PE 版本使用 JavaScript Embedding API 的一个例子
39 0
|
7月前
|
NoSQL JavaScript Java
MongoDB 入门教程系列之三:使用 Restful API 操作 MongoDB
MongoDB 入门教程系列之三:使用 Restful API 操作 MongoDB
78 0
|
7月前
|
SQL 关系型数据库 MySQL
Flask数据库之SQLAlchemy--介绍--链接数据库
Flask数据库之SQLAlchemy--介绍--链接数据库
|
8月前
|
SQL 存储 关系型数据库
【flask-sqlalchemy】SQLAlchemy+PyMysql到mysql的映射
【flask-sqlalchemy】SQLAlchemy+PyMysql到mysql的映射
103 0
|
8月前
|
SQL 关系型数据库 索引
Elasticsearch之Restful操作 2
Elasticsearch之Restful操作
65 0