D K

hw1/app.py

@@ -1,6 +1,39 @@
+import ast
+import http.client
+import math
 from typing import Any, Awaitable, Callable
 
 
+async def compute_fibonacci(num: int) -> int:
+    start = [0, 1]
+    for i in range(2, num + 1):
+        start.append(start[i - 1] + start[i - 2])
+    return start[num]
+
+
+async def compute_factorial(num: int) -> int:
+    return math.factorial(num)
+
+
+async def compute_mean(values: list[float]) -> float:
+    return sum(values) / len(values)
+
+
+async def send_error(send: Callable[[dict[str, Any]], Awaitable[None]], status: int) -> None:
+    await send({"type": "http.response.start", "status": status, "headers": []})
+    await send({"type": "http.response.body", "body": b"Error"})
+
+
+async def handle_lifespan(receive, send):
+    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
+
+
 async def application(
     scope: dict[str, Any],
     receive: Callable[[], Awaitable[dict[str, Any]]],
@@ -12,7 +45,73 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    if scope["type"] == "lifespan":
+        await handle_lifespan(receive, send)
+
+    if scope["path"].startswith("/fibonacci/"):
+        await receive()
+        path: str = scope["path"]
+        path = path.rpartition("/")[2]
+        try:
+            num = int(path)
+        except ValueError:
+            await send_error(send, http.client.UNPROCESSABLE_ENTITY)
+            return
+        if num < 0:
+            await send_error(send, http.client.BAD_REQUEST)
+            return
+        result = await compute_fibonacci(num)
+        await send({"type": "http.response.start", "status": 200, "headers": [[b"content-type", b"application/json"]]})
+        await send({"type": "http.response.body", "body": f'{{"result": {result}}}'.encode()})
+        return
+
+    if scope["path"] == "/factorial":
+        await receive()
+        query_string = scope["query_string"].decode()
+        if not query_string or "n=" not in query_string:
+            await send_error(send, http.client.UNPROCESSABLE_ENTITY)
+            return
+        num_str = query_string.rpartition("=")[2]
+        if not num_str:
+            await send_error(send, http.client.UNPROCESSABLE_ENTITY)
+            return
+        try:
+            num = int(num_str)
+        except ValueError:
+            await send_error(send, http.client.UNPROCESSABLE_ENTITY)
+            return
+        if num < 0:
+            await send_error(send, http.client.BAD_REQUEST)
+            return
+        result = await compute_factorial(num)
+        await send({"type": "http.response.start", "status": 200, "headers": [[b"content-type", b"application/json"]]})
+        await send({"type": "http.response.body", "body": f'{{"result": {result}}}'.encode()})
+        return
+
+    if scope["path"] == "/mean":
+        body = await receive()
+        body_str = body["body"].decode()
+        if not body_str:
+            await send_error(send, http.client.UNPROCESSABLE_ENTITY)
+            return
+        try:
+            lst = ast.literal_eval(body_str)
+        except (ValueError, SyntaxError):
+            await send_error(send, http.client.UNPROCESSABLE_ENTITY)
+            return
+        if not isinstance(lst, list):
+            await send_error(send, http.client.UNPROCESSABLE_ENTITY)
+            return
+        if len(lst) == 0:
+            await send_error(send, http.client.BAD_REQUEST)
+            return
+        result = await compute_mean(lst)
+        await send({"type": "http.response.start", "status": 200, "headers": [[b"content-type", b"application/json"]]})
+        await send({"type": "http.response.body", "body": f'{{"result": {result}}}'.encode()})
+        return
+    await receive()
+    await send_error(send, http.client.NOT_FOUND)
+
 
 if __name__ == "__main__":
     import uvicorn

hw2/hw/shop_api/api/cart_routes.py

@@ -0,0 +1,117 @@
+from http import HTTPStatus
+from typing import Annotated
+
+from fastapi import APIRouter, HTTPException, Query, Response
+from pydantic import NonNegativeFloat, NonNegativeInt, PositiveInt
+
+from shop_api.store.queries import (
+    add_cart,
+    add_item_to_cart,
+    get_cart,
+    get_carts,
+    get_item,
+)
+
+from .contracts import CartIdResponse, CartResponse
+
+router = APIRouter(prefix="/cart")
+
+
+@router.post(
+    "",
+    status_code=HTTPStatus.CREATED,
+)
+async def post_cart(response: Response) -> CartIdResponse:
+    entity = add_cart()
+    response.headers["location"] = f"/cart/{entity.id}"
+    return CartIdResponse(id=entity.id)
+
+
+@router.get(
+    "/{id}",
+    responses={
+        HTTPStatus.OK: {
+            "description": "Successfully returned requested cart",
+        },
+        HTTPStatus.NOT_FOUND: {
+            "description": "Cart not found",
+        },
+    },
+)
+async def get_cart_by_id(id: int) -> CartResponse:
+    entity = get_cart(id)
+
+    if not entity:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Cart with id={id} not found",
+        )
+
+    items_dict = {}
+    for cart_item in entity.info.items:
+        item_entity = get_item(cart_item.id, include_deleted=True)
+        if item_entity:
+            items_dict[cart_item.id] = item_entity
+
+    return CartResponse.from_entity(entity, items_dict)
+
+
+@router.get("")
+async def get_cart_list(
+    offset: Annotated[NonNegativeInt, Query()] = 0,
+    limit: Annotated[PositiveInt, Query()] = 10,
+    min_price: Annotated[NonNegativeFloat | None, Query()] = None,
+    max_price: Annotated[NonNegativeFloat | None, Query()] = None,
+    min_quantity: Annotated[NonNegativeInt | None, Query()] = None,
+    max_quantity: Annotated[NonNegativeInt | None, Query()] = None,
+) -> list[CartResponse]:
+    entities = get_carts(
+        offset=offset,
+        limit=limit,
+        min_price=min_price,
+        max_price=max_price,
+        min_quantity=min_quantity,
+        max_quantity=max_quantity,
+    )
+
+    all_item_ids = set()
+    for cart_entity in entities:
+        for cart_item in cart_entity.info.items:
+            all_item_ids.add(cart_item.id)
+
+    items_dict = {}
+    for item_id in all_item_ids:
+        item_entity = get_item(item_id, include_deleted=True)
+        if item_entity:
+            items_dict[item_id] = item_entity
+
+    return [CartResponse.from_entity(e, items_dict) for e in entities]
+
+
+@router.post(
+    "/{cart_id}/add/{item_id}",
+    responses={
+        HTTPStatus.OK: {
+            "description": "Successfully added item to cart",
+        },
+        HTTPStatus.NOT_FOUND: {
+            "description": "Cart or item not found",
+        },
+    },
+)
+async def add_item_to_cart_handler(cart_id: int, item_id: int) -> CartResponse:
+    entity = add_item_to_cart(cart_id, item_id)
+
+    if not entity:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Cart with id={cart_id} or item with id={item_id} not found",
+        )
+
+    items_dict = {}
+    for cart_item in entity.info.items:
+        item_entity = get_item(cart_item.id, include_deleted=True)
+        if item_entity:
+            items_dict[cart_item.id] = item_entity
+
+    return CartResponse.from_entity(entity, items_dict)

hw2/hw/shop_api/api/chat_routes.py

@@ -0,0 +1,19 @@
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+
+from shop_api.websocket_chat import chat_manager
+
+router = APIRouter()
+
+
+@router.websocket("/chat/{chat_name}")
+async def chat_endpoint(websocket: WebSocket, chat_name: str):
+    room = chat_manager.get_room(chat_name)
+    await room.connect(websocket)
+
+    try:
+        while True:
+            message = await websocket.receive_text()
+            await room.broadcast(message, websocket)
+    except WebSocketDisconnect:
+        await room.disconnect(websocket)
+

hw2/hw/shop_api/api/contracts.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict, NonNegativeFloat
+
+from shop_api.store.models import (
+    CartEntity,
+    CartItemInfo,
+    ItemEntity,
+    ItemInfo,
+    PatchItemInfo,
+)
+
+
+class ItemResponse(BaseModel):
+    id: int
+    name: str
+    price: float
+    deleted: bool = False
+
+    @staticmethod
+    def from_entity(entity: ItemEntity) -> ItemResponse:
+        return ItemResponse(
+            id=entity.id,
+            name=entity.info.name,
+            price=entity.info.price,
+            deleted=entity.info.deleted,
+        )
+
+
+class ItemRequest(BaseModel):
+    name: str
+    price: NonNegativeFloat
+
+    def as_item_info(self) -> ItemInfo:
+        return ItemInfo(name=self.name, price=self.price, deleted=False)
+
+
+class PatchItemRequest(BaseModel):
+    name: str | None = None
+    price: NonNegativeFloat | None = None
+
+    model_config = ConfigDict(extra="forbid")
+
+    def as_patch_item_info(self) -> PatchItemInfo:
+        return PatchItemInfo(name=self.name, price=self.price)
+
+
+class CartItemResponse(BaseModel):
+    id: int
+    name: str
+    quantity: int
+    available: bool
+
+    @staticmethod
+    def from_cart_item_info(info: CartItemInfo) -> CartItemResponse:
+        return CartItemResponse(
+            id=info.id,
+            name=info.name,
+            quantity=info.quantity,
+            available=info.available,
+        )
+
+
+class CartResponse(BaseModel):
+    id: int
+    items: list[CartItemResponse]
+    price: float
+
+    @staticmethod
+    def from_entity(entity: CartEntity, items_dict: dict[int, ItemEntity]) -> CartResponse:
+        cart_items = [
+            CartItemResponse.from_cart_item_info(item) 
+            for item in entity.info.items
+        ]
+
+        total_price = 0.0
+        for item in entity.info.items:
+            if item.id in items_dict and item.available:
+                total_price += items_dict[item.id].info.price * item.quantity
+
+        return CartResponse(
+            id=entity.id,
+            items=cart_items,
+            price=total_price,
+        )
+
+
+class CartIdResponse(BaseModel):
+    id: int