diff --git a/hw1/app.py b/hw1/app.py index 6107b87..478f748 100644 --- a/hw1/app.py +++ b/hw1/app.py @@ -1,4 +1,8 @@ +from http import HTTPStatus from typing import Any, Awaitable, Callable +import json +import math +from urllib.parse import parse_qs async def application( @@ -12,7 +16,202 @@ 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 + + path = scope["path"] + method = scope["method"] + + query_string = scope.get("query_string").decode() + query_params = parse_qs(query_string) + + body = b"" + while True: + message = await receive() + if message["type"] == "http.request": + body += message.get("body") + if not message.get("more_body", False): + break + + path_parts = [p for p in path.split("/") if p] + + try: + if path == "/factorial" and method == "GET": + await handle_factorial(query_params, send) + elif len(path_parts) == 2 and path_parts[0] == "fibonacci" and method == "GET": + await handle_fibonacci(path_parts[1], send) + elif path == "/mean" and method == "GET": + await handle_mean(body, send) + else: + await send_response(send, HTTPStatus.NOT_FOUND, {"error": "Not found"}) + except Exception as e: + await send_response(send, HTTPStatus.INTERNAL_SERVER_ERROR, {"error": str(e)}) + + +async def send_response( + send: Callable[[dict[str, Any]], Awaitable[None]], + status_code: int, + data: dict[str, Any], +): + """Отправляет JSON ответ""" + response_body = json.dumps(data).encode() + + await send({ + "type": "http.response.start", + "status": status_code, + "headers": [ + [b"content-type", b"application/json"], + [b"content-length", str(len(response_body)).encode()], + ], + }) + + await send({ + "type": "http.response.body", + "body": response_body, + }) + + +async def handle_factorial( + query_params: dict[str, list[str]], + send: Callable[[dict[str, Any]], Awaitable[None]], +): + if "n" not in query_params: + await send_response( + send, + HTTPStatus.UNPROCESSABLE_ENTITY, + {"error": "Parameter 'n' is required"}, + ) + return + + value = query_params["n"][0] + + try: + n = int(value) + except (ValueError, TypeError): + await send_response( + send, + HTTPStatus.UNPROCESSABLE_ENTITY, + {"error": "Parameter 'n' must be an integer"}, + ) + return + + if n < 0: + await send_response( + send, + HTTPStatus.BAD_REQUEST, + {"error": "Parameter 'n' must be non-negative"}, + ) + return + + if n > 1000: + await send_response( + send, + HTTPStatus.BAD_REQUEST, + {"error": "Parameter 'n' is too large"}, + ) + return + + result = math.factorial(n) + await send_response(send, HTTPStatus.OK, {"result": result}) + + +async def handle_fibonacci( + n_str: str, + send: Callable[[dict[str, Any]], Awaitable[None]], +): + try: + n = int(n_str) + except (ValueError, TypeError): + await send_response( + send, + HTTPStatus.UNPROCESSABLE_ENTITY, + {"error": "Invalid integer"}, + ) + return + + if n < 0: + await send_response( + send, + HTTPStatus.BAD_REQUEST, + {"error": "Parameter 'n' must be non-negative"}, + ) + return + + if n > 10000: + await send_response( + send, + HTTPStatus.BAD_REQUEST, + {"error": "Parameter 'n' is too large"}, + ) + return + + if n <= 1: + result = n + else: + a, b = 0, 1 + for _ in range(2, n + 1): + a, b = b, a + b + result = b + + await send_response(send, HTTPStatus.OK, {"result": result}) + + +async def handle_mean( + body: bytes, + send: Callable[[dict[str, Any]], Awaitable[None]], +): + if not body: + await send_response( + send, + HTTPStatus.UNPROCESSABLE_ENTITY, + {"error": "Body is required"}, + ) + return + + try: + data = json.loads(body) + except json.JSONDecodeError: + await send_response( + send, + HTTPStatus.UNPROCESSABLE_ENTITY, + {"error": "Invalid JSON"}, + ) + return + + if data is None: + await send_response( + send, + HTTPStatus.UNPROCESSABLE_ENTITY, + {"error": "Body is required"}, + ) + return + + if not isinstance(data, list) or len(data) == 0: + await send_response( + send, + HTTPStatus.BAD_REQUEST, + {"error": "Body must be a non-empty list"}, + ) + return + + for item in data: + if not isinstance(item, (int, float)) or isinstance(item, bool): + await send_response( + send, + HTTPStatus.BAD_REQUEST, + {"error": "All items must be numbers"}, + ) + return + + result = sum(data) / len(data) + await send_response(send, HTTPStatus.OK, {"result": result}) + if __name__ == "__main__": import uvicorn