Hw2 solution

hw1/app.py

@@ -1,5 +1,22 @@
 from typing import Any, Awaitable, Callable
+import json
+import math
+from http import HTTPStatus
+from urllib.parse import parse_qs
 
+def factorial(n: int) -> int:
+    return math.factorial(n)
+
+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 | int]) -> float:
+    if len(nums) == 0: 
+        raise TypeError()
+    return sum(nums) / len(nums)
 
 async def application(
     scope: dict[str, Any],
@@ -12,8 +29,88 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    # assert scope["type"] == "http"
+
+    path_segments = [seg for seg in scope.get("path", "").split("/") if seg]
+
+    status = HTTPStatus.OK
+    response_data = {}
+
+    match path_segments:
+        case ["factorial"]:
+            query_raw = scope.get("query_string", b"").decode("utf-8")
+            query_params = parse_qs(query_raw)
+
+            if "n" in query_params and (n_val := query_params["n"]):
+                try:
+                    n = int(n_val[0])
+                    if n < 0:
+                        status = HTTPStatus.BAD_REQUEST
+                    else:
+                        response_data = {"result": factorial(n)}
+                except ValueError:
+                    status = HTTPStatus.UNPROCESSABLE_ENTITY
+            else:
+                status = HTTPStatus.UNPROCESSABLE_ENTITY
+
+        case ["fibonacci", n_str]:
+            try:
+                n = int(n_str)
+                if n < 0:
+                    status = HTTPStatus.BAD_REQUEST
+                else:
+                    response_data = {"result": fibonacci(n)}
+            except ValueError:
+                status = HTTPStatus.UNPROCESSABLE_ENTITY
+
+        case ["fibonacci"]:
+            status = HTTPStatus.UNPROCESSABLE_ENTITY
+
+        case ["mean"]:
+            body = b""
+            received_any = False
+
+            while True:
+                message = await receive()
+                print("MSG:", message)
+                if message["type"] == "http.request":
+                    received_any = True
+                    body += message.get("body", b"")
+                    if not message.get("more_body", False):
+                        break
+                elif message["type"] == "http.disconnect":
+                    break
+
+            if not received_any or body.strip() == b'null':
+                # тело не передано → 422
+                status = HTTPStatus.UNPROCESSABLE_ENTITY
+                response_data = {"result": None}
+            else:
+                try:
+                    numbers = json.loads(body)
+                    if not isinstance(numbers, list):
+                        raise ValueError
+                    result = mean(numbers)
+                    response_data = {"result": result}
+                except (ValueError, json.JSONDecodeError, TypeError):
+                    # тело есть, но некорректно → 400
+                    status = HTTPStatus.BAD_REQUEST
+                    response_data = {"result": []}
+
+        case _:
+            status = HTTPStatus.NOT_FOUND
 
+    response_body = json.dumps(response_data).encode("utf-8")
+
+    await send({
+        "type": "http.response.start",
+        "status": status.value,
+        "headers": [(b"content-type", b"application/json")],
+    })
+    await send({
+        "type": "http.response.body",
+        "body": response_body,
+    })
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True)

hw2/hw/shop_api/contracts.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from .store.models import Cart, Item
+
+
+class StrictRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+class ItemCreateIn(StrictRequest):
+    name: str = Field(min_length=1)
+    price: float = Field(ge=0)
+
+    def to_domain(self) -> Item:
+        return Item(id=0, name=self.name, price=self.price, deleted=False)
+
+
+class ItemPutIn(StrictRequest):
+    name: str = Field(min_length=1)
+    price: float = Field(ge=0)
+
+
+class ItemPatchIn(StrictRequest):
+    name: str | None = Field(default=None, min_length=1)
+    price: float | None = Field(default=None, ge=0)
+
+
+class ItemOut(BaseModel):
+    id: int
+    name: str
+    price: float
+    deleted: bool
+
+    @classmethod
+    def from_domain(cls, item: Item) -> "ItemOut":
+        return cls(id=item.id, name=item.name, price=item.price, deleted=item.deleted)
+
+
+class CartItemOut(BaseModel):
+    id: int
+    name: str
+    quantity: int
+    available: bool
+
+
+class CartCreateOut(BaseModel):
+    id: int
+
+
+class CartOut(BaseModel):
+    id: int
+    items: list[CartItemOut]
+    price: float
+
+    @classmethod
+    def from_domain(cls, cart: Cart) -> "CartOut":
+        return cls(
+            id=cart.id,
+            items=[
+                CartItemOut(
+                    id=i.item_id,
+                    name=i.name,
+                    quantity=i.quantity,
+                    available=i.available,
+                )
+                for i in cart.items
+            ],
+            price=cart.price,
+        )

hw2/hw/shop_api/main.py

@@ -1,3 +1,5 @@
 from fastapi import FastAPI
+from .routes import router2
 
 app = FastAPI(title="Shop API")
+app.include_router(router2)

hw2/hw/shop_api/routes.py

@@ -0,0 +1,164 @@
+from fastapi import APIRouter, HTTPException, Query, Response, status
+
+from .contracts import (
+    CartCreateOut,
+    CartOut,
+    ItemCreateIn,
+    ItemOut,
+    ItemPatchIn,
+    ItemPutIn,
+)
+from .store.models import Cart, CartItem
+from .store import memory
+
+router2 = APIRouter()
+
+
+@router2.post("/item", response_model=ItemOut, status_code=status.HTTP_201_CREATED)
+def create_item(payload: ItemCreateIn) -> ItemOut:
+    item = payload.to_domain()
+    item.id = next(memory.item_id_generator)
+    memory.items[item.id] = item
+    return ItemOut.from_domain(item)
+
+
+@router2.get("/item/{item_id}", response_model=ItemOut)
+def get_item(item_id: int) -> ItemOut:
+    item = memory.items.get(item_id)
+    if item is None or item.deleted:
+        raise HTTPException(404, "item not found")
+    return ItemOut.from_domain(item)
+
+
+@router2.get("/item", response_model=list[ItemOut])
+def list_items(
+    offset: int = Query(0, ge=0),
+    limit: int = Query(10, gt=0),
+    min_price: float | None = Query(None, ge=0),
+    max_price: float | None = Query(None, ge=0),
+    show_deleted: bool = Query(False),
+) -> list[ItemOut]:
+    all_items = list(memory.items.values())
+
+    if not show_deleted:
+        all_items = [i for i in all_items if not i.deleted]
+    if min_price is not None:
+        all_items = [i for i in all_items if i.price >= min_price]
+    if max_price is not None:
+        all_items = [i for i in all_items if i.price <= max_price]
+
+    page = all_items[offset : offset + limit]
+
+    return [ItemOut.from_domain(i) for i in page]
+
+
+@router2.put("/item/{item_id}", response_model=ItemOut)
+def replace_item(item_id: int, payload: ItemPutIn) -> ItemOut:
+    existing = memory.items.get(item_id)
+    if existing is None or existing.deleted:
+        raise HTTPException(404, "item not found")
+
+    existing.name = payload.name
+    existing.price = payload.price
+    return ItemOut.from_domain(existing)
+
+
+@router2.patch("/item/{item_id}", response_model=ItemOut)
+def patch_item(item_id: int, payload: ItemPatchIn) -> ItemOut | Response:
+    item = memory.items.get(item_id)
+    if item is None:
+        raise HTTPException(404, "item not found")
+    if item.deleted:
+        return Response(status_code=status.HTTP_304_NOT_MODIFIED)
+
+    if "name" in payload.model_fields_set and payload.name is not None:
+        item.name = payload.name
+    if "price" in payload.model_fields_set and payload.price is not None:
+        item.price = payload.price
+
+    return ItemOut.from_domain(item)
+
+
+@router2.delete("/item/{item_id}")
+def delete_item(item_id: int) -> None:
+    item = memory.items.get(item_id)
+    if item is None:
+        raise HTTPException(404, "item not found")
+    item.deleted = True
+
+@router2.post("/cart", response_model=CartCreateOut, status_code=status.HTTP_201_CREATED)
+def create_cart(response: Response) -> CartCreateOut:
+    cart = Cart(id=next(memory.cart_id_generator))
+    memory.carts[cart.id] = cart
+    response.headers["location"] = f"/cart/{cart.id}"
+    return CartCreateOut(id=cart.id)
+
+
+@router2.get("/cart/{cart_id}", response_model=CartOut)
+def get_cart(cart_id: int) -> CartOut:
+    cart = memory.carts.get(cart_id)
+    if cart is None:
+        raise HTTPException(404, "cart not found")
+    return CartOut.from_domain(cart)
+
+
+@router2.get("/cart", response_model=list[CartOut])
+def list_carts(
+    offset: int = Query(0, ge=0),
+    limit: int = Query(10, gt=0),
+    min_price: float | None = Query(None, ge=0),
+    max_price: float | None = Query(None, ge=0),
+    min_quantity: int | None = Query(None, ge=0),
+    max_quantity: int | None = Query(None, ge=0),
+) -> list[CartOut]:
+    all_carts = list(memory.carts.values())
+
+    if min_price is not None:
+        all_carts = [c for c in all_carts if c.price >= min_price]
+    if max_price is not None:
+        all_carts = [c for c in all_carts if c.price <= max_price]
+
+    if min_quantity is not None:
+        all_carts = [
+            c for c in all_carts if sum(i.quantity for i in c.items) >= min_quantity
+        ]
+    if max_quantity is not None:
+        all_carts = [
+            c for c in all_carts if sum(i.quantity for i in c.items) <= max_quantity
+        ]
+
+    page = all_carts[offset : offset + limit]
+
+    return [CartOut.from_domain(c) for c in page]
+
+
+@router2.post("/cart/{cart_id}/add/{item_id}", response_model=CartOut)
+def add_item_to_cart(cart_id: int, item_id: int) -> CartOut:
+    cart = memory.carts.get(cart_id)
+    if cart is None:
+        raise HTTPException(404, "cart not found")
+
+    item = memory.items.get(item_id)
+    if item is None or item.deleted:
+        raise HTTPException(404, "item not found")
+    for ci in cart.items:
+        if ci.item_id == item_id:
+            ci.quantity += 1
+            ci.available = not item.deleted
+            break
+    else:
+        cart.items.append(
+            CartItem(
+                item_id=item.id,
+                name=item.name,
+                quantity=1,
+                available=not item.deleted,
+            )
+        )
+    cart.price = sum(
+        ci.quantity * memory.items[ci.item_id].price
+        for ci in cart.items
+        if ci.item_id in memory.items
+    )
+
+    return CartOut.from_domain(cart)

hw2/hw/shop_api/store/memory.py

@@ -0,0 +1,30 @@
+
+
+from .models import Cart, Item
+
+
+
+def int_item_id_generator():
+    i = 0
+    while True:
+        yield i
+        i += 1
+
+
+item_id_generator = int_item_id_generator()
+
+
+
+
+def int_cart_id_generator():
+    i = 0
+    while True:
+        yield i
+        i += 1
+
+
+cart_id_generator = int_cart_id_generator()
+
+
+carts: dict[int, Cart] = {}
+items: dict[int, Item] = {}