FASTAPI Archives - icodebroker

[PYTHON/FASTAPI] FastAPI 클래스 : StreamingResponse 객체를 사용해 응답 스트리밍 처리하기

■ FastAPI 클래스에서 StreamingResponse 객체를 사용해 응답 스트리밍을 처리하는 방법을 보여준다. ▶ main.py


import json

from pydantic          import BaseModel
from fastapi           import APIRouter
from fastapi           import Path
from fastapi.responses import StreamingResponse
from fastapi           import FastAPI

class TodoItem(BaseModel):
    id   : int
    item : str

    class Config:
        json_schema_extra = {
            "example" : {
                "id"   : 1,
                "item" : "Example Schema!"
            }
        }

class TodoItemUpdate(BaseModel):
    item : str

    class Config:
        json_schema_extra = {
            "example" : {
            "item" : "Read the next chapter of the book"
        }
    }

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")
async def addTodoItem(todoItem : TodoItem) -> dict:
    todoItemList.append(todoItem)
    return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")
async def getTodoItemList() -> dict:
    return {"todoItemList" : todoItemList}

async def enumerateTodoItemListInternal():
    for todoItem in todoItemList:
        yield json.dumps(todoItem.dict()) + "\n"

@apiRouter.get("/todos")
async def enumerateTodoItemList():
    return StreamingResponse(enumerateTodoItemListInternal(), media_type = "application/json")

@apiRouter.get("/todo/{todoItemID}")
async def getTodoItem(todoItemID : int = Path(..., title = "The ID of the todo item to retrieve.")) -> dict:
    for todoItem in todoItemList:
        if todoItem.id == todoItemID:
            return {"todoItem" : todoItem}
    return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.put("/todo/{todoItemID}")
async def updateTodo(todoItemUpdate: TodoItemUpdate, todoItemID : int = Path(..., title = "The ID of the todo item to be updated.")) -> dict:
    for todoItem in todoItemList:
        if todoItem.id == todoItemID:
            todoItem.item = todoItemUpdate.item
            return {"message" : "Todo item updated successfully."}
    return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.delete("/todo/{todoItemID}")
async def deleteTodo(todoItemID : int) -> dict:
  for index in range(len(todoItemList)):
      todoItem = todoItemList[index]
      if todoItem.id == todoItemID:
          todoItemList.pop(index)
          return {"message" : "Todo item deleted successfully."}
  return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.delete("/todo")
async def deleteAllTodoItems() -> dict:
  todoItemList.clear()
  return {"message" : "All Todo items deleted successfully."}

fastAPI = FastAPI()

@fastAPI.get("/")
def root():
    return {"message" : "Hello World"}

@fastAPI.get("/home")
def home():
    return {"message" : "home"}

fastAPI.include_router(apiRouter)

import json

from pydantic import BaseModel

from fastapi import APIRouter

from fastapi import Path

from fastapi.responses import StreamingResponse

from fastapi import FastAPI

class TodoItem(BaseModel):

id : int

item : str

class Config:

json_schema_extra = {

"example" : {

"id" : 1,

"item" : "Example Schema!"

}

class TodoItemUpdate(BaseModel):

item : str

class Config:

json_schema_extra = {

"example" : {

"item" : "Read the next chapter of the book"

}

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")

async def addTodoItem(todoItem : TodoItem) -> dict:

todoItemList.append(todoItem)

return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")

async def getTodoItemList() -> dict:

return {"todoItemList" : todoItemList}

async def enumerateTodoItemListInternal():

for todoItem in todoItemList:

yield json.dumps(todoItem.dict()) + "\n"

@apiRouter.get("/todos")

async def enumerateTodoItemList():

return StreamingResponse(enumerateTodoItemListInternal(), media_type = "application/json")

@apiRouter.get("/todo/{todoItemID}")

async def getTodoItem(todoItemID : int = Path(..., title = "The ID of the todo item to retrieve.")) -> dict:

for todoItem in todoItemList:

if todoItem.id == todoItemID:

return {"todoItem" : todoItem}

return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.put("/todo/{todoItemID}")

async def updateTodo(todoItemUpdate: TodoItemUpdate, todoItemID : int = Path(..., title = "The ID of the todo item to be updated.")) -> dict:

for todoItem in todoItemList:

if todoItem.id == todoItemID:

todoItem.item = todoItemUpdate.item

return {"message" : "Todo item updated successfully."}

return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.delete("/todo/{todoItemID}")

async def deleteTodo(todoItemID : int) -> dict:

for index in range(len(todoItemList)):

todoItem = todoItemList[index]

if todoItem.id == todoItemID:

todoItemList.pop(index)

return {"message" : "Todo item deleted successfully."}

return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.delete("/todo")

async def deleteAllTodoItems() -> dict:

todoItemList.clear()

return {"message" : "All Todo items deleted successfully."}

fastAPI = FastAPI()

@fastAPI.get("/")

def root():

return {"message" : "Hello World"}

@fastAPI.get("/home")

def home():

return {"message" : "home"}

fastAPI.include_router(apiRouter)

▶ requirements.txt


annotated-types==0.6.0
anyio==4.3.0
certifi==2024.2.2
click==8.1.7
colorama==0.4.6
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.2
fastapi==0.111.0
fastapi-cli==0.0.3
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
itsdangerous==2.2.0
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.1
pydantic-extra-types==2.7.0
pydantic-settings==2.2.1
pydantic_core==2.18.2
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.11.0
ujson==5.10.0
uvicorn==0.29.0
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0

annotated-types==0.6.0

anyio==4.3.0

certifi==2024.2.2

click==8.1.7

colorama==0.4.6

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.2

fastapi==0.111.0

fastapi-cli==0.0.3

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

itsdangerous==2.2.0

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.1

pydantic-extra-types==2.7.0

pydantic-settings==2.2.1

pydantic_core==2.18.2

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.11.0

ujson==5.10.0

uvicorn==0.29.0

uvloop==0.19.0

watchfiles==0.21.0

websockets==12.0

▶ 실행 명령


uvicorn main:fastAPI --reload

uvicorn main:fastAPI --reload

[PYTHON/FASTAPI] Request 클래스 : 메소드에서 HTTP Request 객체 직접 참조하기

■ Request 클래스를 사용해 메소드에서 HTTP Request 객체를 직접 참조하는 방법을 보여준다. • FastAPI는 내부적으로 Starlette를 사용하고 있다. • 필요하다면 FastAPI 대신

[PYTHON/LANGCHAIN] langserve 모듈 : langchain 모듈 체인을 REST API 서버로 배포하기

■ langserve 모듈을 사용해 langchain 모듈의 체인을 REST API 서버로 배포하는 방법을 보여준다. ▶ server.py


import os

import uvicorn

from langchain_core.prompts        import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai              import ChatOpenAI
from fastapi                       import FastAPI
from langserve                     import add_routes

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

promptTemplate = ChatPromptTemplate.from_messages(
    [
        ("system", "Translate the following into {language} : "),
        ("user"  , "{text}"                                    )
    ]
)

chatOpenAI = ChatOpenAI(model = "gpt-4")

strOutputParser = StrOutputParser()

runnableSequence = promptTemplate | chatOpenAI | strOutputParser

fastAPI = FastAPI(
  title       = "LangChain Server",
  version     = "1.0",
  description = "A simple API server using LangChain's Runnable interfaces",
)

add_routes(
    fastAPI,
    runnableSequence,
    path = "/chain",
)

if __name__ == "__main__":
    uvicorn.run(fastAPI, host = "0.0.0.0", port = 8000)

import os

import uvicorn

from langchain_core.prompts import ChatPromptTemplate

from langchain_core.output_parsers import StrOutputParser

from langchain_openai import ChatOpenAI

from fastapi import FastAPI

from langserve import add_routes

os.environ["OPENAI_API_KEY"] = "<OPENAI_API_KEY>"

promptTemplate = ChatPromptTemplate.from_messages(

[

("system", "Translate the following into {language} : "),

("user" , "{text}" )

]

)

chatOpenAI = ChatOpenAI(model = "gpt-4")

strOutputParser = StrOutputParser()

runnableSequence = promptTemplate | chatOpenAI | strOutputParser

fastAPI = FastAPI(

title = "LangChain Server",

version = "1.0",

description = "A simple API server using LangChain's Runnable interfaces",

)

add_routes(

fastAPI,

runnableSequence,

path = "/chain",

)

if __name__ == "__main__":

uvicorn.run(fastAPI, host = "0.0.0.0", port = 8000)

▶ client.py


from langserve import RemoteRunnable

remoteRunnable = RemoteRunnable("http://localhost:8000/chain/")

resultString = remoteRunnable.invoke({"language" : "korean", "text" : "hi"})

print(resultString)

from langserve import RemoteRunnable

remoteRunnable = RemoteRunnable("http://localhost:8000/chain/")

resultString = remoteRunnable.invoke({"language" : "korean", "text" : "hi"})

print(resultString)

▶ requirements.txt


aiohttp==3.9.5
aiosignal==1.3.1
annotated-types==0.7.0
anyio==4.4.0
async-timeout==4.0.3
attrs==23.2.0
certifi==2024.6.2
charset-normalizer==3.3.2
distro==1.9.0
exceptiongroup==1.2.1
frozenlist==1.4.1
greenlet==3.0.3
h11==0.14.0
httpcore==1.0.5
httpx==0.27.0
idna==3.7
jsonpatch==1.33
jsonpointer==2.4
langchain==0.2.3
langchain-core==0.2.5
langchain-openai==0.1.8
langchain-text-splitters==0.2.1
langsmith==0.1.75
multidict==6.0.5
numpy==1.26.4
openai==1.33.0
orjson==3.10.3
packaging==23.2
pydantic==2.7.3
pydantic_core==2.18.4
PyYAML==6.0.1
regex==2024.5.15
requests==2.32.3
sniffio==1.3.1
SQLAlchemy==2.0.30
tenacity==8.3.0
tiktoken==0.7.0
tqdm==4.66.4
typing_extensions==4.12.2
urllib3==2.2.1
yarl==1.9.4

aiohttp==3.9.5

aiosignal==1.3.1

annotated-types==0.7.0

anyio==4.4.0

async-timeout==4.0.3

attrs==23.2.0

certifi==2024.6.2

charset-normalizer==3.3.2

distro==1.9.0

exceptiongroup==1.2.1

frozenlist==1.4.1

greenlet==3.0.3

h11==0.14.0

httpcore==1.0.5

httpx==0.27.0

idna==3.7

jsonpatch==1.33

jsonpointer==2.4

langchain==0.2.3

langchain-core==0.2.5

langchain-openai==0.1.8

langchain-text-splitters==0.2.1

langsmith==0.1.75

multidict==6.0.5

numpy==1.26.4

openai==1.33.0

orjson==3.10.3

packaging==23.2

pydantic==2.7.3

pydantic_core==2.18.4

PyYAML==6.0.1

regex==2024.5.15

requests==2.32.3

sniffio==1.3.1

SQLAlchemy==2.0.30

tenacity==8.3.0

tiktoken==0.7.0

tqdm==4.66.4

typing_extensions==4.12.2

urllib3==2.2.1

yarl==1.9.4

[PYTHON/FASTAPI] FastAPI 클래스 : add_middleware 메소드를 사용해 HTTP 요청에 대해 HTTPS 리다이렉션 설정하기

■ FastAPI 클래스의 add_middleware 메소드를 사용해 HTTP 요청에 대해 HTTPS 리다이렉션을 설정하는 방법을 보여준다. ▶ main.py


import uvicorn

from fastapi                          import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

fastAPI = FastAPI()

fastAPI.add_middleware(HTTPSRedirectMiddleware)

@fastAPI.get("/")
async def getHelloWorld():
    return {"message" : "Hello, World!"}

if __name__ == "__main__":
    uvicorn.run(
        "main:fastAPI",
        host         = "0.0.0.0",
        port         = 443,
        reload       = True,
        ssl_keyfile  = "./certificate-private-key.pem",
        ssl_certfile = "./certificate.pem"
    )

import uvicorn

from fastapi import FastAPI

from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

fastAPI = FastAPI()

fastAPI.add_middleware(HTTPSRedirectMiddleware)

@fastAPI.get("/")

async def getHelloWorld():

return {"message" : "Hello, World!"}

if __name__ == "__main__":

uvicorn.run(

"main:fastAPI",

host = "0.0.0.0",

port = 443,

reload = True,

ssl_keyfile = "./certificate-private-key.pem",

ssl_certfile = "./certificate.pem"

)

▶ requirement.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.4
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.3
pydantic_core==2.18.4
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.12.1
ujson==5.10.0
uvicorn==0.30.1
uvloop==0.19.0
watchfiles==0.22.0
websockets==12.0

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.4

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.3

pydantic_core==2.18.4

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.12.1

ujson==5.10.0

uvicorn==0.30.1

uvloop==0.19.0

watchfiles==0.22.0

websockets==12.0

[PYTHON/FASTAPI] FastAPI 클래스 : route 메소드를 사용해 HTTP 요청에 대해 HTTPS 리다이렉션 설정하기

■ FastAPI 클래스의 route 메소드를 사용해 HTTP 요청에 대해 HTTPS 리다이렉션을 설정하는 방법을 보여준다. ▶ redirect_https.py


import uvicorn

from fastapi             import FastAPI
from starlette.requests  import Request
from starlette.responses import RedirectResponse

fastAPI = FastAPI()

@fastAPI.route("/{_:path}")
async def redirectHTTPS(request : Request):
    return RedirectResponse(request.url.replace(scheme = "https"))

if __name__ == "__main__":
    uvicorn.run("redirect_https:fastAPI", host = "0.0.0.0", port = 80)

import uvicorn

from fastapi import FastAPI

from starlette.requests import Request

from starlette.responses import RedirectResponse

fastAPI = FastAPI()

@fastAPI.route("/{_:path}")

async def redirectHTTPS(request : Request):

return RedirectResponse(request.url.replace(scheme = "https"))

if __name__ == "__main__":

uvicorn.run("redirect_https:fastAPI", host = "0.0.0.0", port = 80)

▶ main.py


import uvicorn

from subprocess import Popen
from fastapi    import FastAPI

fastAPI = FastAPI()

@fastAPI.get("/")
async def getHelloWorld():
    return {"message" : "Hello, World!"}

if __name__ == "__main__":
    Popen(["python3", "-m", "redirect_https"])

    uvicorn.run(
        "main:fastAPI",
        host         = "0.0.0.0",
        port         = 443,
        reload       = True,
        ssl_keyfile  = "./certificate-private-key.pem",
        ssl_certfile = "./certificate.pem"
    )

import uvicorn

from subprocess import Popen

from fastapi import FastAPI

fastAPI = FastAPI()

@fastAPI.get("/")

async def getHelloWorld():

return {"message" : "Hello, World!"}

if __name__ == "__main__":

Popen(["python3", "-m", "redirect_https"])

uvicorn.run(

"main:fastAPI",

host = "0.0.0.0",

port = 443,

reload = True,

ssl_keyfile = "./certificate-private-key.pem",

ssl_certfile = "./certificate.pem"

)

▶ requirements.py

[PYTHON/UVICORN] run 함수 : ssl_keyfile/ssl_certfile 인자를 사용해 HTTPS 설정하기

■ run 함수의 ssl_keyfile/ssl_certfile 인자를 사용해 HTTPS를 설정하는 방법을 보여준다. ▶ main.py


import uvicorn

from fastapi import FastAPI

fastAPI = FastAPI()

@fastAPI.get("/")
async def getHelloWorld():
    return {"message" : "Hello, World!"}

if __name__ == "__main__":
    uvicorn.run(
        "main:fastAPI",
        host         = "0.0.0.0",
        port         = 443,
        reload       = True,
        ssl_keyfile  = "./certificate-private-key.pem",
        ssl_certfile = "./certificate.pem"
    )

import uvicorn

from fastapi import FastAPI

fastAPI = FastAPI()

@fastAPI.get("/")

async def getHelloWorld():

return {"message" : "Hello, World!"}

if __name__ == "__main__":

uvicorn.run(

"main:fastAPI",

host = "0.0.0.0",

port = 443,

reload = True,

ssl_keyfile = "./certificate-private-key.pem",

ssl_certfile = "./certificate.pem"

)

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.4
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.3
pydantic_core==2.18.4
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.12.1
ujson==5.10.0
uvicorn==0.30.1
uvloop==0.19.0
watchfiles==0.22.0
websockets==12.0

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.4

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.3

pydantic_core==2.18.4

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.12.1

ujson==5.10.0

uvicorn==0.30.1

uvloop==0.19.0

watchfiles==0.22.0

websockets==12.0

[PYTHON/FASTAPI] FastAPI 클래스 : add_middleware 메소드를 사용해 GZIP 압축 전송 설정하기

■ FastAPI 클래스의 add_middleware 메소드를 사용해 GZIP 압축 전송을 설정하는 방법을 보여준다. ▶ main.py


import uvicorn

from fastapi                 import FastAPI
from fastapi.middleware.gzip import GZipMiddleware

fastAPI = FastAPI()

fastAPI.add_middleware(GZipMiddleware, minimum_size = 1000, compresslevel = 9) # compressionlevel 인자의 디폴트 값은 9로, 이를 낮출 경우 압축 효율은 떨어지지만 압축 속도는 빨라지게 된다.

@fastAPI.get("/")
async def getHelloWorld():
    return {"message" : "Hello, World!"}

if __name__ == "__main__":
    uvicorn.run("main:fastAPI", host = "0.0.0.1", port = 8000, reload = True)

import uvicorn

from fastapi import FastAPI

from fastapi.middleware.gzip import GZipMiddleware

fastAPI = FastAPI()

fastAPI.add_middleware(GZipMiddleware, minimum_size = 1000, compresslevel = 9) # compressionlevel 인자의 디폴트 값은 9로, 이를 낮출 경우 압축 효율은 떨어지지만 압축 속도는 빨라지게 된다.

@fastAPI.get("/")

async def getHelloWorld():

return {"message" : "Hello, World!"}

if __name__ == "__main__":

uvicorn.run("main:fastAPI", host = "0.0.0.1", port = 8000, reload = True)

▶ requirements.txt


annotated-types==0.7.0
anyio==4.4.0
certifi==2024.6.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.4
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.3
pydantic_core==2.18.4
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.12.1
ujson==5.10.0
uvicorn==0.30.1
uvloop==0.19.0
watchfiles==0.22.0
websockets==12.0

annotated-types==0.7.0

anyio==4.4.0

certifi==2024.6.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.4

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.3

pydantic_core==2.18.4

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.12.1

ujson==5.10.0

uvicorn==0.30.1

uvloop==0.19.0

watchfiles==0.22.0

websockets==12.0

[DOCKER/UBUNTU] 파이썬 FastAPI 애플리케이션 도커 파일 샘플

■ FastAPI 서버의 도커 이미지를 만드는 도커 파일 샘플을 보여준다. ▶ dockerfile


# 우분투 이미지를 가져온다.
FROM ubuntu:22.04

# 파이썬을 설치한다.
RUN apt-get update -y
RUN apt-get install -y python3 python3-pip python3-venv
RUN apt-get clean

# 작업 디렉토리를 지정한다.
WORKDIR /testproject

# requirements.txt 파일을 복사하고 파이썬 모듈을 설치한다.
COPY requirements.txt .
RUN pip install --no-cache-dir -r /testproject/requirements.txt

# 소스를 복사한다.
COPY . .

# 노출 포트를 지정한다.
EXPOSE 8000

# FastAPI 애플리케이션을 실행한다.
#CMD ["/bin/bash", "-c", "uvicorn main:fastAPI --host 0.0.0.0 --port 8000"]
 CMD ["uvicorn", "main:fastAPI", "--host", "0.0.0.0", "--port", "8000"]

# 우분투 이미지를 가져온다.

FROM ubuntu:22.04

# 파이썬을 설치한다.

RUN apt-get update -y

RUN apt-get install -y python3 python3-pip python3-venv

RUN apt-get clean

# 작업 디렉토리를 지정한다.

WORKDIR /testproject

# requirements.txt 파일을 복사하고 파이썬 모듈을 설치한다.

COPY requirements.txt .

RUN pip install --no-cache-dir -r /testproject/requirements.txt

# 소스를 복사한다.

COPY . .

# 노출 포트를 지정한다.

EXPOSE 8000

# FastAPI 애플리케이션을 실행한다.

#CMD ["/bin/bash", "-c", "uvicorn main:fastAPI --host 0.0.0.0 --port 8000"]

CMD ["uvicorn", "main:fastAPI", "--host", "0.0.0.0", "--port", "8000"]

[DOCKER/UBUNTU] 파이썬 가상 환경 FastAPI 애플리케이션 도커 파일 샘플

■ 파이썬 가상 환경으로 FastAPI 애플리케이션의 도커 이미지를 만드는 도커 파일 샘플을 보여준다. ▶ dockerfile


# 우분투 이미지를 가져온다.
FROM ubuntu:22.04

# 설치시 확인 메세지를 무시한다. (권장안함)
#ENV DEBIAN_FRONTEND=noninteractive

# 가상환경 디렉터리 경로를 설정한다.
ENV VENV_PATH=/testproject/venv

# 파이썬을 설치한다.
RUN apt-get update -y
RUN apt-get install -y python3 python3-pip python3-venv
RUN apt-get clean

# 작업 디렉토리를 지정한다.
WORKDIR /testproject

# 가상환경을 생성한다.
RUN python3 -m venv $VENV_PATH

# requirements.txt 파일을 복사하고 파이썬 모듈을 설치한다.
COPY requirements.txt .
RUN $VENV_PATH/bin/pip install --no-cache-dir -r requirements.txt

# 소스를 복사한다.
COPY . .

# 노출 포트를 지정한다.
EXPOSE 8000

# 가상환경 활성화하고 FastAPI 애플리케이션을 실행한다.
CMD ["/bin/bash", "-c", "source $VENV_PATH/bin/activate && uvicorn main:fastAPI --host 0.0.0.0 --port 8000"]

# 우분투 이미지를 가져온다.

FROM ubuntu:22.04

# 설치시 확인 메세지를 무시한다. (권장안함)

#ENV DEBIAN_FRONTEND=noninteractive

# 가상환경 디렉터리 경로를 설정한다.

ENV VENV_PATH=/testproject/venv

# 파이썬을 설치한다.

RUN apt-get update -y

RUN apt-get install -y python3 python3-pip python3-venv

RUN apt-get clean

# 작업 디렉토리를 지정한다.

WORKDIR /testproject

# 가상환경을 생성한다.

RUN python3 -m venv $VENV_PATH

# requirements.txt 파일을 복사하고 파이썬 모듈을 설치한다.

COPY requirements.txt .

RUN $VENV_PATH/bin/pip install --no-cache-dir -r requirements.txt

# 소스를 복사한다.

COPY . .

# 노출 포트를 지정한다.

EXPOSE 8000

# 가상환경 활성화하고 FastAPI 애플리케이션을 실행한다.

CMD ["/bin/bash", "-c", "source $VENV_PATH/bin/activate && uvicorn main:fastAPI --host 0.0.0.0 --port 8000"]

[PYTHON/FASTAPI] FastAPI 클래스 : JINJA 템플리트 사용하기

■ FastAPI 클래스에서 JINJA 템플리트를 사용하는 방법을 보여준다. ▶ template/home.html


<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8" />
    <meta content="IE=edge" http-equiv="X-UA-Compatible" />
    <meta content="width=device-width, initial-scale=1.0" name="viewport" />
    <title>Packt Todo Application</title>
    <link crossorigin="anonymous"
        href="https://stackpath.bootstrapcdn.com/bootstrap/4.1.0/css/bootstrap.min.css"
        integrity="sha384-9gVQ4dYFwwWSjIDZnLEWnxCjeSWFphJiwGPXr1jddIhOegiu1FwO5qRGvFXOdJZ4"
        rel="stylesheet" />
    <link crossorigin="anonymous"
        href="https://use.fontawesome.com/releases/v5.0.10/css/all.css"
        integrity="sha384-+d0P83n9kaQMCwj8F4RJB66tzIwOKmrdb46+porD/OvrJ+37WqIM7UoBtwHO6Nlg"
        rel="stylesheet" />
</head>
<body>
    <header>
        <nav class="navar">
            <div class="container-fluid">
                <center>
                    <h1>Packt Todo Application</h1>
                </center>
            </div>
        </nav>
    </header>
    <body>
        <div class="container-fluid">
            {% block todo_container %}{% endblock %}
        </div>
    </body>
</html>

<!DOCTYPE html>

<head>

<title>Packt Todo Application</title>

href="https://stackpath.bootstrapcdn.com/bootstrap/4.1.0/css/bootstrap.min.css"

integrity="sha384-9gVQ4dYFwwWSjIDZnLEWnxCjeSWFphJiwGPXr1jddIhOegiu1FwO5qRGvFXOdJZ4"

rel="stylesheet" />

href="https://use.fontawesome.com/releases/v5.0.10/css/all.css"

integrity="sha384-+d0P83n9kaQMCwj8F4RJB66tzIwOKmrdb46+porD/OvrJ+37WqIM7UoBtwHO6Nlg"

rel="stylesheet" />

</head>

<body>

<h1>Packt Todo Application</h1>

</center>

</div>

</nav>

</header>

<body>

{% block todo_container %}{% endblock %}

</div>

</body>

</html>

▶ template/todo.html


{% extends "home.html" %}

{% block todo_container %}
<main class="container">
    <hr>
    <section class="container-fluid">
        <form method="post">
            <div class="col-auto">
                <div class="input-group mb-3">
                    <input name="item"
                        class="form-control"
                        type="text"
                        aria-describedby="button-addon2"
                        aria-label="Add a todo"
                        placeholder="Purchase Packt's Python workshop course"
                        value="{{ item }}" />
                    &nbsp;
                    &nbsp;
                    <button id="button-addon2"
                        class="btn btn-outline-primary"
                        type="submit"
                        data-mdb-ripple-color="dark">
                        Add Todo
                    </button>
                </div>
            </div>
        </form>
    </section>
    {% if todo %}
    <article class="card container-fluid">
        <br/>
        <h4>Todo ID : {{ todo.id }} </h4>
        <p>
            <strong>
                Item : {{ todo.item }}
            </strong>
        </p>
    </article>
    {% else %}
    <section class="container-fluid">
        <h2 align="center">Todos</h2>
        <br>
        <div class="card">
            <ul class="list-group list-group-flush">
                {% for todo in todoList %}
                <li class="list-group-item">
                    {{ loop.index }}. <a href="/todo/{{ loop.index }}"> {{ todo.item }} </a>
                </li>
                {% endfor %}
            </ul>
        </div>
        {% endif %}
    </section>
</main>
{% endblock %}

{% extends "home.html" %}

{% block todo_container %}

<hr>

<input name="item"

class="form-control"

type="text"

aria-describedby="button-addon2"

aria-label="Add a todo"

placeholder="Purchase Packt's Python workshop course"

value="{{ item }}" />

<button id="button-addon2"

class="btn btn-outline-primary"

type="submit"

data-mdb-ripple-color="dark">

Add Todo

</button>

</div>

</form>

</section>

{% if todo %}

<br/>

<p>

Item : {{ todo.item }}

</strong>

</p>

</article>

{% else %}

<h2 align="center">Todos</h2>

<br>

{% for todo in todoList %}

{{ loop.index }}. <a href="/todo/{{ loop.index }}"> {{ todo.item }} </a>

</li>

{% endfor %}

</ul>

</div>

{% endif %}

</section>

</main>

{% endblock %}

▶ model.py


from typing   import Optional, List
from fastapi  import Form
from pydantic import BaseModel

class Todo(BaseModel):
    id   : Optional[int]
    item : str

    @classmethod
    def as_form(cls, item : str = Form(...)):
        return cls(item = item)

    class Config:
        json_schema_extra = {
            "example" : {
                "id"  : 1,
                "item": "Example schema!"
            }
        }

class TodoItem(BaseModel):
    item : str

    class Config:
        json_schema_extra = {
            "example" : {
                "item" : "Read the next chapter of the book"
            }
        }

class TodoItemList(BaseModel):
    todoList : List[TodoItem]

    class Config:
        json_schema_extra = {
            "example" : {
                "todoList" : [
                    {
                        "item" : "Example schema 1!"
                    },
                    {
                        "item" : "Example schema 2!"
                    }
                ]
            }
        }

from typing import Optional, List

from fastapi import Form

from pydantic import BaseModel

class Todo(BaseModel):

id : Optional[int]

item : str

@classmethod

def as_form(cls, item : str = Form(...)):

return cls(item = item)

class Config:

json_schema_extra = {

"example" : {

"id" : 1,

"item": "Example schema!"

}

class TodoItem(BaseModel):

item : str

class Config:

json_schema_extra = {

"example" : {

"item" : "Read the next chapter of the book"

}

class TodoItemList(BaseModel):

todoList : List[TodoItem]

class Config:

json_schema_extra = {

"example" : {

"todoList" : [

{

"item" : "Example schema 1!"

{

"item" : "Example schema 2!"

}

]

}

▶ todo_router.py


from fastapi            import APIRouter, Path, HTTPException, status, Request, Depends
from fastapi.templating import Jinja2Templates

from model import Todo, TodoItem, TodoItemList

todoList = []

todoAPIRouter = APIRouter()

jinja2Templates = Jinja2Templates(directory = "template/")

@todoAPIRouter.post("/todo")
async def addTodo(request : Request, todo : Todo = Depends(Todo.as_form)):
    todo.id = len(todoList) + 1

    todoList.append(todo)

    return jinja2Templates.TemplateResponse(
        "todo.html",
        {
            "request"  : request,
            "todoList" : todoList
        }
    )

@todoAPIRouter.get("/todo", response_model = TodoItemList)
async def getTodo(request : Request):
    return jinja2Templates.TemplateResponse(
        "todo.html",
        {
            "request"  : request,
            "todoList" : todoList
        }
    )

@todoAPIRouter.get("/todo/{todoID}")
async def get_single_todo(request : Request, todoID : int = Path(..., title = "The ID of the todo to retrieve.")):
    for todo in todoList:
        if todo.id == todoID:
            return jinja2Templates.TemplateResponse(
                "todo.html",
                {
                    "request" : request,
                    "todo"    : todo
                }
            )

    raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo with supplied ID doesn't exist")

@todoAPIRouter.put("/todo/{todoID}")
async def update_todo(request : Request, todoItem : TodoItem, todoID : int = Path(..., title="The ID of the todo to be updated.")) -> dict:
    for todo in todoList:
        if todo.id == todoID:
            todo.item = todoItem.item
            return {"message" : "Todo updated successfully."}

    raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo with supplied ID doesn't exist")

@todoAPIRouter.delete("/todo/{todoID}")
async def delete_single_todo(request : Request, todoID : int) -> dict:
    for index in range(len(todoList)):
        todo = todoList[index]

        if todo.id == todoID:
            todoList.pop(index)
            return {"message" : "Todo deleted successfully."}

    raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo with supplied ID doesn't exist")

@todoAPIRouter.delete("/todo")
async def deleteTodoList() -> dict:
    todoList.clear()

    return {"message" : "Todos deleted successfully."}

from fastapi import APIRouter, Path, HTTPException, status, Request, Depends

from fastapi.templating import Jinja2Templates

from model import Todo, TodoItem, TodoItemList

todoList = []

todoAPIRouter = APIRouter()

jinja2Templates = Jinja2Templates(directory = "template/")

@todoAPIRouter.post("/todo")

async def addTodo(request : Request, todo : Todo = Depends(Todo.as_form)):

todo.id = len(todoList) + 1

todoList.append(todo)

return jinja2Templates.TemplateResponse(

"todo.html",

{

"request" : request,

"todoList" : todoList

}

)

@todoAPIRouter.get("/todo", response_model = TodoItemList)

async def getTodo(request : Request):

return jinja2Templates.TemplateResponse(

"todo.html",

{

"request" : request,

"todoList" : todoList

}

)

@todoAPIRouter.get("/todo/{todoID}")

async def get_single_todo(request : Request, todoID : int = Path(..., title = "The ID of the todo to retrieve.")):

for todo in todoList:

if todo.id == todoID:

return jinja2Templates.TemplateResponse(

"todo.html",

{

"request" : request,

"todo" : todo

}

)

raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo with supplied ID doesn't exist")

@todoAPIRouter.put("/todo/{todoID}")

async def update_todo(request : Request, todoItem : TodoItem, todoID : int = Path(..., title="The ID of the todo to be updated.")) -> dict:

for todo in todoList:

if todo.id == todoID:

todo.item = todoItem.item

return {"message" : "Todo updated successfully."}

raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo with supplied ID doesn't exist")

@todoAPIRouter.delete("/todo/{todoID}")

async def delete_single_todo(request : Request, todoID : int) -> dict:

for index in range(len(todoList)):

todo = todoList[index]

if todo.id == todoID:

todoList.pop(index)

return {"message" : "Todo deleted successfully."}

raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo with supplied ID doesn't exist")

@todoAPIRouter.delete("/todo")

async def deleteTodoList() -> dict:

todoList.clear()

return {"message" : "Todos deleted successfully."}

▶ main.py

[PYTHON/FASTAPI] FastAPI 클래스 : add_middleware 메소드를 사용해 CORS 미들웨어 추가하기

■ FastAPI 클래스의 add_middleware 메소드를 사용해 CORS 미들웨어를 추가하는 방법을 보여준다. ▶ 예제 코드 (PY)


from fastapi                 import FastAPI
from fastapi.middleware.cors import CORSMiddleware

fastAPI = FastAPI()

fastAPI.add_middleware(
    CORSMiddleware,
    allow_origins     = ["*"],
    allow_credentials = True,
    allow_methods     = ["*"],
    allow_headers     = ["*"],
)

from fastapi import FastAPI

from fastapi.middleware.cors import CORSMiddleware

fastAPI = FastAPI()

fastAPI.add_middleware(

CORSMiddleware,

allow_origins = ["*"],

allow_credentials = True,

allow_methods = ["*"],

allow_headers = ["*"],

)

[PYTHON/FASTAPI] FastAPI 클래스 : JWT 인증 애플리케이션 만들기 (MongoDB 연동)

■ FastAPI 클래스를 사용해 JWT 인증 애플리케이션을 만드는 방법을 보여준다. (MongoDB 연동) ▶ .env


DATABASE_URL=mongodb://localhost:27017/testdb
SECRET_KEY=pass1234567

DATABASE_URL=mongodb://localhost:27017/testdb

SECRET_KEY=pass1234567

※ testdb : MongoDB 데이터베이스명 ※ pass1234567

[PYTHON/FASTAPI] RedirectResponse 클래스 : 리다이렉션 설정하기

■ RedirectResponse 클래스를 사용해 리다이렉션을 설정하는 방법을 보여준다. ▶ main.py


from fastapi           import FastAPI
from fastapi.responses import RedirectResponse

import uvicorn

fastAPI = FastAPI()

@fastAPI.get("/")
async def home():
    return RedirectResponse(url = "/home/")

@fastAPI.get("/home")
def home():
    return {"message" : "home"}

if __name__ == "__main__":
    uvicorn.run("main:fastAPI", host = "0.0.0.0", port = 8000, reload = True)

from fastapi import FastAPI

from fastapi.responses import RedirectResponse

import uvicorn

fastAPI = FastAPI()

@fastAPI.get("/")

async def home():

return RedirectResponse(url = "/home/")

@fastAPI.get("/home")

def home():

return {"message" : "home"}

if __name__ == "__main__":

uvicorn.run("main:fastAPI", host = "0.0.0.0", port = 8000, reload = True)

▶ requirements.txt


annotated-types==0.7.0
anyio==4.3.0
certifi==2024.2.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.4
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.1
pydantic_core==2.18.2
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.11.0
ujson==5.10.0
uvicorn==0.29.0
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0

annotated-types==0.7.0

anyio==4.3.0

certifi==2024.2.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.4

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.1

pydantic_core==2.18.2

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.11.0

ujson==5.10.0

uvicorn==0.29.0

uvloop==0.19.0

watchfiles==0.21.0

websockets==12.0

[PYTHON/FASTAPI] FastAPI 클래스 : MongoDB 데이터베이스에서 단순 CRUD 애플리케이션 만들기

■ FastAPI 클래스를 사용해 MongoDB 데이터베이스에서 단순 CRUD 애플리케이션을 만드는 방법을 보여준다. ▶ model/__init__.py

▶ model/event.py


from typing   import Optional, List
from beanie   import Document
from pydantic import BaseModel

class Event(Document):
    title       : str
    image       : str
    description : str
    tagList     : List[str]
    location    : str

    class Settings:
        name = "event"

class EventUpdate(BaseModel):
    title       : Optional[str]
    image       : Optional[str]
    description : Optional[str]
    tagList     : Optional[List[str]]
    location    : Optional[str]

from typing import Optional, List

from beanie import Document

from pydantic import BaseModel

class Event(Document):

title : str

image : str

description : str

tagList : List[str]

location : str

class Settings:

name = "event"

class EventUpdate(BaseModel):

title : Optional[str]

image : Optional[str]

description : Optional[str]

tagList : Optional[List[str]]

location : Optional[str]

▶ model/user.py


from typing   import Optional, List
from beanie   import Document
from pydantic import BaseModel, EmailStr

from model.event import Event

class User(Document):
    email     : EmailStr
    password  : str
    eventList : Optional[List[Event]]

    class Settings:
        name = "user"

class UserSignIn(BaseModel):
    email    : EmailStr
    password : str

from typing import Optional, List

from beanie import Document

from pydantic import BaseModel, EmailStr

from model.event import Event

class User(Document):

email : EmailStr

password : str

eventList : Optional[List[Event]]

class Settings:

name = "user"

class UserSignIn(BaseModel):

email : EmailStr

password : str

▶

[PYTHON/FASTAPI] FastAPI 클래스 : SQLITE 데이터베이스에서 단순 CRUD 애플리케이션 만들기

■ FastAPI 클래스를 사용해 SQLITE 데이터베이스에서 단순 CRUD 애플리케이션을 만드는 방법을 보여준다. ▶ database/connection.py


from sqlmodel import SQLModel, Session, create_engine

databaseFile                 = "test.db"
databaseConnectionString     = f"sqlite:///{databaseFile}"
connectionArgumentDictionary = {"check_same_thread" : False}

engineURL = create_engine(databaseConnectionString, echo = True, connect_args = connectionArgumentDictionary)

def connect():
    SQLModel.metadata.create_all(engineURL)

def getSession():
    with Session(engineURL) as session:
        yield session

from sqlmodel import SQLModel, Session, create_engine

databaseFile = "test.db"

databaseConnectionString = f"sqlite:///{databaseFile}"

connectionArgumentDictionary = {"check_same_thread" : False}

engineURL = create_engine(databaseConnectionString, echo = True, connect_args = connectionArgumentDictionary)

def connect():

SQLModel.metadata.create_all(engineURL)

def getSession():

with Session(engineURL) as session:

yield session

▶ model/event.py


from sqlmodel import JSON, SQLModel, Field, Column
from typing   import List

class Event(SQLModel, table = True):
    id          : int = Field(default = None, primary_key = True)
    title       : str
    image       : str
    description : str
    tagList     : List[str] = Field(sa_column = Column(JSON))
    location    : str

class EventUpdate(SQLModel):
    title       : str | None
    image       : str | None
    description : str | None
    tagList     : List[str] | None
    location    : str | None

from sqlmodel import JSON, SQLModel, Field, Column

from typing import List

class Event(SQLModel, table = True):

id : int = Field(default = None, primary_key = True)

title : str

image : str

description : str

tagList : List[str] = Field(sa_column = Column(JSON))

location : str

class EventUpdate(SQLModel):

title : str | None

image : str | None

description : str | None

tagList : List[str] | None

location : str | None

▶ route/event.py


from fastapi  import APIRouter, Depends, HTTPException, Request, status
from sqlmodel import select
from typing   import List

from database.connection import getSession
from model.event         import Event, EventUpdate

eventList = []

eventAPIRouter = APIRouter(tags = ["Event"])

@eventAPIRouter.post("/new")
async def createEvent(newEvent : Event, session = Depends(getSession)) -> dict:
    session.add(newEvent)
    session.commit()
    session.refresh(newEvent)
    return {"message" : "Event created successfully"}

@eventAPIRouter.get("/", response_model = List[Event])
async def getAllEventList(session = Depends(getSession)) -> List[Event]:
    statement = select(Event)
    eventList = session.exec(statement).all()
    return eventList

@eventAPIRouter.get("/{id}", response_model = Event)
async def getEvent(id : int, session = Depends(getSession)) -> Event:
    event = session.get(Event, id)
    if event:
        return event

    raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist")

@eventAPIRouter.put("/edit/{id}", response_model = Event)
async def updateEvent(id : int, eventUpdate : EventUpdate, session = Depends(getSession)) -> Event:
    event = session.get(Event, id)
    if event:
        event_data = eventUpdate.dict(exclude_unset = True)
        for key, value in event_data.items():
            setattr(event, key, value)
        session.add(event)
        session.commit()
        session.refresh(event)
        return event

    raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist")

@eventAPIRouter.delete("/{id}")
async def deleteEvent(id : int, session = Depends(getSession)) -> dict:
    event = session.get(Event, id)
    if event:
        session.delete(event)
        session.commit()
        return {"message" : "Event deleted successfully"}

    raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist")

from fastapi import APIRouter, Depends, HTTPException, Request, status

from sqlmodel import select

from typing import List

from database.connection import getSession

from model.event import Event, EventUpdate

eventList = []

eventAPIRouter = APIRouter(tags = ["Event"])

@eventAPIRouter.post("/new")

async def createEvent(newEvent : Event, session = Depends(getSession)) -> dict:

session.add(newEvent)

session.commit()

session.refresh(newEvent)

return {"message" : "Event created successfully"}

@eventAPIRouter.get("/", response_model = List[Event])

async def getAllEventList(session = Depends(getSession)) -> List[Event]:

statement = select(Event)

eventList = session.exec(statement).all()

return eventList

@eventAPIRouter.get("/{id}", response_model = Event)

async def getEvent(id : int, session = Depends(getSession)) -> Event:

event = session.get(Event, id)

if event:

return event

raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist")

@eventAPIRouter.put("/edit/{id}", response_model = Event)

async def updateEvent(id : int, eventUpdate : EventUpdate, session = Depends(getSession)) -> Event:

event = session.get(Event, id)

if event:

event_data = eventUpdate.dict(exclude_unset = True)

for key, value in event_data.items():

setattr(event, key, value)

session.add(event)

session.commit()

session.refresh(event)

return event

raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist")

@eventAPIRouter.delete("/{id}")

async def deleteEvent(id : int, session = Depends(getSession)) -> dict:

event = session.get(Event, id)

if event:

session.delete(event)

session.commit()

return {"message" : "Event deleted successfully"}

raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Event with supplied ID does not exist")

▶

[PYTHON/FASTAPI] FastAPI 애플리케이션 구조화하기

■ FastAPI 애플리케이션을 구조화하는 방법을 보여준다. ▶ 디렉토리 및 파일 구조


planner/
    model/
        __init__.py
        event.py
        user.py
    route/
        __init__.ps
        event.py
        user.py
    main.py

planner/

model/

__init__.py

event.py

user.py

route/

__init__.ps

event.py

user.py

main.py

▶ model/__init__.py

▶ model/event.py


from typing   import List
from pydantic import BaseModel

class Event(BaseModel):
    id          : int
    title       : str
    image       : str
    description : str
    tagList     : List[str]
    location    : str

from typing import List

from pydantic import BaseModel

class Event(BaseModel):

id : int

title : str

image : str

description : str

tagList : List[str]

location : str

▶ model/user.py


from pydantic import BaseModel, EmailStr
from typing   import Optional, List

from model.event import Event

class User(BaseModel):
    email     : EmailStr
    password  : str
    eventList : Optional[List[Event]]

class UserSignIn(BaseModel):
    email    : EmailStr
    password : str

from pydantic import BaseModel, EmailStr

from typing import Optional, List

from model.event import Event

class User(BaseModel):

email : EmailStr

password : str

eventList : Optional[List[Event]]

class UserSignIn(BaseModel):

email : EmailStr

password : str

▶

[PYTHON/FASTAPI] HTTPException 클래스 : 커스텀 예외 반환하기

■ HTTPException 클래스를 사용해 커스텀 예외를 반환하는 방법을 보여준다. ▶ main.py


from fastapi  import FastAPI, APIRouter, Path, HTTPException, status
from pydantic import BaseModel
from typing   import List

class TodoItem(BaseModel):
    id   : int
    item : str

class TodoItemUpdate(BaseModel):
    item : str

class TodoItemList(BaseModel):
    todoItemList : List[TodoItem]

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo", status_code = 201)
async def addTodoItem(todoItem : TodoItem) -> dict:
    todoItemList.append(todoItem)
    return {"message" : "Todo item added successfully."}

@apiRouter.get("/todo", response_model = TodoItemList)
async def getTodoItemList() -> dict:
    return {"todoItemList" : todoItemList}

@apiRouter.get("/todo/{todoItemID}")
async def getTodoItem(todoItemID : int = Path(..., title = "The ID of the todo item to retrieve.")) -> dict:
    for todoItem in todoItemList:
        if todoItem.id == todoItemID:
            return {"todoItem" : todoItem}

    raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo item with supplied ID doesn't exist")

@apiRouter.put("/todo/{todoItemID}")
async def updateTodoItem(todoItemUpdate : TodoItemUpdate, todoItemID : int = Path(..., title = "The ID of the todo item to be updated.")) -> dict:
    for todoItem in todoItemList:
        if todoItem.id == todoItemID:
            todoItem.item = todoItemUpdate.item
            return {"message" : "Todo item updated successfully."}

    raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo item with supplied ID doesn't exist")

@apiRouter.delete("/todo/{todoItemID}")
async def deleteTodoItem(todoItemID : int) -> dict:
    for index in range(len(todoItemList)):
        todoItem = todoItemList[index]
        if todoItem.id == todoItemID:
            todoItemList.pop(index)
            return {"message" : "Todo item deleted successfully."}

    raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo item with supplied ID doesn't exist")

@apiRouter.delete("/todo")
async def deleteAllTodoItems() -> dict:
    todoItemList.clear()
    return {"message" : "Todo items deleted successfully."}

fastAPI = FastAPI()

@fastAPI.get("/")
def root():
    return {"message" : "Hello World"}

@fastAPI.get("/home")
def home():
    return {"message" : "home"}

fastAPI.include_router(apiRouter)

from fastapi import FastAPI, APIRouter, Path, HTTPException, status

from pydantic import BaseModel

from typing import List

class TodoItem(BaseModel):

id : int

item : str

class TodoItemUpdate(BaseModel):

item : str

class TodoItemList(BaseModel):

todoItemList : List[TodoItem]

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo", status_code = 201)

async def addTodoItem(todoItem : TodoItem) -> dict:

todoItemList.append(todoItem)

return {"message" : "Todo item added successfully."}

@apiRouter.get("/todo", response_model = TodoItemList)

async def getTodoItemList() -> dict:

return {"todoItemList" : todoItemList}

@apiRouter.get("/todo/{todoItemID}")

async def getTodoItem(todoItemID : int = Path(..., title = "The ID of the todo item to retrieve.")) -> dict:

for todoItem in todoItemList:

if todoItem.id == todoItemID:

return {"todoItem" : todoItem}

raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo item with supplied ID doesn't exist")

@apiRouter.put("/todo/{todoItemID}")

async def updateTodoItem(todoItemUpdate : TodoItemUpdate, todoItemID : int = Path(..., title = "The ID of the todo item to be updated.")) -> dict:

for todoItem in todoItemList:

if todoItem.id == todoItemID:

todoItem.item = todoItemUpdate.item

return {"message" : "Todo item updated successfully."}

raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo item with supplied ID doesn't exist")

@apiRouter.delete("/todo/{todoItemID}")

async def deleteTodoItem(todoItemID : int) -> dict:

for index in range(len(todoItemList)):

todoItem = todoItemList[index]

if todoItem.id == todoItemID:

todoItemList.pop(index)

return {"message" : "Todo item deleted successfully."}

raise HTTPException(status_code = status.HTTP_404_NOT_FOUND, detail = "Todo item with supplied ID doesn't exist")

@apiRouter.delete("/todo")

async def deleteAllTodoItems() -> dict:

todoItemList.clear()

return {"message" : "Todo items deleted successfully."}

fastAPI = FastAPI()

@fastAPI.get("/")

def root():

return {"message" : "Hello World"}

@fastAPI.get("/home")

def home():

return {"message" : "home"}

fastAPI.include_router(apiRouter)

▶ requirements.txt


annotated-types==0.6.0
anyio==4.3.0
certifi==2024.2.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.3
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
itsdangerous==2.2.0
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.1
pydantic-extra-types==2.7.0
pydantic-settings==2.2.1
pydantic_core==2.18.2
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.11.0
ujson==5.10.0
uvicorn==0.29.0
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0

annotated-types==0.6.0

anyio==4.3.0

certifi==2024.2.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.3

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

itsdangerous==2.2.0

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.1

pydantic-extra-types==2.7.0

pydantic-settings==2.2.1

pydantic_core==2.18.2

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.11.0

ujson==5.10.0

uvicorn==0.29.0

uvloop==0.19.0

watchfiles==0.21.0

websockets==12.0

▶ FastAPI 서버 실행 명령


uvicorn main:fastAPI --port 8000 --reload

uvicorn main:fastAPI --port 8000 --reload

▶

[PYTHON/FASTAPI] APIRouter 클래스 : get 메소드에서 response_model 인자 사용하기

■ APIRouter 클래스의 get 메소드에서 response_model 인자를 사용하는 방법을 보여준다. ▶ main.py


from fastapi  import FastAPI, APIRouter
from pydantic import BaseModel
from typing   import List

class TodoItem(BaseModel):
    id   : int
    item : str

class TodoItemList(BaseModel):
    todoItemList : List[TodoItem]

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")
async def addTodo(todoItem : TodoItem) -> dict:
    todoItemList.append(todoItem)
    return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo", response_model = TodoItemList)
async def getTodoItemList() -> dict:
    return {"todoItemList" : todoItemList}

fastAPI = FastAPI()

@fastAPI.get("/")
def root():
    return {"message" : "Hello World"}

@fastAPI.get("/home")
def home():
    return {"message" : "home"}

fastAPI.include_router(apiRouter)

from fastapi import FastAPI, APIRouter

from pydantic import BaseModel

from typing import List

class TodoItem(BaseModel):

id : int

item : str

class TodoItemList(BaseModel):

todoItemList : List[TodoItem]

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")

async def addTodo(todoItem : TodoItem) -> dict:

todoItemList.append(todoItem)

return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo", response_model = TodoItemList)

async def getTodoItemList() -> dict:

return {"todoItemList" : todoItemList}

fastAPI = FastAPI()

@fastAPI.get("/")

def root():

return {"message" : "Hello World"}

@fastAPI.get("/home")

def home():

return {"message" : "home"}

fastAPI.include_router(apiRouter)

▶ requirements.txt


annotated-types==0.6.0
anyio==4.3.0
certifi==2024.2.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.3
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
itsdangerous==2.2.0
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.1
pydantic-extra-types==2.7.0
pydantic-settings==2.2.1
pydantic_core==2.18.2
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.11.0
ujson==5.10.0
uvicorn==0.29.0
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0

annotated-types==0.6.0

anyio==4.3.0

certifi==2024.2.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.3

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

itsdangerous==2.2.0

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.1

pydantic-extra-types==2.7.0

pydantic-settings==2.2.1

pydantic_core==2.18.2

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.11.0

ujson==5.10.0

uvicorn==0.29.0

uvloop==0.19.0

watchfiles==0.21.0

websockets==12.0

▶ FastAPI 서버 실행 명령


uvicorn main:fastAPI --port 8000 --reload

uvicorn main:fastAPI --port 8000 --reload

[PYTHON/FASTAPI] 스웨거/ReDoc 문서에서 샘플 데이터 추가하기

■ 스웨거/ReDoc 문서에서 샘플 데이터를 추가하는 방법을 보여준다. ▶ main.py


from fastapi  import FastAPI, APIRouter, Path
from pydantic import BaseModel

class TodoItem(BaseModel):
    id   : int
    item : str

    class Config:
        json_schema_extra = {
            "example" : {
                "id"   : 1,
                "item" : "Example Schema!"
            }
        }

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")
async def addTodoItem(todoItem : TodoItem) -> dict:
    todoItemList.append(todoItem)
    return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")
async def getTodoItemList() -> dict:
    return {"todoItemList" : todoItemList}

@apiRouter.get("/todo/{todoItemID}")
async def getTodoItem(todoItemID : int = Path(..., title = "The ID of the todo item to retrieve.")) -> dict:
    for todoItem in todoItemList:
        if todoItem.id == todoItemID:
            return {"todoItem" : todoItem}
    return {"message" : "Todo item with supplied ID doesn't exist."}

fastAPI = FastAPI()

@fastAPI.get("/")
def root():
    return {"message" : "Hello World"}

@fastAPI.get("/home")
def home():
    return {"message" : "home"}

fastAPI.include_router(apiRouter)

from fastapi import FastAPI, APIRouter, Path

from pydantic import BaseModel

class TodoItem(BaseModel):

id : int

item : str

class Config:

json_schema_extra = {

"example" : {

"id" : 1,

"item" : "Example Schema!"

}

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")

async def addTodoItem(todoItem : TodoItem) -> dict:

todoItemList.append(todoItem)

return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")

async def getTodoItemList() -> dict:

return {"todoItemList" : todoItemList}

@apiRouter.get("/todo/{todoItemID}")

async def getTodoItem(todoItemID : int = Path(..., title = "The ID of the todo item to retrieve.")) -> dict:

for todoItem in todoItemList:

if todoItem.id == todoItemID:

return {"todoItem" : todoItem}

return {"message" : "Todo item with supplied ID doesn't exist."}

fastAPI = FastAPI()

@fastAPI.get("/")

def root():

return {"message" : "Hello World"}

@fastAPI.get("/home")

def home():

return {"message" : "home"}

fastAPI.include_router(apiRouter)

▶ requirements.txt


annotated-types==0.6.0
anyio==4.3.0
certifi==2024.2.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.3
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
itsdangerous==2.2.0
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.1
pydantic-extra-types==2.7.0
pydantic-settings==2.2.1
pydantic_core==2.18.2
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.11.0
ujson==5.10.0
uvicorn==0.29.0
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0

annotated-types==0.6.0

anyio==4.3.0

certifi==2024.2.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.3

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

itsdangerous==2.2.0

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.1

pydantic-extra-types==2.7.0

pydantic-settings==2.2.1

pydantic_core==2.18.2

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.11.0

ujson==5.10.0

uvicorn==0.29.0

uvloop==0.19.0

watchfiles==0.21.0

websockets==12.0

▶ FastAPI 서버 실행 명령


uvicorn main:fastAPI --port 8000 --reload

uvicorn main:fastAPI --port 8000 --reload

▶ 스웨거/Redoc

[PYTHON/FASTAPI] FastAPI 클래스 : 단순 CURD 애플리케이션 만들기

■ FastAPI 클래스를 사용해 단순 CURD 애플리케이션을 만드는 방법을 보여준다. ▶ main.py


from fastapi  import FastAPI, APIRouter, Path
from pydantic import BaseModel

class TodoItem(BaseModel):
    id   : int
    item : str

    class Config:
        json_schema_extra = {
            "example" : {
                "id"   : 1,
                "item" : "Example Schema!"
            }
        }

class TodoItemUpdate(BaseModel):
    item : str

    class Config:
        json_schema_extra = {
            "example" : {
            "item" : "Read the next chapter of the book"
        }
    }

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")
async def addTodoItem(todoItem : TodoItem) -> dict:
    todoItemList.append(todoItem)
    return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")
async def getTodoItemList() -> dict:
    return {"todoItemList" : todoItemList}

@apiRouter.get("/todo/{todoItemID}")
async def getTodoItem(todoItemID : int = Path(..., title = "The ID of the todo item to retrieve.")) -> dict:
    for todoItem in todoItemList:
        if todoItem.id == todoItemID:
            return {"todoItem" : todoItem}
    return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.put("/todo/{todoItemID}")
async def updateTodo(todoItemUpdate: TodoItemUpdate, todoID : int = Path(..., title = "The ID of the todo item to be updated.")) -> dict:
    for todoItem in todoItemList:
        if todoItem.id == todoItemID:
            todoItem.item = todoItemUpdate.item
            return {"message" : "Todo item updated successfully."}
    return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.delete("/todo/{todoItemID}")
async def deleteTodo(todoItemID : int) -> dict:
  for index in range(len(todoItemList)):
      todoItem = todoItemList[index]
      if todoItem.id == todoItemID:
          todoItemList.pop(index)
          return {"message" : "Todo item deleted successfully."}
  return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.delete("/todo")
async def deleteAllTodoItems() -> dict:
  todoItemList.clear()
  return {"message" : "All Todo items deleted successfully."}

fastAPI = FastAPI()

@fastAPI.get("/")
def root():
    return {"message" : "Hello World"}

@fastAPI.get("/home")
def home():
    return {"message" : "home"}

fastAPI.include_router(apiRouter)

from fastapi import FastAPI, APIRouter, Path

from pydantic import BaseModel

class TodoItem(BaseModel):

id : int

item : str

class Config:

json_schema_extra = {

"example" : {

"id" : 1,

"item" : "Example Schema!"

}

class TodoItemUpdate(BaseModel):

item : str

class Config:

json_schema_extra = {

"example" : {

"item" : "Read the next chapter of the book"

}

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")

async def addTodoItem(todoItem : TodoItem) -> dict:

todoItemList.append(todoItem)

return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")

async def getTodoItemList() -> dict:

return {"todoItemList" : todoItemList}

@apiRouter.get("/todo/{todoItemID}")

async def getTodoItem(todoItemID : int = Path(..., title = "The ID of the todo item to retrieve.")) -> dict:

for todoItem in todoItemList:

if todoItem.id == todoItemID:

return {"todoItem" : todoItem}

return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.put("/todo/{todoItemID}")

async def updateTodo(todoItemUpdate: TodoItemUpdate, todoID : int = Path(..., title = "The ID of the todo item to be updated.")) -> dict:

for todoItem in todoItemList:

if todoItem.id == todoItemID:

todoItem.item = todoItemUpdate.item

return {"message" : "Todo item updated successfully."}

return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.delete("/todo/{todoItemID}")

async def deleteTodo(todoItemID : int) -> dict:

for index in range(len(todoItemList)):

todoItem = todoItemList[index]

if todoItem.id == todoItemID:

todoItemList.pop(index)

return {"message" : "Todo item deleted successfully."}

return {"message" : "Todo item with supplied ID doesn't exist."}

@apiRouter.delete("/todo")

async def deleteAllTodoItems() -> dict:

todoItemList.clear()

return {"message" : "All Todo items deleted successfully."}

fastAPI = FastAPI()

@fastAPI.get("/")

def root():

return {"message" : "Hello World"}

@fastAPI.get("/home")

def home():

return {"message" : "home"}

fastAPI.include_router(apiRouter)

▶ requirements.txt


annotated-types==0.6.0
anyio==4.3.0
certifi==2024.2.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.3
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
itsdangerous==2.2.0
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.1
pydantic-extra-types==2.7.0
pydantic-settings==2.2.1
pydantic_core==2.18.2
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.11.0
ujson==5.10.0
uvicorn==0.29.0
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0

annotated-types==0.6.0

anyio==4.3.0

certifi==2024.2.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.3

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

itsdangerous==2.2.0

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.1

pydantic-extra-types==2.7.0

pydantic-settings==2.2.1

pydantic_core==2.18.2

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.11.0

ujson==5.10.0

uvicorn==0.29.0

uvloop==0.19.0

watchfiles==0.21.0

websockets==12.0

▶ FastAPI 서버 실행 명령


uvicorn main:fastAPI --port 8000 --reload

uvicorn main:fastAPI --port 8000 --reload


from fastapi  import FastAPI, APIRouter
from pydantic import BaseModel

class TodoItem(BaseModel):
    id   : int
    item : str

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")
async def addTodoItem(todoItem : TodoItem) -> dict:
    todoItemList.append(todoItem)
    return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")
async def getTodoItemList() -> dict:
    return {"todoItemList" : todoItemList}

fastAPI = FastAPI()

@fastAPI.get("/")
def root():
    return {"message" : "Hello World"}

@fastAPI.get("/home")
def home():
    return {"message" : "home"}

fastAPI.include_router(apiRouter)

from fastapi import FastAPI, APIRouter

from pydantic import BaseModel

class TodoItem(BaseModel):

id : int

item : str

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")

async def addTodoItem(todoItem : TodoItem) -> dict:

todoItemList.append(todoItem)

return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")

async def getTodoItemList() -> dict:

return {"todoItemList" : todoItemList}

fastAPI = FastAPI()

@fastAPI.get("/")

def root():

return {"message" : "Hello World"}

@fastAPI.get("/home")

def home():

return {"message" : "home"}

fastAPI.include_router(apiRouter)

▶ requirements.txt


annotated-types==0.6.0
anyio==4.3.0
certifi==2024.2.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.3
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
itsdangerous==2.2.0
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.1
pydantic-extra-types==2.7.0
pydantic-settings==2.2.1
pydantic_core==2.18.2
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.11.0
ujson==5.10.0
uvicorn==0.29.0
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0

annotated-types==0.6.0

anyio==4.3.0

certifi==2024.2.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.3

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

itsdangerous==2.2.0

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.1

pydantic-extra-types==2.7.0

pydantic-settings==2.2.1

pydantic_core==2.18.2

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.11.0

ujson==5.10.0

uvicorn==0.29.0

uvloop==0.19.0

watchfiles==0.21.0

websockets==12.0

▶ FastAPI 서버 실행 명령

[PYTHON/FASTAPI] APIRouter 클래스 : 라우팅 사용하기

■ APIRouter 클래스를 사용해 라우팅을 사용하는 방법을 보여준다. ▶ main.py


from fastapi import FastAPI, APIRouter

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")
async def addTodoItem(todoItem : dict) -> dict:
    todoItemList.append(todoItem)
    return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")
async def getTodoItemList() -> dict:
    return {"todoItemList" : todoItemList}

fastAPI = FastAPI()

@fastAPI.get("/")
def root():
    return {"message" : "Hello World"}

@fastAPI.get("/home")
def home():
    return {"message" : "home"}

fastAPI.include_router(apiRouter)

from fastapi import FastAPI, APIRouter

todoItemList = []

apiRouter = APIRouter()

@apiRouter.post("/todo")

async def addTodoItem(todoItem : dict) -> dict:

todoItemList.append(todoItem)

return {"message" : "Todo item added successfully"}

@apiRouter.get("/todo")

async def getTodoItemList() -> dict:

return {"todoItemList" : todoItemList}

fastAPI = FastAPI()

@fastAPI.get("/")

def root():

return {"message" : "Hello World"}

@fastAPI.get("/home")

def home():

return {"message" : "home"}

fastAPI.include_router(apiRouter)

▶ requirements.py


annotated-types==0.6.0
anyio==4.3.0
certifi==2024.2.2
click==8.1.7
dnspython==2.6.1
email_validator==2.1.1
exceptiongroup==1.2.1
fastapi==0.111.0
fastapi-cli==0.0.3
h11==0.14.0
httpcore==1.0.5
httptools==0.6.1
httpx==0.27.0
idna==3.7
itsdangerous==2.2.0
Jinja2==3.1.4
markdown-it-py==3.0.0
MarkupSafe==2.1.5
mdurl==0.1.2
orjson==3.10.3
pydantic==2.7.1
pydantic-extra-types==2.7.0
pydantic-settings==2.2.1
pydantic_core==2.18.2
Pygments==2.18.0
python-dotenv==1.0.1
python-multipart==0.0.9
PyYAML==6.0.1
rich==13.7.1
shellingham==1.5.4
sniffio==1.3.1
starlette==0.37.2
typer==0.12.3
typing_extensions==4.11.0
ujson==5.10.0
uvicorn==0.29.0
uvloop==0.19.0
watchfiles==0.21.0
websockets==12.0

annotated-types==0.6.0

anyio==4.3.0

certifi==2024.2.2

click==8.1.7

dnspython==2.6.1

email_validator==2.1.1

exceptiongroup==1.2.1

fastapi==0.111.0

fastapi-cli==0.0.3

h11==0.14.0

httpcore==1.0.5

httptools==0.6.1

httpx==0.27.0

idna==3.7

itsdangerous==2.2.0

Jinja2==3.1.4

markdown-it-py==3.0.0

MarkupSafe==2.1.5

mdurl==0.1.2

orjson==3.10.3

pydantic==2.7.1

pydantic-extra-types==2.7.0

pydantic-settings==2.2.1

pydantic_core==2.18.2

Pygments==2.18.0

python-dotenv==1.0.1

python-multipart==0.0.9

PyYAML==6.0.1

rich==13.7.1

shellingham==1.5.4

sniffio==1.3.1

starlette==0.37.2

typer==0.12.3

typing_extensions==4.11.0

ujson==5.10.0

uvicorn==0.29.0

uvloop==0.19.0

watchfiles==0.21.0

websockets==12.0

▶ FastAPI 서버 실행 명령


uvicorn main:fastAPI --port 8000 --reload

uvicorn main:fastAPI --port 8000 --reload

▶ CURL

[PYTHON/UVICORN] FastAPI 서버 실행시 포트 설정하기

■ FastAPI 서버 실행시 포트를 설정하는 방법을 보여준다. ▶ 실행 명령


uvicorn main:fastAPI --port 8000 --reload

uvicorn main:fastAPI --port 8000 --reload

※ main : 시작 파일명(main.py) ※ fastAPI : FastAPI 객체명

[태그:] FASTAPI

[PYTHON/FASTAPI] FastAPI 클래스 : StreamingResponse 객체를 사용해 응답 스트리밍 처리하기

[PYTHON/FASTAPI] Request 클래스 : 메소드에서 HTTP Request 객체 직접 참조하기

[PYTHON/LANGCHAIN] langserve 모듈 : langchain 모듈 체인을 REST API 서버로 배포하기

[PYTHON/FASTAPI] FastAPI 클래스 : add_middleware 메소드를 사용해 HTTP 요청에 대해 HTTPS 리다이렉션 설정하기

[PYTHON/FASTAPI] FastAPI 클래스 : route 메소드를 사용해 HTTP 요청에 대해 HTTPS 리다이렉션 설정하기

[PYTHON/UVICORN] run 함수 : ssl_keyfile/ssl_certfile 인자를 사용해 HTTPS 설정하기

[PYTHON/FASTAPI] FastAPI 클래스 : add_middleware 메소드를 사용해 GZIP 압축 전송 설정하기

[DOCKER/UBUNTU] 파이썬 FastAPI 애플리케이션 도커 파일 샘플

[DOCKER/UBUNTU] 파이썬 가상 환경 FastAPI 애플리케이션 도커 파일 샘플

[PYTHON/FASTAPI] FastAPI 클래스 : JINJA 템플리트 사용하기

[PYTHON/FASTAPI] FastAPI 클래스 : add_middleware 메소드를 사용해 CORS 미들웨어 추가하기

[PYTHON/FASTAPI] FastAPI 클래스 : JWT 인증 애플리케이션 만들기 (MongoDB 연동)

[PYTHON/FASTAPI] RedirectResponse 클래스 : 리다이렉션 설정하기

[PYTHON/FASTAPI] FastAPI 클래스 : MongoDB 데이터베이스에서 단순 CRUD 애플리케이션 만들기

[PYTHON/FASTAPI] FastAPI 클래스 : SQLITE 데이터베이스에서 단순 CRUD 애플리케이션 만들기

[PYTHON/FASTAPI] FastAPI 애플리케이션 구조화하기

[PYTHON/FASTAPI] HTTPException 클래스 : 커스텀 예외 반환하기

[PYTHON/FASTAPI] APIRouter 클래스 : get 메소드에서 response_model 인자 사용하기

[PYTHON/FASTAPI] 스웨거/ReDoc 문서에서 샘플 데이터 추가하기

[PYTHON/FASTAPI] FastAPI 클래스 : 단순 CURD 애플리케이션 만들기

[PYTHON/FASTAPI] Path 클래스 : OpenAPI 기반 문서 자동 생성시 라우트 정보 문서화하기

[PYTHON/FASTAPI] FastAPI 클래스 : get 메소드에서 경로 매개 변수 사용하기

[PYTHON/PYDANTIC] BaseModel 클래스 : REST POST 요청 바디 검증하기

[PYTHON/FASTAPI] APIRouter 클래스 : 라우팅 사용하기

[PYTHON/UVICORN] FastAPI 서버 실행시 포트 설정하기