fastapi+sqlserver实现简单接口

  • 作者:ykk
  • 分类: python
  • 发表时间:2020-05-30 21:37:56
  • 阅读量:(668)

1、使用docker启动sqlserver容器。


[root@localhost ~]# docker ps -a
CONTAINER ID        IMAGE                                                  COMMAND                  CREATED             STATUS                       PORTS                    NAMES
45380f7e75bf        mcr.microsoft.com/mssql/server:2019-CU3-ubuntu-18.04   "/opt/mssql/bin/perm…"   3 weeks ago         Exited (255) 8 minutes ago   0.0.0.0:1433->1433/tcp   sql1

[root@localhost ~]# docker restart 45380f7e75bf 
45380f7e75bf
[root@localhost ~]# docker ps -a
CONTAINER ID        IMAGE                                                  COMMAND                  CREATED             STATUS              PORTS                    NAMES
45380f7e75bf        mcr.microsoft.com/mssql/server:2019-CU3-ubuntu-18.04   "/opt/mssql/bin/perm…"   3 weeks ago         Up 3 seconds        0.0.0.0:1433->1433/tcp   sql1
[root@localhost ~]# 

2、连接测试sqlserver是否可用。

3、使用FastAPI实现简单接口。

(1)、sql_server_conn.py 文件  使用PooledDB连接池进行sqlserver连接。


import pymssql
from DBUtils.PooledDB import PooledDB #连接池

class MSSQL:
    def __init__(self, host, user, pwd, db):
        self.host = host
        self.user = user
        self.pwd = pwd
        self.db = db

    def GetConnect(self):
        if not self.db:
            raise (NameError, '没有设置数据库信息')
        self.Pool=PooledDB(creator=pymssql, mincached=2, maxcached=5, maxshared=3, maxconnections=6, blocking=True,
                        host=self.host, user=self.user, \
                        password=self.pwd, database=self.db, charset="GBK",as_dict=True) #指定游标为字典类型
        self.conn = self.Pool.connection()
        cur = self.conn.cursor()
        if not cur:
            raise (NameError, '连接数据库失败')
        else:
            print('数据库连接成功')
            return cur


    def ExecQuery(self, sql):
        cur = self.GetConnect()
        cur.execute(sql)
        resList = cur.fetchall()
        self.conn.close() #查询后不关闭数据库连接
        return resList #list装的是dict



    def ExecNonQuery(self, sql):
        cur = self.GetConnect()
        cur.execute(sql)
        self.conn.commit()
        self.conn.close()



def main():
    ms = MSSQL(host='192.168.0.178', user='SA', pwd='123321Aaa@', db="master")
    sql = ms.ExecQuery('SELECT * FROM   sto')
    print(sql)



if __name__ == '__main__':
    main()

(2)、hzyl_goods_api.py 文件 ,编写接口。


from fastapi import FastAPI
import sql_server_conn


#1、实例化FastAPI对象
app = FastAPI()

#2、实例化sql_server对象
ms = sql_server_conn.MSSQL(host='192.168.0.178', user='SA', pwd='123321Aaa@', db="master")

(3)、创建访问路径,方式为GET请求


#导入http返回模块,"/"根路径返回html。
from fastapi.responses import HTMLResponse
@app.get("/")
async def read_root(): #async 异步

    html_content = """
    <html>
        <head>
            <title>Some api in here</title>
        </head>
        <body>
            <h3>调用方式</h3> 
            <p style="color:red;">http://127.0.0.1:8080/表名/表中的键值 /all为查询表中所有数据<p>
            <p style="color:red;">返回数据为json<p>
        </body>
    </html>
    """
    return HTMLResponse(content=html_content, status_code=200)

(4)、解决datetime类型json转换无法正常显示的问题。


class ComplexEncoder(json.JSONEncoder): #重写类方法 处理datetime无法转换的问题
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.strftime('%Y-%m-%d %H:%M:%S')
        elif isinstance(obj, datetime.date):
            return obj.strftime('%Y-%m-%d')
        else:
            return json.JSONEncoder.default(self, obj)

(5)、下面为具体的实现代码。


#返回的数据转为json
def listToJson(lst):
    import json
    for i in lst:  #i为dict
        str_json= json.dumps(i,cls=ComplexEncoder,ensure_ascii=False)#ensure_ascii=False 转换时指定重写的类方法

    return str_json

@app.get("/sto_a/{item_id}")
def read_item(item_id: str = None):
    if item_id=="all": #当输入为/all时查询所有数据。其他情况查询单条数据。
        list = ms.ExecQuery("SELECT * FROM  sto_a ") #字典对象
    else:
        list = ms.ExecQuery("SELECT * FROM  sto_a WHERE  ID =  '%s'"%(item_id)) #字典对象
    if list:
        json = listToJson(list)
        return json
    else:
        return '未找到记录'

(6)、程序入口


#使用uvicorn进行启动服务
if __name__ == '__main__':


    import uvicorn

    uvicorn.run(app=app,
                host="127.0.0.1",
                port=8080,
                workers=1)

 

(7)、问题补充:

1.测试发现前端返回的json带有“\”

2.优化代码 使用demjson 二次转换


def listToJson(lst):
    json_list =[]  #添加列表返回多个json
    import json
    import demjson #使用demjson二次处理,解决前端出现反斜杠的问题
    for i in lst:  #i为dict
        str_json= demjson.decode(json.dumps(i,cls=ComplexEncoder,ensure_ascii=False)) #ensure_ascii=False 转换时指定重写的类方法
        json_list.append(str_json)
    return json_list

 

(8)、效果展示:

首页

 

单条数据

多条数据

 

4、完整代码。


from fastapi import FastAPI
from fastapi.responses import HTMLResponse
import sql_server_conn
import datetime
import json

app = FastAPI()

ms = sql_server_conn.MSSQL(host='192.168.0.178', user='SA', pwd='123321Aaa@', db="master")


class ComplexEncoder(json.JSONEncoder): #重写类 处理datetime无法转换的问题
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.strftime('%Y-%m-%d %H:%M:%S')
        elif isinstance(obj, datetime.date):
            return obj.strftime('%Y-%m-%d')
        else:
            return json.JSONEncoder.default(self, obj)


def listToJson(lst):
    json_list =[]
    import json
    import demjson #使用demjson二次处理,解决前端出现反斜杠的问题
    for i in lst:  #i为dict
        str_json= demjson.decode(json.dumps(i,cls=ComplexEncoder,ensure_ascii=False)) #ensure_ascii=False 转换时指定重写的类方法
        json_list.append(str_json)
    return json_list


@app.get("/")
async def read_root(): #async 异步

    html_content = """
    <html>
        <head>
            <title>Some api in here</title>
        </head>
        <body>
            <h1>调用方式</h1> 
            <p style="color:red;">http://127.0.0.1:8080/表名/表中的键值 /all为查询表中所有数据<p>
            <p style="color:red;">返回数据为json<p>
        </body>
    </html>
    """
    return HTMLResponse(content=html_content, status_code=200)



@app.get("/sto_a/{item_id}")
def read_item(item_id: str = None):
    if item_id=="all": #当输入为/all时查询所有数据。其他情况查询单条数据。
        list = ms.ExecQuery("SELECT * FROM  sto_a ") #字典对象
    else:
        list = ms.ExecQuery("SELECT * FROM  sto_a WHERE  ID =  '%s'"%(item_id)) #字典对象
    if list:
        json = listToJson(list)
        return json
    else:
        return '未找到记录'

if __name__ == '__main__':


    import uvicorn

    uvicorn.run(app=app,
                host="127.0.0.1",
                port=8080,
                workers=1)






 

 

 

上一篇: request bs4 模块实现简单爬虫

下一篇: 3月31日 星期二 阴

评论 列表: