HW1 + HW2

hw1/app.py

@@ -1,18 +1,102 @@
 from typing import Any, Awaitable, Callable
+import json
+
+
+def response_start(body, code):
+    return {
+        'type': 'http.response.start',
+        'status': code,
+        'headers': [
+            (b'content-type', b'application/json'),
+            (b'content-length', str(len(body)).encode())
+        ]
+    }
+
+
+def response_body(body):
+    return {
+        'type': 'http.response.body',
+        'body': body.encode(),
+    }
+
+
+def calculate_factorial(n):
+    return 1 if n < 2 else calculate_factorial(n - 1) * n
+
+
+def calculate_fibonacci(n):
+    return n if n in (0, 1) else calculate_fibonacci(n - 1) + calculate_fibonacci(n - 2)
+
+
+def calculate_mean(numbers):
+    return sum(numbers) / len(numbers)
 
 
 async def application(
-    scope: dict[str, Any],
-    receive: Callable[[], Awaitable[dict[str, Any]]],
-    send: Callable[[dict[str, Any]], Awaitable[None]],
+        scope: dict[str, Any],
+        receive: Callable[[], Awaitable[dict[str, Any]]],
+        send: Callable[[dict[str, Any]], Awaitable[None]],
 ):
     """
     Args:
         scope: Словарь с информацией о запросе
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    if scope['type'] == 'lifespan':
+        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
+
+    elif scope['type'] == 'http':
+        body = ''
+        code = 200
+        if scope['path'] == '/factorial':
+            if scope['query_string'].startswith(b"n="):
+                key, value = scope['query_string'].decode().split('=')
+                try:
+                    n = int(value)
+                    if n < 0:
+                        code = 400
+                    else:
+                        code = 200
+                        body = json.dumps({"result": calculate_factorial(n)})
+                except ValueError:
+                    code = 422
+            else:
+                code = 422
+
+        elif scope['path'].startswith('/fibonacci/'):
+            try:
+                n = int(scope['path'][len('/fibonacci/'):])
+                if n < 0:
+                    code = 400
+                else:
+                    code = 200
+                    body = json.dumps({"result": calculate_fibonacci(n)})
+            except ValueError:
+                code = 422
+
+        elif scope['path'] == '/mean':
+            message = await receive()
+            request_body = json.loads(message['body'])
+            try:
+                request_body = list(request_body)
+                if len(request_body) == 0:
+                    code = 400
+                else:
+                    body = json.dumps({"result": calculate_mean(request_body)})
+            except TypeError:
+                code = 422
+
+        else:
+            code = 404
+        await send(response_start(body, code))
+        await send(response_body(body))
 
 if __name__ == "__main__":
     import uvicorn

hw2/hw/shop_api/main.py

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

hw2/hw/shop_api/models.py

@@ -0,0 +1,53 @@
+from typing import Optional
+
+from pydantic import (BaseModel, ConfigDict, NonNegativeFloat, NonNegativeInt,
+                      PositiveFloat, PositiveInt)
+
+
+class BaseItem(BaseModel):
+    name: str
+    price: PositiveFloat
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class PatchItem(BaseModel):
+    name: Optional[str] = None
+    price: Optional[PositiveFloat] = None
+
+    model_config = ConfigDict(extra="forbid")
+
+
+class Item(BaseItem):
+    id: int
+    deleted: bool
+
+
+class CartItem(BaseModel):
+    id: int
+    name: str
+    quantity: PositiveInt
+    available: bool
+
+
+class Cart(BaseModel):
+    id: int
+    items: list[CartItem]
+    price: float
+
+
+class CartFilters(BaseModel):
+    offset: NonNegativeInt = 0
+    limit: PositiveInt = 10
+    min_price: NonNegativeFloat | None = None
+    max_price: NonNegativeFloat | None = None
+    min_quantity: NonNegativeInt | None = None
+    max_quantity: NonNegativeInt | None = None
+
+
+class ItemFilters(BaseModel):
+    offset: NonNegativeInt = 0
+    limit: PositiveInt = 10
+    min_price: NonNegativeFloat | None = None
+    max_price: NonNegativeFloat | None = None
+    show_deleted: bool = False

hw2/hw/shop_api/queries.py

@@ -0,0 +1,116 @@
+import itertools
+
+from shop_api.models import (BaseItem, Cart, CartFilters, CartItem, Item, ItemFilters,
+                    PatchItem)
+
+_carts = dict[int, Cart]()
+_items = dict[int, Item]()
+_carts_items_quantity_map = dict[int, dict[int, int]]()
+id_generator_cart = itertools.count(start=1, step=1)
+id_generator_item = itertools.count(start=1, step=1)
+
+
+def create_empty_cart() -> Cart:
+    _id = next(id_generator_cart)
+    _carts[_id] = Cart(id=_id, items=[], price=0.0)
+    _carts_items_quantity_map[_id] = {}
+    return _carts[_id]
+
+
+def generate_cart_items(cart_id: int) -> (list[CartItem], float):
+    cart_items: list[CartItem] = []
+    price: float = 0.0
+    for item_id, item_quantity in _carts_items_quantity_map[cart_id].items():
+        cart_item = CartItem(id=item_id,
+                             name=_items.get(item_id).name,
+                             quantity=item_quantity,
+                             available=not _items.get(item_id).deleted)
+        cart_items.append(cart_item)
+        price += _items.get(cart_item.id).price * item_quantity
+    return cart_items, price
+
+
+def get_cart_by_id(cart_id: int) -> Cart | None:
+    cart = _carts.get(cart_id)
+    if cart and len(_carts_items_quantity_map[cart_id]) > 0:
+        cart.items, cart.price = generate_cart_items(cart_id)
+    return cart
+
+
+def add_item(item: BaseItem) -> Item:
+    _id = next(id_generator_item)
+    _items[_id] = Item(id=_id, name=item.name, price=item.price, deleted=False)
+    return _items[_id]
+
+
+def get_item_by_id(item_id: int) -> Item | None:
+    item = _items.get(item_id)
+    return item
+
+
+def add_to_cart(cart_id: int, item_id: int) -> Cart | None:
+    cart = _carts.get(cart_id)
+    item = _items.get(item_id)
+    if not cart or not item:
+        return None
+    _carts_items_quantity_map[cart_id][item_id] = _carts_items_quantity_map[cart_id].get(item_id, 0) + 1
+    cart.items, cart.price = generate_cart_items(cart_id)
+    return cart
+
+
+def get_carts_filtered(filters: CartFilters) -> list[Cart]:
+    carts = list(_carts.values())
+
+    def matcher(cart: Cart) -> bool:
+        if filters.max_price is not None and cart.price > filters.max_price:
+            return False
+        if filters.min_price is not None and cart.price < filters.min_price:
+            return False
+        if filters.max_quantity is not None and (sum([i.quantity for i in cart.items]) > filters.max_quantity):
+            return False
+        if filters.min_quantity is not None and (sum([i.quantity for i in cart.items]) < filters.min_quantity):
+            return False
+        return True
+
+    carts_filtered = list(filter(matcher, carts))[filters.offset:filters.offset + filters.limit]
+    return carts_filtered
+
+
+def get_items_filtered(filters: ItemFilters) -> list[Item]:
+    items = list(_items.values())
+
+    def matcher(item: Item) -> bool:
+        if filters.max_price is not None and item.price > filters.max_price:
+            return False
+        if filters.min_price is not None and item.price < filters.min_price:
+            return False
+        if not filters.show_deleted and item.deleted:
+            return False
+
+    items_filtered = list(filter(matcher, items))[filters.offset:filters.offset + filters.limit]
+    return items_filtered
+
+
+def delete_item_by_id(item_id: int) -> Item | None:
+    item = _items.get(item_id)
+    if not item:
+        return None
+    item.deleted = True
+    return item
+
+
+def patch_item_query(item_id: int, new_fields: PatchItem) -> Item | None:
+    item = _items.get(item_id)
+    if not item.deleted and new_fields.name:
+        item.name = new_fields.name
+    if not item.deleted and new_fields.price:
+        item.price = new_fields.price
+
+    return item
+
+
+def put_item_query(item_id: int, new_fields: BaseItem) -> Item:
+    item = _items.get(item_id)
+    item.name, item.price = new_fields.name, new_fields.price
+
+    return item

hw2/hw/shop_api/routes.py

@@ -0,0 +1,111 @@
+from http import HTTPStatus
+
+from fastapi import APIRouter, Depends, HTTPException, Response
+from fastapi.responses import JSONResponse
+from shop_api.models import BaseItem, Cart, CartFilters, Item, ItemFilters, PatchItem
+from shop_api.queries import (add_item, add_to_cart, create_empty_cart,
+                     delete_item_by_id, get_cart_by_id, get_carts_filtered,
+                     get_item_by_id, get_items_filtered, patch_item_query,
+                     put_item_query)
+
+router = APIRouter()
+
+
+@router.post("/cart")
+async def create_cart():
+    cart = create_empty_cart()
+    return JSONResponse(content={"id": cart.id}, status_code=HTTPStatus.CREATED,
+                        headers={"location": f"/cart/{cart.id}"})
+
+
+@router.get("/cart/{cart_id}")
+async def get_cart(cart_id: int) -> Cart:
+    cart = get_cart_by_id(cart_id)
+    if cart is None:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Cart with id={cart_id} was not found",
+        )
+    return cart
+
+
+@router.post("/item", status_code=HTTPStatus.CREATED)
+async def create_item(item: BaseItem, response: Response) -> Item:
+    _item = add_item(item)
+    response.headers["location"] = f"/item/{_item.id}"
+    return _item
+
+
+@router.post("/cart/{cart_id}/add/{item_id}")
+async def add_item_to_cart(cart_id: int, item_id: int) -> Cart:
+    cart = add_to_cart(cart_id, item_id)
+    if cart is None:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Could not add item to cart; either item or cart not found",
+        )
+    return cart
+
+
+@router.get("/item/{item_id}")
+async def get_item(item_id: int) -> Item:
+    item = get_item_by_id(item_id)
+    if item is None or item.deleted:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Item with id={item_id} was not found",
+        )
+    return item
+
+
+@router.get("/cart")
+async def get_carts_with_filters(filter_params: CartFilters = Depends()) -> list[Cart]:
+    carts = get_carts_filtered(filter_params)
+    return carts
+
+
+@router.get("/item")
+async def get_items_with_filters(filter_params: ItemFilters = Depends()) -> list[Item]:
+    items = get_items_filtered(filter_params)
+    return items
+
+
+@router.delete("/item/{item_id}")
+async def delete_item(item_id: int) -> Item:
+    item = delete_item_by_id(item_id)
+    if item is None:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Item with id={item_id} was not found",
+        )
+    return item
+
+
+@router.patch("/item/{item_id}")
+async def patch_item(item_id: int, new_item_fields: PatchItem, response: Response) -> Item:
+    item_before = get_item_by_id(item_id)
+    if item_before is None:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Item with id={item_id} was not found",
+        )
+
+    item = patch_item_query(item_id, new_item_fields)
+    if item.deleted:
+        response.status_code = HTTPStatus.NOT_MODIFIED
+        response.body = None
+    return item
+
+
+@router.put("/item/{item_id}")
+async def put_item(item_id: int, new_item_fields: BaseItem) -> Item:
+    item_before = get_item_by_id(item_id)
+    if item_before is None:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Item with id={item_id} was not found",
+        )
+    item = put_item_query(item_id, new_item_fields)
+    return item
+
+