SQL (관계형) 데이터베이스¶
🌐 AI와 사람이 함께한 번역
이 번역은 사람의 안내를 받아 AI가 만들었습니다. 🤝
원문의 의미를 오해하거나 부자연스러워 보이는 등 오류가 있을 수 있습니다. 🤖
FastAPI에서 SQL(관계형) 데이터베이스 사용은 필수가 아닙니다. 하지만 여러분이 원하는어떤 데이터베이스든 사용할 수 있습니다.
여기서는SQLModel을 사용하는 예제를 살펴보겠습니다.
SQLModel은SQLAlchemy와 Pydantic을 기반으로 구축되었습니다.SQL 데이터베이스를 사용해야 하는 FastAPI 애플리케이션에 완벽히 어울리도록FastAPI와 같은 제작자가 만든 도구입니다.
팁
다른 SQL 또는 NoSQL 데이터베이스 라이브러리를 사용할 수도 있습니다 (일부는"ORMs"이라고도 불립니다), FastAPI는 특정 라이브러리의 사용을 강요하지 않습니다. 😎
SQLModel은 SQLAlchemy를 기반으로 하므로, SQLAlchemy에서지원하는 모든 데이터베이스를 손쉽게 사용할 수 있습니다(이것들은 SQLModel에서도 지원됩니다). 예를 들면:
- PostgreSQL
- MySQL
- SQLite
- Oracle
- Microsoft SQL Server 등.
이 예제에서는SQLite를 사용합니다. SQLite는 단일 파일을 사용하고 Python에서 통합 지원하기 때문입니다. 따라서 이 예제를 그대로 복사하여 실행할 수 있습니다.
나중에 프로덕션 애플리케이션에서는PostgreSQL과 같은 데이터베이스 서버를 사용하는 것이 좋습니다.
팁
프론트엔드와 더 많은 도구를 포함하여FastAPI와PostgreSQL을 포함한 공식 프로젝트 생성기가 있습니다:https://github.com/fastapi/full-stack-fastapi-template
이 튜토리얼은 매우 간단하고 짧습니다. 데이터베이스 기본 개념, SQL, 또는 더 고급 기능에 대해 배우고 싶다면,SQLModel 문서를 참고하세요.
SQLModel 설치하기¶
먼저,가상 환경을 생성하고 활성화한 다음,sqlmodel을 설치하세요:
$pipinstallsqlmodel---> 100%단일 모델로 애플리케이션 생성하기¶
우선 단일SQLModel 모델을 사용하여 애플리케이션의 가장 간단한 첫 번째 버전을 생성해보겠습니다.
이후 아래에서여러 모델로 보안과 유연성을 강화하며 개선하겠습니다. 🤓
모델 생성하기¶
SQLModel을 가져오고 데이터베이스 모델을 생성합니다:
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_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()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}Hero 클래스는 Pydantic 모델과 매우 유사합니다 (실제로 내부적으로Pydantic 모델이기도 합니다).
몇 가지 차이점이 있습니다:
table=True는 SQLModel에 이 모델이테이블 모델이며, SQL 데이터베이스의테이블을 나타내야 한다는 것을 알려줍니다. (다른 일반적인 Pydantic 클래스처럼) 단순한데이터 모델이 아닙니다.Field(primary_key=True)는 SQLModel에id가 SQL 데이터베이스의기본 키임을 알려줍니다 (SQL 기본 키에 대한 자세한 내용은 SQLModel 문서를 참고하세요).참고: 기본 키 필드에
int | None을 사용하는 이유는, Python 코드에서id없이 객체를 생성할 수 있게 하기 위해서입니다(id=None). 데이터베이스가저장할 때 생성해 줄 것이라고 가정합니다. SQLModel은 데이터베이스가id를 제공한다는 것을 이해하고, 데이터베이스 스키마에서해당 열을 null이 아닌INTEGER로 정의합니다. 자세한 내용은기본 키에 대한 SQLModel 문서를 참고하세요.Field(index=True)는 SQLModel에 해당 열에 대해SQL 인덱스를 생성하도록 지시합니다. 이를 통해 데이터베이스에서 이 열로 필터링된 데이터를 읽을 때 더 빠르게 조회할 수 있습니다.SQLModel은
str으로 선언된 항목이 SQL 데이터베이스에서TEXT(또는 데이터베이스에 따라VARCHAR) 유형의 열로 저장된다는 것을 인식합니다.
엔진 생성하기¶
SQLModel의engine (내부적으로는 SQLAlchemyengine)은 데이터베이스에 대한연결을 유지하는 역할을 합니다.
코드 전체에서 동일한 데이터베이스에 연결하기 위해하나의 단일engine 객체를 사용합니다.
# Code above omitted 👆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)# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_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()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}check_same_thread=False를 사용하면 FastAPI에서 여러 스레드에서 동일한 SQLite 데이터베이스를 사용할 수 있습니다. 이는하나의 단일 요청이둘 이상의 스레드를 사용할 수 있기 때문에 필요합니다(예: 의존성에서 사용되는 경우).
걱정하지 마세요. 코드가 구조화된 방식으로 인해, 이후에각 요청마다 단일 SQLModel세션을 사용하도록 보장할 것입니다. 실제로 이것이check_same_thread가 하려는 것입니다.
테이블 생성하기¶
그 다음SQLModel.metadata.create_all(engine)을 사용하여 모든테이블 모델의테이블을 생성하는 함수를 추가합니다.
# Code above omitted 👆defcreate_db_and_tables():SQLModel.metadata.create_all(engine)# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_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()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}세션 의존성 생성하기¶
Session은메모리에 객체를 저장하고 데이터에 필요한 모든 변경 사항을 추적한 후,engine을 통해 데이터베이스와 통신합니다.
yield를 사용해 FastAPI의의존성을 생성하여 각 요청마다 새로운Session을 제공합니다. 이는 요청당 하나의 세션만 사용되도록 보장합니다. 🤓
그런 다음 이 의존성을 사용하는 나머지 코드를 간소화하기 위해Annotated 의존성SessionDep을 생성합니다.
# Code above omitted 👆defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_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()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}시작 시 데이터베이스 테이블 생성하기¶
애플리케이션 시작 시 데이터베이스 테이블을 생성합니다.
# Code above omitted 👆app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_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()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}여기서는 애플리케이션 시작 이벤트 시 테이블을 생성합니다.
프로덕션 환경에서는 애플리케이션을 시작하기 전에 실행되는 마이그레이션 스크립트를 사용할 가능성이 높습니다. 🤓
팁
SQLModel은 Alembic을 감싸는 마이그레이션 유틸리티를 제공할 예정입니다. 하지만 현재Alembic을 직접 사용할 수 있습니다.
Hero 생성하기¶
각 SQLModel 모델은 Pydantic 모델이기도 하므로, Pydantic 모델을 사용할 수 있는 동일한타입 어노테이션에서 사용할 수 있습니다.
예를 들어, 파라미터를Hero 타입으로 선언하면JSON 본문에서 값을 읽어옵니다.
마찬가지로, 함수의반환 타입으로 선언하면 해당 데이터의 구조가 자동으로 생성되는 API 문서의 UI에 나타납니다.
# Code above omitted 👆@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_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()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}여기서는SessionDep 의존성(Session)을 사용하여 새로운Hero를Session 인스턴스에 추가하고, 데이터베이스에 변경 사항을 커밋하고,hero 데이터의 최신 상태를 갱신한 다음 이를 반환합니다.
Heroes 조회하기¶
select()를 사용하여 데이터베이스에서Hero를조회할 수 있습니다. 결과에 페이지네이션을 적용하기 위해limit와offset을 포함할 수 있습니다.
# Code above omitted 👆@app.get("/heroes/")defread_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()returnheroes# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_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()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}단일 Hero 조회하기¶
단일Hero를조회할 수도 있습니다.
# Code above omitted 👆@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_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()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}Hero 삭제하기¶
Hero를삭제하는 것도 가능합니다.
# Code above omitted 👆@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:SessionDep)->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_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()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:SessionDep)->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHero(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/")defcreate_hero(hero:Hero,session:Session=Depends(get_session))->Hero:session.add(hero)session.commit()session.refresh(hero)returnhero@app.get("/heroes/")defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),)->list[Hero]:heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}")defread_hero(hero_id:int,session:Session=Depends(get_session))->Hero:hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}애플리케이션 실행하기¶
애플리케이션을 실행할 수 있습니다:
$fastapidevmain.py<span style="color: green;">INFO</span>: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)그런 다음/docs UI로 이동하면,FastAPI가 이모델들을 사용해 API를문서화하고, 데이터를직렬화하고검증하는 데에도 사용하는 것을 확인할 수 있습니다.
여러 모델로 애플리케이션 업데이트¶
이제 이 애플리케이션을 약간리팩터링하여보안과유연성을 개선해 보겠습니다.
이전 애플리케이션을 확인해 보면, 지금까지는 UI에서 클라이언트가 생성할Hero의id를 결정할 수 있게 되어 있는 것을 볼 수 있습니다. 😱
이렇게 해서는 안 됩니다. 클라이언트가 DB에 이미 할당되어 있는id를 덮어쓸 수 있기 때문입니다.id를 결정하는 것은백엔드 또는데이터베이스가 해야 하며,클라이언트가 해서는 안 됩니다.
또한 hero에 대한secret_name을 생성하지만, 지금까지는 이 값을 어디에서나 반환하고 있습니다. 이는 그다지비밀스럽지 않습니다... 😅
이러한 문제는 몇 가지추가 모델을 추가해 해결하겠습니다. 바로 여기서 SQLModel이 빛을 발하게 됩니다. ✨
여러 모델 생성하기¶
SQLModel에서table=True가 설정된 모델 클래스는테이블 모델입니다.
그리고table=True가 없는 모델 클래스는데이터 모델인데, 이것들은 실제로는 (몇 가지 작은 추가 기능이 있는) Pydantic 모델일 뿐입니다. 🤓
SQLModel을 사용하면상속을 통해 모든 경우에 필드를중복 선언하지 않아도 됩니다.
HeroBase - 기본 클래스¶
모든 모델에서공유되는 필드를 가진HeroBase 모델을 시작해 봅시다:
nameage
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}Hero -테이블 모델¶
다음으로 실제테이블 모델인Hero를 생성합니다. 이 모델은 다른 모델에는 항상 포함되는 건 아닌추가 필드를 포함합니다:
idsecret_name
Hero는HeroBase를 상속하므로HeroBase에 선언된 필드도또한 포함합니다. 따라서Hero의 모든 필드는 다음과 같습니다:
idnameagesecret_name
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:str# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}HeroPublic - 공개데이터 모델¶
다음으로HeroPublic 모델을 생성합니다. 이 모델은 API 클라이언트에반환되는 모델입니다.
HeroPublic은HeroBase와 동일한 필드를 가지므로,secret_name은 포함하지 않습니다.
마침내 우리의 heroes의 정체가 보호됩니다! 🥷
또한id: int를 다시 선언합니다. 이를 통해, API 클라이언트와계약을 맺어id가 항상 존재하며 항상int 타입이라는 것을 보장합니다(None이 될 수 없습니다).
팁
반환 모델이 값이 항상 존재하고 항상int(None이 아님)를 보장하는 것은 API 클라이언트에게 매우 유용합니다. 이를 통해 API 클라이언트는 이런 확신을 바탕으로 훨씬 더 간단한 코드를 작성할 수 있습니다.
또한자동으로 생성된 클라이언트는 더 단순한 인터페이스를 제공하므로, API와 소통하는 개발자들이 API를 사용하면서 훨씬 더 좋은 경험을 할 수 있습니다. 😎
HeroPublic의 모든 필드는HeroBase와 동일하며,id는int로 선언됩니다(None이 아님):
idnameage
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:int# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}HeroCreate - hero 생성용데이터 모델¶
이제HeroCreate 모델을 생성합니다. 이 모델은 클라이언트로부터 받은 데이터를검증하는 역할을 합니다.
HeroCreate는HeroBase와 동일한 필드를 가지며,secret_name도 포함합니다.
이제 클라이언트가새 hero를 생성할 때secret_name을 보내면, 데이터베이스에 저장되지만, 그 비밀 이름은 API를 통해 클라이언트에게 반환되지 않습니다.
팁
이 방식은비밀번호를 처리하는 방법과 동일합니다. 비밀번호를 받지만, 이를 API에서 반환하지는 않습니다.
또한 비밀번호 값을 저장하기 전에해싱하여 저장하고,평문으로 저장하지 마십시오.
HeroCreate의 필드는 다음과 같습니다:
nameagesecret_name
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:str# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}HeroUpdate - hero 수정용데이터 모델¶
이전 버전의 애플리케이션에서는hero를 수정할 방법이 없었지만, 이제여러 모델로 이를 할 수 있습니다. 🎉
HeroUpdate데이터 모델은 약간 특별한데, 새 hero를 생성할 때 필요한모든 동일한 필드를 가지지만, 모든 필드가선택적(기본값이 있음)입니다. 이렇게 하면 hero를 수정할 때 수정하려는 필드만 보낼 수 있습니다.
모든필드가 실제로 변경되기 때문에(타입이 이제None을 포함하고, 기본값도 이제None이 됨), 우리는 필드를다시 선언해야 합니다.
HeroBase를 상속할 필요는 없습니다. 모든 필드를 다시 선언하기 때문입니다. 일관성을 위해 상속을 유지하긴 했지만, 필수는 아닙니다. 이는 개인적인 취향의 문제입니다. 🤷
HeroUpdate의 필드는 다음과 같습니다:
nameagesecret_name
# Code above omitted 👆classHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(HeroBase):name:str|None=Noneage:int|None=Nonesecret_name:str|None=None# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}HeroCreate로 생성하고HeroPublic 반환하기¶
이제여러 모델을 사용하므로 애플리케이션의 관련 부분을 업데이트할 수 있습니다.
요청에서HeroCreate데이터 모델을 받아 이를 기반으로Hero테이블 모델을 생성합니다.
이 새테이블 모델Hero는 클라이언트에서 보낸 필드를 가지며, 데이터베이스에서 생성된id도 포함합니다.
그런 다음 함수를 통해 동일한테이블 모델Hero를 그대로 반환합니다. 하지만response_model로HeroPublic데이터 모델을 선언했기 때문에,FastAPI는HeroPublic을 사용하여 데이터를 검증하고 직렬화합니다.
# Code above omitted 👆@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}팁
이제반환 타입 어노테이션-> HeroPublic 대신response_model=HeroPublic을 사용합니다. 반환하는 값이 실제로HeroPublic이아니기 때문입니다.
만약-> HeroPublic으로 선언했다면, 에디터와 린터에서HeroPublic 대신Hero를 반환한다고 (당연히) 불평할 것입니다.
response_model에 선언함으로써FastAPI가 처리하도록 하고, 타입 어노테이션과 에디터 및 다른 도구의 도움에는 영향을 미치지 않도록 합니다.
HeroPublic으로 Heroes 조회하기¶
이전과 동일하게Hero를조회할 수 있습니다. 이번에도response_model=list[HeroPublic]을 사용하여 데이터가 올바르게 검증되고 직렬화되도록 보장합니다.
# Code above omitted 👆@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}HeroPublic으로 단일 Hero 조회하기¶
단일 hero를조회할 수도 있습니다:
# Code above omitted 👆@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}HeroUpdate로 Hero 수정하기¶
hero를 수정할 수도 있습니다. 이를 위해 HTTPPATCH 작업을 사용합니다.
그리고 코드에서는 클라이언트가 보낸 모든 데이터가 담긴dict를 가져오는데,클라이언트가 보낸 데이터만 포함하고, 기본값이어서 들어가 있는 값은 제외합니다. 이를 위해exclude_unset=True를 사용합니다. 이것이 주요 핵심입니다. 🪄
그런 다음,hero_db.sqlmodel_update(hero_data)를 사용하여hero_data의 데이터로hero_db를 업데이트합니다.
# Code above omitted 👆@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db# Code below omitted 👇👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}Hero 다시 삭제하기¶
hero삭제는 이전과 거의 동일합니다.
이번에는 모든 것을 리팩터링하고 싶은 욕구를 만족시키지 못하겠습니다. 😅
# Code above omitted 👆@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}👀 Full file preview
fromtypingimportAnnotatedfromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionSessionDep=Annotated[Session,Depends(get_session)]app=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:SessionDep):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:SessionDep,offset:int=0,limit:Annotated[int,Query(le=100)]=100,):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:SessionDep):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:SessionDep):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}🤓 Other versions and variants
Tip
Prefer to use theAnnotated version if possible.
fromfastapiimportDepends,FastAPI,HTTPException,QueryfromsqlmodelimportField,Session,SQLModel,create_engine,selectclassHeroBase(SQLModel):name:str=Field(index=True)age:int|None=Field(default=None,index=True)classHero(HeroBase,table=True):id:int|None=Field(default=None,primary_key=True)secret_name:strclassHeroPublic(HeroBase):id:intclassHeroCreate(HeroBase):secret_name:strclassHeroUpdate(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)defcreate_db_and_tables():SQLModel.metadata.create_all(engine)defget_session():withSession(engine)assession:yieldsessionapp=FastAPI()@app.on_event("startup")defon_startup():create_db_and_tables()@app.post("/heroes/",response_model=HeroPublic)defcreate_hero(hero:HeroCreate,session:Session=Depends(get_session)):db_hero=Hero.model_validate(hero)session.add(db_hero)session.commit()session.refresh(db_hero)returndb_hero@app.get("/heroes/",response_model=list[HeroPublic])defread_heroes(session:Session=Depends(get_session),offset:int=0,limit:int=Query(default=100,le=100),):heroes=session.exec(select(Hero).offset(offset).limit(limit)).all()returnheroes@app.get("/heroes/{hero_id}",response_model=HeroPublic)defread_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")returnhero@app.patch("/heroes/{hero_id}",response_model=HeroPublic)defupdate_hero(hero_id:int,hero:HeroUpdate,session:Session=Depends(get_session)):hero_db=session.get(Hero,hero_id)ifnothero_db:raiseHTTPException(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)returnhero_db@app.delete("/heroes/{hero_id}")defdelete_hero(hero_id:int,session:Session=Depends(get_session)):hero=session.get(Hero,hero_id)ifnothero:raiseHTTPException(status_code=404,detail="Hero not found")session.delete(hero)session.commit()return{"ok":True}애플리케이션 다시 실행하기¶
애플리케이션을 다시 실행할 수 있습니다:
$fastapidevmain.py<span style="color: green;">INFO</span>: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)/docs API UI로 이동하면 이제 업데이트되어 있고, hero를 생성할 때 클라이언트가id를 보낼 것이라고 기대하지 않는 것 등을 확인할 수 있습니다.
요약¶
SQLModel을 사용하여 SQL 데이터베이스와 상호작용하고,데이터 모델 및테이블 모델로 코드를 간소화할 수 있습니다.
더 많은 내용을 배우려면SQLModel 문서를 참고하세요.FastAPI와 함께 SQLModel을 사용하는 더 긴 미니튜토리얼도 있습니다. 🚀