Implement ASGI application with fibonacci, factorial, and mean endpoints

hw1/app.py

@@ -1,5 +1,7 @@
 from typing import Any, Awaitable, Callable
-
+from urllib.parse import parse_qs
+import math
+import json
 
 async def application(
     scope: dict[str, Any],
@@ -12,7 +14,152 @@ 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"})
+                break
+        return
+
+    if scope["type"] == "http":
+        path = scope['path']
+        if path == '/factorial':
+            query_string = scope["query_string"].decode("utf-8")
+            params = parse_qs(query_string)
+            if "n" not in params or not params["n"][0]:
+                await send_error(send, 422, "Missing or empty parameter n")
+                return
+            try:
+                n = int(params["n"][0])
+            except (ValueError, TypeError):
+                await send_error(send, 422, "n must be an integer")
+                return
+            if n < 0:
+                await send_error(send, 400, "n must be non-negative")
+                return
+
+            result = math.factorial(int(params["n"][0]))
+            response = json.dumps({"result": result})
+
+            await send({
+                    "type": "http.response.start",
+                    "status": 200,
+                    "headers": [
+                        [b"content-type", b"application/json"]
+                    ],
+                })
+            await send({
+                    "type": "http.response.body",
+                    "body": response.encode("utf-8"),
+                })            
+
+        elif path == '/mean':
+            body = b""
+            while True:
+                message = await receive()
+                body += message.get("body", b"")
+                if not message.get("more_body", False):
+                    break
+
+            if not body:
+                await send_error(send, 422, "No JSON given")
+                return
+            try:
+                text = body.decode("utf-8")
+                data = json.loads(text)
+            except Exception:
+                await send_error(send, 422, "Invalid JSON")
+                return
+
+
+            if data is None:
+                await send_error(send, 422, "No JSON given")
+                return
+            if not isinstance(data, list):
+                await send_error(send, 400, "Data must be a list")
+                return
+            if len(data) == 0:
+                await send_error(send, 400, "Empty list")
+                return
+            elif not all(isinstance(x, (int, float)) for x in data):
+                await send_error(send, 400, "All elements must be numbers")
+                return
+            else:
+                result = sum(data)/len(data)
+                response = json.dumps({"result": result})
+
+                await send({
+                    "type": "http.response.start",
+                    "status": 200,
+                    "headers": [
+                        [b"content-type", b"application/json"]
+                    ],
+                })
+                await send({
+                    "type": "http.response.body",
+                    "body": response.encode("utf-8"),
+                })
+
+
+        elif path.startswith('/fibonacci/'):
+            parts = path.split('/')
+            if len(parts) != 3 or not parts[2]:
+                await send_error(send, 422, "Invalid path parameter")
+                return
+            n_str = parts[2]
+            try:
+                n = int(n_str)
+            except ValueError:
+                await send_error(send, 422, "n must be an integer")
+                return
+            if n < 0:
+                await send_error(send, 400, "n must be non-negative")
+                return
+
+            def fib(n):
+                a, b = 0, 1
+                for _ in range(n):
+                    a, b = b, a + b
+                return a
+
+            result = fib(n)
+            response = json.dumps({"result": result})
+
+            await send({
+                "type": "http.response.start",
+                "status": 200,
+                "headers": [
+                    [b"content-type", b"application/json"]
+                ],
+            })
+            await send({
+                "type": "http.response.body",
+                "body": response.encode("utf-8"),
+            })
+            return
+
+        else:
+            await send_error(send, 404, "There's no endpoint like that")
+            return
+
+async def send_error(send, status: int, message: str = ""):
+    await send({
+        "type": "http.response.start",
+        "status": status,
+        "headers": [
+            [b"content-type", b"text/plain; charset=utf-8"]
+        ],
+    })
+    await send({
+        "type": "http.response.body",
+        "body": message.encode("utf-8"),
+    })
+
+
+
 
 if __name__ == "__main__":
     import uvicorn

hw2/hw/shop_api/main.py

@@ -1,3 +1,190 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, Query, HTTPException, Response
+from pydantic import BaseModel
+from typing import Optional, List
 
 app = FastAPI(title="Shop API")
+
+items_memory = {}
+
+class Item(BaseModel):
+    id: int
+    name: str
+    price: float
+    deleted: bool
+
+class ItemCreate(BaseModel):
+    name: str
+    price: float
+
+class ItemPatch(BaseModel):
+    name: Optional[str] = None
+    price: Optional[float] = None
+
+    class Config:
+        extra = "forbid"
+
+class ItemPut(BaseModel):
+    name: str
+    price: float
+    deleted: Optional[bool] = None
+
+item_counter = 0
+
+@app.post('/item', status_code=201)
+def create_item(item: ItemCreate):
+    global item_counter
+    new_item = Item(id=item_counter, name=item.name, price=item.price, deleted=False)
+    items_memory[item_counter] = new_item
+    item_counter += 1
+    return new_item
+
+@app.get('/item/{item_id}')
+def get_item(item_id:int):
+    if item_id not in items_memory:
+        raise HTTPException(status_code=404, detail="Item not found")
+    if items_memory[item_id].deleted:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return items_memory[item_id]
+
+@app.get('/item')
+def get_item_list(
+    limit: Optional[int] = Query(default=10, ge=1),
+    offset: Optional[int] = Query(default=0, ge=0),
+    min_price: Optional[float] = Query(default=None, ge = 0),
+    max_price: Optional[float] = Query(default=None, ge = 0),
+    show_deleted: Optional[bool] = False
+):
+    all_items = list(items_memory.values())
+    if not show_deleted:
+        all_items = [v for v in all_items if not v.deleted]
+    if min_price is not None:
+        all_items = [v for v in all_items if v.price >= min_price]
+    if max_price is not None:
+        all_items = [v for v in all_items if v.price <= max_price]
+    return all_items[offset:offset+limit]
+
+
+@app.put('/item/{item_id}')
+def put_item(item_id: int, item: ItemPut):
+    if item_id not in items_memory:
+      raise HTTPException(status_code=404, detail="Item not found")
+    if item.deleted is not None:
+        items_memory[item_id] = Item(id=item_id, name=item.name, price=item.price, deleted=item.deleted)
+    if item.deleted is None:
+        items_memory[item_id] = Item(id=item_id, name=item.name, price=item.price, deleted=items_memory[item_id].deleted)
+    return items_memory[item_id]
+
+@app.patch('/item/{item_id}')
+def patch_item(item_id: int, item: ItemPatch):
+    if item_id not in items_memory:
+        raise HTTPException(status_code=404, detail="Item not found")
+    # Проверяем, не удалён ли товар
+    if items_memory[item_id].deleted:
+        raise HTTPException(status_code=304, detail="Item is deleted")
+
+    if item.price is not None:
+        items_memory[item_id].price = item.price
+    if item.name is not None:
+        items_memory[item_id].name = item.name
+    return items_memory[item_id]
+
+@app.delete('/item/{item_id}')
+def delete_item(item_id: int):
+    if item_id not in items_memory:
+      raise HTTPException(status_code=404, detail="Item not found")
+    items_memory[item_id].deleted = True
+    return 
+
+
+#  - `PUT /item/{id}` - замена товара по `id` (создание запрещено, только замена существующего)
+#  - `PATCH /item/{id}` - частичное обновление товара по `id` (разрешено менять все поля, кроме `deleted`)
+#  - `DELETE /item/{id}` - удаление товара по `id` (товар помечается как удаленный)
+
+# --- Cart implementation ---
+
+class CartItem(BaseModel):
+    id: int
+    name: str
+    quantity: int
+    available: bool
+
+class Cart(BaseModel):
+    id: int
+    items: List[CartItem] = []
+    price: float = 0.0
+
+carts_memory = {}
+cart_counter = 0
+
+@app.post('/cart', status_code=201)
+def create_cart(response: Response):
+    global cart_counter
+    cart = Cart(id=cart_counter, items=[], price=0.0)
+    carts_memory[cart_counter] = cart
+    cart_counter += 1
+    response.headers["location"] = f"/cart/{cart.id}"
+    return cart
+
+@app.get('/cart/{cart_id}')
+def get_cart(cart_id: int):
+    if cart_id not in carts_memory:
+        raise HTTPException(status_code=404, detail="Cart not found")
+    return carts_memory[cart_id]
+
+@app.get('/cart')
+def get_cart_list(
+    limit: Optional[int] = Query(default=10, ge=1),
+    offset: Optional[int] = Query(default=0, ge=0),
+    min_price: Optional[float] = Query(default=None, ge=0),
+    max_price: Optional[float] = Query(default=None, ge=0),
+    min_quantity: Optional[int] = Query(default=None, ge=0),
+    max_quantity: Optional[int] = Query(default=None, ge=0)
+):
+    all_carts = list(carts_memory.values())
+
+    if min_price is not None:
+        all_carts = [cart for cart in all_carts if cart.price >= min_price]
+    if max_price is not None:
+        all_carts = [cart for cart in all_carts if cart.price <= max_price]
+
+    if min_quantity is not None:
+        all_carts = [cart for cart in all_carts if sum(item.quantity for item in cart.items) >= min_quantity]
+    if max_quantity is not None:
+        all_carts = [cart for cart in all_carts if sum(item.quantity for item in cart.items) <= max_quantity]
+
+    return all_carts[offset:offset+limit]
+
+@app.post('/cart/{cart_id}/add/{item_id}')
+def add_item_to_cart(cart_id: int, item_id: int):
+    if cart_id not in carts_memory:
+        raise HTTPException(status_code=404, detail="Cart not found")
+    if item_id not in items_memory:
+        raise HTTPException(status_code=404, detail="Item not found")
+    if items_memory[item_id].deleted:
+        raise HTTPException(status_code=404, detail="Item not found")
+
+    cart = carts_memory[cart_id]
+    item = items_memory[item_id]
+
+    existing_item = None
+    for cart_item in cart.items:
+        if cart_item.id == item_id:
+            existing_item = cart_item
+            break
+
+    if existing_item:
+        existing_item.quantity += 1
+    else:
+        cart_item = CartItem(
+            id=item_id,
+            name=item.name,
+            quantity=1,
+            available=not item.deleted
+        )
+        cart.items.append(cart_item)
+
+    total_price = sum(items_memory[cart_item.id].price * cart_item.quantity for cart_item in cart.items if cart_item.id in items_memory and not items_memory[cart_item.id].deleted)
+    cart.price = total_price
+
+    return cart
+