一、FastAPI 對MySQL 數據庫的操作

pip install sqlalchemy -i https://pypi.tuna.tsinghua.edu.cn/simplepip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple注意，pymysql需要寫進代碼中，下面會有體現

二、創建數據庫

我相信你們都會所以我就不寫了

三、通過代碼連接數據庫并且測試

#TODO 連接mysql數據庫# 導入FastAPI模塊 from fastapi import FastAPI# 創建app實例 app = FastAPI()## 連接數據庫# 連接mysql數據庫需要導入pymysql模塊 import pymysqlpymysql.install_as_MySQLdb()from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker# 配置數據庫地址：數據庫類型+數據庫驅動名稱://用戶名:密碼@機器地址:端口號/數據庫名 engine = create_engine("mysql://root:1234@127.0.0.1:3306/fastapi", encoding='utf-8') # 把當前的引擎綁定給這個會話； # autocommit：是否自動提交 autoflush：是否自動刷新并加載數據庫 bind：綁定數據庫引擎 Session = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 實例化 session = Session()#TODO 定義表# # declarative_base類維持了一個從類到表的關系，通常一個應用使用一個Base實例，所有實體類都應該繼承此類對象 from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()from sqlalchemy import Column, String, Integer# 創建數據庫模型（定義表結構:表名稱，字段名稱以及字段類型） class User(Base):# 定義表名__tablename__ = 'tbluser'# 定義字段# primary_key=True 設置為主鍵userid = Column(Integer, primary_key=True)username = Column(String(255))# 構造函數def __init__(self, userid, username):self.userid = useridself.username = username# 打印形式def __str__(self):return "id:%s, name:%s" % (str(self.userid), self.username)# 在數據庫中生成表 Base.metadata.create_all(bind=engine)### 添加數據from pydantic import BaseModel# 定義數據模型 class CreatUser(BaseModel):userid: intusername: strdef __str__(self):return "id：%s, name：%s" % (str(self.userid), self.username)

通過uvicorn啟動測試連接并且生成對象表

uvicorn app:app --port 8080 --host 127.0.0.1 --reload

?檢查數據庫看看是否成功

四、增刪改查代碼

### 添加數據from pydantic import BaseModel # 定義數據模型 class CreatUser(BaseModel):userid: intusername: strdef __str__(self):return "id：%s, name：%s" % (str(self.userid), self.username)## 添加單個 @app.post("/user/addUser") async def InserUser(user: CreatUser):try:# 添加數據dataUser = User(userid=user.userid, username=user.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code":"0002","message":"數據庫異常"}return {"code":"0000","message":"添加成功"}from typing import List ## 添加多個 @app.post("/user/addUserList") async def addUserList(*, user: List[CreatUser]):try:# user是一個列表，每個內部元素均為CreatUser類型for u in user:# 自定義的數據模型可以用.訪問屬性dataUser = User(userid=u.userid, username=u.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code": "0002", "message": "數據庫異常"}return {"code": "0000", "message": "添加成功"}### 查詢## 按照user_id查詢 @app.get("/user/{user_id}") async def queryUserByUserId(user_id: int):# 創建Query查詢，filter是where條件，調用one返回唯一行，調用all則是返回所有行try:# one與first的區別：# one：要求結果集中只有一個結果；如果數據庫返回0或2個或更多結果，并且將引發異常，則為錯誤。# first：返回可能更大的結果集中的第一個，如果沒有結果，則返回None。不會引發異常。# filter_by與filter的區別：# filter_by接收的參數形式是關鍵字參數，而filter接收的參數是更加靈活的SQL表達式結構# user1 = session.query(User).filter_by(userid=user_id).first()user1 = session.query(User).filter(User.userid==user_id).first()session.close()# 由于user1只有一個值，所以它直接是一個字典if user1:return {"code":"0000","message":"請求成功","data":user1}else:return {"code":"0001","message":"查詢無結果"}except ArithmeticError:return {"code":"0002","message":"數據庫異常"}## 查詢所有 @app.get("/user/selectall/") async def queryUserByUserId():# 創建Query查詢，filter是where條件，調用one返回唯一行，調用all則是返回所有行try:user1 = session.query(User).all()session.close()# user1 是一個列表，內部元素為字典return {"code": "0000", "message": "請求成功", "data": user1}except ArithmeticError:return {"code":"0002","message":"數據庫異常"}###刪除# 根據user_id刪除單個 @app.delete("/user/deleteUser/{user_id}") async def deleteUser(user_id: int):try:user1 = session.query(User).filter(User.userid == user_id).first()if user1:session.delete(user1)session.commit()session.close()return {"code": "0000", "message": "刪除成功"}else:return {"code": "0001", "message": "參數錯誤"}except ArithmeticError:return {"code": "0002", "message": "數據庫錯誤"}from typing import List ## 刪除多個 @app.delete("/user/deleteUserList") async def deleteUser(user_ids: List[int]):try:for user_id in user_ids:user1 = session.query(User).filter(User.userid == user_id).first()if user1:session.delete(user1)session.commit()session.close()return {"code": "0000", "message": "刪除成功"}except ArithmeticError:return {"code": "0002", "message": "數據庫錯誤"}###修改## 根據user_id修改user_name @app.put("/user/updateUser/") # 定義查詢參數user_id和user_name async def updateUser(user_id: int, user_name: str):try:user1 = session.query(User).filter(User.userid == user_id).first()print(user1)if user1:user1.username = user_namesession.commit()session.close()return {"code": "0000", "message": "修改成功"}else:return {"code": "0001", "message": "參數錯誤"}except ArithmeticError:return {"code": "0002", "message": "數據庫錯誤"}# 方式2： class AlterUser(BaseModel):userid: intusername: str@app.put("/user/updateUser01/") async def deleteUser(user: AlterUser):try:user1 = session.query(User).filter(User.userid == user.userid).first()if user1:user1.username = user.usernamesession.commit()session.close()return {"code": "0000", "message": "修改成功"}else:return {"code": "0001", "message": "參數錯誤"}except ArithmeticError:return {"code": "0002", "message": "數據庫錯誤"}

總結

以上是生活随笔為你收集整理的FastAPI 对MySQL 数据库的操作(win64)的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。