python使用web.py构建web服务并实现对sqlite的增删改查

  • Post author:
  • Post category:python




安装web.py

打开cmd输入

pip install web.py

等待安装完毕,同时需要安装好sqlite,并建好test.db库,创建sys_user(id, login_name, user_name, password)表。



demo

import web

urls = (
    '/upper/(.*)', 'upper',
    '/lower/(.*)', 'lower'
)

app = web.application(urls, globals())

class upper:
    def GET(self, text):
        print('input:' + text)
        return text.upper()


class lower:
    def GET(self, text):
        print('input:' + text)
        return text.lower()


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

例:打开localhost:8080/upper/ABCabc

可以返回ABCABC

这里,urls定义的是访问路径的调用类的映射关系,前面是路径,后面是类名,类里面的GET方法实现了get请求,该路径用post请求是请求不到的。

这里关于web.py第三方类库的使用方法可以参考:

https://webpy.org/cookbook/index.zh-cn




实现简单的增删改查api

首先我们先实现一个python对sqlite数据库操作类

import sqlite3

class sqlliteDBtool:
    def __init__(self):
        """
        初始化函数,创建数据库连接
        """
        self.conn = sqlite3.connect("d:/sqlite/test.db")
        self.conn.row_factory = self.dict_factory
        self.cursor = self.conn.cursor()

    @staticmethod
    def dict_factory(cursor, row):
   		"""
        对sqlite的查询结果转化成python的词典
        :param cursor: 游标
        :param row: 结果集
        :return: 返回词典
        """
        d = {}
        for idx, col in enumerate(cursor.description):
            d[col[0]] = row[idx]
        return d


    def executeUpdate(self, sql, param):
        """
        数据库的插入、修改函数
        :param sql: 传入的sql语句
        :param param: 传入的参数
        :return: 返回操作数据库状态
        """
        try:
            self.cursor.executemany(sql, param)
            i = self.conn.total_changes
        except Exception as e:
            print("错误类型:", e)
            return False
        finally:
            self.conn.commit()
        if i > 0:
            return True
        else:
            return False


    def excuteDelete(self, sql, param):
        """
        操作数据库删除
        :param sql: sql语句
        :param param: 参数
        :return: 返回数据库状态
        """
        try:
            self.cursor.execute(sql, param)
            i = self.conn.total_changes
        except Exception as e:
            print("错误类型:", e)
            return False
        finally:
            self.conn.commit()
        if i > 0:
            return True
        else:
            return False


    def excuteQuery(self, sql, param):
        """
        数据库数据查询
        :param sql: 传入的SQL语句
        :param param: 传入数据
        :return: 返回操作数据库状态
        """
        test = self.cursor.execute(sql, param)
        return test.fetchall()


    def close(self):
        """
        关闭数据库的连接
        :return:
        """
        self.cursor.close()
        self.conn.close()

然后写web服务

import web
from web_test.sqliteDBtool import sqlliteDBtool

urls = (
    '/findUser/(.*)', 'findUser',
    '/addUser', 'addUser',
    '/updateUser', 'updateUser',
    '/deleteUser/(\d)', 'deleteUser'
)


app = web.application(urls, globals())


class findUser:
    def GET(self, loginName):
        #data = web.input()
        db = sqlliteDBtool()
        #param = (data.get("loginName"), )
        if str(loginName).strip().__len__() > 0:
            sql = "select * from sys_user where login_name=?"
            param = (loginName, )
        else:
            sql = "select * from sys_user"
            param = ()
        res = db.excuteQuery(sql, param)
        db.close()
        return res


class addUser:
    def POST(self):
        data = web.input()
        param = [(int(data.get("id")), data.get("login_name"), data.get("user_name"), data.get("password"), )]
        db = sqlliteDBtool()
        res = db.executeUpdate("insert into sys_user(id,login_name,user_name,password) values(?,?,?,?)",
                               param)
        db.close()
        return res


class updateUser:
    def POST(self):
        data = web.input()
        param = [(data.get("login_name"), data.get("user_name"), data.get("password"), int(data.get("id")), )]
        print(param[0][0])
        db = sqlliteDBtool()
        res = db.executeUpdate("update sys_user set login_name=?,user_name=?,password=? where id=?",
                               param)
        db.close()
        return res


class deleteUser:
    def GET(self, id):
        db = sqlliteDBtool()
        param = (id, )
        res = db.excuteDelete("delete from sys_user where id=?", param)
        db.close()
        return res;


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

项目结构如下图:

项目结构

此处没有用到web.py的html模板,只测试了前后端分离的情况,后端提供接口返回json数据。

测试结果如下:

查询

添加一条记录:

添加

查询数据库:

数据库1

可以发现增加成功。

修改和删除也都能通过测试,到此,简单的web接口服务实现完毕。



版权声明:本文为ligai1992原创文章,遵循 CC 4.0 BY-SA 版权协议,转载请附上原文出处链接和本声明。