From 2fb629f4ac5d05d7689feee2f40ff2c780f29057 Mon Sep 17 00:00:00 2001 From: Eugene Vihrev Date: Sun, 28 Sep 2025 00:37:16 +0200 Subject: [PATCH] HW2 completed --- hw2/hw/shop_api/main.py | 144 ++++++++++++++++++++++++++++++++++++- hw2/hw/shop_api/schemas.py | 90 +++++++++++++++++++++++ 2 files changed, 233 insertions(+), 1 deletion(-) create mode 100644 hw2/hw/shop_api/schemas.py diff --git a/hw2/hw/shop_api/main.py b/hw2/hw/shop_api/main.py index f60a8c60..06ae53a2 100644 --- a/hw2/hw/shop_api/main.py +++ b/hw2/hw/shop_api/main.py @@ -1,3 +1,145 @@ -from fastapi import FastAPI +import http +import itertools +import typing + +from fastapi import FastAPI, HTTPException, Response, Query + +import shop_api.schemas as schemas app = FastAPI(title="Shop API") + + +class Store: + def __init__(self): + self.items: dict[int, schemas.Item] = {} + self.carts: dict[int, schemas.Cart] = {} + + self.next_item = 1 + self.next_cart = 1 + + +store = Store() + + +@app.post("/cart", status_code=http.HTTPStatus.CREATED) +async def create_cart(response: Response) -> schemas.WrappedID: + cart_id = store.next_cart + store.carts[cart_id] = schemas.Cart(id=cart_id, items={}) + store.next_cart += 1 + data = schemas.WrappedID(id=cart_id) + response.headers["location"] = f"/cart/{cart_id}" + return data + + +@app.get("/cart/{cart_id}") +async def get_cart(cart_id: int) -> schemas.CartResponse: + if cart_id not in store.carts: + raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND) + return store.carts[cart_id].create_cart_response(store.items)[0] + + +@app.get("/cart") +async def get_carts( + filter: typing.Annotated[schemas.GetCartsRequest, Query()], +) -> list[schemas.CartResponse]: + prepared_carts: typing.Generator[tuple[schemas.CartResponse, int]] = ( + cart.create_cart_response(store.items) for cart in store.carts.values() + ) + filtered_carts = ( + cart + for cart, quantity in prepared_carts + if all( + ( + (filter.min_price is None or cart.price >= filter.min_price), + (filter.max_price is None or cart.price <= filter.max_price), + (filter.min_quantity is None or quantity >= filter.min_quantity), + (filter.max_quantity is None or quantity <= filter.max_quantity), + ) + ) + ) + return list( + itertools.islice(filtered_carts, filter.offset, filter.offset + filter.limit) + ) + + +@app.post("/cart/{cart_id}/add/{item_id}") +async def add_to_cart(cart_id: int, item_id: int): + if cart_id not in store.carts: + raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND) + if item_id not in store.items or store.items[item_id].deleted: + raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND) + if item_id in store.carts[cart_id].items: + store.carts[cart_id].items[item_id] += 1 + else: + store.carts[cart_id].items[item_id] = 0 + return None + + +@app.post("/item", status_code=http.HTTPStatus.CREATED) +async def create_item( + payload: schemas.CreateItemRequest, response: Response +) -> schemas.Item: + item_id = store.next_item + store.items[item_id] = schemas.Item( + id=item_id, name=payload.name, price=payload.price, deleted=False + ) + store.next_item += 1 + response.headers["location"] = f"/item/{item_id}" + return store.items[item_id] + + +@app.get("/item/{item_id}") +async def get_item(item_id: int) -> schemas.Item: + if item_id not in store.items or store.items[item_id].deleted: + raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND) + return store.items[item_id] + + +@app.get("/item") +async def get_items( + filter: typing.Annotated[schemas.GetItemsRequest, Query()], +) -> list[schemas.Item]: + filtered_items = ( + item + for item in store.items.values() + if all( + ( + (filter.min_price is None or item.price >= filter.min_price), + (filter.max_price is None or item.price <= filter.max_price), + (not item.deleted or filter.show_deleted), + ) + ) + ) + return list( + itertools.islice(filtered_items, filter.offset, filter.offset + filter.limit) + ) + + +@app.put("/item/{item_id}") +async def put_item(item_id: int, payload: schemas.CreateItemRequest) -> schemas.Item: + if item_id not in store.items or store.items[item_id].deleted: + raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND) + store.items[item_id].name = payload.name + store.items[item_id].price = payload.price + return store.items[item_id] + + +@app.patch("/item/{item_id}") +async def patch_item(item_id: int, payload: schemas.UpdateItemRequest) -> schemas.Item: + if item_id not in store.items: + raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND) + if store.items[item_id].deleted: + raise HTTPException(status_code=http.HTTPStatus.NOT_MODIFIED) + if payload.name: + store.items[item_id].name = payload.name + if payload.price: + store.items[item_id].price = payload.price + return store.items[item_id] + + +@app.delete("/item/{item_id}") +async def delete_item(item_id: int) -> schemas.Item: + if item_id not in store.items: + raise HTTPException(status_code=http.HTTPStatus.NOT_FOUND) + store.items[item_id].deleted = True + return store.items[item_id] diff --git a/hw2/hw/shop_api/schemas.py b/hw2/hw/shop_api/schemas.py new file mode 100644 index 00000000..c993429f --- /dev/null +++ b/hw2/hw/shop_api/schemas.py @@ -0,0 +1,90 @@ +import pydantic + + +# Requests + + +class GetCartsRequest(pydantic.BaseModel): + offset: pydantic.NonNegativeInt = 0 + limit: pydantic.PositiveInt = 10 + min_price: pydantic.NonNegativeFloat | None = None + max_price: pydantic.NonNegativeFloat | None = None + min_quantity: pydantic.NonNegativeInt | None = None + max_quantity: pydantic.NonNegativeInt | None = None + + +class CreateItemRequest(pydantic.BaseModel): + name: str + price: float + + +class UpdateItemRequest(pydantic.BaseModel): + name: str | None = None + price: float | None = None + + model_config = pydantic.ConfigDict(extra="forbid") + + +class GetItemsRequest(pydantic.BaseModel): + offset: pydantic.NonNegativeInt = 0 + limit: pydantic.PositiveInt = 10 + min_price: pydantic.NonNegativeFloat | None = None + max_price: pydantic.NonNegativeFloat | None = None + show_deleted: bool = False + + +# Responses + + +class WrappedID(pydantic.BaseModel): + id: int + + +class CartResponseItem(pydantic.BaseModel): + id: int + name: str + quantity: int + available: bool + + +class CartResponse(pydantic.BaseModel): + id: int + items: list[CartResponseItem] + price: float + + +# Objects + + +class Item(pydantic.BaseModel): + id: int + name: str + price: float + deleted: bool = False + + +class Cart(pydantic.BaseModel): + id: int + items: dict[int, int] + + def create_cart_response(self, items: dict[int, Item]) -> tuple[CartResponse, int]: + price = 0.0 + total_quantity = 0 + prepated_items = [] + + for item_id, quantity in self.items.items(): + item = items[item_id] + total_quantity += quantity + price += item.price * quantity + prepated_items.append( + CartResponseItem( + id=item.id, + name=item.name, + quantity=quantity, + available=(not item.deleted), + ) + ) + + return CartResponse( + id=self.id, items=prepated_items, price=price + ), total_quantity