Prvý nástrel

🧪 Cvičenie: Asynchrónna služba na spracovanie úloh (FastAPI)

Cieľ

Vytvorte REST API, ktoré simuluje spracovanie dlhých operácií.

Študent si precvičí:

• async def
• await
• asyncio.sleep()
• asyncio.gather()
• tvorbu API vo FastAPI

---

Zadanie

Vytvorte FastAPI aplikáciu s endpointom:

GET /process

Parameter:

/process?times=2,1,3

---

Simulácia úlohy

Použite funkciu:

async def process_task(delay: int):
    await asyncio.sleep(delay)
    return delay

---

Endpoint

Endpoint musí:

1. načítať zoznam časov z parametra times
2. spustiť všetky úlohy súčasne
3. počkať na ich dokončenie
4. vrátiť JSON odpoveď

Príklad:

Request:

GET /process?times=2,1,3

Response:

{
  "results": [2,1,3],
  "count": 3
}

---

Očakávaná implementácia

import asyncio
from fastapi import FastAPI

app = FastAPI()

async def process_task(delay):
    await asyncio.sleep(delay)
    return delay

@app.get("/process")
async def process(times: str):
    delays = [int(x) for x in times.split(",")]

    results = await asyncio.gather(
        *(process_task(d) for d in delays)
    )

    return {
        "results": results,
        "count": len(results)
    }

---

Automatické hodnotenie

Test správnosti

def test_response():
    response = client.get("/process?times=1,2,3")

    assert response.status_code == 200

    data = response.json()

    assert data["results"] == [1,2,3]
    assert data["count"] == 3

---

Test asynchrónnosti

import time

def test_concurrent_execution():
    start = time.perf_counter()

    response = client.get("/process?times=3,2,1")

    duration = time.perf_counter() - start

    assert duration < 4

Ak študent spraví:

for d in delays:
    await process_task(d)

čas bude približne:

6 sekúnd

a test neprejde.

Ak použije:

asyncio.gather(...)

čas bude približne:

3 sekundy

a test prejde.

---

⭐ Pokročilejšia verzia

Endpoint /stats

Pridajte endpoint:

GET /stats?times=2,1,3

Výstup:

{
  "min": 1,
  "max": 3,
  "avg": 2.0
}

Úlohy sa musia opäť spracovať asynchrónne.

---

⭐⭐ Zápočtová verzia

Simulovaný URL checker

Endpoint:

POST /check

Request:

{
  "urls": [
    "google.com",
    "github.com",
    "openai.com"
  ]
}

Simulovaná kontrola:

async def check_url(url):
    await asyncio.sleep(1)
    return {
        "url": url,
        "status": "OK"
    }

Response:

{
  "results": [
    {"url":"google.com","status":"OK"},
    {"url":"github.com","status":"OK"},
    {"url":"openai.com","status":"OK"}
  ]
}

Automatické testy

Kontrolujú:

• HTTP status
• štruktúru JSON
• správne použitie async (čas vykonania)

Previous Post Next Post