FastAPI의 비동기 프로그래밍으로 API 성능을 강화하세요

API 게임을 한 단계 더 발전시킬 준비가 되셨나요?

FastAPI를 사용하면 API를 더 빠르고, 더 빠르게 반응하고, 전문가처럼 무거운 로드를 처리할 수 있습니다.

이 기사에서는 FastAPI에서 비동기 프로그래밍을 활용하여 고성능 API를 구축하는 방법을 보여 드리겠습니다. 결국에는 비동기 엔드포인트를 구현하고 효과적으로 테스트할 수 있는 지식을 갖추게 됩니다.

당신이 배울 내용

당신이 마스터하게 될 내용은 다음과 같습니다:

• 비동기 프로그래밍의 기본과 그것이 중요한 이유.
• 비동기 개발을 위한 FastAPI 환경을 설정하는 방법.
• 실제 사례를 통해 비동기 엔드포인트를 작성하고 테스트합니다.
• HTTP 요청, 파일 처리 및 백그라운드 작업에 비동기 라이브러리를 사용합니다.

FastAPI에서 비동기 프로그래밍을 사용하는 이유는 무엇입니까?

그것은 무엇입니까?

비동기 프로그래밍을 사용하면 작업을 동시에 실행할 수 있습니다. 이는 응답을 기다리는 것이 일반적인 네트워크 요청, 데이터베이스 쿼리 또는 파일 작업과 같은 작업에 특히 유용합니다.

왜 중요한가요?

기존 동기 프로그래밍에서는 작업이 순차적으로 실행되므로 여러 요청을 처리할 때 지연이 발생합니다. 비동기 프로그래밍을 사용하면 여러 사용자에게 동시에 서비스를 제공하여 리소스 활용도를 극대화하고 더 나은 사용자 경험을 보장할 수 있습니다.

FastAPI 환경 설정

이제 소매를 걷어붙이고 놀라운 것을 만들어 봅시다!

먼저 필수 라이브러리를 설치합니다.

pip install "fastapi[standard]" httpx aiofiles pytest

코드

아래는 FastAPI의 비동기 프로그래밍을 보여주는 완전한 예입니다. 코드의 각 부분은 고유한 용도로 사용되며 단계별로 설명하겠습니다.

from fastapi import FastAPI, BackgroundTasks
import httpx
import aiofiles
import pytest
from fastapi.testclient import TestClient

app = FastAPI()

# API Endpoints
@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

# Testing Code
client = TestClient(app)

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_read_file():
    # Create an example file for testing
    with open("example.txt", "w") as file:
        file.write("This is a test file content")

    response = client.get("/read-file")
    assert response.status_code == 200
    assert response.json() == {"content": "This is a test file content"}

def test_schedule_email():
    response = client.post("/send-email/?email=fogigav197@rabitex.com")
    assert response.status_code == 200
    assert response.json() == {"message": "Email scheduled"}

@pytest.mark.asyncio
async def test_call_external_api():
    async with httpx.AsyncClient() as async_client:
        response = await async_client.get("https://jsonplaceholder.typicode.com/posts/1")
    assert response.status_code == 200
    assert "id" in response.json()

무너뜨리기

API 엔드포인트

1.간단한 비동기 엔드포인트

@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

이 엔드포인트는 async def를 사용하여 기본 비동기 경로를 정의하는 방법을 보여줍니다. ID로 항목을 검색합니다.

2.외부 API 호출

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

httpx를 사용하여 비차단 HTTP 요청을 만드는 방법을 보여줍니다. 외부 API 응답을 기다리는 동안 애플리케이션은 다른 요청을 처리할 수 있습니다.

3.비동기 파일 읽기

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

파일을 비동기식으로 읽어서 파일 작업이 애플리케이션을 차단하지 않도록 합니다.

4.백그라운드 작업 실행

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

이 엔드포인트는 이메일을 보내는 백그라운드 작업을 예약하여 메인 스레드가 다른 요청을 처리할 수 있도록 합니다.

코드 테스트

1.기본 엔드포인트 테스트

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

이는 /item/{item_id} 엔드포인트가 예상 데이터를 반환하는지 확인합니다.

2.파일 읽기 테스트

pip install "fastapi[standard]" httpx aiofiles pytest

이것은 테스트 파일을 생성하고 /read-file 엔드포인트가 해당 내용을 올바르게 읽고 반환하는지 확인합니다.

3.백그라운드 작업 테스트

from fastapi import FastAPI, BackgroundTasks
import httpx
import aiofiles
import pytest
from fastapi.testclient import TestClient

app = FastAPI()

# API Endpoints
@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

# Testing Code
client = TestClient(app)

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_read_file():
    # Create an example file for testing
    with open("example.txt", "w") as file:
        file.write("This is a test file content")

    response = client.get("/read-file")
    assert response.status_code == 200
    assert response.json() == {"content": "This is a test file content"}

def test_schedule_email():
    response = client.post("/send-email/?email=fogigav197@rabitex.com")
    assert response.status_code == 200
    assert response.json() == {"message": "Email scheduled"}

@pytest.mark.asyncio
async def test_call_external_api():
    async with httpx.AsyncClient() as async_client:
        response = await async_client.get("https://jsonplaceholder.typicode.com/posts/1")
    assert response.status_code == 200
    assert "id" in response.json()

백그라운드 이메일 작업이 성공적으로 예약되었는지 테스트합니다.

4.외부 API 호출 테스트

@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

이렇게 하면 /external-api 엔드포인트가 외부 소스에서 데이터를 올바르게 가져오도록 보장됩니다.

산출

결론

제공된 코드를 통해 이제 FastAPI를 사용하여 비동기 API를 구축하고 테스트하는 방법을 실질적으로 이해하게 되었습니다. 파일 처리, 외부 API 호출, 백그라운드 작업 예약 등 비동기 프로그래밍을 사용하면 손쉽게 확장 가능한 고성능 애플리케이션을 만들 수 있습니다.

다음 FastAPI 프로젝트를 구축할 준비가 되셨나요? 시작해 봅시다!

읽어주셔서 감사합니다...
즐거운 코딩하세요!

위 내용은 FastAPI의 비동기 프로그래밍으로 API 성능을 강화하세요의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!