HW1+HW2 (first part)

.github/workflows/hw5-tests.yml

@@ -0,0 +1,39 @@
+name: "HW5 Tests"
+
+# Запускаем тесты при изменении файлов в hw5/
+on:
+  pull_request:
+    branches: [ main ]
+    paths: [ 'hw5/**' ]
+  push:
+    branches: [ main ]
+    paths: [ 'hw5/**' ]
+
+jobs:
+  test-hw5:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.12", "3.13"]
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Install dependencies
+      working-directory: hw5
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+
+    - name: Run tests
+      working-directory: hw5
+      env:
+        PYTHONPATH: ${{ github.workspace }}/hw5
+      run: |
+        pytest tests -v

.gitignore

@@ -132,3 +132,6 @@ dmypy.json
 
 # macOS
 .DS_Store
+
+#other
+test.db

hw1/app.py

@@ -1,3 +1,7 @@
+import math
+import json
+from http import HTTPStatus
+from urllib import parse
 from typing import Any, Awaitable, Callable
 
 
@@ -12,8 +16,162 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    if scope["type"] != "http":
+        await send_response(
+            send=send,
+            status=HTTPStatus.INTERNAL_SERVER_ERROR,
+            data={"error": "UnsupportedType", "message": "This application only supports HTTP"},
+        )
+        return
+
+    method = scope["method"]
+    path = scope["path"]
+
+    if method != "GET":
+        await send_response(
+            send=send,
+            status=HTTPStatus.NOT_FOUND,
+            data={"error": "NotFound", "message": f"Path {path} not found for method {method}"},
+        )
+        return
+
+    if path == "/factorial":
+        query_params = parse.parse_qs(scope["query_string"].decode())
+        n_str = query_params.get("n", [None])[0]
+
+        if n_str is None or n_str == "":
+            await send_response(
+                send=send,
+                status=HTTPStatus.UNPROCESSABLE_ENTITY,
+                data={"error": "InvalidParameter", "message": "Query parameter 'n' is required"},
+            )
+            return
+        try:
+            n = int(n_str)
+        except (ValueError, TypeError):
+            await send_response(
+                send=send,
+                status=HTTPStatus.UNPROCESSABLE_ENTITY,
+                data={"error": "InvalidParameter", "message": "'n' must be a valid integer"},
+            )
+            return
+
+        if n < 0:
+            await send_response(
+                send=send,
+                status=HTTPStatus.BAD_REQUEST,
+                data={"error": "InvalidValue", "message": "'n' must be non-negative"},
+            )
+            return
+
+        value = math.factorial(n)
+        await send_response(send=send, status=HTTPStatus.OK, data={"result": value})
+        return
+
+    elif path.startswith("/fibonacci/"):
+        path_parts = path.strip("/").split("/")
+
+        if len(path_parts) != 2:
+            await send_response(
+                send=send, status=HTTPStatus.NOT_FOUND, data={"error": "NotFound", "message": "Invalid path format"}
+            )
+            return
+
+        try:
+            n = int(path_parts[1])
+        except ValueError:
+            await send_response(
+                send=send,
+                status=HTTPStatus.UNPROCESSABLE_ENTITY,
+                data={"error": "InvalidParameter", "message": "Path parameter must be an integer"},
+            )
+            return
+
+        if n < 0:
+            await send_response(
+                send=send,
+                status=HTTPStatus.BAD_REQUEST,
+                data={"error": "InvalidValue", "message": "'n' must be non-negative"},
+            )
+            return
+
+        a, b = 0, 1
+        for _ in range(n):
+            a, b = b, a + b
+
+        await send_response(send=send, status=HTTPStatus.OK, data={"result": a})
+        return
+
+    elif path == "/mean":
+        event = await receive()
+        body = event.get("body", b"")
+        if not body:
+            await send_response(
+                send=send,
+                status=HTTPStatus.UNPROCESSABLE_ENTITY,
+                data={"error": "InvalidParameter", "message": "Request body cannot be empty"},
+            )
+            return
+
+        try:
+            data = json.loads(body)
+        except json.JSONDecodeError:
+            await send_response(
+                send,
+                HTTPStatus.UNPROCESSABLE_ENTITY,
+                {"error": "InvalidJSON", "message": "Could not decode request body"},
+            )
+            return
+
+        if not isinstance(data, list):
+            await send_response(
+                send,
+                HTTPStatus.UNPROCESSABLE_ENTITY,
+                {"error": "InvalidFormat", "message": "Request body must be a JSON array"},
+            )
+            return
+
+        if not data:
+            await send_response(
+                send,
+                HTTPStatus.BAD_REQUEST,
+                {"error": "InvalidValue", "message": "Input array cannot be empty"},
+            )
+            return
+
+        if not all(isinstance(x, (int, float)) for x in data):
+            await send_response(
+                send=send,
+                status=HTTPStatus.UNPROCESSABLE_ENTITY,
+                data={"error": "InvalidParameter", "message": "All elements in the array must be numbers"},
+            )
+            return
+
+        value = sum(data) / len(data)
+        await send_response(send=send, status=HTTPStatus.OK, data={"result": value})
+        return
+
+    await send_response(
+        send=send,
+        status=HTTPStatus.NOT_FOUND,
+        data={"error": "NotFound", "message": f"Path {path} not found"},
+    )
+    return
+
+
+async def send_response(send, status: HTTPStatus, data: dict):
+    body = json.dumps(data).encode("utf-8")
+    await send(
+        {
+            "type": "http.response.start",
+            "status": status.value,
+            "headers": [(b"content-type", b"application/json; charset=utf-8")],
+        }
+    )
+    await send({"type": "http.response.body", "body": body})
+
 
 if __name__ == "__main__":
     import uvicorn
+
     uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True)

hw2/hw/shop_api/cart/contracts.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+from typing import Any
+
+from pydantic import BaseModel
+
+from shop_api.cart.store.models import CartEntity, CartInfo, CartItemInfo
+
+
+class CartItemResponse(BaseModel):
+    id: int
+    name: str
+    quantity: int
+    available: bool
+
+
+# class CartItemRequest(BaseModel):
+#     items: list[CartItemInfo]
+#     price: float
+
+#     def as_cart_info(self) -> CartInfo:
+#         items = [CartItemInfo(**item.dict()) for item in self.items]
+#         # price = 
+#         return CartInfo(items=items, price=self.price)
+
+
+
+class CartResponse(BaseModel):
+    id: int
+    items: list[CartItemInfo]
+    price: float
+
+    @staticmethod
+    def from_entity(entity: CartEntity) -> CartResponse:
+        return CartResponse(
+            id=entity.id,
+            items=entity.info.items,
+            price=entity.info.price,
+        )

hw2/hw/shop_api/cart/store/__init__.py

@@ -0,0 +1,13 @@
+from .models import CartItemInfo, CartEntity, CartInfo
+from .queries import add, delete, get_many, get_one, create
+
+__all__ = [
+    "CartEntity",
+    "CartInfo",
+    "CartItemInfo",
+    "add",
+    "delete",
+    "get_many",
+    "get_one",
+    "create",
+]

hw2/hw/shop_api/cart/store/models.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass
+
+
+@dataclass(slots=True)
+class CartItemInfo:
+    id: int
+    name: str
+    quantity: int
+    available: bool
+
+
+@dataclass(slots=True)
+class CartInfo:
+    items: list[CartItemInfo]
+    price: float
+
+
+@dataclass(slots=True)
+class CartEntity:
+    id: int
+    info: CartInfo

hw2/hw/shop_api/cart/store/queries.py

@@ -0,0 +1,89 @@
+from logging import info
+from typing import Iterable
+
+from shop_api.item.store.models import ItemEntity
+from shop_api.cart.store.models import CartEntity, CartInfo, CartItemInfo
+
+
+_data: dict[int, CartInfo] = {}
+
+
+def int_id_generator() -> Iterable[int]:
+    i = 0
+    while True:
+        yield i
+        i += 1
+
+
+_id_generator = int_id_generator()
+
+
+def create() -> CartEntity:
+    _id = next(_id_generator)
+    info = CartInfo(items=[], price=0.0)
+    _data[_id] = info
+    return CartEntity(id=_id, info=info)
+
+
+def add(cart_id: int, item_entity: ItemEntity) -> CartEntity:
+    cart_info = _data[cart_id]
+    for ci in cart_info.items:
+        if ci.id == item_entity.id:
+            ci.quantity += 1
+            ci.available = not item_entity.info.deleted
+            break
+    else:
+        cart_info.items.append(
+            CartItemInfo(
+                id=item_entity.id,
+                name=item_entity.info.name,
+                quantity=1,
+                available=not item_entity.info.deleted,
+            )
+        )
+    cart_info.price += item_entity.info.price
+    _data[cart_id] = cart_info
+
+    return CartEntity(id=cart_id, info=cart_info)
+
+
+def delete(id: int) -> None:
+    if id in _data:
+        del _data[id]
+
+
+def get_one(id: int) -> CartEntity | None:
+    if id not in _data:
+        return None
+
+    return CartEntity(id=id, info=_data[id])
+
+
+def get_many(
+    offset: int = 0,
+    limit: int = 10,
+    min_price: float = None,
+    max_price: float = None,
+    min_quantity: int = None,
+    max_quantity: int = None,
+) -> Iterable[CartEntity]:
+    curr = 0
+    for id, info in _data.items():
+        if min_price is not None and min_price > info.price:
+            continue
+        if max_price is not None and max_price < info.price:
+            continue
+
+        sum_quantity = 0
+        for item in info.items:
+            sum_quantity += item.quantity
+
+        if min_quantity is not None and min_quantity > sum_quantity:
+            continue
+        if max_quantity is not None and max_quantity < sum_quantity:
+            continue
+
+        if offset <= curr < offset + limit:
+            yield CartEntity(id, info)
+
+        curr += 1