HW1 Майоров Александр

hw1/app.py

@@ -1,5 +1,13 @@
 from typing import Any, Awaitable, Callable
 
+from utils import error_response
+from handlers import handle_factorial, handle_fibonacci, handle_mean
+
+ROUTES = {
+    ('GET', '/factorial'): handle_factorial,
+    ('GET', '/fibonacci'): handle_fibonacci,
+    ('GET', '/mean'): handle_mean,
+}
 
 async def application(
     scope: dict[str, Any],
@@ -12,7 +20,30 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    if scope["type"] == "lifespan":
+        while True:
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            elif message["type"] == "lifespan.shutdown":
+                await send({"type": "lifespan.shutdown.complete"})
+                return
+
+    if scope["type"] != "http":
+        return
+
+    path = scope['path']
+    path_segments = path.strip("/").split('/', 1)
+    normalized_path = '/' + path_segments[0] if path_segments[0] else '/'
+
+    method = scope['method']
+
+    handler = ROUTES.get((method, normalized_path))
+
+    if handler:
+        await handler(scope, receive, send)
+    else:
+        await error_response(send, 404, "Not Found")
 
 if __name__ == "__main__":
     import uvicorn

hw1/handlers.py

@@ -0,0 +1,85 @@
+from typing import Callable
+import json
+from urllib.parse import parse_qs
+
+from utils import json_response, error_response, get_request_body
+from math_utils import calculate_factorial, calculate_fibonacci, calculate_mean
+
+
+async def handle_factorial(scope: dict, _: Callable, send: Callable):
+    """Обработчик для GET /factorial?n=..."""
+    query_string = scope.get('query_string', b'')
+    query_params = parse_qs(query_string)
+    n_values = query_params.get(b'n')
+
+    if not n_values:
+        return await error_response(send, 422, "Missing query parameter 'n'")
+
+    n_str = n_values[0].decode('utf-8')
+    try:
+        n = int(n_str)
+        if n < 0:
+            return await error_response(send, 400, "n must be a non-negative integer")
+
+        result = calculate_factorial(n)
+        response_data = {"n": n, "result": result}
+        await json_response(send, 200, response_data)
+
+    except ValueError:
+        return await error_response(send, 422, "n must be a valid integer")
+
+async def handle_fibonacci(scope: dict, _: Callable, send: Callable):
+    """
+    Обработчик для GET /fibonacci/{n}.
+    Ожидает, что path_parts[1] содержит число n.
+    """
+    path = scope['path']
+    path_parts = path.strip("/").split('/')
+
+    if len(path_parts) != 2 or not path_parts[1]:
+        return await error_response(send, 422, "Missing or invalid path parameter for fibonacci")
+
+    n_str = path_parts[1]
+    try:
+        n = int(n_str)
+        if n < 0:
+            return await error_response(send, 400, "n must be a non-negative integer")
+
+        result = calculate_fibonacci(n)
+        response_data = {"n": n, "result": result}
+        await json_response(send, 200, response_data)
+
+    except ValueError:
+        return await error_response(send, 422, "Path parameter 'n' must be a valid integer")
+
+async def handle_mean(_: dict, receive: Callable, send: Callable):
+    """Обработчик для GET /mean (JSON Body)"""
+    raw_body = await get_request_body(receive)
+
+    try:
+        data = json.loads(raw_body.decode('utf-8')) if raw_body else None
+
+        if data is None:
+            return await error_response(send, 422, "Body must contain valid JSON array.")
+
+        if not isinstance(data, list):
+            return await error_response(send, 422, "Body must be a JSON array of numbers.")
+
+        if not data:
+            return await error_response(send, 400, "Array must not be empty.")
+
+        numbers = []
+        for item in data:
+            if isinstance(item, (int, float)):
+                numbers.append(float(item))
+            else:
+                return await error_response(send, 422, f"Array contains non-numeric value: {item}")
+
+        result = calculate_mean(numbers)
+        response_data = {"numbers": numbers, "result": result}
+        await json_response(send, 200, response_data)
+
+    except json.JSONDecodeError:
+        return await error_response(send, 422, "Invalid JSON format.")
+    except ValueError as e:
+        return await error_response(send, 422, f"Bad Request: {e}")

hw1/math_utils.py

@@ -0,0 +1,23 @@
+from typing import List
+
+def calculate_factorial(n: int) -> int:
+    """Вычисляет факториал неотрицательного целого числа n."""
+    if n == 0:
+        return 1
+    result = 1
+    for i in range(2, n + 1):
+        result *= i
+    return result
+
+def calculate_fibonacci(n: int) -> int:
+    """Вычисляет n-е число Фибоначчи (с 0-го индекса)."""
+    a, b = 0, 1
+    for _ in range(n):
+        a, b = b, a + b
+    return a
+
+def calculate_mean(numbers: List[float]) -> float:
+    """Вычисляет среднее арифметическое списка чисел."""
+    if not numbers:
+        raise ValueError("Array must not be empty.")
+    return sum(numbers) / len(numbers)

hw1/utils.py

@@ -0,0 +1,43 @@
+from typing import Any, Callable
+import json
+
+async def json_response(send: Callable, status: int, data: Any):
+    """
+    Формирует и отправляет HTTP-ответ с JSON-телом.
+
+    Args:
+        send: Корутина для отправки сообщений клиенту.
+        status: HTTP-статус код.
+        data: Данные для сериализации в JSON.
+    """
+    body = json.dumps(data).encode('utf-8')
+    headers = [(b'content-type', b'application/json')]
+
+    await send({"type": "http.response.start", "status": status, "headers": headers})
+    await send({"type": "http.response.body", "body": body})
+
+async def error_response(send: Callable, status: int, detail: str):
+    """
+    Формирует и отправляет ответ с ошибкой в JSON формате.
+
+    Args:
+        send: Корутина для отправки сообщений клиенту.
+        status: HTTP-статус код ошибки.
+        detail: Сообщение об ошибке.
+    """
+    error_message = {"detail": detail}
+    await json_response(send, status, error_message)
+
+async def get_request_body(receive: Callable) -> bytes:
+    """Хелпер для чтения всего тела HTTP-запроса."""
+    raw_body = b""
+    while True:
+        message = await receive()
+        if message['type'] == 'http.request':
+            raw_body += message.get('body', b'')
+            if not message.get('more_body', False):
+                break
+        # Дополнительная обработка на случай пустых или неожиданных сообщений
+        elif message['type'] != 'http.request' and not raw_body:
+            break
+    return raw_body