HWs

hw1/app.py

@@ -1,4 +1,34 @@
+import json
+from http import HTTPStatus
 from typing import Any, Awaitable, Callable
+from urllib.parse import parse_qs
+
+
+async def send_json(send, status: int, payload: dict[str, Any]) -> None:
+    body = json.dumps(payload).encode("utf-8")
+    await send({
+        "type": "http.response.start",
+        "status": status,
+        "headers": [(b"content-type", b"application/json")],
+    })
+    await send({
+        "type": "http.response.body",
+        "body": body,
+    })
+
+
+def factorial(n: int) -> int:
+    result = 1
+    for i in range(2, n + 1):
+        result *= i
+    return result
+
+
+def fibonacci(n: int) -> int:
+    a, b = 0, 1
+    for _ in range(n):
+        a, b = b, a + b
+    return a
 
 
 async def application(
@@ -12,7 +42,71 @@ 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
+
+    method = scope["method"]
+    path = scope["path"]
+
+    if method != "GET":
+        await send_json(send, HTTPStatus.NOT_FOUND, {"error": "not found"})
+        return
+
+    if path == "/factorial":
+        query = parse_qs(scope["query_string"].decode())
+        n_values = query.get("n")
+        if not n_values:
+            await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "n is required"})
+            return
+        try:
+            n = int(n_values[0])
+        except ValueError:
+            await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "n must be an integer"})
+            return
+        if n < 0:
+            await send_json(send, HTTPStatus.BAD_REQUEST, {"error": "n must be non-negative"})
+            return
+        await send_json(send, HTTPStatus.OK, {"result": factorial(n)})
+        return
+
+    if path.startswith("/fibonacci/"):
+        n_str = path.removeprefix("/fibonacci/")
+        try:
+            n = int(n_str)
+        except ValueError:
+            await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "n must be an integer"})
+            return
+        if n < 0:
+            await send_json(send, HTTPStatus.BAD_REQUEST, {"error": "n must be non-negative"})
+            return
+        await send_json(send, HTTPStatus.OK, {"result": fibonacci(n)})
+        return
+
+    if path == "/mean":
+        message = await receive()
+        body = message.get("body", b"")
+        try:
+            numbers = json.loads(body)
+        except ValueError:
+            await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "valid JSON body is required"})
+            return
+        if not isinstance(numbers, list) or not all(isinstance(x, (int, float)) for x in numbers):
+            await send_json(send, HTTPStatus.UNPROCESSABLE_ENTITY, {"error": "body must be a list of numbers"})
+            return
+        if not numbers:
+            await send_json(send, HTTPStatus.BAD_REQUEST, {"error": "list must not be empty"})
+            return
+        await send_json(send, HTTPStatus.OK, {"result": sum(numbers) / len(numbers)})
+        return
+
+    await send_json(send, HTTPStatus.NOT_FOUND, {"error": "not found"})
+
 
 if __name__ == "__main__":
     import uvicorn

hw2/hw/shop_api/main.py

@@ -1,3 +1,193 @@
-from fastapi import FastAPI
+from http import HTTPStatus
+from itertools import count
+
+from fastapi import FastAPI, HTTPException, Query, Response
+from pydantic import BaseModel, ConfigDict
 
 app = FastAPI(title="Shop API")
+
+items_db: dict[int, "Item"] = {}
+carts_db: dict[int, dict[int, int]] = {}
+
+item_id_seq = count(1)
+cart_id_seq = count(1)
+
+
+class Item(BaseModel):
+    id: int
+    name: str
+    price: float
+    deleted: bool = False
+
+
+class ItemCreate(BaseModel):
+    name: str
+    price: float
+
+
+class ItemPatch(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    name: str | None = None
+    price: float | None = None
+
+
+class CartItem(BaseModel):
+    id: int
+    name: str
+    quantity: int
+    available: bool
+
+
+class Cart(BaseModel):
+    id: int
+    items: list[CartItem]
+    price: float
+
+
+def build_cart(cart_id: int) -> Cart:
+    cart_items = []
+    price = 0.0
+
+    for item_id, quantity in carts_db[cart_id].items():
+        item = items_db[item_id]
+        price += item.price * quantity
+        cart_items.append(
+            CartItem(
+                id=item.id,
+                name=item.name,
+                quantity=quantity,
+                available=not item.deleted,
+            )
+        )
+
+    return Cart(id=cart_id, items=cart_items, price=price)
+
+
+@app.post("/cart", status_code=HTTPStatus.CREATED)
+def create_cart(response: Response):
+    cart_id = next(cart_id_seq)
+    carts_db[cart_id] = {}
+    response.headers["location"] = f"/cart/{cart_id}"
+    return {"id": cart_id}
+
+
+@app.get("/cart/{cart_id}")
+def get_cart(cart_id: int) -> Cart:
+    if cart_id not in carts_db:
+        raise HTTPException(HTTPStatus.NOT_FOUND)
+
+    return build_cart(cart_id)
+
+
+@app.get("/cart")
+def get_cart_list(
+    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[Cart]:
+    carts = [build_cart(cart_id) for cart_id in carts_db]
+
+    if min_price is not None:
+        carts = [c for c in carts if c.price >= min_price]
+    if max_price is not None:
+        carts = [c for c in carts if c.price <= max_price]
+
+    if min_quantity is not None:
+        carts = [c for c in carts if sum(i.quantity for i in c.items) >= min_quantity]
+    if max_quantity is not None:
+        carts = [c for c in carts if sum(i.quantity for i in c.items) <= max_quantity]
+
+    return carts[offset : offset + limit]
+
+
+@app.post("/cart/{cart_id}/add/{item_id}")
+def add_item_to_cart(cart_id: int, item_id: int) -> Cart:
+    if cart_id not in carts_db or item_id not in items_db:
+        raise HTTPException(HTTPStatus.NOT_FOUND)
+
+    cart = carts_db[cart_id]
+    cart[item_id] = cart.get(item_id, 0) + 1
+
+    return build_cart(cart_id)
+
+
+@app.post("/item", status_code=HTTPStatus.CREATED)
+def create_item(body: ItemCreate) -> Item:
+    item_id = next(item_id_seq)
+    item = Item(id=item_id, name=body.name, price=body.price)
+    items_db[item_id] = item
+
+    return item
+
+
+@app.get("/item/{item_id}")
+def get_item(item_id: int) -> Item:
+    item = items_db.get(item_id)
+
+    if item is None or item.deleted:
+        raise HTTPException(HTTPStatus.NOT_FOUND)
+
+    return item
+
+
+@app.get("/item")
+def get_item_list(
+    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 = False,
+) -> list[Item]:
+    items = list(items_db.values())
+
+    if not show_deleted:
+        items = [i for i in items if not i.deleted]
+
+    if min_price is not None:
+        items = [i for i in items if i.price >= min_price]
+    if max_price is not None:
+        items = [i for i in items if i.price <= max_price]
+
+    return items[offset : offset + limit]
+
+
+@app.put("/item/{item_id}")
+def put_item(item_id: int, body: ItemCreate) -> Item:
+    if item_id not in items_db:
+        raise HTTPException(HTTPStatus.NOT_FOUND)
+
+    item = items_db[item_id]
+    item.name = body.name
+    item.price = body.price
+
+    return item
+
+
+@app.patch("/item/{item_id}")
+def patch_item(item_id: int, body: ItemPatch):
+    if item_id not in items_db:
+        raise HTTPException(HTTPStatus.NOT_FOUND)
+
+    item = items_db[item_id]
+
+    if item.deleted:
+        return Response(status_code=HTTPStatus.NOT_MODIFIED)
+
+    for field, value in body.model_dump(exclude_unset=True).items():
+        setattr(item, field, value)
+
+    return item
+
+
+@app.delete("/item/{item_id}")
+def delete_item(item_id: int):
+    item = items_db.get(item_id)
+
+    if item is not None:
+        item.deleted = True
+
+    return {"id": item_id, "deleted": True}

lecture3/Dockerfile

@@ -20,4 +20,4 @@ RUN pip install -r requirements.txt
 
 FROM base as local
 
-CMD ["uvicorn", "demo_service.api:app", "--port", "8080", "--host", "0.0.0.0"]
+CMD ["uvicorn", "shop_api.main:app", "--port", "8080", "--host", "0.0.0.0"]