HW-2 Bogdan Minko

hw1/app.py

@@ -1,4 +1,141 @@
 from typing import Any, Awaitable, Callable
+from dataclasses import dataclass
+import math
+import statistics
+import json
+from urllib.parse import parse_qs
+
+fibonacci_array = [0, 1]
+
+
+@dataclass
+class int_model:
+    n: int
+
+    def fibonacci(self) -> int:
+        while len(fibonacci_array) <= self.n:
+            next_fib = fibonacci_array[-1] + fibonacci_array[-2]
+            fibonacci_array.append(next_fib)
+
+        return fibonacci_array[self.n]
+
+    def factorial(self) -> int:
+        result = math.factorial(self.n)
+        return result
+
+
+@dataclass
+class mean_model:
+    n: list[float]
+
+    def mean(self) -> float:
+        result = statistics.mean(self.n)
+        return result
+
+
+async def send_json_response(send, status_code, data):
+    """Отправляет ответ"""
+    response_body = json.dumps(data, ensure_ascii=False).encode("utf-8")
+
+    await send(
+        {
+            "type": "http.response.start",
+            "status": status_code,
+            "headers": [
+                [b"content-type", b"application/json"],
+                [b"content-length", str(len(response_body)).encode()],
+            ],
+        }
+    )
+
+    await send(
+        {
+            "type": "http.response.body",
+            "body": response_body,
+        }
+    )
+
+
+async def send_error_response(send, status_code, error_message):
+    """Отправляет ответ с ошибкой"""
+    await send_json_response(send, status_code, {"error": error_message})
+
+
+async def get_request_body(receive):
+    """Получает тело запроса"""
+    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
+    return body
+
+
+async def route_fibonacci(send, path: str):
+    """GET /fibonacci/{n} - возвращает n-е число Фибоначчи"""
+    n_str = path.split("/fibonacci/")[-1]
+
+    try:
+        n = int(n_str)
+        if n < 0:
+            await send_error_response(send, 400, "n should be greater or equal to 0")
+            return
+        result = int_model(n=n).fibonacci()
+        await send_json_response(send, 200, {"result": result})
+    except ValueError:
+        await send_error_response(send, 422, "Invalid number format")
+    except AssertionError as e:
+        await send_error_response(send, 400, str(e))
+
+
+async def route_factorial(send, query_params: dict):
+    """GET /factorial?n=число - возвращает факториал числа"""
+
+    if "n" not in query_params:
+        await send_error_response(send, 422, "Parameter 'n' is required")
+        return
+
+    try:
+        n = int(query_params["n"][0])
+        if n < 0:
+            await send_error_response(send, 400, "n should be greater or equal to 0")
+            return
+        result = int_model(n=n).factorial()
+        await send_json_response(send, 200, {"result": result})
+    except ValueError:
+        await send_error_response(send, 422, "Invalid number format")
+    except AssertionError as e:
+        await send_error_response(send, 400, str(e))
+
+
+async def route_mean(send, receive):
+    """GET /mean с JSON body - возвращает среднее арифметическое"""
+
+    body = await get_request_body(receive)
+    if not body:
+        await send_error_response(send, 422, "JSON body is required")
+        return
+    try:
+        data = json.loads(body.decode("utf-8"))
+
+        if not isinstance(data, list):
+            await send_error_response(send, 422, "Expected a list of numbers")
+            return
+
+        if len(data) == 0:
+            await send_error_response(send, 400, "Numbers list cannot be empty")
+            return
+
+        numbers = [float(item) for item in data]
+        result = mean_model(n=numbers).mean()
+        await send_json_response(send, 200, {"result": result})
+
+    except json.JSONDecodeError:
+        await send_error_response(send, 422, "Invalid JSON format")
+    except (ValueError, TypeError):
+        await send_error_response(send, 422, "All elements must be numbers")
 
 
 async def application(
@@ -12,8 +149,53 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+
+    if scope["type"] == "lifespan":
+        while True:
+            message = await receive()
+            if message["type"] == "lifespan.startup":
+                print("Application is starting up...")
+                await send({"type": "lifespan.startup.complete"})
+            elif message["type"] == "lifespan.shutdown":
+                print("Application is shutting down...")
+                await send({"type": "lifespan.shutdown.complete"})
+                return
+
+    # Проверяем тип запроса
+    if scope["type"] != "http":
+        await send_error_response(send, 422, "Unsupported request type")
+        return
+
+    # Проверяем метод запроса
+    if scope["method"] != "GET":
+        await send_error_response(send, 404, "Method not allowed")
+        return
+
+    # Получаем данные запроса
+    path = scope["path"]
+    query_string = scope["query_string"].decode("utf-8")
+    query_params = parse_qs(query_string) if query_string else {}
+
+    try:
+        match path:
+            case p if p.startswith("/fibonacci/"):
+                await route_fibonacci(send, path)
+
+            case "/factorial":
+                await route_factorial(send, query_params)
+
+            case "/mean":
+                await route_mean(send, receive)
+
+            case _:
+                await send_error_response(send, 404, "Endpoint not found")
+
+    except Exception as e:
+        await send_error_response(send, 500, f"Internal server error: {str(e)}")
+
 
 if __name__ == "__main__":
+    print("\n🚀 Запуск ASGI сервера...")
     import uvicorn
+
     uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True)

hw2/hw/pyproject.toml

@@ -0,0 +1,14 @@
+[project]
+name = "hw"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "faker>=37.8.0",
+    "fastapi>=0.117.1",
+    "httpx>=0.27.2",
+    "pytest>=7.4.0",
+    "pytest-asyncio>=0.21.0",
+    "uvicorn>=0.24.0",
+]

hw2/hw/shop_api/cart_routes.py

@@ -0,0 +1,42 @@
+from fastapi import APIRouter, Response, HTTPException, Depends
+from http import HTTPStatus
+from .store.storage import local_storage
+from .contracts import IdModel, CartResponseModel, ListQueryModel
+cart_router = APIRouter(prefix="/cart")
+
+@cart_router.post("/",
+                  status_code=HTTPStatus.CREATED)
+async def post_cart(response: Response): 
+    cart = local_storage.create_cart()
+    response.headers["location"] = f"/cart/{cart.id}"
+    return IdModel(id=cart.id)
+
+
+@cart_router.post("/{cart_id}/add/{item_id}")
+async def add_item_to_cart(cart_id: int, item_id: int):
+    success = local_storage.add_item_to_cart(cart_id, item_id)
+    if not success:
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
+    return {"message": "Item added to cart"}
+
+@cart_router.get("/{cart_id}",
+                 status_code=HTTPStatus.OK)
+async def get_cart(cart_id: int) -> CartResponseModel:
+    try: 
+       cart = local_storage.get_cart(id=cart_id)
+       return CartResponseModel.from_entity(cart)
+    except KeyError:
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="cart not found")
+
+
+@cart_router.get("/")
+async def get_item_list(query: ListQueryModel = Depends()) -> list[CartResponseModel]:
+    carts = local_storage.get_carts(
+        offset=query.offset,
+        limit=query.limit,
+        min_price=query.min_price,
+        max_price=query.max_price,
+        min_quantity=query.min_quantity,
+        max_quantity=query.max_quantity
+    )
+    return [CartResponseModel.from_entity(cart) for cart in carts]

hw2/hw/shop_api/contracts.py

@@ -0,0 +1,71 @@
+from pydantic import BaseModel, Field, ConfigDict
+from typing import Optional, List
+from .store.storage import CartData, ItemnInCartData, ItemData, ItemsData
+
+class IdModel(BaseModel):
+    """
+    Model handles with id 
+    """
+    id: int = Field(description="Returns id")
+
+class ItemRequest(BaseModel):
+    name: str = Field(description="item name")
+    price: float = Field(gt=0, description="item price")
+
+class ItemPatchRequest(BaseModel):
+    model_config = ConfigDict(extra='forbid')
+    name: Optional[str] = None
+    price: Optional[float] = Field(None, gt=0.0)
+
+class ItemModel(BaseModel):
+    """
+    Model defines items
+    """
+
+    id: int = Field(description="item id, int")
+    name: str = Field(description="item name, str")
+    price: float = Field(description="item price, float")
+    deleted: bool = Field(default=False, description="item available. bool")    
+    @staticmethod
+    def from_entity(entity: ItemData) -> "ItemModel":
+        return ItemModel(
+            id=entity.id,
+            name=entity.name,
+            price=entity.price,
+            deleted=entity.deleted
+        )
+
+class ListQueryModel(BaseModel):
+    offset: int = Field(0, ge=0, description="Number of items to skip")
+    limit: int = Field(10, gt=0, description="Maximum number of items to return")
+    min_price: Optional[float] = Field(None, ge=0, description="Minimum price filter")
+    max_price: Optional[float] = Field(None, ge=0, description="Maximum price filter")
+    show_deleted: bool = Field(False, description="Include deleted items")
+    min_quantity: Optional[int] = Field(None, ge=0, description="Minimum quantity filter")
+    max_quantity: Optional[int] = Field(None, ge=0, description="Maximum quantity filter")
+
+class CartItemModel(BaseModel):
+    """Model defines items in cart"""
+    id: int = Field(description="item id")
+    name: str = Field(description="item name") 
+    quantity: int = Field(description="quantity in cart")
+    available: bool = Field(description="is item available (not deleted)")
+
+class CartResponseModel(BaseModel):
+    """Models defines Carts"""
+    id: int = Field(description="cart id")
+    items: List[CartItemModel] = Field(description="items in cart")
+    price: float = Field(description="total cart price")
+
+    @staticmethod
+    def from_entity(entity: CartData) -> "CartResponseModel":
+        return CartResponseModel(
+            id=entity.id,
+            items=[CartItemModel(
+                id=item.id,
+                name=item.name, 
+                quantity=item.quantity,
+                available=item.available
+            ) for item in entity.items],
+            price=entity.price
+        )

hw2/hw/shop_api/item_routes.py

@@ -0,0 +1,53 @@
+from fastapi import APIRouter, HTTPException, Depends
+from http import HTTPStatus
+from .store.storage import local_storage
+from .contracts import ItemPatchRequest, ItemRequest, ItemModel, ListQueryModel
+item_router = APIRouter(prefix="/item")
+
+@item_router.post("/",
+                  status_code=HTTPStatus.CREATED)
+async def post_item(item_data: ItemRequest) -> ItemModel:
+    item = local_storage.add_item(name=item_data.name, price=item_data.price)
+    return ItemModel.from_entity(entity=item)
+
+@item_router.get("/{item_id}",
+                 status_code=HTTPStatus.OK)
+async def get_item(item_id: int) -> ItemModel:
+    try: 
+       item = local_storage.get_item(id=item_id)
+       return ItemModel.from_entity(item)
+    except KeyError:
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="item not found")
+
+@item_router.get("/")
+async def get_item_list(query: ListQueryModel = Depends()) -> list[ItemModel]:
+    items = local_storage.get_items(
+        offset=query.offset,
+        limit=query.limit,
+        min_price=query.min_price,
+        max_price=query.max_price,
+        show_deleted=query.show_deleted
+    )
+    return [ItemModel.from_entity(item) for item in items]
+
+@item_router.put("/{item_id}")
+async def put_item(item_id: int, item_data:ItemRequest) -> ItemModel:
+    item = local_storage.put_item(item_id = item_id,
+                                  name = item_data.name,
+                                  price = item_data.price)
+    return ItemModel.from_entity(item)
+
+@item_router.patch("/{item_id}")
+async def patch_item(item_id: int, item_data: ItemPatchRequest) -> ItemModel:
+    item = local_storage.patch_item(item_id=item_id,
+                                    name = item_data.name,
+                                    price = item_data.price)
+    if item.deleted:
+        raise HTTPException(status_code=HTTPStatus.NOT_MODIFIED)
+    return ItemModel.from_entity(item)              
+
+@item_router.delete("/{item_id}")
+async def patch_item(item_id: int) -> ItemModel:
+    item = local_storage.soft_delete_item(item_id=item_id)
+    return ItemModel.from_entity(item)              
+

hw2/hw/shop_api/main.py

@@ -1,3 +1,11 @@
 from fastapi import FastAPI
-
+from .cart_routes import cart_router
+from .item_routes import item_router
 app = FastAPI(title="Shop API")
+
+app.include_router(cart_router)
+app.include_router(item_router)
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)
+