[250829금2012H] Backend & FastAPI (3)

윤승호·2025년 8월 30일

어렵다 어려워

학습시간 09:00~03:00(당일18H/누적2012H)


◆ 학습내용

실습


1. 기본 API

문제

아래 기능을 갖는 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 객체 반환

제품 ID 조회 API (@app.get("/products/{product_id}"))

  • 경로 매개변수(Path Parameter) {product_id} 사용
  • 함수 매개변수 product_id: int 로 타입 지정
  • FastAPI가 URL 경로의 문자열을 정수 타입으로 자동 변환 및 검증
  • 입력받은 ID를 포함한 제품 정보 JSON 객체 반환

카테고리별 제품 목록 조회 API (@app.get("/categories/{category_name}/products"))

  • 경로 매개변수(Path Parameter) {category_name} 사용
  • read_category_products 함수의 category_name 인자로 URL 경로 값 전달
  • 입력받은 카테고리명에 해당하는 예시 상품 목록 JSON 객체 반환

계산기 API (@app.get("/calculator/add"))

  • 쿼리 매개변수(Query Parameter) ab 사용
  • 요청 URL 예시: /calculator/add?a=10&b=20
  • FastAPI가 a=10, b=20 값을 add_numbers 함수의 매개변수 a, b에 자동 매핑
  • 두 숫자의 덧셈 결과를 포함한 JSON 객체 반환

2. 책 리뷰 API

문제

아래 기능을 갖는 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()}

Pydantic 모델 정의

  • Book: 책의 기본 정보를 나타내는 모델. is_available 필드는 기본값으로 True를 가짐
  • Review: 리뷰 정보를 나타내는 모델. rating 필드는 Field를 사용하여 1에서 5 사이의 값만 받도록 유효성 검사 규칙을 추가
  • BookWithReviews: 책 한 권과 여러 개의 리뷰 리스트를 함께 다루기 위한 중첩 모델. book 필드는 Book 모델을, reviews 필드는 Review 모델의 리스트(List[Review])를 타입으로 가짐

POST /books (책 생성 API)

  • 요청 본문(Request Body)으로 Book 모델 형식의 JSON 데이터를 받음
  • 전역 변수 book_id_counter를 사용해 새로운 책 ID를 생성
  • 인메모리 데이터베이스 books_dbreviews_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를 요청 본문으로 받음
  • 한 번의 API 요청으로 책 정보와 여러 개의 리뷰 정보를 동시에 처리
  • 새로운 책 ID를 생성한 후, 페이로드에서 분리한 책 정보와 리뷰 리스트를 각각의 DB에 저장
  • 201 Created 상태 코드와 함께 생성된 책 ID를 포함한 전체 데이터를 반환

3. To do 리스트 API

문제

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)

Pydantic 모델 정의

  • Task: To-Do 작업의 기본 스키마. description은 선택 사항, completed는 기본값 False를 가짐
  • TaskUpdate: 부분 수정을 위한 스키마. 모든 필드를 Optional로 선언하여 클라이언트가 원하는 필드만 보낼 수 있도록 허용

API 엔드포인트 설명

  • POST /tasks

    • Task 모델을 요청 본문으로 받아 새로운 Task를 생성
    • 성공 시 201 Created 상태 코드 반환
  • GET /tasks

    • 인메모리 DB tasks_db에 저장된 모든 Task의 목록을 리스트 형태로 반환
  • GET /tasks/completed

    • completed 필드가 True인 Task들만 필터링하여 리스트로 반환
    • FastAPI 경로 우선순위 규칙에 따라, 동적 경로인 /tasks/{task_id} 보다 먼저 선언해야 정상 동작
  • GET /tasks/{task_id}

    • 경로 매개변수로 받은 task_id에 해당하는 Task를 조회
    • ID가 존재하지 않을 경우 404 Not Found 에러 발생
  • PUT /tasks/{task_id}

    • 특정 Task의 모든 필드를 요청 본문의 Task 모델 데이터로 완전히 교체
    • ID가 존재하지 않을 경우 404 Not Found 에러 발생
  • PATCH /tasks/{task_id}

    • 특정 Task의 일부 필드만 수정
    • TaskUpdate 모델을 사용하여 변경하려는 필드만 선택적으로 받음
    • model_dump(exclude_unset=True)를 통해 클라이언트가 보낸 데이터만 추출하여 업데이트
  • DELETE /tasks/{task_id}

    • 특정 task_id를 가진 Task를 DB에서 삭제
    • 성공적으로 삭제되면 본문 내용 없이 204 No Content 상태 코드를 반환

4. 학생 관리 API

문제

# 학생 관리 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

Pydantic 모델 정의

  • GradeEnum
    • Enum을 사용하여 grade 필드에 들어올 수 있는 값을 "1", "2", "3", "4"로 명확하게 제한
    • 이외의 값이 들어오면 유효성 검사 단계에서 에러 발생
  • Student
    • Field를 사용하여 각 필드에 상세한 유효성 검사 규칙을 적용
    • name: 최소 1자, 최대 50자
    • age: 5 이상, 100 이하
    • subjects: 최소 1개, 최대 100개의 요소를 가진 리스트
    • gpa: 0.0 이상, 4.0 이하
  • StudentResponse
    • Student 모델을 상속받아 DB에 저장된 후 부여되는 id 필드를 추가
    • 클라이언트에게 응답을 보낼 때 사용하는 모델

API 엔드포인트 설명

  • POST /students (학생 생성)

    • 요청 본문으로 Student 모델 형식의 데이터를 받음
    • FastAPI가 Pydantic 모델에 정의된 모든 유효성 검사 규칙(길이, 범위, Enum 등)을 자동으로 수행
    • 검증 통과 시, 새로운 학생 ID를 부여하고 인메모리 DB에 저장한 후 StudentResponse 형태로 반환
  • GET /students (학생 복합 검색)

    • 쿼리 파라미터를 사용하여 여러 조건으로 학생을 검색하는 기능
    • min_age, max_age, grade, subject는 모두 Optional로 선언되어 선택적으로 사용 가능
    • 필터링 로직
      • DB에 있는 모든 학생 데이터를 순회
      • min_age, max_age 파라미터가 있으면 나이 범위 조건을 검사
      • grade 파라미터가 있으면 학년 일치 여부를 검사
      • subject 파라미터가 있으면, 해당 학생의 subjects 리스트에 과목이 포함되어 있는지 검사
      • 모든 조건을 통과하는 학생만 결과 리스트에 추가하여 반환

5. Validator 데이터 검증 API

문제

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 @validator

  • Field만으로 구현하기 어려운 복잡한 유효성 검사 로직을 직접 함수로 작성할 때 사용하는 데코레이터
  • 특정 필드의 값을 받아서 원하는 조건으로 검증하고, 통과하면 값을 그대로 반환하고 실패하면 ValueError를 발생시키는 패턴으로 사용

name_validation 구현

  • @validator('name')
    • 이 데코레이터는 바로 아래 선언된 함수가 name 필드에 대한 검증 로직임을 Pydantic에 알림
  • validate_name_no_spaces(cls, v)
    • cls는 모델 클래스 자체, vname 필드에 들어온 값(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를 발생시켜 실패 처리

6. 날씨 조회 async API

문제

날씨 조회 어플리케이션을 위한 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}

  • 처리 방식: 순차적 실행
  • I/O 모델: Blocking I/O
    • httpx.get 호출 시, 외부 API의 응답이 올 때까지 코드 실행이 멈춤
  • 작업 흐름
    • for 루프를 통해 각 도시의 API를 순서대로 호출
    • 한 도시의 요청 및 응답 과정이 완전히 끝나야 다음 도시 요청을 시작
  • 예상 소요 시간
    • (도시 수) × (개별 요청 시간)
    • 예: 3개 도시 조회 시 약 3초 소요

/weather_async/{cities}

  • 처리 방식: 동시성(Concurrency) 실행
  • I/O 모델: Non-blocking I/O
    • await client.get 호출 시, 응답을 기다리는 동안 CPU 제어권을 이벤트 루프에 넘겨 다른 작업을 처리
  • 작업 흐름
    • asyncio.gather를 통해 모든 도시의 API 요청 작업을 거의 동시에 시작
    • 모든 작업의 응답이 완료될 때까지 한 번에 기다림
  • 예상 소요 시간
    • 여러 요청 중 가장 오래 걸리는 단일 요청 시간에 근사
    • 예: 3개 도시 조회 시 약 1초 소요
구분동기 (Sync)비동기 (Async)
I/O 작업Blocking (대기)Non-blocking (다른 작업 전환)
실행 흐름순차적동시적
자원 효율성대기 시간 동안 스레드 점유대기 시간 동안 다른 작업 수행
총 소요 시간작업 수 × 작업 시간가장 긴 단일 작업 시간
적합한 경우CPU-bound 작업, 간단한 로직I/O-bound 작업 (네트워크, DB)
profile
나는 AI 엔지니어가 된다.

0개의 댓글