Python FastAPI 框架 操作Mysql数据库 增删改查
2 比 1 更容易理解,可以先看2(單文件級別)
1、FastAPI 框架 操作Mysql數(shù)據(jù)庫(項目多文件級別)
FastAPI 可以使用任何您想要的關系型數(shù)據(jù)庫。
在這里,讓我們看一個使用著SQLAlchemy的示例。
您可以很容易地將SQLAlchemy支持任何數(shù)據(jù)庫,像:
- PostgreSQL
- MySQL
- SQLite
- Oracle
- Microsoft SQL Server,等等其它數(shù)據(jù)庫
在此示例中,我們將使用SQLite,因為它使用單個文件并且 在Python中具有集成支持。因此,您可以復制此示例并按原樣來運行它。
稍后,對于您的產(chǎn)品級別的應用程序,您可能會要使用像PostgreSQL這樣的數(shù)據(jù)庫服務器。
1.0 創(chuàng)建mysql數(shù)據(jù)庫
-
創(chuàng)建test數(shù)據(jù)庫,數(shù)據(jù)庫創(chuàng)建users表和items表
-
users表
- items表
1.1 測試項目文件結構
對于這些示例,假設您有一個名為的目錄my_super_project,其中包含一個名為的子目錄sql_app,其結構如下:
. └── sql_app├── __init__.py├── crud.py├── database.py├── main.py├── models.py└── schemas.py該文件__init__.py只是一個空文件,但它告訴 Python 其中sql_app的所有模塊(Python 文件)都是一個包。
1.2 數(shù)據(jù)庫配置 database.py
# 1、導入 SQLAlchemy 部件 from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 連接mysql數(shù)據(jù)庫需要導入pymysql模塊 import pymysql pymysql.install_as_MySQLdb()# 2、為 SQLAlchemy 定義數(shù)據(jù)庫 URL地址 # 配置數(shù)據(jù)庫地址:數(shù)據(jù)庫類型+數(shù)據(jù)庫驅(qū)動名稱://用戶名:密碼@機器地址:端口號/數(shù)據(jù)庫名 SQLALCHEMY_DATABASE_URL = "mysql://test:123456@127.0.0.1:3306/test"# 3、創(chuàng)建 SQLAlchemy 引擎 engine = create_engine(SQLALCHEMY_DATABASE_URL, encoding='utf-8')# 4、創(chuàng)建數(shù)據(jù)庫會話 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)# 5、創(chuàng)建一個Base類declarative_base # 稍后我們將用這個類繼承,來創(chuàng)建每個數(shù)據(jù)庫模型或類(ORM 模型) Base = declarative_base()1.3 創(chuàng)建數(shù)據(jù)庫模型 models.py
用Base類來創(chuàng)建 SQLAlchemy 模型
我們將使用我們之前創(chuàng)建的Base類來創(chuàng)建 SQLAlchemy 模型。
SQLAlchemy 使用的“模型”這個術語 來指代與數(shù)據(jù)庫交互的這些類和實例。
而 Pydantic 也使用“模型”這個術語 來指代不同的東西,即數(shù)據(jù)驗證、轉(zhuǎn)換以及文檔類和實例。
創(chuàng)建關系
items = relationship("Item", back_populates="owner")- 當訪問 user 中的屬性items時,如 中my_user.items,它將有一個ItemSQLAlchemy 模型列表(來自items表),這些模型具有指向users表中此記錄的外鍵
- 當您訪問my_user.items時,SQLAlchemy 實際上會從items表中的獲取一批記錄并在此處填充進去。
- 同樣,當訪問 Item中的屬性owner時,它將包含表中的-UserSQLAlchemy 模型users。使用owner_id屬性/列及其外鍵來了解要從users表中獲取哪條記錄。
1.4 創(chuàng)建 Pydantic 模型 schemas.py
現(xiàn)在讓我們查看一下文件sql_app/schemas.py。
為了避免 SQLAlchemy模型和 Pydantic模型之間的混淆,我們將有models.py(SQLAlchemy 模型的文件)和schemas.py( Pydantic 模型的文件)。
這些 Pydantic 模型或多或少地定義了一個“schema”(一個有效的數(shù)據(jù)形狀)。
因此,這將幫助我們在使用兩者時避免混淆。
創(chuàng)建初始 Pydantic模型/模式?
創(chuàng)建一個ItemBase和UserBasePydantic模型(或者我們說“schema”)以及在創(chuàng)建或讀取數(shù)據(jù)時具有共同的屬性。
ItemCreate為 創(chuàng)建一個UserCreate繼承自它們的所有屬性(因此它們將具有相同的屬性),以及創(chuàng)建所需的任何其他數(shù)據(jù)(屬性)。
因此在創(chuàng)建時也應當有一個password屬性。
但是為了安全起見,password不會出現(xiàn)在其他同類 Pydantic模型中,例如用戶請求時不應該從 API 返回響應中包含它。
from typing import List, Union# 1、創(chuàng)建初始 Pydantic模型/模式 from pydantic import BaseModel# 1、創(chuàng)建初始 Pydantic模型/模式 class ItemBase(BaseModel):title: strdescription: Union[str, None] = None# 1、創(chuàng)建初始 Pydantic模型/模式 class ItemCreate(ItemBase):pass# 2、創(chuàng)建用于讀取/返回的Pydantic模型/模式 class Item(ItemBase):id: intowner_id: intclass Config:orm_mode = True# 1、創(chuàng)建初始 Pydantic模型/模式 class UserBase(BaseModel):email: str# 1、創(chuàng)建初始 Pydantic模型/模式 class UserCreate(UserBase):password: str# 2、創(chuàng)建用于讀取/返回的Pydantic模型/模式 class User(UserBase):id: intis_active: boolitems: List[Item] = []class Config:orm_mode = True請注意,讀取用戶(從 API 返回)時將使用不包括password的User Pydantic模型。
SQLAlchemy 風格和 Pydantic 風格
請注意,SQLAlchemy模型使用 =來定義屬性,并將類型作為參數(shù)傳遞給Column,例如:
name = Column(String)雖然 Pydantic模型使用: 聲明類型,但新的類型注釋語法/類型提示是:
name: str請牢記這一點,這樣您在使用:還是=時就不會感到困惑。
1.5 CRUD工具 crud.py
從 sqlalchemy.orm中導入Session,這將允許您聲明db參數(shù)的類型,并在您的函數(shù)中進行更好的類型檢查和完成。
導入之前的models(SQLAlchemy 模型)和schemas(Pydantic模型/模式)。
from sqlalchemy.orm import Sessionfrom . import models, schemasdef get_user(db: Session, user_id: int):return db.query(models.User).filter(models.User.id == user_id).first()# 通過 ID 和電子郵件查詢單個用戶 def get_user_by_email(db: Session, email: str):return db.query(models.User).filter(models.User.email == email).first()# 查詢多個用戶 def get_users(db: Session, skip: int = 0, limit: int = 100):return db.query(models.User).offset(skip).limit(limit).all()def create_user(db: Session, user: schemas.UserCreate):fake_hashed_password = user.password + "notreallyhashed"# 使用您的數(shù)據(jù)創(chuàng)建一個 SQLAlchemy 模型實例。db_user = models.User(email=user.email, hashed_password=fake_hashed_password)# 使用add來將該實例對象添加到您的數(shù)據(jù)庫。db.add(db_user)# 使用commit來對數(shù)據(jù)庫的事務提交(以便保存它們)。db.commit()# 使用refresh來刷新您的數(shù)據(jù)庫實例(以便它包含來自數(shù)據(jù)庫的任何新數(shù)據(jù),例如生成的 ID)。db.refresh(db_user)return db_user# 查詢多個項目 def get_items(db: Session, skip: int = 0, limit: int = 100):return db.query(models.Item).offset(skip).limit(limit).all()def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):db_item = models.Item(**item.dict(), owner_id=user_id)db.add(db_item)db.commit()db.refresh(db_item)return db_item1.6 主FastAPI應用程序 main.py
from typing import Listfrom fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Sessionfrom . import crud, models, schemas from .database import SessionLocal, engine# 1、創(chuàng)建數(shù)據(jù)庫表 models.Base.metadata.create_all(bind=engine)app = FastAPI()# 2、創(chuàng)建依賴項 # Dependency def get_db():# 我們需要每個請求有一個獨立的數(shù)據(jù)庫會話/連接(SessionLocal),# 在所有請求中使用相同的會話,然后在請求完成后關閉它。db = SessionLocal()# 我們的依賴項將創(chuàng)建一個新的 SQLAlchemy SessionLocal,# 它將在單個請求中使用,然后在請求完成后關閉它。try:yield dbfinally:db.close()# 4、創(chuàng)建您的FastAPI 路徑操作 # 3、db: Session = Depends(get_db) :當在路徑操作函數(shù)中使用依賴項時,我們使用Session,直接從 SQLAlchemy 導入的類型聲明它。 @app.post("/users/", response_model=schemas.User) def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):db_user = crud.get_user_by_email(db, email=user.email)if db_user:raise HTTPException(status_code=400, detail="Email already registered")return crud.create_user(db=db, user=user)# 4、創(chuàng)建您的FastAPI 路徑操作 @app.get("/users/", response_model=List[schemas.User]) def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):users = crud.get_users(db, skip=skip, limit=limit)return users# 4、創(chuàng)建您的FastAPI 路徑操作 @app.get("/users/{user_id}", response_model=schemas.User) def read_user(user_id: int, db: Session = Depends(get_db)):db_user = crud.get_user(db, user_id=user_id)if db_user is None:raise HTTPException(status_code=404, detail="User not found")return db_user# 4、創(chuàng)建您的FastAPI 路徑操作 @app.post("/users/{user_id}/items/", response_model=schemas.Item) def create_item_for_user(user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db) ):return crud.create_user_item(db=db, item=item, user_id=user_id)# 4、創(chuàng)建您的FastAPI 路徑操作 @app.get("/items/", response_model=List[schemas.Item]) def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):items = crud.get_items(db, skip=skip, limit=limit)return items1.6 關于 def 對比 async def
在這里,我們在路徑操作函數(shù)和依賴項中都使用著 SQLAlchemy 模型,它將與外部數(shù)據(jù)庫進行通信。
這會需要一些“等待時間”。
但是由于 SQLAlchemy 不具有await直接使用的兼容性,因此類似于:
user = await db.query(User).first()…相反,我們可以使用:
user = db.query(User).first()然后我們應該聲明路徑操作函數(shù)和不帶 的依賴關系async def,只需使用普通的def,如下:
@app.get("/users/{user_id}", response_model=schemas.User) def read_user(user_id: int, db: Session = Depends(get_db)):db_user = crud.get_user(db, user_id=user_id)...如果您需要異步連接到關系數(shù)據(jù)庫,請參閱Async SQL (Relational) Databases
如果您很好奇并且擁有深厚的技術知識,您可以在Async文檔中查看有關如何處理 async def于def差別的技術細節(jié)。
1.7 執(zhí)行項目
- 在sql_app目錄外執(zhí)行
2、FastAPI 對MySQL 數(shù)據(jù)庫的操作(單文件級別)
(317條消息) FastAPI 對MySQL 數(shù)據(jù)庫的操作(win64)_fastapi 連接mysql_梧鈷的博客-CSDN博客
2.1 安裝依賴
$ pip install sqlalchemy -i https://pypi.tuna.tsinghua.edu.cn/simple $ pip install pymysql -i https://pypi.tuna.tsinghua.edu.cn/simple2.2 測試連接
- main.py
- 運行
2.3 增刪改查代碼
增
from pydantic import BaseModel # 定義數(shù)據(jù)模型 class CreatUser(BaseModel):userid: intusername: strdef __str__(self):return "id:%s, name:%s" % (str(self.userid), self.username)# 添加單個數(shù)據(jù) @app.post("/user/addUser") async def InserUser(user: CreatUser):try:# 添加數(shù)據(jù)dataUser = User(userid=user.userid, username=user.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫異常"}return {"code": "0000", "message": "添加成功"}from typing import List# 添加多個數(shù)據(jù) @app.post("/user/addUserList") async def addUserList(*, user: List[CreatUser]):try:# user是一個列表,每個內(nèi)部元素均為CreatUser類型for u in user:# 自定義的數(shù)據(jù)模型可以用.訪問屬性dataUser = User(userid=u.userid, username=u.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫異常"}return {"code": "0000", "message": "添加成功"}查
# 按照user_id查詢 @app.get("/user/{user_id}") async def queryUserByUserId(user_id: int):# 創(chuàng)建Query查詢,filter是where條件,調(diào)用one返回唯一行,調(diào)用all則是返回所有行try:# one與first的區(qū)別:# one:要求結果集中只有一個結果;如果數(shù)據(jù)庫返回0或2個或更多結果,并且將引發(fā)異常,則為錯誤。# first:返回可能更大的結果集中的第一個,如果沒有結果,則返回None。不會引發(fā)異常。# filter_by與filter的區(qū)別:# filter_by接收的參數(shù)形式是關鍵字參數(shù),而filter接收的參數(shù)是更加靈活的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": "數(shù)據(jù)庫異常"}## 查詢所有 @app.get("/user/selectall/") async def queryUserByUserId():# 創(chuàng)建Query查詢,filter是where條件,調(diào)用one返回唯一行,調(diào)用all則是返回所有行try:user1 = session.query(User).all()session.close()# user1 是一個列表,內(nèi)部元素為字典return {"code": "0000", "message": "請求成功", "data": user1}except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫異常"}刪
# 根據(jù)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": "參數(shù)錯誤"}except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫錯誤"}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": "數(shù)據(jù)庫錯誤"}改
# 根據(jù)user_id修改user_name @app.put("/user/updateUser/") # 定義查詢參數(shù)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": "參數(shù)錯誤"}except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫錯誤"}# 方式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": "參數(shù)錯誤"}except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫錯誤"}完整代碼
# 導入FastAPI模塊 from fastapi import FastAPI app = FastAPI()# 1、連接mysql數(shù)據(jù)庫需要導入pymysql模塊 import pymysql pymysql.install_as_MySQLdb()# 2、配置數(shù)據(jù)庫 from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker# 配置數(shù)據(jù)庫地址:數(shù)據(jù)庫類型+數(shù)據(jù)庫驅(qū)動名稱://用戶名:密碼@機器地址:端口號/數(shù)據(jù)庫名 SQLALCHEMY_DATABASE_URL = "mysql://test:123456@127.0.0.1:3306/test" engine = create_engine(SQLALCHEMY_DATABASE_URL, encoding='utf-8')# 3、把當前的引擎綁定給這個會話; # autocommit:是否自動提交 autoflush:是否自動刷新并加載數(shù)據(jù)庫 bind:綁定數(shù)據(jù)庫引擎 Session = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 實例化 session = Session()# 4、創(chuàng)建Base實例 # declarative_base類維持了一個從類到表的關系,通常一個應用使用一個Base實例,所有實體類都應該繼承此類對象 from sqlalchemy.ext.declarative import declarative_base Base = declarative_base()# 5、創(chuàng)建數(shù)據(jù)庫模型(定義表結構:表名稱,字段名稱以及字段類型) from sqlalchemy import Column, String, Integerclass User(Base):# 定義表名__tablename__ = 'user'# 定義字段# primary_key=True 設置為主鍵userid = Column(Integer, primary_key=True)username = Column(String(255))# 構造函數(shù)def __init__(self, userid, username):self.userid = useridself.username = username# 打印形式def __str__(self):return "id:%s, name:%s" % (str(self.userid), self.username)# 6、在數(shù)據(jù)庫中生成表 Base.metadata.create_all(bind=engine)# 【增】 from pydantic import BaseModel # 定義數(shù)據(jù)模型 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:# 添加數(shù)據(jù)dataUser = User(userid=user.userid, username=user.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫異常"}return {"code": "0000", "message": "添加成功"}from typing import List# 添加多個 @app.post("/user/addUserList") async def addUserList(*, user: List[CreatUser]):try:# user是一個列表,每個內(nèi)部元素均為CreatUser類型for u in user:# 自定義的數(shù)據(jù)模型可以用.訪問屬性dataUser = User(userid=u.userid, username=u.username)session.add(dataUser)session.commit()session.close()except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫異常"}return {"code": "0000", "message": "添加成功"}# 【查】# 按照user_id查詢 @app.get("/user/{user_id}") async def queryUserByUserId(user_id: int):# 創(chuàng)建Query查詢,filter是where條件,調(diào)用one返回唯一行,調(diào)用all則是返回所有行try:# one與first的區(qū)別:# one:要求結果集中只有一個結果;如果數(shù)據(jù)庫返回0或2個或更多結果,并且將引發(fā)異常,則為錯誤。# first:返回可能更大的結果集中的第一個,如果沒有結果,則返回None。不會引發(fā)異常。# filter_by與filter的區(qū)別:# filter_by接收的參數(shù)形式是關鍵字參數(shù),而filter接收的參數(shù)是更加靈活的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": "數(shù)據(jù)庫異常"}## 查詢所有 @app.get("/user/selectall/") async def queryUserByUserId():# 創(chuàng)建Query查詢,filter是where條件,調(diào)用one返回唯一行,調(diào)用all則是返回所有行try:user1 = session.query(User).all()session.close()# user1 是一個列表,內(nèi)部元素為字典return {"code": "0000", "message": "請求成功", "data": user1}except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫異常"}#【刪】# 根據(jù)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": "參數(shù)錯誤"}except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫錯誤"}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": "數(shù)據(jù)庫錯誤"}# 【改】## 根據(jù)user_id修改user_name @app.put("/user/updateUser/") # 定義查詢參數(shù)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": "參數(shù)錯誤"}except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫錯誤"}# 方式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": "參數(shù)錯誤"}except ArithmeticError:return {"code": "0002", "message": "數(shù)據(jù)庫錯誤"}運行測試
uvicorn main:app --reload總結
以上是生活随笔為你收集整理的Python FastAPI 框架 操作Mysql数据库 增删改查的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: C# 字节数组截取
- 下一篇: 导、地线用途及技术特性