์ด์ ๊ธ
https://koreatstm.tistory.com/295
Fast API - ๊ธฐ์ด
๋ง์ ํ๋ ์ ์ํฌ๋ฅผ ์ฌ์ฉํด์์ง๋ง, ๊น์ด๊ฐ ๋ถ์กฑํ๋ค๊ณ ํญ์ ์๊ฐํ๋ค. ์ด์ ๋ ๊ทธ๊ฒ๋ค์ ๋ํ ๊น์ด๋ฅผ ๋ ์ฑ์๋ณด๊ณ ์ ํ๋ค.๋ฌด์๋ณด๋ค ์ด๋ค ์ง๋ฌธ์ ๊ดํด์๋ ์ด์ ๋ฅผ ๋ฌผ์ ๋ ๋ฐ๋ก ๋ต ํ ์ ์๋ ์ค
koreatstm.tistory.com
ํ ํ๋ฆฟ
Jinja2๋?
- Jinja2๋ Python ์น ํ๋ ์์ํฌ์์ ๋์ ์น ํ์ด์ง ์์ฑ์ ์ํ ํ ํ๋ฆฟ ์์ง
- ํ ํ๋ฆฟ ํ์ผ ๋ด์์ {{ ๋ณ์๋ช }}์ผ๋ก ๋์ ๋ฐ์ดํฐ ์ฝ์ ๊ฐ๋ฅ
- ์ฃผ์ ๊ตฌ๋ถ ๊ธฐํธ
- {{...}}: ๋ณ์๋ ํํ์ ์ถ๋ ฅ
- {%...%}: ์ ์ด๋ฌธ ์์ฑ ์ ์ฌ์ฉ
- {#...#}: ์ฃผ์ ์ฒ๋ฆฌ
FastAPI์์ Jinja2 ํ ํ๋ฆฟ ์ฌ์ฉ
- templates ๋๋ ํฐ๋ฆฌ
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/")
def read_root(request: Request):
items = ["์ฌ๊ณผ", "๋ฐ๋๋", "์ค๋ ์ง"]
return templates.TemplateResponse("index.html", {"request": request, "items": items})<!DOCTYPE html>
<html>
<body>
<ul>
{% for item in items %}
<li>{{ item }}</li>
{% endfor %}
</ul>
</body>
</html>์ด๋: {{username}} ๋ถ๋ถ์ด ์ค์ ๊ฐ์ผ๋ก ๋์ฒด๋์ด ๋ ๋๋ง ๋จ
ํ ํ๋ฆฟ ๋ ๋๋ง: HTML ํ ํ๋ฆฟ์ ๋ฐ์ดํฐ ์ฝ์ ํ๋ ค๋ฉด TemplateResponse๋ฉ์๋ ์ฌ์ฉ
- ๋ ๋๋ง: ๋ณ์๋ ๋ฐ์ดํฐ๊ฐ ๋ค์ด๊ฐ ํ ํ๋ฆฟ์ ์ค์ ๊ฐ์ด ์ฑ์์ง ์์ฑ๋ ๋ฌธ์(HTML ๋ฑ)๋ก ๋ณํํ๋ ๊ฒ
- FastAPI์์๋ Jinja2Templates ํด๋์ค๋ก ํ ํ๋ฆฟ ๋๋ ํฐ๋ฆฌ ๋ฐ ์ต์ ์ง์ ํ, TemplateResponse๋ก HTML ๋ ๋๋ง
- ํ ํ๋ฆฟ์ผ๋ก ์ ๋ฌ๋ ๋ณ์๋ค์ ๋ด์ ๋์ ๋๋ฆฌ {request~ username}์ ๋๋ฒ์งธ ์ธ์๋ก ๋ฐ์์, ์ด๋ฅผ ๋์ ์ผ๋ก ๋ฐ์ดํฐ ์ฒ๋ฆฌ
HTML ํ ํ๋ฆฟ ํ์ผ(Jinja2)์ ๋์ ๋ฐ์ดํฐ๋ฅผ ๋ฃ์ด ์น ๋ธ๋ผ์ฐ์ ์ ๋ณด์ฌ์ค ๋ ์๋ตํ๋ ๊ฐ์ฒด
์ด๋, ๋ฐ๋์ request ๊ฐ์ฒด๋ฅผ context์ ํฌํจ์์ผ์ผํ๋ค. ๋์ด ์ธํธ!!
TemplateResponse๋ฅผ ์ฌ์ฉํ ๋ Request ๊ฐ์ฒด๋ฅผ ๋์ ๋๋ฆฌ์ ํฌํจ์์ผ์ผ ํ๋ค. Request๊ฐ์ฒด์ ์์กดํ๊ธฐ ๋๋ฌธ
Jinja2Templates์ ์ฃผ์ ์ต์
| ์ต์ ๋ช | ์ค๋ช | ๊ธฐ๋ณธ๊ฐ |
| directory | ํ ํ๋ฆฟ ํ์ผ์ด ์์นํ ๋๋ ํฐ๋ฆฌ ๊ฒฝ๋ก | "templates" |
| encoding | ํ ํ๋ฆฟ ํ์ผ์ ์ธ์ฝ๋ฉ ๋ฐฉ์ | "utf-8" |
| auto_reload | ํ ํ๋ฆฟ ํ์ผ์ด ๋ณ๊ฒฝ๋ ๋ ์๋์ผ๋ก ๋ค์ ๋ก๋ํ ์ง ์ฌ๋ถ | False |
- ์ฃผ์ ์ต์ : directory(๊ฒฝ๋ก), encoding(์ธ์ฝ๋ฉ), auto_reload(์๋ ์ฌ๋ก๋ฉ)
- templates ๋๋ ํฐ๋ฆฌ ๋ด์ HTML ํ์ผ์์ ๋ฐ์ดํฐ๋ฅผ ๋ฐ์ ๋์ ์ผ๋ก ํ์ด์ง๋ฅผ ์์ฑํด ๋ฐํ
# ๊ฒฝ๋ก ๋งค๊ฐ๋ณ์
@app.get("/user/{username}")
def root(request: Request,username:str):
return templates.TemplateResponse("index.html", {"request": request, "username":username})
# ์ฟผ๋ฆฌ ๋งค๊ฐ๋ณ์
# /user?username=ํ๊ธธ๋ → username์ด "ํ๊ธธ๋"์ผ๋ก ํ
ํ๋ฆฟ์ ์ ๋ฌ๋จ
@app.get("/user")
def root(request: Request, username: str):
return templates.TemplateResponse("index.html", {"request": request, "username": username})
1. ํํฐ: ๋ณ์์ ํ์ดํ(|) ๊ธฐํธ๋ฅผ ์ฌ์ฉํด ๊ฐ์ ๋ณํ
{{ username|lower }} {# ์๋ฌธ์๋ก ๋ณํ #}
{{ price|round(2) }} {# ์์์ ๋์งธ์๋ฆฌ๊น์ง ๋ฐ์ฌ๋ฆผ #}2. ๋งคํฌ๋ก: ๋ฐ๋ณต๋๋ ํ ํ๋ฆฟ ์ฝ๋๋ฅผ ํจ์์ฒ๋ผ ์ฌ์ฌ์ฉ
<!-- macros.html ๋งคํฌ๋ก ์ ์ธ -->
{% macro print_name(name) %}
<p>์ด๋ฆ์ {{ name }} ์
๋๋ค.</p>
{% endmacro %}
<!-- index.html ๋งคํฌ๋ก ํธ์ถ -->
{% from "macros.html" import print_name %}
{{ print_name("ํ๊ธธ๋") }}
{{ print_name("๊น์ฒ ์") }}3. block: ํ ํ๋ฆฟ ์์์์ ์์ ํ ํ๋ฆฟ์ด ๋ด์ฉ์ ๋ฎ์ด์ธ ์ ์๋ ์์ญ์ ์ง์ -> ํค๋, ํธํฐ๋ฅผ ๊ณตํต ์ฌ์ฉ(์ ๋ฒ ๋งํด๋ฐฉ ํ๋ก์ ํธ์์ ์์ฃผ ์ ์ฉํ๊ฒ ์ฌ์ฉํ๋ค.)
{# base.html #}
<body>
{% block content %}{% endblock %} <!-- ์์๋ฐ๋ ํ
ํ๋ฆฟ์์ ๋ด์ฉ ์ถ๊ฐ-->
</body>
{# index.html #}
{% extends "base.html" %}
{% block content %}
<h1>์ฌ๊ธฐ์ ๋ด์ฉ ์์ฑ</h1>
{% endblock %}4. import: ๋ค๋ฅธ ํ ํ๋ฆฟ ํ์ผ์ ๋งคํฌ๋ก๋ ๋ณ์๋ฅผ ๊ฐ์ ธ์ด
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
app = FastAPI()
templates = Jinja2Templates(directory="templates")
@app.get("/")
def read_root(request: Request):
return templates.TemplateResponse("index.html", {"request": request})index.html
{% import "macros.html" as macros %}
<h2>import๋ก ๋งคํฌ๋ก ์ฌ์ฉ ์์</h2>
{{ macros.print_name("ํ๊ธธ๋") }}
{{ macros.print_name("๊น์ฒ ์") }}macros.html
{% macro print_name(name) %}
<p>์ด๋ฆ์ {{ name }} ์
๋๋ค.</p>
{% endmacro %}5. set: ํ ํ๋ฆฟ ๋ด์์ ๋ณ์๋ฅผ ์ ์ธํ๊ฑฐ๋ ๊ฐ์ ์ ์ฅํ ๋ ์ฌ์ฉ
{% set total = price * quantity %}
์ดํฉ: {{ total }}6. do: ํํ์์ ๊ฒฐ๊ณผ๋ฅผ ์ถ๋ ฅํ์ง ์๊ณ , ๋ถ์ ํจ๊ณผ๋ง ๋ฐ์์ํค๊ณ ์ถ์ ๋ ์ฌ์ฉ
{% do my_list.append('item') %}7. with: ๋ธ๋ก ๋ด๋ถ์์๋ง ์ฌ์ฉํ ์์ ๋ณ์ ๋ฒ์๋ฅผ ์ง์ ํ ๋ ์ฌ์ฉ
{% with total=price*quantity %}
์ดํฉ: {{ total }}
{% endwith %}
์ ์ ํ์ผ
์ ์ ํ์ผ(static files)์ด๋?
- ์๋ฒ๊ฐ ๋ณ๋์ ์ฒ๋ฆฌ ์์ด ๊ทธ๋๋ก ํด๋ผ์ด์ธํธ์๊ฒ ์ ๋ฌํ๋ ํ์ผ
- FastAPI์์๋ starlette์ StaticFiles ํด๋์ค๋ฅผ ์ด์ฉํด ์ ์ ํ์ผ์ ์ฝ๊ฒ ์๋น์คํ ์ ์๋ค.
- FastAPI๋ Starlette์ StaticFiles๋ฅผ ๊ทธ๋๋ก ๊ฐ์ ธ์์, ๊ฐ๋ฐ์๊ฐ ๋ ํธํ๊ฒ ์ธ ์ ์๊ฒ fastapi.staticfiles๋ก ํ ๋ฒ ๋ ๊ฐ์ผ ๊ฒ๋ฟ์ด๋ค.
- ์ฆ, ๋ ์ค ์ด๋ ๊ฒ์ ์จ๋ ๋์๊ณผ ๊ธฐ๋ฅ, ๋ด๋ถ ๊ตฌํ์ด ์์ ํ ๋์ผ
- from fastapi.staticfiles import StaticFiles
- from starlette.staticfiles import StaticFiles
from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
app = FastAPI()
# 'static' ํด๋๋ฅผ '/static' ๊ฒฝ๋ก๋ก ์๋น์ค
app.mount("/static", StaticFiles(directory="static"), name="static")
templates = Jinja2Templates(directory="templates")
@app.get("/")
def read_root(request: Request):
return templates.TemplateResponse("index.html", {"request": request})<!DOCTYPE html>
<html>
<head>
<title>์ ์ ํ์ผ </title>
</head>
<body>
<h1>์ ์ ํ์ผ(static files) ์ฌ์ฉ </h1>
<img src="/static/test.png" alt="์ด๋ฏธ์ง">
</body>
</html>
APIRouter
APIRouter๋?
- APIRouter๋ FastAPI์์ ์ฌ๋ฌ ์๋ํฌ์ธํธ๋ฅผ ๋ ผ๋ฆฌ์ ์ผ๋ก ๋ฌถ์ด์ ๊ด๋ฆฌํ ์ ์๊ฒ ํด์ฃผ๋ ํด๋์ค
- ๋๊ท๋ชจ ํ๋ก์ ํธ์์ ๋ผ์ฐํฐ๋ฅผ ํ์ผ๋ณ/๊ธฐ๋ฅ๋ณ๋ก ๋ถ๋ฆฌํด app.include_router()๋ก ๋ฉ์ธ ์ฑ์ ๋ฑ๋กํ๋ค.
- ์ด๋ฅผ ํตํด ๋ผ์ฐํธ ์์ค์์ ์์กด์ฑ๊ณผ ๋ฏธ๋ค์จ์ด๋ฅผ ์ ์ฉํ ์ ์๋ค.
- ๋ฏธ๋ค์จ์ด: ํน์ ๊ฒฝ๋ก ์๋์ ์ํด ์ฒ๋ฆฌ๋๊ธฐ ์ , ๋ชจ๋ ์์ฒญ์ ๋ํด์ ๋์ํ๋ ํจ์. ๋ํ ๋ชจ๋ ์๋ต์ด ๋ฐํ๋๊ธฐ ์ ์๋ ๋์ผํ๊ฒ ๋์ํ๋ค.
- ๋ฏธ๋ค์จ์ด๋ ์์ฉ ํ๋ก๊ทธ๋จ์ผ๋ก ์ค๋ ์์ฒญ์ ๊ฐ์ ธ์จ๋ค.
- ์์ฒญ์ ์์ฉ ํ๋ก๊ทธ๋จ์ ๊ฒฝ๋ก ์๋์ผ๋ก ์ ๋ฌํ์ฌ ์ฒ๋ฆฌ
- ์๋ต์ ๋ฐํ
- ๋ฏธ๋ค์จ์ด: ํน์ ๊ฒฝ๋ก ์๋์ ์ํด ์ฒ๋ฆฌ๋๊ธฐ ์ , ๋ชจ๋ ์์ฒญ์ ๋ํด์ ๋์ํ๋ ํจ์. ๋ํ ๋ชจ๋ ์๋ต์ด ๋ฐํ๋๊ธฐ ์ ์๋ ๋์ผํ๊ฒ ๋์ํ๋ค.
- APIRouter์ ๋ฏธ๋ค์จ์ด
- FastAPI์ ๋ฏธ๋ค์จ์ด๋ ์ฑ ์ ์ฒด์ ์ ์ฉ๋๋ฉฐ, ๋ผ์ฐํฐ๋ณ๋ก ์ง์ ๋ฏธ๋ค์จ์ด๋ฅผ ๋ค๋ฅด๊ฒ ์ ์ฉํ ์๋ ์๋ค.
- ํ์ง๋ง APIRouter์ dependencies ์ต์ ์ ์ฌ์ฉํ๋ฉด, ๋ผ์ฐํฐ ๋จ์๋ก ๊ณตํต ๋ก์ง(์: ์ธ์ฆ, ๊ถํ ์ฒดํฌ ๋ฑ)์ "์์กด์ฑ ํจ์"๋ก ์ ์ฉํ ์ ์๋ค.
- ์์กด์ฑ ํจ์๋ ๋ฏธ๋ค์จ์ด์ ๋น์ทํ๊ฒ ๋์ํ์ง๋ง, ๋ผ์ฐํฐ๋ณ/์๋ํฌ์ธํธ๋ณ๋ก ์ธ๋ฐํ๊ฒ ์ ์ฉํ ์ ์๋ค๋ ์ ์ด ๋ค๋ฅด๋ค.
from fastapi import FastAPI, APIRouter
app = FastAPI()
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/")
async def read_users():
return [{"username": "Rick"}, {"username": "Morty"}]
app.include_router(router)- prefix="/users": ๋ชจ๋ ๊ฒฝ๋ก์ /users๊ฐ ์ ๋์ฌ๋ก ๋ถ์ (/users/)
- tags=["users"]: API ๋ฌธ์์์ ๊ทธ๋ฃนํ์ ์ฌ์ฉ
items_router = APIRouter(prefix="/items", tags=["items"])
@items_router.get("/")
async def read_items():
return [{"item_id": "Foo"}, {"item_id": "Bar"}]
app.include_router(router)
app.include_router(items_router)- ์ฌ๋ฌ ๋ผ์ฐํฐ๋ฅผ ๊ฐ๊ฐ ๋ค๋ฅธ ์ ๋์ฌ๋ก ๋ฑ๋ก ๊ฐ๋ฅ
- ๋์ผ ๋ผ์ฐํฐ๋ฅผ ์ฌ๋ฌ ์ ๋์ฌ๋ก ์ค๋ณต ๋ฑ๋กํด API ๋ฒ์ ๊ด๋ฆฌ๋ ๊ฐ๋ฅ
1. ๊ธฐ๋ณธ FastAPI ๋ผ์ฐํฐ
from fastapi import FastAPI
app = FastAPI()
@app.get("/users")
def get_users():
return [{"name": "Alice"}]
@app.get("/items")
def get_items():
return [{"item": "Book"}]- ๋ชจ๋ ์๋ํฌ์ธํธ๊ฐ ํ ํ์ผ์ ๋ชจ์ฌ ์์
- ํ๋ก์ ํธ๊ฐ ์ปค์ง์๋ก ์ฝ๋๊ฐ ๋ณต์กํด์ง๊ณ ๊ด๋ฆฌ๊ฐ ์ด๋ ค์
2. APIRouter ์ฌ์ฉ
# user_router.py
from fastapi import APIRouter
router = APIRouter()
@router.get("/")
def get_users():
return [{"name": "Alice"}]# items_router.py
from fastapi import APIRouter
router = APIRouter()
@router.get("/")
def get_items():
return [{"item": "Book"}]# main.py
from fastapi import FastAPI
from users_router import router as users_router
from items_router import router as items_router
app = FastAPI()
app.include_router(users_router, prefix="/users")
app.include_router(items_router, prefix="/items")- ๊ฐ ๊ธฐ๋ฅ๋ณ๋ก ํ์ผ์ ๋๋ ์ ๊ด๋ฆฌ
- ๋ฉ์ธ ์ฑ์ include_router๋ก ๋ฑ๋ก
- ๋ผ์ฐํฐ๋ณ๋ก prefix, ํ๊ทธ, ๋ฏธ๋ค์จ์ด ๋ฑ ๋ณ๋ ์ค์ ๊ฐ๋ฅ
- ๊ท๋ชจ๊ฐ ์ปค์ง๋ฉด: APIRouter๋ก ๊ธฐ๋ฅ๋ณ๋ก ๋ถ๋ฆฌ·๋ชจ๋ํํด์ผ ์ ์ง๋ณด์์ ํ์ฅ์ฑ์ด ํจ์ฌ ์ข์์ง
1. @app.~ ๋ฐ์ฝ๋ ์ดํฐ
@app.get("/hello")
def hello():
return {"msg": "hello"}- ์ค๋ช : FastAPI ์ ํ๋ฆฌ์ผ์ด์ (app)์ ์ง์ ์๋ํฌ์ธํธ๋ฅผ ๋ฑ๋กํ๋ ๋ฐฉ์
- ์ฉ์ด: "์ ํ๋ฆฌ์ผ์ด์ ๋ผ์ฐํฐ", "๋ฉ์ธ ๋ผ์ฐํฐ" ๋ผ ํ๋ค.
2. @router.~ ๋ฐ์ฝ๋ ์ดํฐ
from fastapi import APIRouter
router = APIRouter()
@router.get("/foo")
def foo():
return {"msg": "foo"}- ์ค๋ช : APIRouter ๊ฐ์ฒด์ ์๋ํฌ์ธํธ๋ฅผ ๋ฑ๋กํ๋ ๋ฐฉ์
- ์ฉ์ด: "๋ฏธ๋ค์จ์ด ๋ผ์ฐํฐ", "๋ผ์ฐํฐ ๊ฐ์ฒด", "์๋ธ ๋ผ์ฐํฐ", "๊ธฐ๋ฅ๋ณ ๋ผ์ฐํฐ" ๋ผ ํ๋ค.
from fastapi import FastAPI, APIRouter
app = FastAPI()
router = APIRouter()
@app.get("/main")
def main_route():
return {"msg": "main route"}
@router.get("/sub")
def sub_route():
return {"msg": "sub route"}
app.include_router(router, prefix="/api")
3. ๋ฏธ๋ค์จ์ด ์ค์
- FastAPI๋ ๋ผ์ฐํฐ ๋จ์๋ก ๋ฏธ๋ค์จ์ด๋ฅผ ์ง์ ์ถ๊ฐํ ์ ์๋ค.
- ๋ฏธ๋ค์จ์ด๋ FastAPI ์ธ์คํด์ค(app)์ ์ถ๊ฐํด์ผ ํ๋ฉฐ, ์ด ์ธ์คํด์ค์ ์ฌ๋ฌ APIRouter๋ฅผ ํฌํจ์ํจ๋ค.
from fastapi.middleware.trustedhost import TrustedHostMiddleware
app.add_middleware(
TrustedHostMiddleware, allowed_hosts=["localhost", "example.com"]
)- TrustedHostMiddleware๋ ์์ฒญ์ Host ํค๋๊ฐ ์ง์ ๋ ๋๋ฉ์ธ/์๋ธ๋๋ฉ์ธ์ผ ๋๋ง ํ์ฉํ๋ ๋ณด์ ๊ธฐ๋ฅ
4. ์์กด์ฑ ํจ์(Depends)์ APIRouter
- ์์กด์ฑ ํจ์๋ ๋ผ์ฐํฐ ์ ์ฒด ๋๋ ํน์ ์๋ํฌ์ธํธ๊ฐ ์คํ๋๊ธฐ ์ ์ ๊ณตํต ๋ก์ง์ ์คํํ๋ ํจ์
from fastapi import FastAPI, APIRouter, Depends, HTTPException, Header
# 1. ํ ํฐ ์ฒดํฌ ํจ์
def token_check(authorization: str = Header(None)):
if authorization != "Bearer mysecrettoken":
raise HTTPException(status_code=401, detail="Invalid or missing token")
# 2. /api ์ ์ฉ ๋ผ์ฐํฐ ์์ฑ, ์์กด์ฑ ๋ถ์ฐฉ
api_router = APIRouter(
prefix="/api",
dependencies=[Depends(token_check)]
)
@api_router.get("/hello")
def hello():
return {"message": "API ํ ํฐ ์ธ์ฆ ์ฑ๊ณต!"}
# 3. ์ผ๋ฐ ๋ผ์ฐํฐ (ํ ํฐ ์ฒดํฌ ์์)
public_router = APIRouter()
@public_router.get("/public")
def public():
return {"message": "๋๊ตฌ๋ ์ ๊ทผ ๊ฐ๋ฅ"}
# 4. ์ฑ์ ๋ผ์ฐํฐ ๋ฑ๋ก
app = FastAPI()
app.include_router(api_router) # /api/* ๊ฒฝ๋ก๋ง ํ ํฐ ์ฒดํฌ
app.include_router(public_router) # /public ๊ฒฝ๋ก๋ ํ ํฐ ์ฒดํฌ ์์- token_check ํจ์: ์์ฒญ ํค๋์ Authorization ๊ฐ์ด "Bearer mysecrettoken"์ด ์๋๋ฉด 401 ์๋ฌ ๋ฐ์
- dependencies=[Depends(token_check)]: ์ด ๋ผ์ฐํฐ์ ์๋ํฌ์ธํธ์์ token_check๊ฐ ์๋์ผ๋ก ์คํ๋จ
- /public → ํ ํฐ ์์ด ๋๊ตฌ๋ ์ ๊ทผ ๊ฐ๋ฅ
- /api/hello → ํ ํฐ์ด ๋ง์์ผ ์ ๊ทผ ๊ฐ๋ฅ
- ๋ผ์ฐํฐ์ ๋ฑ๋กํ๋ฉด ํด๋น ๋ผ์ฐํฐ์ ๋ชจ๋ ์๋ํฌ์ธํธ์์ ์๋ ์ ์ฉ๋๋ค.
- ์์กด์ฑ ์ฃผ์
(Dependency Injection, DI)
- ์ฝ๋๊ฐ ๋์ํ๋ ๋ฐ ํ์ํ ์ธ๋ถ ์์์ ์ง์ ์์ฑํ๊ฑฐ๋ ๊ด๋ฆฌํ์ง ์๊ณ , ์ธ๋ถ์์(ํ๋ ์์ํฌ๊ฐ) ๋์ ๋ง๋ค์ด์ "์ฃผ์ "ํด์ฃผ๋ ํ๋ก๊ทธ๋๋ฐ ๊ธฐ๋ฒ
- ์์กด์ฑ: ์ด๋ค ํจ์๋ ํด๋์ค๊ฐ ๋์ํ๋ ค๋ฉด ํ์ํ ๊ฒ(์: DB ์ฐ๊ฒฐ, ์ธ์ฆ ์ ๋ณด, ๊ณตํต ํ๋ผ๋ฏธํฐ ๋ฑ)
- ์ฃผ์ : ๊ทธ ํ์ํ ๊ฒ์ ์ง์ ๋ง๋ค์ง ์๊ณ , ์ธ๋ถ์์(์ฃผ๋ก ํ๋ ์์ํฌ๊ฐ) ๋์ ๋ฃ์ด์ฃผ๋ ๊ฒ
- ์ ์ฐ๋๊ฐ?
- ์ฝ๋์ ์ฌ์ฌ์ฉ์ฑ, ์ ์ง๋ณด์์ฑ, ํ ์คํธ ์ฉ์ด์ฑ์ด ํฌ๊ฒ ํฅ์๋จ
- ์ฌ๋ฌ ์๋ํฌ์ธํธ์์ ๋ฐ๋ณต๋๋ ๋ก์ง(์: DB ์ฐ๊ฒฐ, ์ธ์ฆ, ๊ณตํต ํ๋ผ๋ฏธํฐ ๋ฑ)์ ํ ๋ฒ์ ๊ด๋ฆฌ ๊ฐ๋ฅ
- ๊ฐ ํจ์๊ฐ "๋ฌด์์ด ํ์ํ์ง"๋ง ์ ์ธํ๋ฉด, ์ค์ ๋ก ์ด๋ป๊ฒ ๊ฐ์ ธ์ค๋์ง๋ ํ๋ ์์ํฌ๊ฐ ์ฒ๋ฆฌ
1. Depends๋?
- FastAPI์์ ์์กด์ฑ ์ฃผ์ (Dependency Injection)์ ์ํ ํจ์
- ์ธ์ฆ, DB ์ฐ๊ฒฐ, ๊ณตํต ํ๋ผ๋ฏธํฐ ๊ฒ์ฆ ๋ฑ "์ฌ๋ฌ ์๋ํฌ์ธํธ์์ ๋ฐ๋ณต๋๋ ๊ณตํต ๋ก์ง"์ ํ ๋ฒ์ ์ฒ๋ฆฌํ ์ ์๋ค.
- ์๋ํฌ์ธํธ ํจ์์ ํ๋ผ๋ฏธํฐ, ๋ผ์ฐํฐ, ์ฑ ์ ์ฒด์ ๋ถ์ผ ์ ์๋ค.
2. APIRouter์ Depends ์ ์ฉ
router = APIRouter(dependencies=[Depends(ํจ์)])์ด๋ ๊ฒ ํ๋ฉด, ํด๋น ๋ผ์ฐํฐ์ ๋ฑ๋ก๋ ๋ชจ๋ ์๋ํฌ์ธํธ์์ ์ง์ ํ ์์กด์ฑ ํจ์๊ฐ ์๋์ผ๋ก ์คํ๋๋ค.
5. ๋ผ์ฐํธ ์ค์ ์์
- ์์ ๋ผ์ฐํฐ(APIRouter)์ ์ค์ ํ prefix, tags, dependencies ๋ฑ์ ํ์ ๋ผ์ฐํฐ/์๋ํฌ์ธํธ์ ์์๋๋ค.
- ์ฌ๋ฌ APIRouter๋ฅผ ๊ณ์ธต์ ์ผ๋ก ๊ตฌ์ฑํ ์ ์๋ค.
from fastapi import FastAPI, APIRouter, Depends
def auth_dependency():
print("์ธ์ฆ ์ฒดํฌ ์คํ!")
# ํ์ ๋ผ์ฐํฐ ์ ์
sub_router = APIRouter()
@sub_router.get("/profile")
def get_profile():
return {"msg": "ํ๋กํ ์ ๋ณด"}
# ์์ ๋ผ์ฐํฐ ์ ์ (prefix, tags, dependencies ์ค์ )
parent_router = APIRouter(
prefix="/users",
tags=["users"],
dependencies=[Depends(auth_dependency)]
)
# ํ์ ๋ผ์ฐํฐ๋ฅผ ์์ ๋ผ์ฐํฐ์ ํฌํจ (include_router)
parent_router.include_router(sub_router, prefix="/me", tags=["private"])
# FastAPI ์ฑ์ ์์ ๋ผ์ฐํฐ๋ฅผ ํฌํจ
app = FastAPI()
app.include_router(parent_router)- ์์ ๋ผ์ฐํฐ(parent_router)
- prefix: /users
- tags: ["users"]
- dependencies: [Depends(auth_dependency)] (๋ชจ๋ ํ์ ๋ผ์ฐํธ์ ์ธ์ฆ ์ฒดํฌ ์ ์ฉ)
- ํ์ ๋ผ์ฐํฐ(sub_router)
- /profile ์๋ํฌ์ธํธ๋ง ์ ์
- ๋ผ์ฐํฐ ๊ณ์ธต ๊ตฌ์กฐ
- /users/me/profile
- tags: ["users", "private"]
- dependencies: auth_dependency๊ฐ ์๋ ์คํ๋จ
- /users/me/profile
- /users/me/profile๋ก ์์ฒญ ์:auth_dependency()๊ฐ ๋จผ์ ์คํ๋จ (์์กด์ฑ ์์)์ค์ ์๋ํฌ์ธํธ ํจ์๊ฐ ์คํ๋จSwagger UI์์ tags๋ ์์/๋ณํฉ๋จ
๋ฐฑ๊ทธ๋ผ์ด๋ ํ์คํฌ
BackgroundTasks๋?
- FastAPI ๋ด์ฅ ํด๋์ค
- ๋น๋๊ธฐ ์ฒ๋ฆฌ๋ก, ์๋ต์ ๋จผ์ ๋ฐํํ๊ณ ๋ณ๋์ ์ค๋ ๋์์ ์์ ์ ์คํ
- HTTP ์๋ต์ ๋จผ์ ๋ณด๋ด๊ณ ์๋ฒ์์ ๊ณ์ ์์ ์ ์คํํ ์ ์๊ฒ ํด์ค
- ์์: ์ด๋ฉ์ผ ๋ฐ์ก, ๋ก๊ทธ ๊ธฐ๋ก ๋ฑ ์๊ฐ์ด ์ค๋ ๊ฑธ๋ฆฌ์ง๋ง ์ฆ์ ์๋ต์ด ํ์ํ ์์ ์ ์ ํฉ
์ฅ์
- ๋น๋๊ธฐ ์ฒ๋ฆฌ๋ก ์ฌ์ฉ์ ๋๊ธฐ ์๊ฐ ์ต์ํ
- ๋ฆฌ์์ค ์ต์ ํ: ์๋ฒ๊ฐ ๊ธด ์์ ์ ๋ฌถ์ด์ง ์์
- ๊ฐ๋จํ ์์ ์ Celery ๊ฐ์ ์ธ๋ถ ํ ์์ด ์ฒ๋ฆฌ ๊ฐ๋ฅ
์ฌ์ฉ๋ฒ ๋ฐ ์ธ์ ์ ๋ฌ
1. ํจ์๋ง ์ ๋ฌ
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
def simple_task():
print("๋จ์ ๋ฐฑ๊ทธ๋ผ์ด๋ ์์
์คํ")
@app.get("/only-func")
async def only_func(background_tasks: BackgroundTasks):
background_tasks.add_task(simple_task)
return {"message": "ํจ์๋ง ์ ๋ฌ"}2. ์์น๊ธฐ๋ฐ ์ธ์ ์ ๋ฌ
def task_with_args(a, b):
print(f"a: {a}, b: {b}")
@app.get("/args")
async def with_args(background_tasks: BackgroundTasks):
background_tasks.add_task(task_with_args, 10, 20)
return {"message": "์์น ์ธ์ ์ ๋ฌ"}3. ํค์๋ ์ธ์ ์ ๋ฌ
def task_with_kwargs(a, b=0):
print(f"a: {a}, b: {b}")
@app.get("/kwargs")
async def with_kwargs(background_tasks: BackgroundTasks):
background_tasks.add_task(task_with_kwargs, a=5, b=15)
return {"message": "ํค์๋ ์ธ์ ์ ๋ฌ"}4. ํจ์+์์น+ํค์๋ ์ธ์ ํผํฉ ์ ๋ฌ
def mixed_task(a, b, c=0):
print(f"a: {a}, b: {b}, c: {c}")
@app.get("/mixed")
async def mixed(background_tasks: BackgroundTasks):
background_tasks.add_task(mixed_task, 1, 2, c=3)
return {"message": "ํผํฉ ์ธ์ ์ ๋ฌ"}
- BackgroundTasks๋ก ๋น๋๊ธฐ ๋ฐฑ๊ทธ๋ผ์ด๋ ์์ ์ฝ๊ฒ ์ฒ๋ฆฌ
- add_task(func, *args, **kwargs)๋ก ๋ค์ํ ์ธ์ ์ ๋ฌ ๊ฐ๋ฅ
- ์๋ต๊ณผ ๋์์ ๋ณ๋ ์ค๋ ๋์์ ์์ ์คํ, ์๋ฒ ๋ฆฌ์์ค ํจ์จ์ ์ฌ์ฉ
์คํธ๋ฆฌ๋ฐ ์๋ต
์คํธ๋ฆฌ๋ฐ ์๋ต์ด๋?
- ์๋ฒ๊ฐ ๋ฐ์ดํฐ๋ฅผ ํ ๋ฒ์ ๋ชจ๋ ๋ณด๋ด๋ ๋์ , ๋ฐ์ดํฐ๋ฅผ ์ผ์ ๋จ์(์ฒญํฌ, ํ, ํ๋ ์)๋ก ๋๋์ด ์์ฐจ์ ์ผ๋ก ํด๋ผ์ด์ธํธ์ ์ ์กํ๋ ๋ฐฉ์
์ ์คํธ๋ฆฌ๋ฐ ์๋ต์ ์ฐ๋?
- ๋ฉ๋ชจ๋ฆฌ ์ ์ฝ: ๋ฐ์ดํฐ๋ฅผ ํ ๋ฒ์ ๋ชจ๋ ๋ฉ๋ชจ๋ฆฌ์ ์ฌ๋ฆฌ์ง ์๊ณ , ์์ฑ๋๋ ๋๋ก ๋ฐ๋ก๋ฐ๋ก ์ ์ก
- ๋ ์ดํด์ ๊ฐ์: ์ผ๋ถ ๋ฐ์ดํฐ๊ฐ ์ค๋น๋๋ ์ฆ์ ๋จผ์ ์ ์ก๋์ด, ์ฌ์ฉ์๋ ๋น ๋ฅด๊ฒ ๊ฒฐ๊ณผ๋ฅผ ๋ณผ ์ ์์
- ๋์ฉ๋ ์ฒ๋ฆฌ: ์์ญ~์๋ฐฑ MB ์ด์์ ํ์ผ, ์ค์๊ฐ ๋ฐ์ดํฐ ๋ฑ๋ ๋ฌด๋ฆฌ ์์ด ์ฒ๋ฆฌ
StreamingResponse์ ๋์ ์๋ฆฌ
- StreamingResponse๋ FastAPI๊ฐ ์ ๊ณตํ๋ ํน์ ์๋ต ํด๋์ค
- ๋ด๋ถ์ ์ผ๋ก ์ดํฐ๋ฌ๋ธ(์ ๋๋ ์ดํฐ, ํ์ผ ๊ฐ์ฒด ๋ฑ)์ ๋ฐ์, yield๋ก ๋ฐํ๋ ๋ฐ์ดํฐ ์กฐ๊ฐ์ ํด๋ผ์ด์ธํธ์ ์์ฐจ์ ์ผ๋ก ๋ด๋ณด๋ธ๋ค.
- yield๋ ํจ์ ์คํ์ ์ผ์ ์ค์งํ๊ณ , ๊ฐ์ ํด๋ผ์ด์ธํธ์ ์ ๋ฌํ ๋ค ๋ค์ ํธ์ถ ์ ๊ทธ ๋ค์ ์ค๋ถํฐ ๊ณ์ ์คํํ๋ค.
- media_type(์: "text/csv", "application/json", "video/mp4" ๋ฑ)๊ณผ headers๋ฅผ ์ง์ ํด, ํด๋ผ์ด์ธํธ๊ฐ ๋ฐ์ดํฐ ํ์๊ณผ ํ์ผ๋ช ์ ์ฌ๋ฐ๋ฅด๊ฒ ์ธ์ํ๋๋ก ํ ์ ์๋ค.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import csv
import io
app = FastAPI()
def csv_row_generator():
# ํค๋
output = io.StringIO()
writer = csv.writer(output)
writer.writerow(["id", "name", "score"])
yield output.getvalue()
output.seek(0)
output.truncate(0)
# ๋ฐ์ดํฐ ํ์ ๋ฐ๋ณต์ ์ผ๋ก ์์ฑ
for i in range(1, 101):
writer.writerow([i, f"user_{i}", i * 10])
yield output.getvalue()
output.seek(0)
output.truncate(0)
@app.get("/download/csv")
def download_csv():
headers = {
"Content-Disposition": 'attachment; filename="data.csv"'
}
return StreamingResponse(
csv_row_generator(),
media_type="text/csv",
headers=headers
)- csv_row_generator
- io.StringIO ๊ฐ์ฒด์ csv.writer๋ก ํ ์ค์ฉ ๋ฐ์ดํฐ๋ฅผ ์
- ๊ฐ ํ์ด ์์ฑ๋ ๋๋ง๋ค yield๋ก ํ์ฌ๊น์ง์ ๋ด์ฉ์ ๋ฐํ
- yield ์ดํ StringIO๋ฅผ ๋น์์ ๋ค์ ํ ์ค๋น
- StreamingResponse
- generator๋ฅผ ๋ฐ์์, ํด๋ผ์ด์ธํธ์ ํ ์ค์ฉ(ํน์ ์ฒญํฌ ๋จ์๋ก) ์ ์ก
- media_type๊ณผ headers๋ก ํ์ผ ํ์๊ณผ ํ์ผ๋ช ์ง์
- ํด๋ผ์ด์ธํธ UX
- ์ฌ์ฉ์๋ ๋ค์ด๋ก๋๊ฐ ์์๋์๋ง์ ์ผ๋ถ ๋ฐ์ดํฐ๋ถํฐ ๋ฐ๋ก ๋ฐ๊ธฐ ์์
- ๋์ฉ๋ CSV๋ ๋ฉ๋ชจ๋ฆฌ ๋ถ์กฑ ์์ด ์ฒ๋ฆฌ ๊ฐ๋ฅ
์น ์์ผ
์น์์ผ์ด๋?
- ์ค์๊ฐ, ์๋ฐฉํฅ ํต์ ํ๋กํ ์ฝ
- ์๋ฐฉํฅ ํต์ ์ด๋ ํด๋ผ์ด์ธํธ๊ฐ ์๋ฒ๋ก ๋ฉ์์ง๋ฅผ ๋ณด๋ผ ์ ์๊ณ ์๋ฒ๋ ํด๋ผ์ด์ธํธ๋ก ๋ฉ์์ง๋ฅผ ๋ณด๋ผ ์ ์์์ ์๋ฏธํ๋ค.
- ๋จ์ผ ์ฐฝ๋ง ์ด์ด๋, ๋ธ๋ผ์ฐ์ ์์ ์๋ฒ๋ก ๋ฉ์์ง๋ฅผ ๋ณด๋ผ ์ ์๊ณ ์๋ฒ๊ฐ ์๋ต ๋ฉ์์ง๋ฅผ ํด๋ผ์ด์ธํธ๋ก ๋ณด๋ผ ์ ์๋ค.
- ์ฆ, ๋จ์ผ ์ฐ๊ฒฐ์์๋ ํด๋ผ์ด์ธํธ์ ์๋ฒ๊ฐ ์๋ก ๋ฉ์์ง๋ฅผ ์ฃผ๊ณ ๋ฐ๋ ๊ฒ ์์ฒด๊ฐ ์น์์ผ์ "์๋ฐฉํฅ ํต์ "
- ์น์์ผ์ ํด๋ผ์ด์ธํธ(๋ธ๋ผ์ฐ์ ๋ฑ)์ ์๋ฒ๊ฐ ํ ๋ฒ ์ฐ๊ฒฐ๋๋ฉด, ๊ทธ ์ฐ๊ฒฐ์ ๊ณ์ ์ ์งํ๋ฉฐ ์๋ก ์์ ๋กญ๊ฒ ๋ฐ์ดํฐ๋ฅผ ์ฃผ๊ณ ๋ฐ์ ์ ์๋ ํ๋กํ ์ฝ
- HTTP์ ๋ฌ๋ฆฌ, ์๋ฒ๊ฐ ํด๋ผ์ด์ธํธ์๊ฒ ๋จผ์ ๋ฐ์ดํฐ๋ฅผ ๋ณด๋ผ ์๋ ์๊ณ , ํด๋ผ์ด์ธํธ๋ ์ธ์ ๋ ๋ฉ์์ง๋ฅผ ๋ณด๋ผ ์ ์๋ค.
- ์ ์ฐ๋๊ฐ?
- ์ค์๊ฐ ๋ฐ์ดํฐ ์ ์ก์ด ํ์ํ ๋(์ฑํ , ์ค์๊ฐ ์๋ฆผ, ๊ฒ์, ์ฃผ์ ์ ๋ณด ๋ฑ)
- HTTP๋ณด๋ค ํค๋๊ฐ ์๊ณ , ์ฐ๊ฒฐ์ ์ ์งํ๋ฏ๋ก ๋ฐ๋ณต์ ์ธ ์ฐ๊ฒฐ/ํด์ ์ ๋๋ ์ค๋ฒํค๋๊ฐ ์ค์ด๋ฆ(ํจ์จ์ฑ)
- ํธ๋์
ฐ์ดํฌ(Handshake) ๊ณผ์
- ์น์์ผ ์ฐ๊ฒฐ์ ์ฒ์์ HTTP๋ก ์์ํด์, ํน์ํ ํค๋(Connection: Upgrade, Upgrade: websocket ๋ฑ)๋ฅผ ํตํด ์น์์ผ์ผ๋ก ์ ๊ทธ๋ ์ด๋๋๋ค.
- ์๋ฒ๊ฐ 101 Switching Protocols๋ก ์๋ตํ๋ฉด, ๊ทธ ๋ค๋ก๋ ์น์์ผ ํ๋กํ ์ฝ๋ก ์ค์๊ฐ ํต์ ์ด ์์๋๋ค.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>WebSocket Test</title>
</head>
<body>
<h1>WebSocket Test</h1>
<div>
<button onclick="connectWS()">์ฐ๊ฒฐ</button>
<button onclick="disconnectWS()">์ฐ๊ฒฐ ํด์ </button>
<input type="text" id="msgInput" placeholder="๋ฉ์์ง ์
๋ ฅ">
<button onclick="sendMsg()">์ ์ก</button>
</div>
<div>
<span id="status">์ฐ๊ฒฐ ์ํ: ๋๊น</span>
</div>
<ul id="log"></ul>
<script>
let ws = null;
function connectWS() {
ws = new WebSocket("ws://localhost:8000/ws");
ws.onopen = () => document.getElementById('status').innerText = "์ฐ๊ฒฐ ์ํ: ์ฐ๊ฒฐ๋จ";
ws.onclose = () => document.getElementById('status').innerText = "์ฐ๊ฒฐ ์ํ: ๋๊น";
ws.onmessage = (event) => {
const log = document.getElementById('log');
const li = document.createElement('li');
li.innerText = event.data;
log.appendChild(li);
};
}
function disconnectWS() {
if (ws) ws.close();
}
function sendMsg() {
if (ws && ws.readyState === WebSocket.OPEN) {
const msg = document.getElementById('msgInput').value;
ws.send(msg);
}
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await websocket.send_text(f"๋ฐ์ ๋ฉ์์ง: {data}")
except WebSocketDisconnect:
pass- ์น์์ผ ๋ผ์ฐํธ ์์ฑ: @app.websocket("/ws") ๋ฐ์ฝ๋ ์ดํฐ๋ก ์น์์ผ ๋ผ์ฐํธ ์์ฑ
- ๋น๋๊ธฐ ํจ์ ์ ์: ์น์์ผ ์ฐ๊ฒฐ ์ฒ๋ฆฌ๋ ๋น๋๊ธฐ ํจ์๋ก ์์ฑํด์ผ ํ๋ค.
- ์ฐ๊ฒฐ ์๋ฆฝ: await websocket.accept()๋ก ์ฐ๊ฒฐ์ ๋น๋๊ธฐ์ ์ผ๋ก ์๋ฝ
- ๋ฐ์ดํฐ ์ก์์ ๋ฌดํ ๋ฃจํ์์ await websocket.receive_text()๋ก ๋ฉ์์ง๋ฅผ ๋ฐ๊ณ ,await websocket.send_text()๋ก ๋ฉ์์ง๋ฅผ ๋ณด๋ผ ์ ์๋ค.
- ์ฐ๊ฒฐ ์ข ๋ฃ ์ฒ๋ฆฌ: ํด๋ผ์ด์ธํธ๊ฐ ์ฐ๊ฒฐ์ ๋์ผ๋ฉด WebSocketDisconnect ์์ธ๊ฐ ๋ฐ์ํ๋ฉฐ, ์ด๋ฅผ ์ก์ ์ฒ๋ฆฌํ๋ค.
์ฐธ๊ณ
https://fastapi.tiangolo.com/advanced/middleware/
https://fastapi.tiangolo.com/tutorial/static-files/
์ง์์ ์ผ๋ก ์ ๋ฐ์ดํธ ์ค์ธ ๊ธ์ ๋๋ค.
'๐ผ ๋ฐฑ์ค๋ > Fast API' ์นดํ ๊ณ ๋ฆฌ์ ๋ค๋ฅธ ๊ธ
| Fast API - ๋ชจ๋ธ ์๋น (2) | 2025.06.25 |
|---|---|
| Fast API - ORM (0) | 2025.06.24 |
| Fast API - ์ฑ๋ฅ ๊ฐ์ (0) | 2025.06.23 |
| Fast API - ๊ธฐ์ด (3) | 2025.06.18 |
