Implement ASGI application with fibonacci, factorial, and mean endpoints

hw1/app.py

@@ -1,4 +1,7 @@
 from typing import Any, Awaitable, Callable
+import json
+from math import prod
+from urllib.parse import parse_qs
 
 
 async def application(
@@ -12,8 +15,117 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+
+    if scope.get("type") != "http":
+        await _send_json(send, {"detail": "unsupported scope type"}, status=422)
+        return
+
+    method = scope.get("method", "GET").upper()
+    path: str = scope.get("path", "/")
+    query = parse_qs((scope.get("query_string") or b"").decode(
+        "utf-8"), keep_blank_values=True)
+
+    if path.startswith("/fibonacci/"):
+        n_str = path[len("/fibonacci/"):].strip("/")
+        if n_str == "":
+            await _send_json(send, {"detail": "n is required"}, status=422)
+            return
+        try:
+            n = int(n_str)
+        except ValueError:
+            await _send_json(send, {"detail": "n must be an integer"}, status=422)
+            return
+        if n < 0:
+            await _send_json(send, {"detail": "n must be non-negative"}, status=400)
+            return
+
+        a, b = 0, 1
+        for _ in range(n):
+            a, b = b, a + b
+        await _send_json(send, {"result": a}, status=200)
+        return
+
+    if path == "/factorial":
+        if method != "GET":
+            await _send_json(send, {"detail": "not found"}, status=404)
+            return
+
+        raw = query.get("n", [None])[0]
+        if raw is None or raw == "":
+            await _send_json(send, {"detail": "n is required"}, status=422)
+            return
+        try:
+            n = int(raw)
+        except ValueError:
+            await _send_json(send, {"detail": "n must be an integer"}, status=422)
+            return
+        if n < 0:
+            await _send_json(send, {"detail": "n must be non-negative"}, status=400)
+            return
+
+        result = 1 if n == 0 else prod(range(1, n + 1))
+        await _send_json(send, {"result": result}, status=200)
+        return
+
+    if path == "/mean":
+        if method != "GET":
+            await _send_json(send, {"detail": "not found"}, status=404)
+            return
+
+        body_bytes = await _read_body(receive)
+        if not body_bytes:
+            await _send_json(send, {"detail": "json body required (array of numbers)"}, status=422)
+            return
+
+        try:
+            data = json.loads(body_bytes.decode("utf-8"))
+        except json.JSONDecodeError:
+            await _send_json(send, {"detail": "invalid JSON"}, status=422)
+            return
+
+        if not isinstance(data, list):
+            await _send_json(send, {"detail": "body must be a JSON array"}, status=422)
+            return
+
+        if len(data) == 0:
+            await _send_json(send, {"detail": "array must not be empty"}, status=400)
+            return
+
+        try:
+            nums = [float(x) for x in data]
+        except (TypeError, ValueError):
+            await _send_json(send, {"detail": "array must contain only numbers"}, status=422)
+            return
+
+        mean_value = sum(nums) / len(nums)
+        await _send_json(send, {"result": mean_value}, status=200)
+        return
+
+    await _send_json(send, {"detail": "not found"}, status=404)
+
+
+async def _read_body(receive: Callable[[], Awaitable[dict[str, Any]]]) -> bytes:
+    body = b""
+    more = True
+    while more:
+        message = await receive()
+        if message["type"] != "http.request":
+            break
+        body += message.get("body", b"")
+        more = message.get("more_body", False)
+    return body
+
+
+async def _send_json(send: Callable[[dict[str, Any]], Awaitable[None]], payload: dict, status: int = 200):
+    body = json.dumps(payload).encode("utf-8")
+    headers = [
+        (b"content-type", b"application/json; charset=utf-8"),
+        (b"content-length", str(len(body)).encode("ascii")),
+    ]
+    await send({"type": "http.response.start", "status": status, "headers": headers})
+    await send({"type": "http.response.body", "body": body})
+
 
 if __name__ == "__main__":
     import uvicorn
-    uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True)
+    uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True)

hw2/hw/Dockerfile

@@ -0,0 +1,23 @@
+FROM python:3.12 AS base
+
+ARG PYTHONFAULTHANDLER=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONHASHSEED=random \
+    PIP_NO_CACHE_DIR=on \
+    PIP_DISABLE_PIP_VERSION_CHECK=on \
+    PIP_DEFAULT_TIMEOUT=500
+
+RUN apt-get update && apt-get install -y gcc
+RUN python -m pip install --upgrade pip
+
+WORKDIR $APP_ROOT/src
+COPY . ./
+
+ENV VIRTUAL_ENV=$APP_ROOT/src/.venv \
+    PATH=$APP_ROOT/src/.venv/bin:$PATH
+
+RUN pip install -r requirements.txt
+
+FROM base as local
+
+CMD ["uvicorn", "shop_api.main:app", "--port", "8080", "--host", "0.0.0.0"]

hw2/hw/docker-compose.yml

@@ -0,0 +1,31 @@
+version: "3"
+
+services:
+
+  local:
+    build:
+      context: .
+      dockerfile: ./Dockerfile
+      target: local
+    restart: always
+    ports:
+      - 8080:8080
+
+  grafana:
+    image: grafana/grafana:latest
+    ports:
+      - 3000:3000
+    restart: always
+
+  prometheus:
+    image: prom/prometheus
+    volumes:
+      - ./settings/prometheus/:/etc/prometheus/
+    command:
+      - "--config.file=/etc/prometheus/prometheus.yml"
+      - "--storage.tsdb.path=/prometheus"
+      - "--web.console.libraries=/usr/share/prometheus/console_libraries"
+      - "--web.console.templates=/usr/share/prometheus/consoles"
+    ports:
+      - 9090:9090
+    restart: always

hw2/hw/ping.py

@@ -0,0 +1,51 @@
+import time
+
+import requests
+
+BASE = 'http://localhost:8080'
+
+
+def run_tests():
+    # создаем товар
+    r = requests.post(f'{BASE}/item', json={'name': 'Test Item', 'price': 10.5})
+    print('POST /item', r.status_code, r.json())
+    item_id = r.json()['id']
+
+    # получаем товар
+    r = requests.get(f'{BASE}/item/{item_id}')
+    print('GET /item/{id}', r.status_code, r.json())
+
+    # обновляем товар
+    r = requests.patch(f'{BASE}/item/{item_id}', json={'price': 12.0})
+    print('PATCH /item/{id}', r.status_code, r.json())
+
+    # список товаров
+    r = requests.get(f'{BASE}/item')
+    print('GET /item', r.status_code, len(r.json()))
+
+    # создаем корзину
+    r = requests.post(f'{BASE}/cart')
+    print('POST /cart', r.status_code, r.json())
+    cart_id = r.json()['id']
+
+    # получаем корзину
+    r = requests.get(f'{BASE}/cart/{cart_id}')
+    print('GET /cart/{id}', r.status_code, r.json())
+
+    # добавляем товар в корзину
+    r = requests.post(f'{BASE}/cart/{cart_id}/add/{item_id}')
+    print('POST /cart/{cart_id}/add/{item_id}', r.status_code, r.json())
+
+    # список корзин
+    r = requests.get(f'{BASE}/cart')
+    print('GET /cart', r.status_code, len(r.json()))
+
+
+if __name__ == '__main__':
+    while True:
+        try:
+            run_tests()
+        except Exception as e:
+            print('Error:', e)
+
+        time.sleep(1)

hw2/hw/requirements.txt

@@ -1,6 +1,7 @@
 # Основные зависимости для ASGI приложения
 fastapi>=0.117.1
 uvicorn>=0.24.0
+prometheus-fastapi-instrumentator==7.1.0 # для prometheus + grafana (hm3)
 
 # Зависимости для тестирования
 pytest>=7.4.0