網(wǎng)站建設(shè)費(fèi)用如何做賬務(wù)處理baidu com百度一下
官網(wǎng)文檔:https://fastapi.tiangolo.com/zh/tutorial/sql-databases/
SQL (關(guān)系型) 數(shù)據(jù)庫(kù)?
FastAPI不需要你使用SQL(關(guān)系型)數(shù)據(jù)庫(kù)。
但是您可以使用任何您想要的關(guān)系型數(shù)據(jù)庫(kù)。
這里我們將看到一個(gè)使用SQLModel的示例。
SQLModel是在SQLAlchemy和Pydantic的基礎(chǔ)上構(gòu)建的。它是由FastAPI的同一作者制作的,與需要使用SQL數(shù)據(jù)庫(kù)的FastAPI應(yīng)用程序完美匹配。
小貼士
你可以使用任何其他你想要的SQL或NoSQL數(shù)據(jù)庫(kù)庫(kù)(在某些情況下稱為“ORM”),FastAPI不會(huì)強(qiáng)迫你使用任何東西。
由于SQLModel基于SQLAlchemy,您可以輕松使用SQLAlchemi支持的任何數(shù)據(jù)庫(kù)(這使得它們也受SQLModel支持)您可以很容易地將其調(diào)整為任何SQLAlchemy支持的數(shù)據(jù)庫(kù),如:
- PostgreSQL
- MySQL
- SQLite
- Oracle
- Microsoft SQL Server,等等其它數(shù)據(jù)庫(kù)
在此示例中,我們將使用SQLite,因?yàn)樗褂脝蝹€(gè)文件并且 在Python中具有集成支持。因此,您可以復(fù)制此示例并按原樣來(lái)運(yùn)行它。
稍后,對(duì)于您的產(chǎn)品級(jí)別的應(yīng)用程序,您可能會(huì)要使用像PostgreSQL這樣的數(shù)據(jù)庫(kù)服務(wù)器。
Tip
這兒有一個(gè)FastAPI和PostgreSQL的官方項(xiàng)目生成器,全部基于Docker,包括前端和更多工具:https://github.com/tiangolo/full-stack-fastapi-postgresql
這是一個(gè)非常簡(jiǎn)單而簡(jiǎn)短的教程,如果你想了解數(shù)據(jù)庫(kù)、SQL或更高級(jí)的功能,請(qǐng)參閱SQLModel文檔。
安裝SQLModel
首先,確保創(chuàng)建虛擬環(huán)境,激活它,然后安裝sqlmodel:
pip install sqlmodel
Successfully installed SQLAlchemy-2.0.36 sqlmodel-0.0.22
使用單個(gè)模型創(chuàng)建應(yīng)用程序
我們將首先使用單個(gè)SQLModel模型創(chuàng)建該應(yīng)用程序最簡(jiǎn)單的第一個(gè)版本。
稍后,我們將通過(guò)以下多種型號(hào)來(lái)提高它的安全性和多功能性。🤓
from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass Hero(SQLModel, table=True):id: int | None = Field(default=None, primary_key=True)name: str = Field(index=True)age: int | None = Field(default=None, index=True)secret_name: strsqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)def create_db_and_tables():SQLModel.metadata.create_all(engine)def get_session():with Session(engine) as session:yield sessionSessionDep = Annotated[Session, Depends(get_session)]app = FastAPI()@app.on_event("startup")
def on_startup():create_db_and_tables()@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:session.add(hero)session.commit()session.refresh(hero)return hero@app.get("/heroes/")
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}
創(chuàng)建模型
導(dǎo)入SQLModel并創(chuàng)建數(shù)據(jù)庫(kù)模型:
from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass Hero(SQLModel, table=True):id: int | None = Field(default=None, primary_key=True)name: str = Field(index=True)age: int | None = Field(default=None, index=True)secret_name: str
🤓 其他版本和變體
Hero類與Pydantic模型非常相似(事實(shí)上,在下面,它實(shí)際上是一個(gè)Pydantic模式)。
存在一些差異:
- table=True告訴SQLModel這是一個(gè)表模型,它應(yīng)該表示SQL數(shù)據(jù)庫(kù)中的一個(gè)表,它不僅僅是一個(gè)數(shù)據(jù)模型(就像任何其他常規(guī)Pydantic類一樣)。
- 字段(primary_key=True)告訴SQLModel id是SQL數(shù)據(jù)庫(kù)中的主鍵(您可以在SQLModel文檔中了解有關(guān)SQL主鍵的更多信息)。
- 通過(guò)將類型設(shè)置為int|None,SQLModel將知道該列在SQL數(shù)據(jù)庫(kù)中應(yīng)該是INTEGER,并且應(yīng)該是NULLABLE。
- 字段(index=True)告訴SQLModel,它應(yīng)該為該列創(chuàng)建SQL索引,這樣在讀取由該列篩選的數(shù)據(jù)時(shí)可以更快地在數(shù)據(jù)庫(kù)中查找。
- SQLModel將知道聲明為str的內(nèi)容將是TEXT類型的SQL列(或VARCHAR,具體取決于數(shù)據(jù)庫(kù))。
創(chuàng)建引擎
SQLModel引擎(其下實(shí)際上是SQLAlchemy引擎)負(fù)責(zé)保存與數(shù)據(jù)庫(kù)的連接。
您將有一個(gè)單一的引擎對(duì)象,用于所有代碼連接到同一個(gè)數(shù)據(jù)庫(kù)。
sqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)
使用check_same_thread=False允許FastAPI在不同線程中使用相同的SQLite數(shù)據(jù)庫(kù)。這是必要的,因?yàn)橐粋€(gè)請(qǐng)求可能會(huì)使用多個(gè)線程(例如在依賴關(guān)系中)。
別擔(dān)心,根據(jù)代碼的結(jié)構(gòu)方式,我們將確保稍后每個(gè)請(qǐng)求使用一個(gè)SQLModel會(huì)話,這實(shí)際上是check_same_thread試圖實(shí)現(xiàn)的。
創(chuàng)建表格
然后,我們添加一個(gè)函數(shù),該函數(shù)使用SQLModel.media.create_all(engine)為所有表模型創(chuàng)建表。
def create_db_and_tables():SQLModel.metadata.create_all(engine)
創(chuàng)建會(huì)話依賴關(guān)系
會(huì)話是將對(duì)象存儲(chǔ)在內(nèi)存中并跟蹤數(shù)據(jù)中所需的任何更改,然后使用引擎與數(shù)據(jù)庫(kù)通信。
我們將使用yield創(chuàng)建一個(gè)FastAPI依賴關(guān)系,為每個(gè)請(qǐng)求提供一個(gè)新的Session。這就是確保我們每個(gè)請(qǐng)求使用單個(gè)會(huì)話的原因。🤓
然后,我們創(chuàng)建一個(gè)帶注釋的依賴項(xiàng)SessionDep,以簡(jiǎn)化將使用此依賴項(xiàng)的其余代碼。
def get_session():with Session(engine) as session:yield sessionSessionDep = Annotated[Session, Depends(get_session)]
啟動(dòng)時(shí)創(chuàng)建數(shù)據(jù)庫(kù)表
我們將在應(yīng)用程序啟動(dòng)時(shí)創(chuàng)建數(shù)據(jù)庫(kù)表。
app = FastAPI()@app.on_event("startup")
def on_startup():create_db_and_tables()
在這里,我們?cè)趹?yīng)用程序啟動(dòng)事件上創(chuàng)建表。
對(duì)于生產(chǎn)環(huán)境,您可能會(huì)使用在啟動(dòng)應(yīng)用程序之前運(yùn)行的遷移腳本。🤓
小貼士
SQLModel將有封裝Alembic的遷移實(shí)用程序,但現(xiàn)在,您可以直接使用Alembic。
創(chuàng)建英雄庫(kù)
因?yàn)槊總€(gè)SQLModel模型也是一個(gè)Pydantic模型,所以您可以在使用Pydantics模型的相同類型注釋中使用它。
例如,如果你聲明一個(gè)Hero類型的參數(shù),它將從JSON正文中讀取。
同樣,您可以將其聲明為函數(shù)的返回類型,然后數(shù)據(jù)的形狀將顯示在自動(dòng)API文檔UI中。
@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep) -> Hero:session.add(hero)session.commit()session.refresh(hero)return hero
在這里,我們使用SessionDep依賴項(xiàng)(Session)將新的Hero添加到Session實(shí)例中,將更改提交到數(shù)據(jù)庫(kù)中,刷新Hero中的數(shù)據(jù),然后返回它。
讀英雄庫(kù)
我們可以使用select()從數(shù)據(jù)庫(kù)中讀取Heros。我們可以包含一個(gè)限制和偏移量來(lái)對(duì)結(jié)果進(jìn)行分頁(yè)。
@app.get("/heroes/")
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
) -> list[Hero]:heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes
讀一個(gè)英雄條目
我們可以讀一個(gè)英雄。
@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep) -> Hero:hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero
如果不是英雄:
引發(fā)HTTPException(狀態(tài)碼=404,詳細(xì)信息=“未找到英雄”)
刪除英雄
我們也可以刪除英雄。
@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}
運(yùn)行應(yīng)用程序
您可以運(yùn)行該應(yīng)用程序:
fastapi-dev-main.py
然后轉(zhuǎn)到/docs UI,您將看到FastAPI正在使用這些模型來(lái)記錄API,它也將使用它們來(lái)序列化和驗(yàn)證數(shù)據(jù)。
使用多個(gè)模型更新應(yīng)用程序
現(xiàn)在,讓我們稍微重構(gòu)一下這個(gè)應(yīng)用程序,以提高安全性和多功能性。
如果你查看之前的應(yīng)用程序,在UI中你可以看到,到目前為止,它讓客戶端決定要?jiǎng)?chuàng)建的英雄的id。😱
我們不應(yīng)該讓這種情況發(fā)生,他們可能會(huì)覆蓋我們已經(jīng)在數(shù)據(jù)庫(kù)中分配的id。決定id應(yīng)該由后端或數(shù)據(jù)庫(kù)完成,而不是由客戶端完成。
此外,我們?yōu)橛⑿蹌?chuàng)建了一個(gè)secret_name,但到目前為止,我們到處都在返回它,這不是什么秘密。。。😅
我們將通過(guò)添加一些額外的模型來(lái)解決這些問(wèn)題。SQLModel將在這里大放異彩。?
源代碼:
from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass HeroBase(SQLModel):name: str = Field(index=True)age: int | None = Field(default=None, index=True)class Hero(HeroBase, table=True):id: int | None = Field(default=None, primary_key=True)secret_name: strclass HeroPublic(HeroBase):id: intclass HeroCreate(HeroBase):secret_name: strclass HeroUpdate(HeroBase):name: str | None = Noneage: int | None = Nonesecret_name: str | None = Nonesqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)def create_db_and_tables():SQLModel.metadata.create_all(engine)def get_session():with Session(engine) as session:yield sessionSessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()@app.on_event("startup")
def on_startup():create_db_and_tables()@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):db_hero = Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)return db_hero@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
):heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):hero_db = session.get(Hero, hero_id)if not hero_db:raise HTTPException(status_code=404, detail="Hero not found")hero_data = hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)return hero_db@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}
創(chuàng)建多個(gè)模型
在SQLModel中,任何具有table=True的模型類都是表模型。
任何沒(méi)有table=True的模型類都是數(shù)據(jù)模型,這些模型實(shí)際上只是Pydantic模型(帶有一些小的額外功能)。🤓
使用SQLModel,我們可以使用繼承來(lái)避免在所有情況下復(fù)制所有字段。
HeroBase-基類
讓我們從一個(gè)HeroBase模型開(kāi)始,該模型包含所有模型共享的所有字段:
名稱
年齡
name
age
class HeroBase(SQLModel):name: str = Field(index=True)age: int | None = Field(default=None, index=True)
英雄-桌子模型
然后,讓我們創(chuàng)建Hero,即實(shí)際的表模型,其中包含其他模型中并不總是包含的額外字段:
身份證件
秘密名稱
id
secret_name
因?yàn)镠ero繼承自HeroBase,所以它也有在HeroBase中聲明的字段,所以Hero的所有字段都是:
身份證件
名稱
年齡
秘密名稱
id
name
age
secret_name
class HeroBase(SQLModel):name: str = Field(index=True)age: int | None = Field(default=None, index=True)class Hero(HeroBase, table=True):id: int | None = Field(default=None, primary_key=True)secret_name: str
HeroPublic-公共數(shù)據(jù)模型
接下來(lái),我們創(chuàng)建一個(gè)HeroPublic模型,該模型將返回給API的客戶端。
它具有與HeroBase相同的字段,因此不包括secret_name。
最后,我們英雄的身份得到了保護(hù)!🥷
它還重新聲明id:int。通過(guò)這樣做,我們與API客戶端簽訂了合同,這樣他們就可以總是期望id在那里并且是int(永遠(yuǎn)不會(huì)是None)。
小貼士
讓返回模型確保一個(gè)值總是可用的,并且總是int(而不是None)對(duì)API客戶端非常有用,他們可以編寫更簡(jiǎn)單的具有這種確定性的代碼。
此外,自動(dòng)生成的客戶端將具有更簡(jiǎn)單的接口,因此與您的API通信的開(kāi)發(fā)人員可以更好地使用您的API。😎
HeroPublic中的所有字段都與HeroBase中的相同,id聲明為int(不是None):
身份證件
名稱
年齡
秘密名稱
id
name
age
secret_name
class HeroBase(SQLModel):name: str = Field(index=True)age: int | None = Field(default=None, index=True)class Hero(HeroBase, table=True):id: int | None = Field(default=None, primary_key=True)secret_name: strclass HeroPublic(HeroBase):id: int
HeroCreate-創(chuàng)建英雄的數(shù)據(jù)模型
現(xiàn)在我們創(chuàng)建一個(gè)HeroCreate模型,這個(gè)模型將驗(yàn)證來(lái)自客戶端的數(shù)據(jù)。
它具有與HeroBase相同的字段,還具有secret_name。
現(xiàn)在,當(dāng)客戶端創(chuàng)建一個(gè)新英雄時(shí),他們將發(fā)送secret_name,它將存儲(chǔ)在數(shù)據(jù)庫(kù)中,但這些秘密名稱不會(huì)在API中返回給客戶端。
小貼士
這就是你處理密碼的方式。接收它們,但不要在API中返回它們。
您還可以在存儲(chǔ)密碼之前對(duì)其值進(jìn)行哈希運(yùn)算,切勿以純文本形式存儲(chǔ)。
HeroCreate的字段包括:
名稱
年齡
秘密名稱
name
age
secret_name
class HeroBase(SQLModel):name: str = Field(index=True)age: int | None = Field(default=None, index=True)class Hero(HeroBase, table=True):id: int | None = Field(default=None, primary_key=True)secret_name: strclass HeroPublic(HeroBase):id: intclass HeroCreate(HeroBase):secret_name: str
HeroUpdate-更新英雄的數(shù)據(jù)模型
在之前的應(yīng)用程序版本中,我們沒(méi)有更新英雄的方法,但現(xiàn)在有了多個(gè)模型,我們可以做到。🎉
HeroUpdate數(shù)據(jù)模型有點(diǎn)特殊,它具有創(chuàng)建新英雄所需的所有相同字段,但所有字段都是可選的(它們都有一個(gè)默認(rèn)值)。這樣,當(dāng)你更新英雄時(shí),你可以只發(fā)送你想要更新的字段。
因?yàn)樗凶侄螌?shí)際上都發(fā)生了變化(類型現(xiàn)在包括None,它們現(xiàn)在的默認(rèn)值為None),我們需要重新聲明它們。
我們真的不需要從HeroBase繼承,因?yàn)槲覀冋谥匦侣暶魉凶侄?。為了保持一致?#xff0c;我會(huì)讓它繼承,但這不是必需的。這更多的是個(gè)人品味的問(wèn)題。🤷
HeroUpdate的字段包括:
名稱
年齡
秘密名稱
name
age
secret_name
class HeroUpdate(HeroBase):name: str | None = Noneage: int | None = Nonesecret_name: str | None = None
使用HeroCreate創(chuàng)建并返回一個(gè)HeroPublic
現(xiàn)在我們有了多個(gè)模型,我們可以更新應(yīng)用程序中使用它們的部分。
我們?cè)谡?qǐng)求中接收HeroCreate數(shù)據(jù)模型,并從中創(chuàng)建Hero表模型。
這個(gè)新的表模型Hero將具有客戶端發(fā)送的字段,并且還將具有數(shù)據(jù)庫(kù)生成的id。
然后,我們返回與函數(shù)中相同的表模型Hero。但是,當(dāng)我們使用HeroPublic數(shù)據(jù)模型聲明response_model時(shí),FastAPI將使用HeroPublic來(lái)驗(yàn)證和序列化數(shù)據(jù)。
@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):db_hero = Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)return db_hero
小貼士
現(xiàn)在我們使用response_model=HeroPublic而不是返回類型注釋->HeroPublic,因?yàn)槲覀兎祷氐闹祵?shí)際上不是HeroPublic。
如果我們聲明了->HeroPublic,你的編輯和linter會(huì)抱怨(這是理所當(dāng)然的)你返回的是Hero而不是HeroPublic。
通過(guò)在response_model中聲明它,我們告訴FastAPI去做它的事情,而不會(huì)干擾類型注釋以及編輯器和其他工具的幫助。
用HeroPublic閱讀英雄
我們可以像以前一樣讀取Heros,同樣,我們使用response_model=list[HeroPublic]來(lái)確保數(shù)據(jù)被正確驗(yàn)證和序列化。
@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
):heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes
與HeroPublic一起閱讀《一個(gè)英雄》
我們可以讀一個(gè)英雄:
@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero
使用HeroUpdate更新英雄
我們可以更新英雄。為此,我們使用HTTP PATCH操作。
在代碼中,我們得到一個(gè)包含客戶端發(fā)送的所有數(shù)據(jù)的字典,只有客戶端發(fā)送的數(shù)據(jù),不包括任何僅作為默認(rèn)值的值。為此,我們使用exclude_unset=True。這是主要的伎倆。🪄
然后,我們使用hero_db.sqlmodel_update(hero_data)用hero_da中的數(shù)據(jù)更新hero_db。
@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):hero_db = session.get(Hero, hero_id)if not hero_db:raise HTTPException(status_code=404, detail="Hero not found")hero_data = hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)return hero_db
再次刪除英雄
刪除英雄幾乎是一樣的。
我們不會(huì)滿足在這個(gè)項(xiàng)目中重構(gòu)所有內(nèi)容的愿望。😅
@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}
再次運(yùn)行應(yīng)用程序
您可以再次運(yùn)行該應(yīng)用程序:
fastapi dev main.py
輸出信息:Uvicorn正在運(yùn)行http://127.0.0.1:8000(按CTRL+C退出)
如果你轉(zhuǎn)到/docs API UI,你會(huì)看到它現(xiàn)在已經(jīng)更新,并且它不會(huì)期望在創(chuàng)建英雄時(shí)從客戶端接收id,等等。

回顧
您可以使用SQLModel與SQL數(shù)據(jù)庫(kù)交互,并使用數(shù)據(jù)模型和表模型簡(jiǎn)化代碼。
您可以在SQLModel文檔中了解更多信息,其中有一個(gè)關(guān)于使用SQLModel和FastAPI的較長(zhǎng)迷你教程。🚀
實(shí)踐
?安裝SQLModel
首先,確保創(chuàng)建虛擬環(huán)境,激活它,然后安裝sqlmodel:
pip install sqlmodel
源代碼
存儲(chǔ)文件到sql.py
from typing import Annotatedfrom fastapi import Depends, FastAPI, HTTPException, Query
from sqlmodel import Field, Session, SQLModel, create_engine, selectclass HeroBase(SQLModel):name: str = Field(index=True)age: int | None = Field(default=None, index=True)class Hero(HeroBase, table=True):id: int | None = Field(default=None, primary_key=True)secret_name: strclass HeroPublic(HeroBase):id: intclass HeroCreate(HeroBase):secret_name: strclass HeroUpdate(HeroBase):name: str | None = Noneage: int | None = Nonesecret_name: str | None = Nonesqlite_file_name = "database.db"
sqlite_url = f"sqlite:///{sqlite_file_name}"connect_args = {"check_same_thread": False}
engine = create_engine(sqlite_url, connect_args=connect_args)def create_db_and_tables():SQLModel.metadata.create_all(engine)def get_session():with Session(engine) as session:yield sessionSessionDep = Annotated[Session, Depends(get_session)]
app = FastAPI()@app.on_event("startup")
def on_startup():create_db_and_tables()@app.post("/heroes/", response_model=HeroPublic)
def create_hero(hero: HeroCreate, session: SessionDep):db_hero = Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)return db_hero@app.get("/heroes/", response_model=list[HeroPublic])
def read_heroes(session: SessionDep,offset: int = 0,limit: Annotated[int, Query(le=100)] = 100,
):heroes = session.exec(select(Hero).offset(offset).limit(limit)).all()return heroes@app.get("/heroes/{hero_id}", response_model=HeroPublic)
def read_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")return hero@app.patch("/heroes/{hero_id}", response_model=HeroPublic)
def update_hero(hero_id: int, hero: HeroUpdate, session: SessionDep):hero_db = session.get(Hero, hero_id)if not hero_db:raise HTTPException(status_code=404, detail="Hero not found")hero_data = hero.model_dump(exclude_unset=True)hero_db.sqlmodel_update(hero_data)session.add(hero_db)session.commit()session.refresh(hero_db)return hero_db@app.delete("/heroes/{hero_id}")
def delete_hero(hero_id: int, session: SessionDep):hero = session.get(Hero, hero_id)if not hero:raise HTTPException(status_code=404, detail="Hero not found")session.delete(hero)session.commit()return {"ok": True}
啟動(dòng)服務(wù)
執(zhí)行命令:
fastapi dev sql.py
執(zhí)行后顯示:
INFO Importing from /Users/skywalk/work/fastapi ╭─ Python module file ─╮ │ │ │ 🐍 sql.py │ │ │ ╰──────────────────────╯ INFO Importing module sql
INFO Found importable FastAPI app ╭─ Importable FastAPI app ─╮ │ │ │ from sql import app │ │ │ ╰──────────────────────────╯ INFO Using import string sql:app ╭────────── FastAPI CLI - Development mode ───────────╮ │ │ │ Serving at: http://127.0.0.1:8000 │ │ │ │ API docs: http://127.0.0.1:8000/docs │ │ │ │ Running in development mode, for production use: │ │ │ │ fastapi run │ │ │ ╰─────────────────────────────────────────────────────╯ INFO: Will watch for changes in these directories: ['/Users/xxx/work/fastapi']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [37935] using WatchFiles
INFO: Started server process [37941]
INFO: Waiting for application startup.
INFO: Application startup complete.
測(cè)試
瀏覽docs頁(yè)面:
執(zhí)行curl添加指令
curl -X 'POST' \'http://127.0.0.1:8000/heroes/' \-H 'accept: application/json' \-H 'Content-Type: application/json' \-d '{"name": "string","age": 0,"secret_name": "string"
}'
?輸出:
{"name":"string","age":0,"id":2}
證明一條信息被添加
查看一下:
curl http://127.0.0.1:8000/heroes/
[{"name":"string","age":0,"id":1},{"name":"string","age":0,"id":2}]
果然多了一條信息。