1.开发案例

1.1.项目一:任务列表

1.1.1.开发步骤

        (1)系统规划

        任务列表(当前任务、已完成、未完成)、创建/编辑任务、废弃任务、执行任务、删除任务

       (2) 业务逻辑

        创建任务,状态待执行

        状态待执行,操作执行,状态执行中

        状态待执行,操作废弃,状态已废弃

        状态已废弃,操作删除,列表删除数据

        状态执行中,操作完成,状态已完成

        

        (3)模块设计

        (4)产品原型

        (5)代码实现

1.1.2.数据库设计

CREATE TABLE 'task' (

'id' int(11) NOT NULL AUTO_INCREMENT,

'name' varchar(100) NOT NULL COMMENT '任务名称',

'desc' varchar(255) DEFAULT NULL COMMENT '任务描述',

'start_time' timestamp NULL DEFAULT NULL COMMENT '执行开始时间',

'end_time' timestamp NULL DEFAULT NULL COMMENT '执行结束时间',

'assign' varchar(50) NOT NULL COMMENT '执行人',

'status' enum('DISCARD','FINISHED','INPROCESS','INIT') NOT NULL DEFAULT 'INIT'

COMMENT '状态',

'created_time' timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,

'is_del' tinyint(1) DEFAULT '0' COMMENT ' 逻辑删除标识。0:未删除,1:已删除',

PRIMARY KEY ('id')

) ENGINE=InnoDB DEFAULT CHARSET=utf8;

1.1.3.前端模板

(1)安装vue-element-admin框架

(2)添加路由

编辑src/router/index.js文件,并在constantRoutes列表中追加一个路由子项:

{

path: '/todo',

component: Layout,

redirect: '/todo/index',

hidden: false,

children: [

{

path: 'index',

component: () => import('@/views/todo/index.vue'),

name: 'Profile',

meta: { title: '任务列表', icon: 'list'}

}

]

},

   (3)添加页面、页面元素、事件处理

在src/views下创建一个文件夹todo,在该文件夹下创建一个文件index.vue

   (4)发送ajax请求到服务器端,提交表单数据、获取任务列表

在src/api目录下新建一个todo.js的文件;

定义了一个发送ajax请求的submit函数,该函数实际上调用了封装好axios组件的request函数,并将请求相关的url、method、headers、data数据传递给该底层函数;

之后,在src/views/todo/index.vue页面中引入定义好的submit函数,并将onSubmit处理函数中的内容进行替换。

import request from '@/utils/request'

export function submit(data) {

return request({

url: '/api/todo',

method: 'post',

headers: {

'Content-Type': 'application/json'

},

data

})

}

export function pullData(params) {

return request({

url: '/api/todo/list',

method: 'get',

params

})

}

export function execute(data) {

return request({

url: '/api/todo/execute',

method: 'post',

data

})

}

export function deletetodo(data) {

return request({

url: '/api/todo/delete',

method: 'delete',

data

})

}

1.1.4.后端开发

(1)app.py

配置路由及处理函数

from controller.todo import todo, get_todo_list, execute, delete

from flask import Flask

from controller.index import index

from config import Config

import os

conf = Config.get(os.environ.get('FLASK_ENV'), Config['default'])

def create_app():

app = Flask(__name__, static_url_path='/do_not_use_this_path__')

app.config.from_object(conf)

# 保存任务

app.route('/todo', methods=['POST'])(todo)

# 任务列表

app.route('/todo/list')(get_todo_list)

app.route('/')(index)

# 执行/完成/废止任务

app.route('/todo/execute', methods=['POST'])(execute)

# 删除任务

app.route('/todo/delete', methods=['delete'])(delete)

return app

if __name__ == '__main__':

create_app().run(host='0.0.0.0', port=9528, debug=True, threaded=True)

(2)controller文件下,todo.py文件实现处理函数

from flask import jsonify, request

from server.controller import warp_date_field

from server.model.todo import update_todo, create_todo, get_current_todo, get_unfinish_todo, get_finished_todo, \

execute_todo, delete_todo

def todo():

data = request.json

if data.get('id'): # 更新任务

ret = update_todo(data)

else: # 新增任务

ret = create_todo(data)

code = 20000 if ret else -1

return jsonify({

"code": code,

"msg": '',

"data": ret

})

def get_todo_list():

tab = request.args.get('tab')

if tab == 'current':

ret = get_current_todo()

elif tab == 'unfinish':

ret = get_unfinish_todo()

elif tab == 'finished':

ret = get_finished_todo()

else:

ret = []

print("处理前:", ret)

ret = warp_date_field(ret)

print("处理后:", ret)

return jsonify({

"code": 20000,

"msg": '',

"data": ret

})

def execute():

data = request.json

if data.get('id'):

ret = execute_todo(data)

msg = "操作成功"

else:

msg = "任务id不存在"

code = 20000 if ret else -1

return jsonify({

"code": code,

"msg": msg,

"data": ret

})

def delete():

data = request.json

if data.get('id'):

ret = delete_todo(data)

msg = "操作成功"

else:

msg = "任务id不存在"

code = 20000 if ret else -1

return jsonify({

"code": code,

"msg": msg,

"data": ret

})

(3)model文件下,__init__.py实现连接数据库

import records

def get_db():

conn = 'mysql+pymysql://root:a.12345678@localhost:3306/test_task'

return records.Database(conn)

(4)model文件下,todo.py实现操作数据库

import logging

import time

from . import get_db

def update_todo(data):

sql = '''update task set name=:name, description=:description, start_time=:start_time,

end_time=:end_time, assign=:assign, status=:status where id=:id'''

try:

data_list = []

data_list.insert(0, data)

get_db().bulk_query(sql, data_list)

return True

except Exception as e:

logging.exception(e)

return False

def create_todo(data):

sql = '''insert into task(name, description, start_time, end_time, assign, status)

values(:name, :description, :start_time, :end_time, :assign, :status)'''

try:

data_list = []

data_list.insert(0, data)

get_db().bulk_query(sql, data_list)

rows = get_db().query("select id from task order by id desc")

insert_id = rows.first(as_dict=True)

print(insert_id)

return insert_id

except Exception as e:

logging.exception(e)

return False

def get_current_todo():

sql = '''select * from task where status in ('INIT', 'INPROCESS') and start_time < :today and :today < end_time'''

today = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

try:

rows = get_db().query(sql, today=today).all(as_dict=True)

return rows

except Exception as e:

logging.exception(e)

return []

def get_unfinish_todo():

sql = '''select * from task where status in ('INIT', 'INPROCESS')'''

try:

rows = get_db().query(sql).all(as_dict=True)

print("查询的:", rows)

return rows

except Exception as e:

logging.exception(e)

return []

def get_finished_todo():

sql = '''select * from task where status in ('FINISHED', 'DISCARD')'''

try:

rows = get_db().query(sql).all(as_dict=True)

return rows

except Exception as e:

logging.exception(e)

return []

def execute_todo(data):

sql = '''update task set status=:status where id=:id'''

try:

data_list = []

data_list.insert(0, data)

get_db().bulk_query(sql, data_list)

return True

except Exception as e:

logging.exception(e)

return False

def delete_todo(data):

sql = '''delete from task where id=:id'''

try:

data_list = []

data_list.insert(0, data)

get_db().bulk_query(sql, data_list)

return True

except Exception as e:

logging.exception(e)

return False

1.1.5.统一端口

nginx配置reverse-procy.conf文件:

server {     listen 80;     server_name localhost;    # 这是外网访问进来时的连接地址     location / {  # 转发前端服务        proxy_set_header Host $host;        proxy_set_header X-Real-IP $remote_addr;        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;        proxy_pass http://127.0.0.1:9527/; }     location /api/ { # 转发后端服务        proxy_set_header Host $host;        proxy_set_header X-Real-IP $remote_addr;        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;        proxy_pass http://127.0.0.1:9528/; }      }

参考链接

评论可见,请评论后查看内容,谢谢!!!
 您阅读本篇文章共花了: