2 比 1 更容易理解，可以先看2（單文件級別）

1、FastAPI 框架 操作Mysql數(shù)據(jù)庫(項(xiàng)目多文件級別)

FastAPI 可以使用任何您想要的關(guān)系型數(shù)據(jù)庫。
在這里，讓我們看一個使用著SQLAlchemy的示例。
您可以很容易地將SQLAlchemy支持任何數(shù)據(jù)庫，像：

• PostgreSQL
• MySQL
• SQLite
• Oracle
• Microsoft SQL Server，等等其它數(shù)據(jù)庫
  在此示例中，我們將使用SQLite，因?yàn)樗褂脝蝹€文件并且 在Python中具有集成支持。因此，您可以復(fù)制此示例并按原樣來運(yùn)行它。

稍后，對于您的產(chǎn)品級別的應(yīng)用程序，您可能會要使用像PostgreSQL這樣的數(shù)據(jù)庫服務(wù)器。

1.0 創(chuàng)建mysql數(shù)據(jù)庫

• 創(chuàng)建test數(shù)據(jù)庫，數(shù)據(jù)庫創(chuàng)建users表和items表

• users表

• items表

1.1 測試項(xiàng)目文件結(jié)構(gòu)

對于這些示例，假設(shè)您有一個名為的目錄my_super_project，其中包含一個名為的子目錄sql_app，其結(jié)構(gòu)如下：

. └── 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、導(dǎo)入 SQLAlchemy 部件 from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 連接mysql數(shù)據(jù)庫需要導(dǎo)入pymysql模塊 import pymysql pymysql.install_as_MySQLdb()# 2、為 SQLAlchemy 定義數(shù)據(jù)庫 URL地址 # 配置數(shù)據(jù)庫地址：數(shù)據(jù)庫類型+數(shù)據(jù)庫驅(qū)動名稱://用戶名:密碼@機(jī)器地址:端口號/數(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ù)語 來指代與數(shù)據(jù)庫交互的這些類和實(shí)例。
而 Pydantic 也使用“模型”這個術(shù)語 來指代不同的東西，即數(shù)據(jù)驗(yàn)證、轉(zhuǎn)換以及文檔類和實(shí)例。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String from sqlalchemy.orm import relationship# 1、從database.py導(dǎo)入Base類 from .database import Base# User繼承Base類 class User(Base):# 表名__tablename__ = "users"# 2、創(chuàng)建模型屬性/列，使用Column來表示 SQLAlchemy 中的默認(rèn)值。id = Column(Integer, primary_key=True, index=True)email = Column(String, unique=True, index=True)hashed_password = Column(String)is_active = Column(Boolean, default=True)# 3、創(chuàng)建關(guān)系# 當(dāng)訪問 user 中的屬性items時，如 中my_user.items，它將有一個ItemSQLAlchemy 模型列表（來自items表），這些模型具有指向users表中此記錄的外鍵# 當(dāng)您訪問my_user.items時，SQLAlchemy 實(shí)際上會從items表中的獲取一批記錄并在此處填充進(jìn)去。# 同樣，當(dāng)訪問 Item中的屬性owner時，它將包含表中的UserSQLAlchemy 模型users。使用owner_id屬性/列及其外鍵來了解要從users表中獲取哪條記錄。items = relationship("Item", back_populates="owner")# Item繼承Base類 class Item(Base):__tablename__ = "items"id = Column(Integer, primary_key=True, index=True)title = Column(String, index=True)description = Column(String, index=True)owner_id = Column(Integer, ForeignKey("users.id"))owner = relationship("User", back_populates="items")

創(chuàng)建關(guān)系

items = relationship("Item", back_populates="owner")

• 當(dāng)訪問 user 中的屬性items時，如 中my_user.items，它將有一個ItemSQLAlchemy 模型列表（來自items表），這些模型具有指向users表中此記錄的外鍵
• 當(dāng)您訪問my_user.items時，SQLAlchemy 實(shí)際上會從items表中的獲取一批記錄并在此處填充進(jìn)去。
• 同樣，當(dāng)訪問 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)建時也應(yīng)當(dāng)有一個password屬性。

但是為了安全起見，password不會出現(xiàn)在其他同類 Pydantic模型中，例如用戶請求時不應(yīng)該從 API 返回響應(yīng)中包含它。

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 風(fēng)格和 Pydantic 風(fēng)格

請注意，SQLAlchemy模型使用 =來定義屬性，并將類型作為參數(shù)傳遞給Column，例如：

name = Column(String)

雖然 Pydantic模型使用: 聲明類型，但新的類型注釋語法/類型提示是：

name: str

請牢記這一點(diǎn)，這樣您在使用:還是=時就不會感到困惑。

1.5 CRUD工具 crud.py

從 sqlalchemy.orm中導(dǎo)入Session，這將允許您聲明db參數(shù)的類型，并在您的函數(shù)中進(jìn)行更好的類型檢查和完成。

導(dǎo)入之前的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 模型實(shí)例。db_user = models.User(email=user.email, hashed_password=fake_hashed_password)# 使用add來將該實(shí)例對象添加到您的數(shù)據(jù)庫。db.add(db_user)# 使用commit來對數(shù)據(jù)庫的事務(wù)提交（以便保存它們）。db.commit()# 使用refresh來刷新您的數(shù)據(jù)庫實(shí)例（以便它包含來自數(shù)據(jù)庫的任何新數(shù)據(jù)，例如生成的 ID）。db.refresh(db_user)return db_user# 查詢多個項(xiàng)目 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_item

1.6 主FastAPI應(yīng)用程序 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)建依賴項(xiàng) # Dependency def get_db():# 我們需要每個請求有一個獨(dú)立的數(shù)據(jù)庫會話/連接（SessionLocal），# 在所有請求中使用相同的會話，然后在請求完成后關(guān)閉它。db = SessionLocal()# 我們的依賴項(xiàng)將創(chuàng)建一個新的 SQLAlchemy SessionLocal，# 它將在單個請求中使用，然后在請求完成后關(guān)閉它。try:yield dbfinally:db.close()# 4、創(chuàng)建您的FastAPI 路徑操作 # 3、db: Session = Depends(get_db) ：當(dāng)在路徑操作函數(shù)中使用依賴項(xiàng)時，我們使用Session，直接從 SQLAlchemy 導(dǎo)入的類型聲明它。 @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 items

1.6 關(guān)于 def 對比 async def

在這里，我們在路徑操作函數(shù)和依賴項(xiàng)中都使用著 SQLAlchemy 模型，它將與外部數(shù)據(jù)庫進(jìn)行通信。

這會需要一些“等待時間”。

但是由于 SQLAlchemy 不具有await直接使用的兼容性，因此類似于：

user = await db.query(User).first()

…相反，我們可以使用：

user = db.query(User).first()

然后我們應(yīng)該聲明路徑操作函數(shù)和不帶 的依賴關(guān)系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)...

如果您需要異步連接到關(guān)系數(shù)據(jù)庫，請參閱Async SQL (Relational) Databases

如果您很好奇并且擁有深厚的技術(shù)知識，您可以在Async文檔中查看有關(guān)如何處理 async def于def差別的技術(shù)細(xì)節(jié)。

1.7 執(zhí)行項(xiàng)目

• 在sql_app目錄外執(zhí)行

uvicorn sql_app.main:app --reload INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

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/simple

2.2 測試連接

• main.py

# 導(dǎo)入FastAPI模塊 from fastapi import FastAPI app = FastAPI()# 1、連接mysql數(shù)據(jù)庫需要導(dǎo)入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ū)動名稱://用戶名:密碼@機(jī)器地址:端口號/數(shù)據(jù)庫名 SQLALCHEMY_DATABASE_URL = "mysql://test:123456@127.0.0.1:3306/test" engine = create_engine(SQLALCHEMY_DATABASE_URL, encoding='utf-8')# 3、把當(dāng)前的引擎綁定給這個會話； # autocommit：是否自動提交 autoflush：是否自動刷新并加載數(shù)據(jù)庫 bind：綁定數(shù)據(jù)庫引擎 Session = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 實(shí)例化 session = Session()# 4、創(chuàng)建Base實(shí)例 # declarative_base類維持了一個從類到表的關(guān)系，通常一個應(yīng)用使用一個Base實(shí)例，所有實(shí)體類都應(yīng)該繼承此類對象 from sqlalchemy.ext.declarative import declarative_base Base = declarative_base()# 5、創(chuàng)建數(shù)據(jù)庫模型（定義表結(jié)構(gòu):表名稱，字段名稱以及字段類型） from sqlalchemy import Column, String, Integerclass User(Base):# 定義表名__tablename__ = 'user'# 定義字段# primary_key=True 設(shè)置為主鍵userid = Column(Integer, primary_key=True)username = Column(String(255))# 構(gòu)造函數(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)

• 運(yùn)行

$ uvicorn main:app --reload

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：要求結(jié)果集中只有一個結(jié)果；如果數(shù)據(jù)庫返回0或2個或更多結(jié)果，并且將引發(fā)異常，則為錯誤。# first：返回可能更大的結(jié)果集中的第一個，如果沒有結(jié)果，則返回None。不會引發(fā)異常。# filter_by與filter的區(qū)別：# filter_by接收的參數(shù)形式是關(guān)鍵字參數(shù)，而filter接收的參數(shù)是更加靈活的SQL表達(dá)式結(jié)構(gòu)# 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": "查詢無結(jié)果"}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ù)庫錯誤"}

完整代碼

# 導(dǎo)入FastAPI模塊 from fastapi import FastAPI app = FastAPI()# 1、連接mysql數(shù)據(jù)庫需要導(dǎo)入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ū)動名稱://用戶名:密碼@機(jī)器地址:端口號/數(shù)據(jù)庫名 SQLALCHEMY_DATABASE_URL = "mysql://test:123456@127.0.0.1:3306/test" engine = create_engine(SQLALCHEMY_DATABASE_URL, encoding='utf-8')# 3、把當(dāng)前的引擎綁定給這個會話； # autocommit：是否自動提交 autoflush：是否自動刷新并加載數(shù)據(jù)庫 bind：綁定數(shù)據(jù)庫引擎 Session = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 實(shí)例化 session = Session()# 4、創(chuàng)建Base實(shí)例 # declarative_base類維持了一個從類到表的關(guān)系，通常一個應(yīng)用使用一個Base實(shí)例，所有實(shí)體類都應(yīng)該繼承此類對象 from sqlalchemy.ext.declarative import declarative_base Base = declarative_base()# 5、創(chuàng)建數(shù)據(jù)庫模型（定義表結(jié)構(gòu):表名稱，字段名稱以及字段類型） from sqlalchemy import Column, String, Integerclass User(Base):# 定義表名__tablename__ = 'user'# 定義字段# primary_key=True 設(shè)置為主鍵userid = Column(Integer, primary_key=True)username = Column(String(255))# 構(gòu)造函數(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：要求結(jié)果集中只有一個結(jié)果；如果數(shù)據(jù)庫返回0或2個或更多結(jié)果，并且將引發(fā)異常，則為錯誤。# first：返回可能更大的結(jié)果集中的第一個，如果沒有結(jié)果，則返回None。不會引發(fā)異常。# filter_by與filter的區(qū)別：# filter_by接收的參數(shù)形式是關(guān)鍵字參數(shù)，而filter接收的參數(shù)是更加靈活的SQL表達(dá)式結(jié)構(gòu)# 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": "查詢無結(jié)果"}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ù)庫錯誤"}

運(yùn)行測試

uvicorn main:app --reload

總結(jié)

以上是生活随笔為你收集整理的Python FastAPI 框架 操作Mysql数据库 增删改查的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。