HW1 & HW2 & HW3 & HW4 solutions

.github/workflows/hw5-tests.yml

@@ -0,0 +1,39 @@
+name: "HW5 Tests"
+
+on:
+  pull_request:
+    branches: [ main ]
+    paths: [ 'hw5_own/**' ]
+  push:
+    branches: [ main ]
+    paths: [ 'hw5_own/**' ]
+
+jobs:
+  test-hw2:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.12", "3.13"]
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install dependencies
+      working-directory: hw5_own
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+
+    - name: Run tests
+      working-directory: hw5_own
+      env:
+        PYTHONPATH: ${{ github.workspace }}/hw5_own
+      run: |
+        pytest test_homework5.py -v
+        pytest test_websocket_chat.py -v

hw1/app.py

@@ -1,5 +1,100 @@
 from typing import Any, Awaitable, Callable
 
+import json  # https://docs.python.org/3/library/json.html - это стандартная библиотека в составе python (я не хочу делать алгос по десериализации JSON)
+
+
+# Числовые коды (если нельзя использовать http.HTTPStatus)
+OK, BAD_REQUEST, NOT_FOUND, UNPROC = 200, 400, 404, 422
+
+
+# Специальные исключения
+class ENotFound(Exception): ...
+
+
+class EBadRequest(Exception): ...
+
+
+# --- helpers & math -----------------------------------------------------------
+def read_query(qs: bytes) -> dict[str, str]:
+    """Вместо urllib вот такая заглушка"""
+    if not qs:
+        return {}
+    out: dict[str, str] = {}
+    s = qs.decode()
+    for pair in s.split("&"):
+        if not pair:
+            continue
+        k, v = (pair.split("=", 1) + [""])[:2]
+        out[k] = v
+    return out
+
+
+def factorial(n: int) -> int:
+    """Факториал"""
+    r = 1
+    for i in range(2, n + 1):
+        r *= i
+    return r
+
+
+def fibonacci(n: int) -> int:
+    """Фибоначчи"""
+    a, b = 0, 1
+    for _ in range(n):
+        a, b = b, a + b
+    return a
+
+
+def mean(nums: list[float]) -> float:
+    """Среднее"""
+    return sum(nums) / len(nums)
+
+
+async def read_body(receive: Callable[[], Awaitable[dict[str, Any]]]) -> bytes:
+    """
+    Args:
+        receive: Корутина для получения сообщений от клиента
+    """
+    body = b""
+    while True:
+        msg = await receive()
+        if msg["type"] == "http.request":
+            body += msg.get("body", b"")
+            if not msg.get("more_body"):
+                break
+        else:
+            break
+    return body
+
+
+async def respond(
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+    status: int,
+    data: Any | None = None,
+):
+    """
+    Args:
+        send: Словарь с информацией о запросе
+        status: статус код
+        data: любая инфа, которую в байтах хотим закинуть
+    """
+    if data is None:
+        headers = [(b"content-type", b"text/plain"), (b"content-length", b"0")]
+        await send(
+            {"type": "http.response.start", "status": status, "headers": headers}
+        )
+        await send({"type": "http.response.body", "body": b""})
+    else:
+        body = json.dumps({"result": data}).encode()
+        headers = [
+            (b"content-type", b"application/json"),
+            (b"content-length", str(len(body)).encode()),
+        ]
+        await send(
+            {"type": "http.response.start", "status": status, "headers": headers}
+        )
+        await send({"type": "http.response.body", "body": body})
+
 
 async def application(
     scope: dict[str, Any],
@@ -12,8 +107,72 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    # По поводу lifespan: https://www.youtube.com/watch?v=VJ963Z6lsQ4
+    if scope["type"] == "lifespan":
+        while True:
+            message = await receive()
+            t = message.get("type")
+            if t == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            elif t == "lifespan.shutdown":
+                await send({"type": "lifespan.shutdown.complete"})
+                return
+        return
+
+    if scope["type"] != "http":
+        return
+
+    method, path = scope["method"], scope["path"]
+    body = await read_body(receive)
+
+    try:
+        if method != "GET":
+            raise ENotFound()
+
+        # /factorial?n=...
+        if path == "/factorial":
+            n_raw = read_query(scope.get("query_string", b"")).get("n")
+            if n_raw is None:
+                raise ValueError("missing n")  # -> 422
+            n = int(n_raw)  # ValueError -> 422
+            if n < 0:
+                raise EBadRequest()  # -> 400
+            return await respond(send, OK, factorial(n))
+
+        # /fibonacci/{n}
+        if path.startswith("/fibonacci/"):
+            tail = path.split("/", 2)[2]
+            n = int(tail)  # ValueError -> 422
+            if n < 0:
+                raise EBadRequest()  # -> 400
+            return await respond(send, OK, fibonacci(n))
+
+        # /mean (GET body = JSON array)
+        if path == "/mean":
+            if not body:
+                raise ValueError("empty body")  # -> 422
+            data = json.loads(body.decode())  # JSONDecodeError -> 422
+            if not isinstance(data, list):
+                raise ValueError("not a list")  # -> 422
+            if not data:
+                raise EBadRequest()  # -> 400
+            nums = [float(x) for x in data]  # ValueError/TypeError -> 422
+            return await respond(send, OK, mean(nums))
+
+        raise ENotFound()
+
+    except ENotFound:
+        return await respond(send, NOT_FOUND)
+    except EBadRequest:
+        return await respond(send, BAD_REQUEST)
+    except (ValueError, TypeError, json.JSONDecodeError):
+        return await respond(send, UNPROC)
+    except Exception:
+        # На всякий случай — считаем прочие сбои как 400
+        return await respond(send, BAD_REQUEST)
+
 
 if __name__ == "__main__":
     import uvicorn
+
     uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True)

hw2/hw/requirements_with_websockets.txt

@@ -0,0 +1,10 @@
+# Основные зависимости для ASGI приложения
+fastapi>=0.117.1
+uvicorn>=0.24.0
+websockets>=15.0.1
+
+# Зависимости для тестирования
+pytest>=7.4.0
+pytest-asyncio>=0.21.0
+httpx>=0.27.2
+Faker>=37.8.0

hw2/hw/shop_api/core/schemas.py

@@ -0,0 +1,49 @@
+# from __future__ import annotations
+from typing import Optional, Annotated
+from pydantic import BaseModel, Field, ConfigDict
+
+ItemName = Annotated[str, Field(description="Наименование товара", min_length=1)]
+ItemId = Annotated[int, Field(description="Идентификатор корзины", ge=0)]
+ItemPrice = Annotated[float, Field(description="Цена товара", ge=0)]
+
+CartId = Annotated[int, Field(description="Идентификатор корзины")]
+
+
+# ---- Item ----
+class ItemOut(BaseModel):
+    id: ItemId
+    name: ItemName
+    price: ItemPrice
+    deleted: bool = Field(description="Удален ли товар", default=False)
+
+
+class ItemCreate(BaseModel):
+    name: ItemName
+    price: ItemPrice
+
+
+class ItemPut(BaseModel):
+    name: ItemName
+    price: ItemPrice
+
+
+class ItemPatch(BaseModel):
+    # Разрешаем частичное обновление ТОЛЬКО name/price; лишние поля → 422
+    model_config = ConfigDict(extra="forbid")
+
+    name: Optional[ItemName] = None
+    price: Optional[ItemPrice] = None
+
+
+# ---- Cart ----
+class CartItemView(BaseModel):
+    id: ItemId
+    name: ItemName
+    quantity: int = Field(description="Количество товара в корзине", ge=0)
+    available: bool = Field(description="Доступен ли товар")
+
+
+class CartView(BaseModel):
+    id: int
+    items: list[CartItemView]
+    price: float = Field(description="Общая сумма заказа", ge=0.0)