어렵다 어려워
학습시간 09:00~03:00(당일18H/누적2012H)
실습
아래 기능을 갖는 API를 구현해 주세요.
1. root : 웰컴메시지
2. products/{product_id} : 제품 id로 정보를 조회합니다.
3. categories/{category_name}/products : 카테고리별 제품 목록을 조회합니다.
4. calculator/add : 두 숫자를 받아서 덧셈 결과를 반환합니다. a b / x y
from fastapi import FastAPI
app = FastAPI()
# 1. root : 웰컴메시지
@app.get("/")
def read_root():
return {"message": "Hello FastAPI!"}
# 2. products/{product_id} : 제품 id로 정보를 조회합니다.
@app.get("/products/{product_id}")
def read_product(product_id: int):
return {"product_id": product_id, "name": f"제품-{product_id}", "description": f"{product_id}번 제품의 상세 정보입니다."}
# 3. categories/{category_name}/products : 카테고리별 제품 목록을 조회합니다.
@app.get("/categories/{category_name}/products")
def read_category_products(category_name: str):
return {"category": category_name, "products": [f"{category_name}-상품A", f"{category_name}-상품B", f"{category_name}-상품C"]}
# 4. calculator/add : 두 숫자를 받아서 덧셈 결과를 반환합니다.
@app.get("/calculator/add")
def add_numbers(a: int, b: int):
result = a + b
return {"a": a, "b": b, "result": result}
root API (@app.get("/"))@app.get("/") 데코레이터로 루트 경로 GET 요청 처리read_root 함수를 실행하여 환영 메시지가 담긴 JSON 객체 반환@app.get("/products/{product_id}")){product_id} 사용product_id: int 로 타입 지정@app.get("/categories/{category_name}/products")){category_name} 사용read_category_products 함수의 category_name 인자로 URL 경로 값 전달@app.get("/calculator/add"))a와 b 사용/calculator/add?a=10&b=20a=10, b=20 값을 add_numbers 함수의 매개변수 a, b에 자동 매핑아래 기능을 갖는 API를 구현해 주세요.
1. Book 모델 (title, author, isbn, price, published_year, is_available=True)
2. Review 모델 (rating : 1-5, comment, reviewer_name)
3. BookWithReviews 모델 (Book + reviews 리스트)
4. POST /books : 책을 생성 (user created)
5. POST /books/{book_id}/reviews : {book_id}를 갖는 책에 리뷰를 추가
6. POST /books/complete : 책이랑 리뷰를 함께 생성
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field
from typing import List, Dict
app = FastAPI()
# --- Pydantic 모델 정의 ---
class Book(BaseModel):
title: str
author: str
isbn: str
price: float
published_year: int
is_available: bool = True
class Review(BaseModel):
rating: int = Field(ge=1, le=5) # 1 이상 5 이하의 값만 허용
comment: str
reviewer_name: str
class BookWithReviews(BaseModel):
book: Book
reviews: List[Review]
# --- 인메모리 데이터베이스 및 ID 카운터 ---
books_db: Dict[int, Book] = {}
reviews_db: Dict[int, List[Review]] = {} # key: book_id, value: list of reviews
book_id_counter = 0
# --- API 엔드포인트 구현 ---
# 4. POST /books : 책을 생성
@app.post("/books", status_code=status.HTTP_201_CREATED)
def create_book(book: Book):
global book_id_counter
book_id_counter += 1
new_book_id = book_id_counter
books_db[new_book_id] = book
reviews_db[new_book_id] = [] # 새 책에 대한 리뷰 리스트 초기화
return {"book_id": new_book_id, **book.model_dump()}
# 5. POST /books/{book_id}/reviews : 책에 리뷰를 추가
@app.post("/books/{book_id}/reviews", status_code=status.HTTP_201_CREATED)
def create_review_for_book(book_id: int, review: Review):
if book_id not in books_db:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Book with id {book_id} not found"
)
reviews_db[book_id].append(review)
return review
# 6. POST /books/complete : 책이랑 리뷰를 함께 생성
@app.post("/books/complete", status_code=status.HTTP_201_CREATED)
def create_book_with_reviews(payload: BookWithReviews):
global book_id_counter
book_id_counter += 1
new_book_id = book_id_counter
# 페이로드에서 책과 리뷰 정보 분리
book_data = payload.book
reviews_data = payload.reviews
# 책 정보 저장 및 리뷰 리스트 초기화
books_db[new_book_id] = book_data
reviews_db[new_book_id] = reviews_data
return {"book_id": new_book_id, **payload.model_dump()}
is_available 필드는 기본값으로 True를 가짐rating 필드는 Field를 사용하여 1에서 5 사이의 값만 받도록 유효성 검사 규칙을 추가book 필드는 Book 모델을, reviews 필드는 Review 모델의 리스트(List[Review])를 타입으로 가짐POST /books (책 생성 API)Book 모델 형식의 JSON 데이터를 받음book_id_counter를 사용해 새로운 책 ID를 생성books_db와 reviews_db에 새로운 책 정보와 빈 리뷰 리스트를 저장201 Created 상태 코드와 함께 생성된 책 ID와 정보를 반환POST /books/{book_id}/reviews (리뷰 추가 API){book_id}로 리뷰를 추가할 책을 특정Review 모델 형식의 JSON 데이터를 받음books_db에 해당 book_id가 존재하지 않을 경우, 404 Not Found 에러를 발생book_id가 존재하면, reviews_db의 해당 책 리뷰 리스트에 새로운 리뷰를 추가201 Created 상태 코드와 함께 생성된 리뷰 정보를 반환POST /books/complete (책과 리뷰 동시 생성 API)BookWithReviews를 요청 본문으로 받음201 Created 상태 코드와 함께 생성된 책 ID를 포함한 전체 데이터를 반환CRUD API를 구현해 주세요.
1. Task 모델 (title, description, completed=False, priority)
2. TaskUpdate 모델 (모든 필드는 optional)
3. POST /tasks : task create : 201 status code
4. GET /tasks : read all task
5. GET /tasks/{task_id} : read {task_id} task
6. PUT /tasks/{task_id} : {task_id} task 전체를 업데이트
7. PATCH /tasks/{task_id} : {task_id} task 부분 업데이트
8. DELETE /tasks/{task_id} : {task_id} task 삭제 : 204 status code
9. GET /tasks/completed : 완료된 작업만 조회
from fastapi import FastAPI, HTTPException, status, Response
from pydantic import BaseModel
from typing import List, Optional, Dict
app = FastAPI()
# --- Pydantic 모델 정의 ---
class Task(BaseModel):
title: str
description: Optional[str] = None
completed: bool = False
priority: int
class TaskUpdate(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
completed: Optional[bool] = None
priority: Optional[int] = None
# --- 인메모리 데이터베이스 및 ID 카운터 ---
tasks_db: Dict[int, Task] = {}
task_id_counter = 0
# --- API 엔드포인트 구현 ---
# 3. POST /tasks : task 생성
@app.post("/tasks", response_model=Task, status_code=status.HTTP_201_CREATED)
def create_task(task: Task):
global task_id_counter
task_id_counter += 1
tasks_db[task_id_counter] = task
return task
# 4. GET /tasks : 모든 task 조회
@app.get("/tasks", response_model=List[Task])
def read_all_tasks():
return list(tasks_db.values())
# 9. GET /tasks/completed : 완료된 task만 조회
# 중요: /tasks/{task_id} 보다 먼저 정의되어야 경로 충돌이 발생하지 않음
@app.get("/tasks/completed", response_model=List[Task])
def read_completed_tasks():
completed_tasks = [task for task in tasks_db.values() if task.completed]
return completed_tasks
# 5. GET /tasks/{task_id} : 특정 task 조회
@app.get("/tasks/{task_id}", response_model=Task)
def read_task(task_id: int):
if task_id not in tasks_db:
raise HTTPException(status_code=404, detail="Task not found")
return tasks_db[task_id]
# 6. PUT /tasks/{task_id} : task 전체 업데이트
@app.put("/tasks/{task_id}", response_model=Task)
def update_task_full(task_id: int, task: Task):
if task_id not in tasks_db:
raise HTTPException(status_code=404, detail="Task not found")
tasks_db[task_id] = task
return tasks_db[task_id]
# 7. PATCH /tasks/{task_id} : task 부분 업데이트
@app.patch("/tasks/{task_id}", response_model=Task)
def update_task_partial(task_id: int, task_update: TaskUpdate):
if task_id not in tasks_db:
raise HTTPException(status_code=404, detail="Task not found")
stored_task = tasks_db[task_id]
update_data = task_update.model_dump(exclude_unset=True)
updated_task = stored_task.model_copy(update=update_data)
tasks_db[task_id] = updated_task
return updated_task
# 8. DELETE /tasks/{task_id} : task 삭제
@app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_task(task_id: int):
if task_id not in tasks_db:
raise HTTPException(status_code=404, detail="Task not found")
del tasks_db[task_id]
return Response(status_code=status.HTTP_204_NO_CONTENT)
description은 선택 사항, completed는 기본값 False를 가짐Optional로 선언하여 클라이언트가 원하는 필드만 보낼 수 있도록 허용POST /tasks
Task 모델을 요청 본문으로 받아 새로운 Task를 생성201 Created 상태 코드 반환GET /tasks
tasks_db에 저장된 모든 Task의 목록을 리스트 형태로 반환GET /tasks/completed
completed 필드가 True인 Task들만 필터링하여 리스트로 반환/tasks/{task_id} 보다 먼저 선언해야 정상 동작GET /tasks/{task_id}
task_id에 해당하는 Task를 조회404 Not Found 에러 발생PUT /tasks/{task_id}
Task 모델 데이터로 완전히 교체404 Not Found 에러 발생PATCH /tasks/{task_id}
TaskUpdate 모델을 사용하여 변경하려는 필드만 선택적으로 받음model_dump(exclude_unset=True)를 통해 클라이언트가 보낸 데이터만 추출하여 업데이트DELETE /tasks/{task_id}
task_id를 가진 Task를 DB에서 삭제204 No Content 상태 코드를 반환# 학생 관리 API를 만들어주세요
1. Student 모델 : name, age, grade, subjects, gpa
- name : str, 1 ~ 50
- age : int, 5~100
- grade : str (학년) : "1", 2, 3, 4
- subjects : List[str] : 최소 1개, 최대 100개
- gpa : float, 0.0 ~ 4.0
2. 복합검색이 가능하도록 API를 만들어주세요 (나이범위, grade, subject)
from fastapi import FastAPI, HTTPException, status, Query
from pydantic import BaseModel, Field
from enum import Enum
from typing import List, Optional, Dict
app = FastAPI()
# --- Pydantic 모델 및 Enum 정의 ---
class GradeEnum(str, Enum):
GRADE_1 = "1"
GRADE_2 = "2"
GRADE_3 = "3"
GRADE_4 = "4"
class Student(BaseModel):
name: str = Field(min_length=1, max_length=50)
age: int = Field(ge=5, le=100)
grade: GradeEnum
subjects: List[str] = Field(min_length=1, max_length=100)
gpa: float = Field(ge=0.0, le=4.0)
# 응답 시 id를 포함하기 위한 모델
class StudentResponse(Student):
id: int
# --- 인메모리 데이터베이스 및 샘플 데이터 ---
students_db: Dict[int, Student] = {
1: Student(name="홍길동", age=17, grade="1", subjects=["수학", "과학"], gpa=3.5),
2: Student(name="이순신", age=18, grade="2", subjects=["역사", "국어"], gpa=3.8),
3: Student(name="유관순", age=17, grade="1", subjects=["역사", "영어", "과학"], gpa=3.9),
4: Student(name="강감찬", age=19, grade="3", subjects=["수학", "전략"], gpa=3.2),
}
student_id_counter = 4
# --- API 엔드포인트 구현 ---
@app.post("/students", response_model=StudentResponse, status_code=status.HTTP_201_CREATED)
def create_student(student: Student):
global student_id_counter
student_id_counter += 1
students_db[student_id_counter] = student
return StudentResponse(id=student_id_counter, **student.model_dump())
@app.get("/students", response_model=List[StudentResponse])
def search_students(
min_age: Optional[int] = Query(None, ge=5),
max_age: Optional[int] = Query(None, le=100),
grade: Optional[GradeEnum] = None,
subject: Optional[str] = Query(None, min_length=1)
):
results = []
for student_id, student in students_db.items():
# 나이 범위 필터링
if min_age is not None and student.age < min_age:
continue
if max_age is not None and student.age > max_age:
continue
# 학년 필터링
if grade is not None and student.grade != grade:
continue
# 과목 필터링
if subject is not None and subject not in student.subjects:
continue
results.append(StudentResponse(id=student_id, **student.model_dump()))
return results
GradeEnumEnum을 사용하여 grade 필드에 들어올 수 있는 값을 "1", "2", "3", "4"로 명확하게 제한StudentField를 사용하여 각 필드에 상세한 유효성 검사 규칙을 적용name: 최소 1자, 최대 50자age: 5 이상, 100 이하subjects: 최소 1개, 최대 100개의 요소를 가진 리스트gpa: 0.0 이상, 4.0 이하StudentResponseStudent 모델을 상속받아 DB에 저장된 후 부여되는 id 필드를 추가POST /students (학생 생성)
Student 모델 형식의 데이터를 받음StudentResponse 형태로 반환GET /students (학생 복합 검색)
min_age, max_age, grade, subject는 모두 Optional로 선언되어 선택적으로 사용 가능min_age, max_age 파라미터가 있으면 나이 범위 조건을 검사grade 파라미터가 있으면 학년 일치 여부를 검사subject 파라미터가 있으면, 해당 학생의 subjects 리스트에 과목이 포함되어 있는지 검사class Student(BaseModel):
name: str = Field(..., min_length=1, max_length=50)
age: int = Field(..., ge=5, le=100)
grade: Grade
subjects: List[str] = Field(..., min_items=1, max_items=10)
gpa: float = Field(..., ge=0.0, le=4.0)
# validator 를 이용해서 name, subject 값을 검증하는 함수를 만들어주세요
1. name_validation : name의 value 에 공백이 있는지/없는지를 확인
2. subject_validation : subject 이 ["math", "korean", "english", "coding", "science"] 에 해당되지 않으면 valueerror
from pydantic import BaseModel, Field, validator
from typing import List
from enum import Enum
# 이전 실습의 Grade Enum 재사용
class Grade(str, Enum):
GRADE_1 = "1"
GRADE_2 = "2"
GRADE_3 = "3"
GRADE_4 = "4"
class Student(BaseModel):
name: str = Field(..., min_length=1, max_length=50)
age: int = Field(..., ge=5, le=100)
grade: Grade
subjects: List[str] = Field(..., min_length=1, max_length=10) # min_items가 Pydantic v1 방식이므로 v2 방식인 min_length로 수정
gpa: float = Field(..., ge=0.0, le=4.0)
# 1. name_validation : 이름에 공백이 있는지 검증
@validator('name')
def validate_name_no_spaces(cls, v):
if ' ' in v:
raise ValueError('이름에 공백을 포함할 수 없습니다.')
return v
# 2. subject_validation : 과목 목록이 허용된 값인지 검증
@validator('subjects', each_item=True)
def validate_subject_in_allowed_list(cls, v):
allowed_subjects = {"math", "korean", "english", "coding", "science"}
if v not in allowed_subjects:
raise ValueError(f"'{v}'는 허용된 과목이 아닙니다. 허용 목록: {allowed_subjects}")
return v
# --- 검증 테스트 ---
if __name__ == "__main__":
print("--- 1. 정상 데이터 테스트 ---")
try:
valid_student = Student(
name="홍길동",
age=17,
grade="1",
subjects=["math", "science"],
gpa=3.5
)
print("성공: 유효한 학생 데이터입니다.")
print(valid_student.model_dump_json(indent=2))
except ValueError as e:
print(f"실패: {e}")
print("\n--- 2. 이름에 공백이 있는 경우 테스트 ---")
try:
invalid_name_student = Student(
name="홍 길동",
age=17,
grade="1",
subjects=["math", "science"],
gpa=3.5
)
except ValueError as e:
print(f"성공 (에러 발생 예상): {e}")
print("\n--- 3. 허용되지 않은 과목이 있는 경우 테스트 ---")
try:
invalid_subject_student = Student(
name="이순신",
age=18,
grade="2",
subjects=["math", "history"],
gpa=3.8
)
except ValueError as e:
print(f"성공 (에러 발생 예상): {e}")
Pydantic @validatorField만으로 구현하기 어려운 복잡한 유효성 검사 로직을 직접 함수로 작성할 때 사용하는 데코레이터ValueError를 발생시키는 패턴으로 사용name_validation 구현@validator('name')name 필드에 대한 검증 로직임을 Pydantic에 알림validate_name_no_spaces(cls, v)cls는 모델 클래스 자체, v는 name 필드에 들어온 값(value)을 의미if ' ' in v: 코드로 값에 공백이 포함되어 있는지 확인raise ValueError를 통해 에러를 발생시켜 유효성 검사를 실패 처리subject_validation 구현@validator('subjects', each_item=True)subjects 필드는 리스트(List[str])이므로, 리스트 안의 각 항목을 하나씩 검증해야 함each_item=True 옵션이 바로 이 역할을 수행. 리스트의 모든 아이템에 대해 아래 검증 함수를 각각 실행하라는 의미validate_subject_in_allowed_list(cls, v)v는 리스트 전체가 아니라 리스트의 개별 항목(과목명 문자열)임allowed_subjects 라는 허용된 과목 set을 정의 (리스트보다 set이 포함 여부 확인에 더 효율적)if v not in allowed_subjects: 코드로 해당 과목이 허용 목록에 있는지 확인ValueError를 발생시켜 실패 처리날씨 조회 어플리케이션을 위한 API를 만들어주세요
https://httpbin.org/delay/1
1. GET : /weather_sync/{cities}
2. GET : /weather_async/{cities}
# 필요 라이브러리: pip install httpx
import time
import asyncio
import httpx
from fastapi import FastAPI
app = FastAPI()
# 외부 API 호출 시뮬레이션 (동기)
def fetch_weather_sync(city: str, client: httpx.Client):
url = "https://httpbin.org/delay/1"
response = client.get(url)
return {"city": city, "weather": "맑음", "status": response.status_code}
# 외부 API 호출 시뮬레이션 (비동기)
async def fetch_weather_async(city: str, client: httpx.AsyncClient):
url = "https://httpbin.org/delay/1"
response = await client.get(url)
return {"city": city, "weather": "맑음", "status": response.status_code}
# 1. GET : /weather_sync/{cities} (동기 처리)
@app.get("/weather_sync/{cities}")
def get_weather_sync(cities: str):
start_time = time.time()
city_list = cities.split(',')
weather_data = []
with httpx.Client() as client:
for city in city_list:
# Blocking I/O: 응답 대기 중 실행 흐름이 멈춤
weather_data.append(fetch_weather_sync(city, client))
total_time = time.time() - start_time
return {"data": weather_data, "total_time": f"{total_time:.2f} 초"}
# 2. GET : /weather_async/{cities} (비동기 처리)
@app.get("/weather_async/{cities}")
async def get_weather_async(cities: str):
start_time = time.time()
city_list = cities.split(',')
async with httpx.AsyncClient() as client:
# Non-blocking I/O: 모든 작업을 동시에 실행
tasks = [fetch_weather_async(city, client) for city in city_list]
weather_data = await asyncio.gather(*tasks)
total_time = time.time() - start_time
return {"data": weather_data, "total_time": f"{total_time:.2f} 초"}
/weather_sync/{cities}httpx.get 호출 시, 외부 API의 응답이 올 때까지 코드 실행이 멈춤for 루프를 통해 각 도시의 API를 순서대로 호출/weather_async/{cities}await client.get 호출 시, 응답을 기다리는 동안 CPU 제어권을 이벤트 루프에 넘겨 다른 작업을 처리asyncio.gather를 통해 모든 도시의 API 요청 작업을 거의 동시에 시작| 구분 | 동기 (Sync) | 비동기 (Async) |
|---|---|---|
| I/O 작업 | Blocking (대기) | Non-blocking (다른 작업 전환) |
| 실행 흐름 | 순차적 | 동시적 |
| 자원 효율성 | 대기 시간 동안 스레드 점유 | 대기 시간 동안 다른 작업 수행 |
| 총 소요 시간 | 작업 수 × 작업 시간 | 가장 긴 단일 작업 시간 |
| 적합한 경우 | CPU-bound 작업, 간단한 로직 | I/O-bound 작업 (네트워크, DB) |