diff --git a/hw1/app.py b/hw1/app.py index 6107b87..5bba371 100644 --- a/hw1/app.py +++ b/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) diff --git a/hw2/hw/shop_api/contracts.py b/hw2/hw/shop_api/contracts.py new file mode 100644 index 0000000..a49ccf2 --- /dev/null +++ b/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, + ) \ No newline at end of file diff --git a/hw2/hw/shop_api/main.py b/hw2/hw/shop_api/main.py index f60a8c6..df7d789 100644 --- a/hw2/hw/shop_api/main.py +++ b/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) diff --git a/hw2/hw/shop_api/routes.py b/hw2/hw/shop_api/routes.py new file mode 100644 index 0000000..46308e8 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/hw2/hw/shop_api/store/__init__.py b/hw2/hw/shop_api/store/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/hw2/hw/shop_api/store/memory.py b/hw2/hw/shop_api/store/memory.py new file mode 100644 index 0000000..6bad553 --- /dev/null +++ b/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] = {} \ No newline at end of file diff --git a/hw2/hw/shop_api/store/models.py b/hw2/hw/shop_api/store/models.py new file mode 100644 index 0000000..4d7d940 --- /dev/null +++ b/hw2/hw/shop_api/store/models.py @@ -0,0 +1,22 @@ +from __future__ import annotations +from pydantic import BaseModel + + +class Item(BaseModel): + id: int + name: str + price: float + deleted: bool = False + + +class CartItem(BaseModel): + item_id: int + name: str + quantity: int + available: bool + + +class Cart(BaseModel): + id: int + items: list[CartItem] = [] + price: float = 0.0 \ No newline at end of file