Flask vue-element-admin InfluxDB 开发一个任务管理系统包括登录，任务列表，任务详情；清写出具体开发流程和代码

开发流程：

1. 确定需求，设计数据库表结构；
2. 搭建 Flask 后端框架，实现用户注册、登录功能，使用 JWT 进行身份验证；
3. 设计任务列表和任务详情页面，使用 Vue.js 和 Element UI 开发前端页面；
4. 使用 InfluxDB 存储任务数据，编写 InfluxDB 数据库操作代码；
5. 后端提供接口，前端调用后端接口，实现任务列表和任务详情展示；
6. 编写代码测试，部署上线。

代码实现：

1. 设计数据库表结构

CREATE TABLE users (
  id INTEGER PRIMARY KEY,
  username TEXT NOT NULL,
  password TEXT NOT NULL
);

CREATE TABLE tasks (
  id INTEGER PRIMARY KEY,
  title TEXT NOT NULL,
  description TEXT,
  status TEXT NOT NULL,
  created_at TIMESTAMP NOT NULL,
  updated_at TIMESTAMP NOT NULL,
  user_id INTEGER NOT NULL,
  FOREIGN KEY (user_id) REFERENCES users (id)
);

2. 后端 Flask 框架代码

from flask import Flask, jsonify, request
from flask_jwt_extended import JWTManager, jwt_required, create_access_token
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime
from influxdb import InfluxDBClient

app = Flask(__name__)
app.config['JWT_SECRET_KEY'] = 'secret_key'
jwt = JWTManager(app)

client = InfluxDBClient(host='localhost', port=8086, username='root', password='root', database='tasks')

@app.route('/api/register', methods=['POST'])
def register():
    username = request.json.get('username')
    password = request.json.get('password')
    hashed_password = generate_password_hash(password)
    client.write_points([{
        "measurement": "users",
        "tags": {
            "username": username
        },
        "fields": {
            "password": hashed_password
        }
    }])
    return jsonify({'message': 'User created successfully!'}), 201

@app.route('/api/login', methods=['POST'])
def login():
    username = request.json.get('username')
    password = request.json.get('password')
    result = client.query(f'SELECT password FROM users WHERE username=\'{username}\'')
    if len(result) == 0:
        return jsonify({'message': 'Invalid username or password!'}), 401
    pwd_hash = result[0][0]['password']
    if check_password_hash(pwd_hash, password):
        access_token = create_access_token(identity=username)
        return jsonify(access_token=access_token), 200
    else:
        return jsonify({'message': 'Invalid username or password!'}), 401

@app.route('/api/tasks', methods=['GET'])
@jwt_required()
def get_tasks():
    result = client.query('SELECT * FROM tasks')
    tasks = []
    for row in result:
        task = {
            'id': row['id'],
            'title': row['title'],
            'description': row['description'],
            'status': row['status'],
            'created_at': row['created_at'],
            'updated_at': row['updated_at'],
            'user_id': row['user_id']
        }
        tasks.append(task)
    return jsonify(tasks), 200

@app.route('/api/tasks/<int:task_id>', methods=['GET'])
@jwt_required()
def get_task(task_id):
    result = client.query(f'SELECT * FROM tasks WHERE id={task_id}')
    if len(result) == 0:
        return jsonify({'message': 'Task not found!'}), 404
    row = result[0]
    task = {
        'id': row['id'],
        'title': row['title'],
        'description': row['description'],
        'status': row['status'],
        'created_at': row['created_at'],
        'updated_at': row['updated_at'],
        'user_id': row['user_id']
    }
    return jsonify(task), 200

@app.route('/api/tasks', methods=['POST'])
@jwt_required()
def create_task():
    title = request.json.get('title')
    description = request.json.get('description')
    status = request.json.get('status')
    created_at = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
    updated_at = created_at
    user_id = request.json.get('user_id')
    client.write_points([{
        "measurement": "tasks",
        "fields": {
            "title": title,
            "description": description,
            "status": status,
            "created_at": created_at,
            "updated_at": updated_at,
            "user_id": user_id
        }
    }])
    return jsonify({'message': 'Task created successfully!'}), 201

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

3. 前端 Vue.js 和 Element UI 代码

<template>
  <div>
    <el-button type="primary" @click="showCreateDialog">Create Task</el-button>
    <el-table :data="tasks" style="width: 100%">
      <el-table-column prop="id" label="ID"></el-table-column>
      <el-table-column prop="title" label="Title"></el-table-column>
      <el-table-column prop="description" label="Description"></el-table-column>
      <el-table-column prop="status" label="Status"></el-table-column>
      <el-table-column prop="created_at" label="Created At"></el-table-column>
      <el-table-column prop="updated_at" label="Updated At"></el-table-column>
      <el-table-column prop="user_id" label="User ID"></el-table-column>
      <el-table-column label="Operations">
        <template slot-scope="scope">
          <el-button type="text" @click="showDetailDialog(scope.row.id)">Detail</el-button>
        </template>
      </el-table-column>
    </el-table>
    <el-dialog :visible.sync="createDialogVisible" title="Create Task">
      <el-form :model="form" label-position="left" label-width="80px">
        <el-form-item label="Title">
          <el-input v-model="form.title"></el-input>
        </el-form-item>
        <el-form-item label="Description">
          <el-input v-model="form.description"></el-input>
        </el-form-item>
        <el-form-item label="Status">
          <el-select v-model="form.status">
            <el-option label="Open" value="open"></el-option>
            <el-option label="In Progress" value="in_progress"></el-option>
            <el-option label="Closed" value="closed"></el-option>
          </el-select>
        </el-form-item>
        <el-form-item label="User ID">
          <el-input v-model="form.user_id"></el-input>
        </el-form-item>
      </el-form>
      <div slot="footer">
        <el-button @click="createTask">Create</el-button>
        <el-button @click="createDialogVisible = false">Cancel</el-button>
      </div>
    </el-dialog>
    <el-dialog :visible.sync="detailDialogVisible" title="Task Detail">
      <el-card>
        <p>ID: {{ detailTask.id }}</p>
        <p>Title: {{ detailTask.title }}</p>
        <p>Description: {{ detailTask.description }}</p>
        <p>Status: {{ detailTask.status }}</p>
        <p>Created At: {{ detailTask.created_at }}</p>
        <p>Updated At: {{ detailTask.updated_at }}</p>
        <p>User ID: {{ detailTask.user_id }}</p>
      </el-card>
      <div slot="footer">
        <el-button @click="detailDialogVisible = false">Close</el-button>
      </div>
    </el-dialog>
  </div>
</template>

<script>
import axios from 'axios'

export default {
  name: 'TaskList',
  data () {
    return {
      tasks: [],
      createDialogVisible: false,
      detailDialogVisible: false,
      form: {
        title: '',
        description: '',
        status: '',
        user_id: ''
      },
      detailTask: {}
    }
  },
  methods: {
    loadTasks () {
      axios.get('/api/tasks').then(response => {
        this.tasks = response.data
      }).catch(error => {
        console.error(error)
      })
    },
    showCreateDialog () {
      this.createDialogVisible = true
    },
    createTask () {
      axios.post('/api/tasks', this.form).then(response => {
        console.log(response.data)
        this.createDialogVisible = false
        this.loadTasks()
      }).catch(error => {
        console.error(error)
      })
    },
    showDetailDialog (id) {
      axios.get(`/api/tasks/${id}`).then(response => {
        this.detailTask = response.data
        this.detailDialogVisible = true
      }).catch(error => {
        console.error(error)
      })
    }
  },
  mounted () {
    this.loadTasks()
  }
}
</script>

4. InfluxDB 数据库操作代码

from influxdb import InfluxDBClient
from datetime import datetime

client = InfluxDBClient(host='localhost', port=8086, username='root', password='root', database='tasks')

def create_user(username, password):
    hashed_password = generate_password_hash(password)
    client.write_points([{
        "measurement": "users",
        "tags": {
            "username": username
        },
        "fields": {
            "password": hashed_password
        }
    }])

def get_user_password(username):
    result = client.query(f'SELECT password FROM users WHERE username=\'{username}\'')
    if len(result) == 0:
        return None
    pwd_hash = result[0][0]['password']
    return pwd_hash

def get_tasks():
    result = client.query('SELECT * FROM tasks')
    tasks = []
    for row in result:
        task = {
            'id': row['id'],
            'title': row['title'],
            'description': row['description'],
            'status': row['status'],
            'created_at': row['created_at'],
            'updated_at': row['updated_at'],
            'user_id': row['user_id']
        }
        tasks.append(task)
    return tasks

def get_task_by_id(task_id):
    result = client.query(f'SELECT * FROM tasks WHERE id={task_id}')
    if len(result) == 0:
        return None
    row = result[0]
    task = {
        'id': row['id'],
        'title': row['title'],
        'description': row['description'],
        'status': row['status'],
        'created_at': row['created_at'],
        'updated_at': row['updated_at'],
        'user_id': row['user_id']
    }
    return task

def create_task(title, description, status, user_id):
    created_at = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')
    updated_at = created_at
    client.write_points([{
        "measurement": "tasks",
        "fields": {
            "title": title,
            "description": description,
            "status": status,
            "created_at": created_at,
            "updated_at": updated_at,
            "user_id": user_id
        }
    }])