[HW1] Implement ASGI application with fibonacci, factorial, and mean endpoints

hw1/app.py

@@ -1,5 +1,33 @@
+from http import HTTPStatus
+import json
+from utils.utils import (
+    check_request_valid, 
+    create_message, 
+    create_start_message
+)
+from utils.math_functions import (
+    factorial, 
+    fibonacci, 
+    mean, 
+    validate_list, 
+    validate_number
+)
 from typing import Any, Awaitable, Callable
 
+import urllib
+
+
+async def read_body(
+        receive: Callable[[], Awaitable[dict[str, Any]]]
+) -> str:
+    body = b""
+    while True:
+        message = await receive()
+        body += message.get("body", b"")
+        if not message.get("more_body", False):
+            break
+    return body
+
 
 async def application(
     scope: dict[str, Any],
@@ -12,7 +40,93 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+
+    if scope['type'] == 'http':
+        method = scope["method"]
+        path = scope["path"]
+
+        if not check_request_valid(method, path):
+            status_code = HTTPStatus.NOT_FOUND.value
+            body = {
+                "result": HTTPStatus.NOT_FOUND.phrase
+            }
+
+        input_body = await read_body(receive)
+
+        if path == "/factorial":
+            query_string = scope["query_string"]
+            parsed_query = urllib.parse.parse_qs(query_string.decode("utf-8"))
+            if parsed_query.get("n"):
+                n = parsed_query["n"][0]
+            else:
+                n = None
+
+            validation_result = validate_number(n)
+            if validation_result:
+                status_code, data = validation_result
+            else:
+                data = factorial(n)
+                status_code = HTTPStatus.OK.value
+
+            body = {
+                "result": data
+            }
+
+        elif "/fibonacci" in path:
+            n = path.split("/")[-1]
+
+            validation_result = validate_number(n)
+            if validation_result:
+                status_code, data = validation_result
+            else:
+                data = fibonacci(n)
+                status_code = HTTPStatus.OK.value
+
+            body = {
+                "result": data
+            }
+
+        elif path == "/mean":
+            input_body = input_body.decode()
+            try:
+                json_data = json.loads(input_body)
+            except json.decoder.JSONDecodeError:
+                json_data = []
+
+            validation_result = validate_list(json_data)
+
+            if validation_result:
+                status_code, data = validation_result
+            else:
+                data = mean(json_data)
+                status_code = HTTPStatus.OK.value
+
+            body = {
+                "result": data
+            }
+
+        await send(
+            create_start_message(
+                status_code=status_code
+            )
+        )
+
+        await send(
+            create_message(
+                body=body
+            )
+        )
+
+    elif scope["type"] == "lifespan":
+        while True:
+            message = await receive()
+            t = message.get("type")
+            if t == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            elif t == "lifespan.shutdown":
+                await send({"type": "lifespan.shutdown.complete"})
+                return
+
 
 if __name__ == "__main__":
     import uvicorn

hw1/utils/math_functions.py

@@ -0,0 +1,55 @@
+from http import HTTPStatus
+from typing import Any, Optional, Union
+
+from utils.utils import is_digit
+
+
+def validate_number(
+        n: str
+) -> Optional[tuple[int, str]]:
+    if not n or not is_digit(n):
+        return HTTPStatus.UNPROCESSABLE_ENTITY.value, HTTPStatus.UNPROCESSABLE_ENTITY.phrase
+
+    n = int(n)
+
+    if n < 0:
+        return HTTPStatus.BAD_REQUEST.value, HTTPStatus.BAD_REQUEST.phrase
+
+
+def validate_list(
+        l: Any
+) -> Optional[tuple[int, str]]:
+    if isinstance(l, list) and l == []:
+        return HTTPStatus.BAD_REQUEST.value, HTTPStatus.BAD_REQUEST.phrase
+
+    if not l:
+        return HTTPStatus.UNPROCESSABLE_ENTITY.value, HTTPStatus.UNPROCESSABLE_ENTITY.phrase
+
+
+def factorial(
+        n: str
+) -> int:
+    n = int(n)
+
+    factorial = 1
+    for i in range(1, n + 1):
+        factorial *= i
+
+    return factorial
+
+
+def fibonacci(
+        n: str
+) -> int:
+    n = int(n)
+
+    if n <= 1:
+        return n
+    else:
+        return fibonacci(n - 1) + fibonacci(n - 2)
+
+
+def mean(
+        nums: list[Union[int, float]]
+) -> Union[int, float]:
+    return sum(nums)/len(nums)

hw1/utils/utils.py

@@ -0,0 +1,38 @@
+import json
+
+
+def is_digit(
+        str_num: str
+) -> bool:
+    if str_num.startswith("-"):
+        str_num = str_num[1:]
+
+    return str_num.isdigit()
+
+
+def check_request_valid(
+        method: str,
+        path: str
+) -> bool:
+    return method == "GET" and path in ["/factorial", "/fibonacci", "/mean"]
+
+
+def create_start_message(
+        status_code: int = 200
+) -> dict:
+    return {
+        'type': 'http.response.start',
+        'status': status_code,
+        'headers': [
+            (b'content-type', b'text/plain'),
+        ]
+    }
+
+
+def create_message(
+        body: str
+) -> dict:
+    return {
+        'type': 'http.response.body',
+        'body': json.dumps(body).encode("utf-8"),
+    }

hw2/.github/workflows/test.yml

@@ -0,0 +1,36 @@
+name: HW2 tests
+
+on:
+  push:
+    branches: [ "main", "master" ] 
+  pull_request:
+    branches: [ "main", "master" ] 
+
+jobs:
+  build-and-test:
+    name: Build and Test
+
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        python-version: ["3.10", "3.11", "3.12"]
+
+    steps:
+      - name: Check out repository code
+        uses: actions/checkout@v4
+
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+          pip install pytest pytest-mock pytest-cov 
+
+      - name: Run tests
+        run: |
+          pytest

hw2/hw/shop_api/handlers/cart.py

@@ -0,0 +1,108 @@
+from http import HTTPStatus
+from typing import Annotated, List
+import uuid
+
+from fastapi import APIRouter, HTTPException, Query, Response
+from fastapi.responses import JSONResponse
+from pydantic import NonNegativeFloat, NonNegativeInt, PositiveFloat, PositiveInt
+
+from shop_api.models.cart import CartOutSchema
+from shop_api import local_data
+
+
+router = APIRouter(prefix="/cart")
+
+
+@router.post(
+        "",
+        response_model=CartOutSchema,
+        status_code=HTTPStatus.CREATED
+)
+async def add_cart(response: Response):
+    cart_id = str(uuid.uuid4())
+    cart_data = {"id": cart_id}
+
+    local_data.add_single_cart(cart_data=cart_data)
+
+    response.headers["Location"] = f"/cart/{cart_id}"
+
+    return cart_data
+
+
+@router.get(
+    "/{cart_id}",
+    response_model=CartOutSchema,
+    status_code=HTTPStatus.OK
+)
+async def get_cart_by_id(cart_id: str):
+    cart_data = local_data.get_single_cart(cart_id=cart_id)
+
+    return cart_data
+
+
+@router.get(
+        "",
+        response_model=List[CartOutSchema],
+        status_code=HTTPStatus.OK
+)
+async def get_all_carts(
+    offset: Annotated[NonNegativeInt, Query()] = 0,
+    limit: Annotated[PositiveInt, Query()] = 10,
+    min_price: Annotated[NonNegativeFloat, Query()] = None,
+    max_price: Annotated[NonNegativeFloat, Query()] = None,
+    min_quantity: Annotated[NonNegativeInt, Query()] = None,
+    max_quantity: Annotated[NonNegativeInt, Query()] = None,
+):
+    all_carts = local_data.get_all_carts()
+
+    filtered_carts: List[CartOutSchema] = []
+    for cart in all_carts:
+        if min_price is not None and cart.price < min_price:
+            continue
+        if max_price is not None and cart.price > max_price:
+            continue
+        if min_quantity is not None and sum([item.quantity for item in cart.items]) < min_quantity:
+            continue
+        if max_quantity is not None and sum([item.quantity for item in cart.items]) > max_quantity:
+            continue
+
+        filtered_carts.append(cart)
+
+    filtered_carts = filtered_carts[offset: offset + limit]
+
+    return filtered_carts
+
+
+@router.post(
+        "/{cart_id}/add/{item_id}",
+        response_model=CartOutSchema,
+        status_code=HTTPStatus.OK
+)
+async def add_item_to_cart(
+    cart_id: str,
+    item_id: str
+):
+    cart = local_data.get_single_cart(cart_id=cart_id)
+
+    if cart is None:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Cart with {cart_id=!r} wasn't found",
+        )
+
+    item_ids = local_data.get_all_item_ids_for_cart(
+        cart_id=cart_id
+    )
+
+    if item_id in item_ids:
+        for item in cart.items:
+            if item_id == item.id:
+                item.quantity += 1
+    else:
+        cart.items.append(
+            local_data.get_single_item(
+                item_id=item_id
+            )
+        )
+
+    return cart