python flask sqlite http服务

python flask 服务

为了让拼接项目能够简单实用http服务，用python flask 来做一个http服务，使用python的原因是拼接项目本身很多人工智能的服务是用python来写。数据库为了简单实用sqlite，python自带sqlite3 ,很方便

db.sql

CREATE TABLE if not exists project(id INTEGER PRIMARY KEY AUTOINCREMENT,projectname TEXT UNIQUE NOT NULL,picnum TEXT NOT NULL,memo TEXT,created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);CREATE TABLE if not exists rtsp(id INTEGER PRIMARY KEY AUTOINCREMENT,projectid INTEGER NOT NULL,created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,urls TEXT NOT NULL,picpath TEXT NOT NULL,FOREIGN KEY (projectid) REFERENCES project(id)
);

上面的表在不存在的时候就会创建，存成db.sql

show me the code

代码如下

from __future__ import with_statement
from contextlib import closing  
import sqlite3
from flask import Flask, request, session, g, redirect, url_for, \abort, render_template, flash,jsonify# configuration
DATABASE = './project.db'
DEBUG = Trueapp = Flask(__name__)
app.config.from_object(__name__)app.config.from_envvar('FLASKR_SETTINGS', silent=True)def connect_db():return sqlite3.connect(app.config['DATABASE'])def init_db():with closing(connect_db()) as db:with app.open_resource('db.sql',"r") as f:#str = f.read()#print(str)db.cursor().executescript(f.read())db.commit()@app.before_request
def before_request():g.db = connect_db()@app.after_request
def after_request(response):g.db.close()return response@app.route('/')
def show_first():return "this is a test"@app.route('/project', methods=['GET'])
def show_project():cur = g.db.execute('select id,projectname,picnum,memo,created from project order by id desc')entries = [dict(id=row[0], projectname=row[1],picnum= row[2],memo=row[3],created=row[4]) for row in cur.fetchall()]return jsonify(entries)@app.route('/project/add', methods=['POST'])
def add_entry():#if not session.get('logged_in'):#    abort(401)g.db.execute('insert into project (projectname,picnum,memo) values (?,?,?)',[request.form['projectname'], request.form['picnum'],request.form['memo'] ])g.db.commit()flash('New project was successfully posted')return redirect(url_for('show_first'))@app.route('/project/del/id>', methods=['POST'])
def del_entry():#if not session.get('logged_in'):#    abort(401)g.db.execute('delete from project where id = ?',[id])g.db.commit()flash('project was successfully delete')return redirect(url_for('show_first'))#需要登录或者单点登录的时候把以下代码准备加上
@app.route('/login', methods=['GET', 'POST'])
def login():error = Noneif request.method == 'POST':if request.form['username'] != app.config['USERNAME']:error = 'Invalid username'elif request.form['password'] != app.config['PASSWORD']:error = 'Invalid password'else:session['logged_in'] = Trueflash('You were logged in')return redirect(url_for('show_first'))return render_template('login.html', error=error)#需要登录或者单点登录的时候把以下代码准备加上
@app.route('/logout')
def logout():session.pop('logged_in', None)flash('You were logged out')return redirect(url_for('show_first'))if __name__ == '__main__':init_db()app.run()