1.fastapi安装及访问页面

环境

Python3.8

pip3 install fastapi

pip3 install uvicorn

pip3 install -i https://pypi.douban.com/simple 更换阿里源

安装python3.8后使用pip命令安装 fastapi 第三方模块以及 uvicorn 模块

导入 fastapi

from fastapi import FastAPI

app = FastAPI()

@app.get("/")

async def root():

return {"message": "Hello World"}

写入到名为web.py中在终端使用命令启动web

uvicorn web:app --reload

uvicorn main:app --host ‘192.168.93.131’ --port 8080 --reload Linux上这样启动不然会冲突

访问：loaclhost:8000

页面展示Hello World信息

web访问完成

2.与mysq数据库连接

安装 sqlalchemy 模块

建造User的映射

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base): #构造映射字段

__tablename__ = 'tbluser' #对应mysql的tbluser表

userid = Column(Integer,primary_key=True) #表中字段userid

username = Column(String(255))#表中username字段

def __init__(self,userid,username): #构造函数

self.userid = userid

self.username = username

def __str__(self): #打印形式

return "id：%s,名称：%s" % (str(self.userid) ,self.username)

链接mysql用pymysql方式连接

安装pymysql

pip install pymysql

from sqlalchemy import Column, String, create_engine, Integer

#数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名

engine =create_engine("mysql+pymysql://root:123456@localhost:3306/fastapi",encoding='utf-8')

# 把当前的引擎绑定给这个会话

Session = sessionmaker(bind=engine)

# 实例化

session = Session()

查询user

#查询

@app.get("/user/{user_id}")

async def queryUserByUserId(user_id):

# 创建Query查询，filter是where条件，调用one返回唯一行，调用all则是返回所有行

try:

user1 = session.query(User).filter(User.userid==user_id).one()

if user1 ==None:

return "查询无结果"

# user2 = session.query(User).filter(User.userid == '2').one()

# vars1 = {"第一位":user1,"第二位":user2}

# userList = session.query(User).all()

# for user in userList:

# print(user)

# user = engine.connect().execute("select * from tbluser where userid = 1;")

# user = session.execute("select * from user where id = 1;")

session.close()

except ArithmeticError:

print("Error: 查询失败")

return user1

增加单个User CreatUser是用来存储前端穿过来的对象，无法用User这个对象，后续添加的话需要将CreatUser里面值重新赋值给User对象，进行添加

class CreatUser(BaseModel):

userid : int

username : str

def __str__(self):

return "id：%s,名称：%s" % (str(self.userid) ,self.username)

# 增加一个

@app.post("/user/addUser")

async def InserUser(user: CreatUser):

try:

dataUser = User(userid=user.userid,username=user.username)#将CreatUser接收到的值宠幸赋值给User

session.add(dataUser)

session.commit()#每次添加都要commit提交

session.close()#关闭

except ArithmeticError:

print("Error: 添加失败")

return "添加成功"

from typing import List #导入typing模块 的List

# 增加多个

@app.post("/user/addUserList")

async def addUserList(*,user: List[CreatUser]): # 接受List列表 前提需要导入List

try:

for u in user:

dataUser = User(userid=u.userid, username=u.username)

session.add(dataUser) #我觉得这里这样处理应该太消耗资源了,等待更新

session.commit()

session.close()

except ArithmeticError:

print("Error: 添加失败")

return "添加成功"

修改

#修改

@app.put("/user/updateUser")

async def updateUser(user_id: int ,updateUser:schema.CreatUser):

try:

queryUser = session.query(models.User).filter(models.User.userid==user_id).one()

update_dict = updateUser.dict(exclude_unset=True) #变成字典,修改所有

for k, v in update_dict.items():

setattr(queryUser, k, v)

session.commit()

session.refresh(queryUser)

session.close()

except ArithmeticError:

print("Error: 修改失败")

return queryUser

删除:逻辑删除

#删除

@app.delete("/user/deleteUser")

async def deleteUser(user_id,deleteUser: schema.CreatUser):

queryUser = session.query(models.User).filter(models.User.userid == user_id).one()

setattr(queryUser, "isDelete", 0) #修改isDelete的值为0

session.commit()

session.close()

return {'Result': '删除成功'}

总结：

class User(Base): #这里Base用来从数据库获取User对象进行字段映射展示的

class CreatUser(BaseModel): #接受前台传入的对象，需要把值重新给User保存到数据库

每次查询操作都需要关闭session

提交修改操作需要commit，close