From 8fa28cc1e6632ab9574ddc93bb45c97f7320b4b3 Mon Sep 17 00:00:00 2001 From: Laskin Pavel Date: Sun, 27 Sep 2026 23:44:15 +0300 Subject: [PATCH 1/2] Implement ASGI application with fibonacci, factorial, and mean endpoints --- hw1/app.py | 120 +++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 116 insertions(+), 4 deletions(-) diff --git a/hw1/app.py b/hw1/app.py index 6107b87..dfbee55 100644 --- a/hw1/app.py +++ b/hw1/app.py @@ -1,10 +1,30 @@ +import json +from urllib.parse import parse_qs from typing import Any, Awaitable, Callable +async def send_response(send: Callable, status: int, data: dict | None = None): + body = json.dumps(data).encode('utf-8') if data is not None else b'' + headers = [ + (b'content-type', b'application/json'), + (b'content-length', str(len(body)).encode('ascii')), + ] + + await send({ + 'type': 'http.response.start', + 'status': status, + 'headers': headers, + }) + + await send({ + 'type': 'http.response.body', + 'body': body, + }) + 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: @@ -13,7 +33,99 @@ async def application( 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 + return + + if scope['type'] != 'http': + return + path = scope['path'] + method = scope['method'] + + if path == '/factorial': + query_string = scope.get('query_string', b'').decode('utf-8') + params = parse_qs(query_string) + + if 'n' not in params or not params['n'][0]: + await send_response(send, 422, {"detail": "no param n"}) + return + + try: + n = int(params['n'][0]) + except ValueError: + await send_response(send, 422, {"detail": "param n is not int"}) + return + + if n < 0: + await send_response(send, 400, {"detail": "param n < 0"}) + return + + ans = 1 + for i in range(2, n + 1): + ans *= i + + await send_response(send, 200, {"result": ans}) + + + elif path.startswith('/fibonacci/'): + try: + n = int(path.split('/')[-1]) + except ValueError: + await send_response(send, 422, {"detail": "bad param n"}) + return + + if n < 0: + await send_response(send, 400, {"detail": "param n < 0"}) + return + + if n == 0: + ans = 0 + elif n == 1: + ans = 1 + else: + a = 0 + b = 1 + for i in range(2, n + 1): + a, b = b, a + b + ans = b + + await send_response(send, 200, {"result": ans}) + + + elif path == '/mean': + body = b'' + while True: + message = await receive() + if message['type'] == 'http.request': + body += message.get('body', b'') + if not message.get('more_body', False): + break + + try: + data = json.loads(body) + if not isinstance(data, list): + raise ValueError + except (json.JSONDecodeError, ValueError): + await send_response(send, 422, {"detail": "bad body"}) + return + + if len(data) == 0: + await send_response(send, 400, {"detail": "data is empty"}) + return + + ans = sum(data) / len(data) + await send_response(send, 200, {"result": ans}) + + else: + await send_response(send, 404, {"detail": "Not Found"}) + if __name__ == "__main__": import uvicorn - uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True) + uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True) \ No newline at end of file From 146ec6c0f07e1b774fe513558b0fd1fb8176738a Mon Sep 17 00:00:00 2001 From: Laskin Pavel Date: Mon, 5 Oct 2026 01:56:00 +0500 Subject: [PATCH 2/2] hw2 --- hw1/app.py | 6 +- hw2/hw/shop_api/main.py | 187 +++++++++++++++++++++++++++++++++++++++- 2 files changed, 189 insertions(+), 4 deletions(-) diff --git a/hw1/app.py b/hw1/app.py index dfbee55..74015b1 100644 --- a/hw1/app.py +++ b/hw1/app.py @@ -22,9 +22,9 @@ async def send_response(send: Callable, status: int, data: dict | None = None): }) 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: diff --git a/hw2/hw/shop_api/main.py b/hw2/hw/shop_api/main.py index f60a8c6..a54a5ac 100644 --- a/hw2/hw/shop_api/main.py +++ b/hw2/hw/shop_api/main.py @@ -1,3 +1,188 @@ -from fastapi import FastAPI +from typing import Any, Dict, Optional +from fastapi import FastAPI, HTTPException, Query, Body, Response, status +from pydantic import BaseModel, ConfigDict app = FastAPI(title="Shop API") + +items: Dict[int, Dict[str, Any]] = {} +carts: Dict[int, Dict[str, Any]] = {} + +next_id = 1 +next_cart_id = 1 + +class ItemPatch(BaseModel): + model_config = ConfigDict(extra="forbid") + name: Optional[str] = None + price: Optional[float] = None + +def get_item_raw(item_id: int) -> Dict[str, Any]: + item = items.get(item_id) + if item is None: + raise HTTPException(status_code=404, detail="Item not found") + return item + +def get_item(item_id: int) -> Dict[str, Any]: + item = items.get(item_id) + if item is None or item["deleted"]: + raise HTTPException(status_code=404, detail="Item not found") + return item + +def get_cart(cart_id: int) -> Dict[str, Any]: + cart = carts.get(cart_id) + if cart is None: + raise HTTPException(status_code=404, detail="Cart not found") + return cart + +def dict_of_cart(cart_id: int) -> Dict[str, Any]: + cart = get_cart(cart_id) + out_items = [] + total_price = 0 + for cart_item in cart["items"]: + item = items.get(cart_item["id"]) + available = not item["deleted"] + out_items.append({ + "id": item["id"], + "name": item["name"], + "quantity": cart_item["quantity"], + "available": available, + }) + if available: + total_price += item["price"] * cart_item["quantity"] + return { + "id": cart_id, + "items": out_items, + "price": total_price, + } + +def cart_q(cart_id: int) -> int: + cart = get_cart(cart_id) + return sum(ci["quantity"] for ci in cart["items"]) + +@app.post("/cart", status_code=status.HTTP_201_CREATED) +def create_cart(response: Response): + global next_cart_id + cart_id = next_cart_id + next_cart_id += 1 + carts[cart_id] = { + "id": cart_id, + "items": [], + } + response.headers["Location"] = f"/cart/{cart_id}" + return {"id": cart_id} + +@app.get("/cart/{cart_id}") +def get_cart_e(cart_id: int): + return dict_of_cart(cart_id) + +@app.get("/cart") +def list_carts( + offset: int = Query(0, ge=0), + limit: int = Query(10, gt=0), + min_price: Optional[float] = Query(None, ge=0), + max_price: Optional[float] = Query(None, ge=0), + min_quantity: Optional[int] = Query(None, ge=0), + max_quantity: Optional[int] = Query(None, ge=0), +): + result = [] + for cart_id in sorted(carts): + cart = dict_of_cart(cart_id) + price = cart["price"] + quantity = cart_q(cart_id) + if min_price is not None and price < min_price: + continue + if max_price is not None and price > max_price: + continue + if min_quantity is not None and quantity < min_quantity: + continue + if max_quantity is not None and quantity > max_quantity: + continue + result.append(cart) + return result[offset: offset + limit] + +@app.post("/cart/{cart_id}/add/{item_id}") +def add_item_to_cart(cart_id: int, item_id: int): + cart = get_cart(cart_id) + get_item(item_id) + for cart_item in cart["items"]: + if cart_item["id"] == item_id: + cart_item["quantity"] += 1 + break + else: + cart["items"].append({ + "id": item_id, + "quantity": 1, + }) + + return dict_of_cart(cart_id) + +@app.post("/item", status_code=status.HTTP_201_CREATED) +def create_item(body: Dict[str, Any] = Body(...)): + global next_id + price = float(body["price"]) + item_id = next_id + next_id += 1 + item = { + "id": item_id, + "name": body["name"], + "price": price, + "deleted": False, + } + items[item_id] = item + return item + +@app.get("/item/{item_id}") +def get_item_e(item_id: int): + return get_item(item_id) + +@app.get("/item") +def list_items( + offset: int = Query(0, ge=0), + limit: int = Query(10, gt=0), + min_price: Optional[float] = Query(None, ge=0), + max_price: Optional[float] = Query(None, ge=0), + show_deleted: bool = Query(False), +): + result = [] + for item_id in sorted(items): + item = items[item_id] + if not show_deleted and item["deleted"]: + continue + if min_price is not None and item["price"] < min_price: + continue + if max_price is not None and item["price"] > max_price: + continue + result.append(item) + return result[offset: offset + limit] + +@app.put("/item/{item_id}") +def replace_item(item_id: int, body: Dict[str, Any] = Body(...)): + get_item(item_id) + if "name" not in body or "price" not in body: + raise HTTPException(status_code=422, detail="name and price are required") + price = float(body["price"]) + item = { + "id": item_id, + "name": body["name"], + "price": price, + "deleted": False, + } + items[item_id] = item + return item + +@app.patch("/item/{item_id}") +def patch_item(item_id: int, body: ItemPatch = Body(...)): + item = get_item_raw(item_id) + if item["deleted"]: + return Response(status_code=status.HTTP_304_NOT_MODIFIED) + data = body.model_dump(exclude_unset=True) + if "name" in data: + item["name"] = data["name"] + if "price" in data: + item["price"] = float(data["price"]) + return item + +@app.delete("/item/{item_id}") +def delete_item(item_id: int): + item = get_item_raw(item_id) + item["deleted"] = True + return item