Python FastAPI와 비동기 프로그래밍 기초

1. 비동기 프로그래밍 기초

비동기 함수와 코루틴

• def 키워드로 정의한 함수는 기본적으로 동기 방식으로 동작합니다.

def do_sync():
    print("do_sync")

do_sync()  # 동기 함수 호출

• async 키워드를 사용하면 함수가 비동기 처리됩니다. 이러한 함수를 **코루틴(coroutine)**이라고 부릅니다.

async def do_async():
    print("do_async")

do_async()  # 호출 시 실행되지 않고 coroutine 객체가 반환됩니다.

.

코루틴의 실행

• 비동기 함수는 일반적인 방식으로 호출하면 실행되지 않습니다.
• await 키워드를 사용하여 호출해야 실행됩니다.

 

import asyncio

async def do_async():
    print("do_async")

async def main_async():
    await do_async()  # 비동기 함수 호출

asyncio.run(main_async())  # 비동기 실행

#출력
#do_async

• await: 비동기 작업을 일시 중단하고 다른 작업을 처리할 수 있도록 함.

---

2. FastAPI를 활용한 RESTful API

FastAPI는 비동기 프로그래밍과 잘 통합되어 있으며, 쉽고 빠르게 API를 개발할 수 있습니다.

기본 설정

• users: API에서 사용할 예제 데이터(딕셔너리 형태).
• FastAPI 애플리케이션 생성:

  from fastapi import FastAPI
  
  app = FastAPI()

---

3. CRUD API 구현

(1) GET 요청: 사용자 정보 조회

FastAPI의 경로 파라미터와 타입 어노테이션을 활용하여 데이터를 조회합니다.

# 예제 데이터
users = {
    0: {"userid": "apple", "name": "김사과"},
    1: {"userid": "banana", "name": "반하나"},
    2: {"userid": "orange", "name": "오렌지"}
}

@app.get("/users/{id}")
async def find_user(id: int):  # id는 정수 타입
    user = users[id]  # 딕셔너리에서 사용자 검색
    return user  # JSON 형태로 반환​

(2) GET 요청: 특정 키로 데이터 조회

@app.get("/users/{id}/{key}")
async def find_user_by_key(id: int, key: str):  # key는 문자열 타입
    user = users[id][key]  # 특정 키로 값 검색
    return user

• Swagger UI 테스트:
  • /docs 경로에서 자동으로 생성된 문서와 API를 테스트할 수 있습니다.

(3) GET 요청: 이름으로 사용자 조회

@app.get("/id-by-name")
async def find_user_by_name(name: str):
    for idx, user in users.items():  # users 딕셔너리를 순회
        if user['name'] == name:
            return user
    return {"error": "데이터를 찾지 못함"}

---

(4) POST 요청: 사용자 생성

• pydantic의 BaseModel을 활용하여 JSON 데이터를 검증하고 파싱합니다.

from pydantic import BaseModel

class User(BaseModel):
    userid: str
    name: str

@app.post("/users/{id}")
async def create_user(id: int, user: User):
    if id in users:  # 이미 존재하는 ID 검사
        return {"error": "이미 존재하는 키"}
    users[id] = user.__dict__  # JSON 데이터를 딕셔너리로 변환
    return {"success": "ok"}

---

(5) PUT 요청: 사용자 업데이트

• Optional: 값을 선택적으로 제공할 수 있도록 설정.
• 누락된 필드는 기존 데이터를 유지.

from typing import Optional

class UserForUpdate(BaseModel):
    userid: Optional[str] = None
    name: Optional[str] = None

@app.put("/users/{id}")
async def update_user(id: int, user: UserForUpdate):
    if id not in users:
        return {"error": "id가 존재하지 않음"}

    # 제공된 필드만 업데이트
    if user.userid:
        users[id]['userid'] = user.userid
    if user.name:
        users[id]['name'] = user.name

    return {"success": "ok"}​

---

(6) DELETE 요청: 사용자 삭제

• pop 메서드를 사용하여 데이터를 삭제합니다.

@app.delete("/users/{id}")
async def delete_item(id: int):
    users.pop(id, None)  # ID 삭제
    return {"success": "ok"}​

---

4. 실행 방법

1. 설치

pip install fastapi uvicorn​

2. 실행

uvicorn main:app --reload​

• main:app:
  • main.py 파일에서 app 객체를 실행.
• --reload:
  • 코드 변경 시 서버가 자동으로 재시작.

3. 테스트

• Swagger UI: http://127.0.0.1:8000/docs
• 예제 요청:
  • GET: http://127.0.0.1:8000/users/0
  • POST: 사용자 생성.
  • PUT: 사용자 업데이트.
  • DELETE: 사용자 삭제.

---

5. 주요 개념 요약

1. 코루틴과 비동기 함수:
   • async와 await를 통해 비동기 작업 처리.
2. FastAPI의 기능:
   • 경로 파라미터, 쿼리 매개변수, JSON 데이터 처리.
3. pydantic:
   • 데이터 유효성 검사와 타입 지정.
4. Swagger UI:
   • API 문서화와 테스트를 위한 자동화 도구.