代码：
from flask import Flask, make_response, request import os.path import MySQLdb import json import hashlib import random import math import os from datetime import datetimeapp = Flask(__name__)class DateEncoder(json.JSONEncoder):def default(self, obj):if isinstance(obj, datetime):return obj.strftime("%Y-%m-%d %H:%M:%S")else:return json.JSONEncoder.default(self, obj)secretKey = 'saacac3423@21212' pagesize = 20def getConn():conn = MySQLdb.Connection('127.0.0.1', 'root', '123456', 'my_bbs')cursor = conn.cursor(cursorclass = MySQLdb.cursors.DictCursor)return (conn, cursor)def getloginuserinfo(sessionId):(conn, cursor) = getConn()try:sessionIdHead = request.cookies.get("sessionId")except:sessionIdHead = ''if sessionIdHead is not None and sessionIdHead != '':sessionId = sessionIdHeadsql = "select id,username,nickname,addTime,sessionId from user where sessionId='%s'" % sessionIdcursor.execute(sql)data = cursor.fetchone()if data is None:data = {'id' : 0, 'username' : '', 'nickname' : '', 'addTime' : '', 'sessionId' : ''}return datadef response(code, msg, data):if code != 0:result = {'code' : code, 'msg' : msg, 'data' : None}else:result = {'code' : 0, 'msg' : '', 'data' : data}result = json.dumps(result, cls = DateEncoder, ensure_ascii = False)response = make_response(result)response.headers['Server'] = 'flask-Web-Framework'return responsedef error(code, msg):return response(code, msg, None)def success( data = {}):return response(0, '', data)@app.route('/') def index():(conn, cursor) = getConn()result = "此站接口使用python.flask实现，<a href='api.html' target='_blank'>接口列表</a>"response = make_response(result)response.headers['Server'] = 'flask-Web-Framework'return response@app.route('/user/register') def register():(conn, cursor) = getConn()username = request.args.get("username", "")password = request.args.get("password", "")nickname = request.args.get("nickname", "")sql = "select id,username,nickname,addTime from user where username='%s'" % usernamecursor.execute(sql)data = cursor.fetchone()if data != None:return error(1, '用户名已经存在')try:passwordMd5 = hashlib.md5(password.encode(encoding='utf-8')).hexdigest()sql = "insert into user(username, password, nickname) value('%s', '%s', '%s')" % (username, passwordMd5, nickname)cursor.execute(sql)conn.commit()insertId = cursor.lastrowidreturn success(insertId)except MySQLdb.Error as e:conn.rollback()return error(1, '注册失败')@app.route('/user/login') def login():(conn, cursor) = getConn()username = request.args.get("username", "")password = request.args.get("password", "")passwordMd5 = hashlib.md5(password.encode(encoding='utf-8')).hexdigest()sql = "select id,username,nickname,addTime from user where username='%s' and password='%s'" % (username, passwordMd5)cursor.execute(sql)data = cursor.fetchone()if data == None:return error(1, '用户名或者密码错误')tmpSessionId = secretKey + str(data['id']) + str(data['addTime'])tmpSessionId = hashlib.md5(tmpSessionId.encode(encoding='utf-8')).hexdigest()try:sql = "update user set sessionId='%s' where id=%s" % (tmpSessionId, data['id'])cursor.execute(sql)conn.commit()data['sessionId'] = tmpSessionIdreturn success(data)except MySQLdb.Error as e:conn.rollback()return error(1, '保存会话id失败')@app.route('/user/logout') def logout():(conn, cursor) = getConn()sessionId = request.args.get("sessionId", "")data = getloginuserinfo(sessionId)if data == None:return success(None)if data['sessionId'] == '':return success(data)try:sql = "update user set sessionId='' where sessionId='%s'" % sessionIdcursor.execute(sql)conn.commit()data['sessionId'] = ''return success(data)except MySQLdb.Error as e:conn.rollback()return error(1, '删除会话id失败')@app.route('/user/getuserinfo') def getuserinfo():(conn, cursor) = getConn()sessionId = request.args.get("sessionId", "")userinfo = getloginuserinfo(sessionId)return success(userinfo)@app.route('/post/list') def postlist():(conn, cursor) = getConn()page = request.args.get("page", "1")keyword = request.args.get("keyword", "")page = int(page)if page <= 0:page = 1addsql = " isDel=0 "if keyword is not None and keyword != '':addsql = " isDel=0 and title like '%"+keyword+"%' "start = (page - 1) * pagesizesql1 = "select count(1) as count from content where %s" % addsqlcursor.execute(sql1)countdata = cursor.fetchone()totalpage = math.ceil(countdata['count'] / float(pagesize))data = []if totalpage > 0:sql2 = "select id,title,userId,userNickename,replyNum,updateTime from content where %s order by updateTime desc limit %s,%s" % (addsql, start, pagesize)cursor.execute(sql2)data = cursor.fetchall()return success({'totalpage' : totalpage, 'data' : data})@app.route('/post/detail') def postdetail():(conn, cursor) = getConn()id = request.args.get("id", "0")sql = "select id,title,content,userId,userNickename,replyNum,updateTime from content where isDel=0 and id=%s" % idcursor.execute(sql)data = cursor.fetchone()return success(data)@app.route('/post/add') def postadd():(conn, cursor) = getConn()title = request.args.get("title", "")content = request.args.get("content", "")sessionId = request.args.get("sessionId", "")userinfo = getloginuserinfo(sessionId)userId = userinfo['id']userNickename = userinfo['nickname']if userId <= 0:return error(1, '请先登录')try:sql = "insert into content(title, content, userId, userNickename) value('%s', '%s', %s, '%s')" % (title, content, userId, userNickename)cursor.execute(sql)conn.commit()insertId = cursor.lastrowidreturn success(insertId)except MySQLdb.Error as e:conn.rollback()return error(1, '发帖失败')@app.route('/post/edit') def postedit():(conn, cursor) = getConn()id = request.args.get("id", "0")title = request.args.get("title", "")content = request.args.get("content", "")sessionId = request.args.get("sessionId", "")userinfo = getloginuserinfo(sessionId)userId = userinfo['id']userNickename = userinfo['nickname']if userId <= 0:return error(1, '请先登录')try:sql = "update content set title='%s',content='%s',userId=%s,userNickename='%s' where id=%s and userId=%s" % (title, content, userId, userNickename, id, userId)cursor.execute(sql)conn.commit()return success(None)except MySQLdb.Error as e:conn.rollback()return error(1, '编辑帖子失败')@app.route('/post/delete') def postdelete():(conn, cursor) = getConn()id = request.args.get("id", "0")sessionId = request.args.get("sessionId", "")userinfo = getloginuserinfo(sessionId)userId = userinfo['id']userNickename = userinfo['nickname']if userId <= 0:return error(1, '请先登录')try:sql = "update content set isDel=1 where id=%s and userId=%s" % (id, userId)cursor.execute(sql)conn.commit()return success(None)except MySQLdb.Error as e:conn.rollback()return error(1, '删除帖子失败')@app.route('/reply/list') def replylist():(conn, cursor) = getConn()page = request.args.get("page", "1")contentId = request.args.get("contentId", "0")page = int(page)if page <= 0:page = 1start = (page - 1) * pagesizesql1 = "select count(1) as count from reply where isDel=0 and contentId=%s" % contentIdcursor.execute(sql1)countdata = cursor.fetchone()totalpage = math.ceil(countdata['count'] / float(pagesize))data = []if totalpage > 0:sql2 = "select id,content,replyUserId,replyUserNickename,addTime from reply where isDel=0 and contentId=%s order by id asc limit %s,%s" % (contentId, start, pagesize)cursor.execute(sql2)data = cursor.fetchall()return success({'totalpage' : totalpage, 'data' : data})@app.route('/reply/detail') def replydetail():(conn, cursor) = getConn()id = request.args.get("id", "0")sql = "select id,content,replyUserId,replyUserNickename,addTime from reply where isDel=0 and id=%s" % idcursor.execute(sql)data = cursor.fetchone()return success(data)@app.route('/reply/add') def replyadd():(conn, cursor) = getConn()contentId = request.args.get("contentId", "0")content = request.args.get("content", "")sessionId = request.args.get("sessionId", "")userinfo = getloginuserinfo(sessionId)userId = userinfo['id']userNickename = userinfo['nickname']if userId <= 0:return error(1, '请先登录')try:sql2 = "update content set replyNum=replyNum+1 where id=%s" % contentIdcursor.execute(sql2)sql1 = "insert into reply(contentId, content, replyUserId, replyUserNickename) value(%s, '%s', %s, '%s')" % (contentId, content, userId, userNickename)cursor.execute(sql1)conn.commit()insertId = cursor.lastrowidreturn success(insertId)except MySQLdb.Error as e:conn.rollback()return error(1, '回复失败')@app.route('/reply/edit') def replyedit():(conn, cursor) = getConn()id = request.args.get("id", "0")content = request.args.get("content", "")sessionId = request.args.get("sessionId", "")userinfo = getloginuserinfo(sessionId)userId = userinfo['id']userNickename = userinfo['nickname']if userId <= 0:return error(1, '请先登录')try:sql = "update reply set content='%s',replyUserId=%s,replyUserNickename='%s' where id=%s and replyUserId=%s" % (content, userId, userNickename, id, userId)cursor.execute(sql)conn.commit()return success(None)except MySQLdb.Error as e:conn.rollback()return error(1, '编辑回复失败')@app.route('/reply/delete') def replydelete():(conn, cursor) = getConn()id = request.args.get("id", "0")sessionId = request.args.get("sessionId", "")userinfo = getloginuserinfo(sessionId)userId = userinfo['id']userNickename = userinfo['nickname']if userId <= 0:return error(1, '请先登录')sql = "select id,content,replyUserId,replyUserNickename,addTime,contentId from reply where isDel=0 and id=%s" % idcursor.execute(sql)contentdata = cursor.fetchone()if contentdata is None:return error(1, '回复不存在')try:sql2 = "update content set replyNum=replyNum-1 where id=%s" % contentdata['contentId']cursor.execute(sql2)sql1 = "update reply set isDel=1 where id=%s and replyUserId=%s" % (id, userId)cursor.execute(sql1)conn.commit()return success(None)except MySQLdb.Error as e:conn.rollback()return error(1, '删除回复失败')if __name__ == "__main__":app.run(port=1089)
输出：
D:\workspace\studys\study_pys\pc_app\dist>D:\software\Python310\python.exe D:\workspace\studys\study_bbs\start_web_flask.py* Serving Flask app 'start_web_flask'* Debug mode: off WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.* Running on http://127.0.0.1:1089 Press CTRL+C to quit
