feat: Implement ASGI application with fibonacci, factorial, and mean endpoints

hw1/app.py

@@ -1,4 +1,7 @@
+import json
+import math
 from typing import Any, Awaitable, Callable
+from urllib.parse import parse_qs
 
 
 async def application(
@@ -12,7 +15,148 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+
+    # Parse request
+    method = scope.get("method", "GET")
+    path = scope.get("path", "/")
+    query_string = scope.get("query_string", b"").decode("utf-8")
+
+    # Helper function to send JSON response
+    async def send_json_response(status_code: int, data: dict):
+        await send({
+            "type": "http.response.start",
+            "status": status_code,
+            "headers": [
+                [b"content-type", b"application/json"],
+            ],
+        })
+        await send({
+            "type": "http.response.body",
+            "body": json.dumps(data).encode("utf-8"),
+        })
+
+    # Helper function to send error response
+    async def send_error_response(status_code: int, message: str = ""):
+        await send({
+            "type": "http.response.start",
+            "status": status_code,
+            "headers": [
+                [b"content-type", b"application/json"],
+            ],
+        })
+        body = json.dumps({"error": message}) if message else b""
+        await send({
+            "type": "http.response.body",
+            "body": body.encode("utf-8") if isinstance(body, str) else body,
+        })
+
+    # Helper function to read request body
+    async def read_body():
+        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
+
+    # Only handle GET requests
+    if method != "GET":
+        await send_error_response(404)
+        return
+
+    # Route handling
+    if path.startswith("/fibonacci/"):
+        # Extract n from path
+        try:
+            n_str = path[11:]  # Remove "/fibonacci/"
+            if not n_str:
+                await send_error_response(422, "Invalid path parameter")
+                return
+
+            n = int(n_str)
+            if n < 0:
+                await send_error_response(400, "Parameter n must be non-negative")
+                return
+
+            # Calculate fibonacci
+            def fibonacci(num):
+                if num <= 1:
+                    return num
+                a, b = 0, 1
+                for _ in range(2, num + 1):
+                    a, b = b, a + b
+                return b
+
+            result = fibonacci(n)
+            await send_json_response(200, {"result": result})
+
+        except ValueError:
+            await send_error_response(422, "Invalid path parameter")
+
+    elif path == "/factorial":
+        # Parse query parameters
+        query_params = parse_qs(query_string)
+
+        if "n" not in query_params:
+            await send_error_response(422, "Missing required parameter 'n'")
+            return
+
+        try:
+            n_values = query_params["n"]
+            if not n_values or not n_values[0]:
+                await send_error_response(422, "Parameter 'n' cannot be empty")
+                return
+
+            n = int(n_values[0])
+            if n < 0:
+                await send_error_response(400, "Parameter n must be non-negative")
+                return
+
+            # Calculate factorial
+            result = math.factorial(n)
+            await send_json_response(200, {"result": result})
+
+        except ValueError:
+            await send_error_response(422, "Invalid parameter value")
+
+    elif path == "/mean":
+        # Read and parse JSON body
+        try:
+            body = await read_body()
+            if not body:
+                await send_error_response(422, "Missing request body")
+                return
+
+            numbers = json.loads(body.decode("utf-8"))
+
+            if not isinstance(numbers, list):
+                await send_error_response(422, "Request body must be a JSON array")
+                return
+
+            if len(numbers) == 0:
+                await send_error_response(400, "Array cannot be empty")
+                return
+
+            # Validate all elements are numbers
+            for num in numbers:
+                if not isinstance(num, (int, float)):
+                    await send_error_response(422, "All array elements must be numbers")
+                    return
+
+            # Calculate mean
+            result = sum(numbers) / len(numbers)
+            await send_json_response(200, {"result": result})
+
+        except json.JSONDecodeError:
+            await send_error_response(422, "Invalid JSON")
+        except Exception:
+            await send_error_response(422, "Invalid request")
+
+    else:
+        # 404 for any other path
+        await send_error_response(404)
 
 if __name__ == "__main__":
     import uvicorn

hw2/hw/requirements.txt

@@ -1,6 +1,10 @@
 # Основные зависимости для ASGI приложения
 fastapi>=0.117.1
 uvicorn>=0.24.0
+websockets>=12.0
+
+# База данных
+sqlalchemy>=2.0.0
 
 # Зависимости для тестирования
 pytest>=7.4.0

hw2/hw/shop_api/api/__init__.py

@@ -0,0 +1,4 @@
+from . import carts, items
+
+__all__ = ["items", "carts"]
+

hw2/hw/shop_api/api/carts.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+from http import HTTPStatus
+from typing import Annotated, Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Response
+from sqlalchemy.orm import Session
+
+from .. import models, schemas
+from ..database import get_db
+
+router = APIRouter(prefix="/cart", tags=["carts"])
+
+
+@router.post("", response_model=schemas.CartCreateResponse, status_code=HTTPStatus.CREATED)
+def create_cart(response: Response, db: Session = Depends(get_db)):
+    """Создание новой корзины"""
+    db_cart = models.Cart()
+    db.add(db_cart)
+    db.commit()
+    db.refresh(db_cart)
+    response.headers["location"] = f"/cart/{db_cart.id}"
+    return db_cart
+
+
+@router.get("/{cart_id}", response_model=schemas.CartResponse)
+def get_cart(cart_id: int, db: Session = Depends(get_db)):
+    """Получение корзины по ID"""
+    cart = db.query(models.Cart).filter(models.Cart.id == cart_id).first()
+    if not cart:
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Cart not found")
+
+    items = []
+    total_price = 0.0
+
+    for cart_item in cart.cart_items:
+        item = cart_item.item
+        items.append(
+            schemas.CartItemResponse(
+                id=item.id,
+                name=item.name,
+                quantity=cart_item.quantity,
+                available=not item.deleted,
+            )
+        )
+        if not item.deleted:
+            total_price += item.price * cart_item.quantity
+
+    return schemas.CartResponse(id=cart.id, items=items, price=total_price)
+
+
+@router.get("", response_model=list[schemas.CartResponse])
+def get_carts(
+    offset: Annotated[int, Query(ge=0)] = 0,
+    limit: Annotated[int, Query(gt=0)] = 10,
+    min_price: Annotated[Optional[float], Query(ge=0)] = None,
+    max_price: Annotated[Optional[float], Query(ge=0)] = None,
+    min_quantity: Annotated[Optional[int], Query(ge=0)] = None,
+    max_quantity: Annotated[Optional[int], Query(ge=0)] = None,
+    db: Session = Depends(get_db),
+):
+    """Получение списка корзин с фильтрацией"""
+    carts = db.query(models.Cart).all()
+
+    cart_responses = []
+    for cart in carts:
+        items = []
+        total_price = 0.0
+        total_quantity = 0
+
+        for cart_item in cart.cart_items:
+            item = cart_item.item
+            items.append(
+                schemas.CartItemResponse(
+                    id=item.id,
+                    name=item.name,
+                    quantity=cart_item.quantity,
+                    available=not item.deleted,
+                )
+            )
+            if not item.deleted:
+                total_price += item.price * cart_item.quantity
+            total_quantity += cart_item.quantity
+
+        cart_response = schemas.CartResponse(id=cart.id, items=items, price=total_price)
+
+        # Apply filters
+        if min_price is not None and total_price < min_price:
+            continue
+        if max_price is not None and total_price > max_price:
+            continue
+        if min_quantity is not None and total_quantity < min_quantity:
+            continue
+        if max_quantity is not None and total_quantity > max_quantity:
+            continue
+
+        cart_responses.append(cart_response)
+
+    return cart_responses[offset : offset + limit]
+
+
+@router.post("/{cart_id}/add/{item_id}")
+def add_item_to_cart(cart_id: int, item_id: int, db: Session = Depends(get_db)):
+    """Добавление товара в корзину"""
+    cart = db.query(models.Cart).filter(models.Cart.id == cart_id).first()
+    if not cart:
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Cart not found")
+
+    item = db.query(models.Item).filter(models.Item.id == item_id).first()
+    if not item:
+        raise HTTPException(status_code=HTTPStatus.NOT_FOUND, detail="Item not found")
+
+    # Check if item already in cart
+    cart_item = (
+        db.query(models.CartItem)
+        .filter(models.CartItem.cart_id == cart_id, models.CartItem.item_id == item_id)
+        .first()
+    )
+
+    if cart_item:
+        cart_item.quantity += 1
+    else:
+        cart_item = models.CartItem(cart_id=cart_id, item_id=item_id, quantity=1)
+        db.add(cart_item)
+
+    db.commit()
+    return {"message": "Item added to cart"}
+