Implement ASGI application with fibonacci, factorial, and mean endpoints

hw1/app.py

@@ -1,6 +1,80 @@
+import json
 from typing import Any, Awaitable, Callable
 
 
+def get_fibonacci_n(path: str) -> int:
+    _, n = path.rsplit("/", 1)
+    return int(n)
+
+
+def fibonacci(n: int) -> int:
+    if n < 0:
+        raise ValueError("n should be non-negative")
+    if n < 2:
+        return n
+    a, b = 0, 1
+    for _ in range(n - 2):
+        a, b = a + b, a
+    return b
+
+
+def mean(numbers: list[int]) -> float:
+    if not numbers:
+        raise ValueError("Empty list")
+    return sum(numbers) / len(numbers)
+
+
+def factorial(n: int) -> int:
+    if n < 0:
+        raise ValueError("n should be non-negative")
+    res = 1
+    mul = 1
+    for _ in range(n):
+        res *= mul
+        mul += 1
+    return res
+
+
+def parse_query_params(query_params: bytes) -> dict[bytes, bytes]:
+    if not query_params:
+        return {}
+    query_groups = [query_group.split(b"=") for query_group in query_params.split(b"&")]
+    return {query_group[0]: query_group[1] for query_group in query_groups}
+
+
+async def _send_response(send, message: str, status: int) -> None:
+    await send({
+        "type": "http.response.start",
+        "status": status,
+        "headers": [(b"content-type", b"text/plain")],
+    })
+    await send({
+        "type": "http.response.body",
+        "body": message.encode(),
+    })
+
+
+async def _send_result(send, result) -> None:
+    await send({
+        "type": "http.response.start",
+        "status": 200,
+        "headers": [(b"content-type", b"application/json")],
+    })
+    await send({
+        "type": "http.response.body",
+        "body": json.dumps({"result": result}).encode(),
+    })
+
+
+async def get_json_body(receive):
+    res = []
+    while True:
+        message = await receive()
+        res.append(message["body"])
+        if not message.get("more_body", False):
+            return json.loads(b"".join(res))
+
+
 async def application(
     scope: dict[str, Any],
     receive: Callable[[], Awaitable[dict[str, Any]]],
@@ -12,7 +86,58 @@ async def application(
         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'})
+                return
+
+    send_response = lambda *args, **kwargs: _send_response(send, *args, **kwargs)
+    send_result = lambda *args, **kwargs: _send_result(send, *args, **kwargs)
+    path = scope["path"]
+    method_name = path.split("/")[1]
+    if method_name not in ("fibonacci", "mean", "factorial"):
+        return await send_response("Wrong method", 404)
+    elif method_name == "fibonacci":
+        try:
+            n = get_fibonacci_n(path)
+        except ValueError:
+            return await send_response("Could not parse n", 422)
+        try:
+            res = fibonacci(n)
+        except ValueError:
+            return await send_response("Could not calculate fibonacci", 400)
+        return await send_result(res)
+    query_params = parse_query_params(scope["query_string"])
+    if method_name == "mean":
+        numbers = await get_json_body(receive)
+        if not numbers:
+            if numbers is None:
+                return await send_response("empty body", 422)
+            if numbers is None:
+                return await send_response("empty list", 400)
+        try:
+            res = mean(numbers)
+        except ValueError:
+            return await send_response("Empty list", 400)
+        return await send_result(res)
+    elif method_name == "factorial":
+        if b"n" not in query_params:
+            return await send_response("n required for factorial", 422)
+        try:
+            n = int(query_params[b"n"])
+        except ValueError:
+            return await send_response("Could not parse n", 422)
+        try:
+            res = factorial(n)
+        except ValueError:
+            return await send_response("Could not calculate factorial", 400)
+        return await send_result(res)
+    assert 0
+
 
 if __name__ == "__main__":
     import uvicorn

hw2/hw/shop_api/api/__init__.py

@@ -0,0 +1,4 @@
+from shop_api.api.cart.routes import router as cart_router
+from shop_api.api.item.routes import router as item_router
+
+__all__ = ["cart_router", "item_router"]

hw2/hw/shop_api/api/cart/__init__.py

@@ -0,0 +1 @@
+from shop_api.api.cart import routes

hw2/hw/shop_api/api/cart/routes.py

@@ -0,0 +1,54 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Query, Response, HTTPException, status
+from pydantic import NonNegativeInt, PositiveInt
+
+from shop_api.store.queries import get_carts as _get_carts
+from shop_api.store.queries import get_items as _get_items
+from shop_api.store.queries import create_cart as _create_cart
+from shop_api.store.queries import add_item_to_cart as _add_item_to_cart
+from shop_api.store.models import Cart
+
+
+router = APIRouter(prefix="/cart")
+
+
+@router.post("/", status_code=status.HTTP_201_CREATED)
+async def create_cart(response: Response) -> Cart:
+    cart = _create_cart()
+    response.headers["Location"] = f"/cart/{cart.id}"
+    return cart
+
+
+@router.get("/")
+async def get_carts(
+    offset: Annotated[NonNegativeInt, Query()] = 0,
+    limit: Annotated[PositiveInt, Query()] = 10,
+    min_price: Annotated[float | None, Query(gt=0)] = None,
+    max_price: Annotated[float | None, Query(gt=0)] = None,
+    min_quantity: Annotated[int | None, Query(gt=0)] = None,
+    max_quantity: Annotated[int | None, Query(ge=0)] = None,
+) -> list[Cart]:
+    return _get_carts(
+        offset, limit, min_price, max_price, min_quantity, max_quantity
+    )
+
+
+@router.get("/{id}")
+async def get_cart(id: int) -> Cart:
+    carts = _get_carts(offset=id, limit=1)
+    if not carts:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+    return carts[0]
+
+
+@router.post("/{cart_id}/add/{item_id}", status_code=status.HTTP_201_CREATED)
+async def add_item_to_cart(cart_id: int, item_id: int) -> Cart:
+    cart = _get_carts(offset=cart_id, limit=1)
+    if not cart:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Cart not found")
+    item = _get_items(offset=item_id, limit=1)
+    if not item:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Item not found")
+    _add_item_to_cart(cart[0], item[0])
+    return cart[0]

hw2/hw/shop_api/api/item/contracts.py

@@ -0,0 +1,20 @@
+from pydantic import BaseModel
+
+
+class ItemPostRequest(BaseModel):
+    name: str
+    price: float
+
+
+class ItemPutRequest(BaseModel):
+    name: str
+    price: float
+
+
+class ItemPatchRequest(BaseModel):
+    name: str | None = None
+    price: float | None = None
+
+    model_config = {
+        "extra": "forbid",
+    }

hw2/hw/shop_api/api/item/routes.py

@@ -0,0 +1,68 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Query, Response, HTTPException, status
+from pydantic import NonNegativeInt, PositiveInt
+
+from shop_api.store.queries import get_items as _get_items
+from shop_api.store.queries import create_item as _create_item
+from shop_api.store.queries import replace_item as _replace_item
+from shop_api.store.queries import patch_item as _patch_item
+from shop_api.store.models import Item
+from shop_api.api.item.contracts import ItemPostRequest, ItemPutRequest, ItemPatchRequest
+
+
+router = APIRouter(prefix="/item")
+
+
+@router.post("/", status_code=status.HTTP_201_CREATED)
+async def create_item(request: ItemPostRequest, response: Response) -> Item:
+    item = _create_item(request.name, request.price)
+    response.headers["Location"] = f"/item/{item.id}"
+    return item
+
+
+@router.get("/")
+async def get_items(
+    offset: Annotated[NonNegativeInt, Query()] = 0,
+    limit: Annotated[PositiveInt, Query()] = 10,
+    min_price: Annotated[float | None, Query(gt=0)] = None,
+    max_price: Annotated[float | None, Query(gt=0)] = None,
+    show_deleted: Annotated[bool, Query()] = False,
+) -> list[Item]:
+    return _get_items(
+        offset, limit, min_price, max_price, show_deleted
+    )
+
+
+@router.get("/{id}")
+async def get_item(id: int) -> Item:
+    items = _get_items(offset=id, limit=1)
+    if not items:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+    return items[0]
+
+
+@router.put("/{id}")
+async def put_item(id: int, request: ItemPutRequest) -> Item:
+    items = _get_items(offset=id, limit=1)
+    if not items:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
+    _replace_item(items[0].id, request.name, request.price)
+    return items[0]
+
+
+@router.patch("/{id}")
+async def patch_item(id: int, request: ItemPatchRequest) -> Item:
+    items = _get_items(offset=id, limit=1)
+    if not items:
+        raise HTTPException(status_code=status.HTTP_304_NOT_MODIFIED)
+    _patch_item(items[0].id, request.name, request.price)
+    return items[0]
+
+
+@router.delete("/{id}")
+async def delete_item(id: int) -> None:
+    items = _get_items(offset=id, limit=1)
+    if not items:
+        return
+    items[0].deleted = True

hw2/hw/shop_api/main.py

@@ -1,3 +1,8 @@
 from fastapi import FastAPI
 
+from shop_api.api import cart_router, item_router
+
 app = FastAPI(title="Shop API")
+
+app.include_router(cart_router)
+app.include_router(item_router)

hw2/hw/shop_api/store/models.py

@@ -0,0 +1,21 @@
+from pydantic import BaseModel
+
+
+class Item(BaseModel):
+    id: int
+    name: str
+    price: float
+    deleted: bool = False
+
+
+class CartItem(BaseModel):
+    id: int
+    name: str
+    quantity: int
+    available: bool
+
+
+class Cart(BaseModel):
+    id: int
+    items: list[CartItem] = []
+    price: float = 0.0