Edgar Oganisyan HW1 + (HW2 + websckets)

hw1/app.py

@@ -1,5 +1,17 @@
+from http import HTTPStatus
 from typing import Any, Awaitable, Callable
 
+from routes.fibonacci import handle_fibonacci
+from routes.factorial import handle_factorial
+from routes.mean import handle_mean
+
+from helpers import send_json
+
+
+ROUTES = {
+    ("GET", "/factorial"): handle_factorial,
+    ("GET", "/mean"): handle_mean,
+}
 
 async def application(
     scope: dict[str, Any],
@@ -12,7 +24,35 @@ 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"})
+                break
+        return
+
+    if scope["type"] != "http":
+        return
+
+    method: str = scope["method"]
+    path: str = scope["path"]
+
+    # Сначала проверяем точные маршруты
+    handler = ROUTES.get((method, path))
+    if handler is not None:
+        await handler(scope, receive, send)
+        return
+
+    # Проверяем маршруты с параметрами
+    if method == "GET" and path.startswith("/fibonacci/"):
+        await handle_fibonacci(scope, receive, send)
+        return
+
+    return await send_json(send, HTTPStatus.NOT_FOUND, f"Not Found: {method} {path}")
+
 
 if __name__ == "__main__":
     import uvicorn

hw1/helpers.py

@@ -0,0 +1,51 @@
+import json
+from http import HTTPStatus
+from typing import Any, Awaitable, Callable
+
+async def read_body(receive: Callable[[], Awaitable[dict[str, Any]]]) -> bytes:
+    """
+    Читает тело HTTP-запроса целиком, собирая все чанки http.request
+    до тех пор, пока more_body == False.
+    """
+    body = bytearray()
+    while True:
+        event = await receive()
+        assert event["type"] == "http.request"
+        body.extend(event.get("body", b""))
+        if not event.get("more_body", False):
+            break
+
+    return bytes(body)
+
+
+async def send_json(
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+    status: HTTPStatus | int,
+    response: dict[str, Any] | str,
+    headers: list[tuple[bytes, bytes]] | None = None,
+) -> None:    
+    hdrs: list[tuple[bytes, bytes]] = [
+        (b"content-type", b"application/json"),
+    ]
+    if headers:
+        hdrs.extend(headers)
+
+    if isinstance(response, dict):
+        body_content = json.dumps(response).encode("utf-8")
+    else:
+        body_content = str(response).encode("utf-8")
+
+    await send(
+        {
+            "type": "http.response.start",
+            "status": int(status),
+            "headers": hdrs,
+        }
+    )
+    await send(
+        {
+            "type": "http.response.body",
+            "body": body_content,
+            "more_body": False,
+        }
+    )

hw1/routes/factorial.py

@@ -0,0 +1,37 @@
+from http import HTTPStatus
+from typing import Any, Awaitable, Callable
+from urllib.parse import parse_qs
+
+from helpers import send_json
+
+
+async def handle_factorial(
+    scope: dict[str, Any],
+    receive: Callable[[], Awaitable[dict[str, Any]]],
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+):
+    query_string = scope.get("query_string", b"").decode("utf-8")
+    query_params = parse_qs(query_string)
+
+    if "n" not in query_params:
+        return await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "Missing parameter 'n'"})
+
+    try:
+        n = int(query_params["n"][0])
+    except (ValueError, IndexError):
+        return await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "Parameter 'n' must be an integer"})
+
+    if n < 0:
+        return await send_json(send, HTTPStatus.BAD_REQUEST, {"error": "Parameter 'n' must be non-negative"})
+
+    result = calculate_factorial(n)
+    return await send_json(send, HTTPStatus.OK, {"result": result})
+
+def calculate_factorial(n: int) -> int:
+    if n == 0:
+        return 1
+
+    res = 1
+    for i in range(1, n + 1):
+        res *= i
+    return res

hw1/routes/fibonacci.py

@@ -0,0 +1,37 @@
+from http import HTTPStatus
+from typing import Any, Awaitable, Callable
+
+from helpers import send_json
+
+
+async def handle_fibonacci(
+    scope: dict[str, Any],
+    receive: Callable[[], Awaitable[dict[str, Any]]],
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+):
+    path = scope.get("path", "")
+
+    path_parts = path.split("/")
+    if len(path_parts) != 3:
+        return await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "Invalid path format"})
+
+    try:
+        n = int(path_parts[2])
+    except ValueError:
+        return await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "Parameter must be an integer"})
+
+    if n < 0:
+        return await send_json(send, HTTPStatus.BAD_REQUEST, {"error": "Parameter must be non-negative"})
+
+    result = calculate_fibonacci(n)
+    return await send_json(send, HTTPStatus.OK, {"result": result})
+
+
+def calculate_fibonacci(n: int) -> int:
+    if n <= 1:
+        return n
+
+    a, b = 0, 1
+    for _ in range(2, n + 1):
+        a, b = b, a + b
+    return b

hw1/routes/mean.py

@@ -0,0 +1,38 @@
+import json
+from http import HTTPStatus
+from typing import Any, Awaitable, Callable
+
+from helpers import read_body, send_json
+
+
+async def handle_mean(
+    scope: dict[str, Any],
+    receive: Callable[[], Awaitable[dict[str, Any]]],
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+):
+    body = await read_body(receive)
+
+    if not body:
+        return await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "Request body is required"})
+
+    try:
+        data = json.loads(body)
+    except json.JSONDecodeError:
+        return await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "Invalid JSON"})
+
+    if not isinstance(data, list):
+        return await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "Data must be a list"})
+
+    if len(data) == 0:
+        return await send_json(send, HTTPStatus.BAD_REQUEST, {"error": "List cannot be empty"})
+
+    for item in data:
+        if not isinstance(item, (int, float)):
+            return await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "All elements must be numbers"})
+
+    result = calculate_mean(data)
+    return await send_json(send, HTTPStatus.OK, {"result": result})
+
+
+def calculate_mean(numbers: list[int | float]) -> float:
+    return sum(numbers) / len(numbers)

hw2/grpc_example/ping_pb2.pyi

@@ -0,0 +1,17 @@
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from typing import ClassVar as _ClassVar, Optional as _Optional
+
+DESCRIPTOR: _descriptor.FileDescriptor
+
+class PingRequest(_message.Message):
+    __slots__ = ("message",)
+    MESSAGE_FIELD_NUMBER: _ClassVar[int]
+    message: str
+    def __init__(self, message: _Optional[str] = ...) -> None: ...
+
+class PongResponse(_message.Message):
+    __slots__ = ("message",)
+    MESSAGE_FIELD_NUMBER: _ClassVar[int]
+    message: str
+    def __init__(self, message: _Optional[str] = ...) -> None: ...