REST Archives - icodebroker

[TOOL/CURL] curl 명령 : OLLAMA 서버 통신하기

■ curl 명령을 사용해 OLLAMA 서버와 통신하는 방법을 보여준다. ▶ 실행 명령


[윈도우즈]

curl -X POST http://localhost:11434/api/generate -H "Content-Type: application/json" -d "{\"model\":\"bnksys/eeve-yanolja-v1:latest\",\"prompt\":\"Hello\"}"

[리눅스]

curl http://127.0.0.1:11434/api/generate -d "{'model': 'bnksys/eeve-yanolja-v1:latest', 'prompt': 'Hello'}"

[윈도우즈]

curl -X POST http://localhost:11434/api/generate -H "Content-Type: application/json" -d "{\"model\":\"bnksys/eeve-yanolja-v1:latest\",\"prompt\":\"Hello\"}"

[리눅스]

curl http://127.0.0.1:11434/api/generate -d "{'model': 'bnksys/eeve-yanolja-v1:latest', 'prompt': 'Hello'}"

[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

[C#/COMMON/.NET8] HttpClient 클래스 : PostAsJsonAsync/GetFromJsonAsync/GetAsync/PutAsJsonAsync/DeleteAsync 메소드를 사용해 REST API 호출하기

■ HttpClient 클래스의 PostAsJsonAsync/GetFromJsonAsync/GetAsync/PutAsJsonAsync/DeleteAsync 메소드를 사용해 REST API를 호출하는 방법을 보여준다. ※ REST API 서버는 파이썬 서버를 사용했다 : https://icodebroker.com/archives/107331 ▶ TodoItem.cs

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

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

[TOOL/CURL] curl 명령 : HTTP DELETE 요청에서 데이터 삭제 요청하기

■ curl 명령을 사용해 HTTP DELETE 요청에서 데이터 삭제를 요청하는 방법을 보여준다. 1. [명령 프롬프트]를 실행한다. 2. [명령 프롬프트]에서 아래 명령을 실행한다.

[TOOL/CURL] curl 명령 : HTTP PUT 요청에서 데이터 수정 요청하기

■ curl 명령을 사용해 HTTP PUT 요청에서 데이터 수정을 요청하는 방법을 보여준다. 1. [명령 프롬프트]를 실행한다. 2. [명령 프롬프트]에서 아래 명령을 실행한다.

[TOOL/CURL] curl 명령 : HTTP POST 요청에서 JWT 토큰을 포함한 신규 데이터 추가 요청하기

■ curl 명령을 사용해 HTTP POST 요청에서 JWT 토큰을 포함한 신규 데이터 추가를 요청하는 방법을 보여준다. 1. [명령 프롬프트]를 실행한다. 2. [명령

[TOOL/CURL] curl 명령 : HTTP POST 요청에서 FORM 인증 방식으로 사용자 로그인하기

■ curl 명령을 사용해 HTTP POST 요청에서 FORM 인증 방식으로 사용자 로그인하는 방법을 보여준다. 1. [명령 프롬프트]를 실행한다. 2. [명령 프롬프트]에서 아래

[TOOL/CURL] curl 명령 : HTTP POST 요청에서 신규 사용자 추가 요청하기

■ curl 명령을 사용해 HTTP POST 요청에서 신규 사용자 추가를 요청하는 방법을 보여준다. 1. [명령 프롬프트]를 실행한다. 2. [명령 프롬프트]에서 아래 명령을

[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

[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/UVICORN] run 함수 : FastAPI 서버 실행하기

■ run 함수를 사용해 FastAPI 서버를 실행하는 방법을 보여준다. ▶ main.py


from fastapi import FastAPI

import uvicorn

fastAPI = FastAPI()

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

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

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

from fastapi import FastAPI

import uvicorn

fastAPI = FastAPI()

@fastAPI.get("/")

def root():

return {"message" : "Hello World"}

@fastAPI.get("/home")

def home():

return {"message" : "home"}

if __name__ == "__main__":

uvicorn.run("main:fastAPI", host = "127.0.0.1", 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.12.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.12.0

ujson==5.10.0

uvicorn==0.29.0

uvloop==0.19.0

watchfiles==0.21.0

websockets==12.0

[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