HW 1, HW 2 + chat, HW4, HW5

.github/workflows/hw5-tests.yml

@@ -0,0 +1,38 @@
+name: "HW5 Tests"
+
+on:
+  pull_request:
+    branches: [ main ]
+    paths: [ 'hw2/hw/**' ]
+  push:
+    branches: [ main ]
+    paths: [ 'hw2/hw/**' ]
+
+jobs:
+  test-hw5:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.12"]
+
+    steps:
+    - name: Check out repository 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: hw2/hw
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+
+    - name: Run tests
+      working-directory: hw2/hw
+      env:
+        PYTHONPATH: ${{ github.workspace }}/hw2/hw
+      run: |
+        pytest tests/test_api.py --cov=shop_api --cov-report=term-missing --cov-fail-under=95

hw1/app.py

@@ -1,4 +1,27 @@
 from typing import Any, Awaitable, Callable
+import json
+import math
+
+
+def fibonacchi(n):
+    if n < 0:
+        raise Exception()
+    prev, ans = 0, 1
+    for _ in range(n):
+        prev, ans = ans, ans + prev
+    return ans
+
+
+def factorial(n):
+    if n < 0:
+        raise Exception()
+    return math.factorial(n)
+
+
+def mean(arr):
+    if len(arr) == 0:
+        raise Exception()
+    return sum(arr) / len(arr)
 
 
 async def application(
@@ -12,7 +35,53 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    if 'path' in scope and scope['method'] == 'GET':
+        splitted_path = scope['path'].split('/')
+    else:
+        splitted_path = [None, None]
+    body = {"result": None}
+    status = 200
+    function = None
+    try:
+        if splitted_path[1] == 'fibonacci':
+            val = splitted_path[2]
+            assert '.' not in val
+            param = int(val)
+            function = fibonacchi
+        elif splitted_path[1] == 'factorial':
+            key, val = scope['query_string'].decode('utf-8').split('=')
+            assert key == 'n'
+            assert '.' not in val
+            param = int(val)
+            function = factorial
+        elif splitted_path[1] == 'mean':
+            if scope['query_string'] != b'':
+                key, val = scope['query_string'].decode('utf-8').split('=')
+                assert key == 'numbers'
+            else:
+                val = await receive()
+                val = val['body'].decode('utf-8')
+                assert val[0] == '[' and val[-1] == ']'
+                val = val[1:-1]
+            param = []
+            if val != '':
+                for num in val.split(','):
+                    param.append(float(num))
+            function = mean
+        else:
+            status = 404
+    except:
+        status = 422
+    if function is not None:
+        try:
+            body = {"result": function(param)}
+        except:
+            status = 400
+    await send({'type': 'http.response.start',
+                'status': status,
+               'headers': [[b'content-type', b'application/json']]})
+    await send({'type': 'http.response.body', 'body': json.dumps(body).encode('utf-8')})
+
 
 if __name__ == "__main__":
     import uvicorn

hw2/hw/chat/client.py

@@ -0,0 +1,35 @@
+import asyncio
+import websockets
+
+
+async def receive_messages(websocket):
+    try:
+        async for message in websocket:
+            print(message)
+    except:
+        pass
+
+
+async def send_messages(websocket):
+    loop = asyncio.get_running_loop()
+    while True:
+        try:
+            message = await loop.run_in_executor(None, lambda: input())
+            await websocket.send(message)
+        except:
+            break
+
+
+async def main():
+    chat_name = input("Chat name: ").strip()
+
+    uri = f"ws://127.0.0.1:8000/chat/{chat_name}"
+    async with websockets.connect(uri) as websocket:
+        await asyncio.gather(
+            receive_messages(websocket),
+            send_messages(websocket)
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

hw2/hw/chat/server.py

@@ -0,0 +1,47 @@
+from collections import defaultdict
+from typing import Dict, List
+
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+import uuid
+
+
+app = FastAPI()
+
+
+class ConnectionManager:
+    def __init__(self):
+        self.active_connections: Dict[str, List[WebSocket]] = defaultdict(list)
+
+    async def connect(self, websocket: WebSocket, chat_name: str) -> str:
+        await websocket.accept()
+        username = str(uuid.uuid4())
+        self.active_connections[chat_name].append(websocket)
+        return username
+
+    def disconnect(self, websocket: WebSocket, chat_name: str):
+        self.active_connections[chat_name].remove(websocket)
+        if chat_name in self.active_connections and len(self.active_connections[chat_name]) == 0:
+            del self.active_connections[chat_name]
+
+    async def broadcast(self, message: str, chat_name: str, websocket = None):
+        for connection in self.active_connections[chat_name]:
+            if connection != websocket:
+                await connection.send_text(message)
+
+
+manager = ConnectionManager()
+
+
+@app.websocket("/chat/{chat_name}")
+async def websocket_endpoint(websocket: WebSocket, chat_name: str):
+    username = await manager.connect(websocket, chat_name)
+
+    await manager.broadcast(f"--- {username} joined the chat ---", chat_name)
+
+    try:
+        while True:
+            data = await websocket.receive_text()
+            await manager.broadcast(f"{username} :: {data}", chat_name, websocket)
+    except WebSocketDisconnect:
+        manager.disconnect(websocket, chat_name)
+        await manager.broadcast(f"--- {username} left the chat ---", chat_name)

hw2/hw/requirements.txt

@@ -5,5 +5,6 @@ uvicorn>=0.24.0
 # Зависимости для тестирования
 pytest>=7.4.0
 pytest-asyncio>=0.21.0
+pytest-cov
 httpx>=0.27.2
 Faker>=37.8.0

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

@@ -0,0 +1 @@
+from .routes import router

hw2/hw/shop_api/api/cart/contracts.py

@@ -0,0 +1,33 @@
+from typing import List
+
+from pydantic import BaseModel
+
+from hw2.hw.shop_api.store.models import (
+    CartEntity,
+    AddItemInfo
+)
+
+
+class ItemCartResponse(BaseModel):
+    id: int
+    name: str
+    quantity: int
+    available: bool
+
+
+class CartResponse(BaseModel):
+    id: int
+    items: List[ItemCartResponse]
+    price: float
+
+    @staticmethod
+    def from_entity(entity: CartEntity):
+        return CartResponse(
+            id=entity.id,
+            items=[ItemCartResponse(id=item_id,
+                                    name=info.name,
+                                    quantity=info.quantity,
+                                    available=info.available)
+                   for item_id, info in entity.info.items.items()],
+            price=entity.info.price
+        )

hw2/hw/shop_api/api/cart/routes.py

@@ -0,0 +1,78 @@
+from http import HTTPStatus
+from typing import Annotated
+
+from fastapi import APIRouter, HTTPException, Query, Response, Request
+from pydantic import NonNegativeInt, PositiveInt, NonNegativeFloat
+
+from hw2.hw.shop_api import store
+
+from .contracts import CartResponse
+
+router = APIRouter(prefix="/cart")
+
+
+@router.post(
+    "/",
+    status_code=HTTPStatus.CREATED,
+)
+async def post_cart(_: Request, response: Response) -> CartResponse:
+    entity = store.add_cart()
+
+    # as REST states one should provide uri to newly created resource in location header
+    response.headers["location"] = f"/cart/{entity.id}"
+
+    return CartResponse.from_entity(entity)
+
+
+@router.get(
+    "/{id}",
+    responses={
+        HTTPStatus.OK: {
+            "description": "Successfully returned requested cart",
+        },
+        HTTPStatus.NOT_FOUND: {
+            "description": "Failed to return requested cart as one was not found",
+        },
+    },
+)
+async def get_cart_by_id(id: int) -> CartResponse:
+    entity = store.get_one_cart(id)
+
+    if not entity:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Request resource /cart/{id} was not found",
+        )
+
+    return CartResponse.from_entity(entity)
+
+
+@router.get("/")
+async def get_cart_list(
+    offset: Annotated[NonNegativeInt, Query()] = 0,
+    limit: Annotated[PositiveInt, Query()] = 10,
+    min_price: Annotated[NonNegativeFloat, Query()] = 0,
+    max_price: Annotated[NonNegativeFloat, Query()] = 1e10,
+    min_quantity: Annotated[NonNegativeInt, Query()] = 0,
+    max_quantity: Annotated[NonNegativeInt, Query()] = 1e10,
+) -> list[CartResponse]:
+    return [CartResponse.from_entity(e) for e in store.get_many_carts(offset, limit, min_price, max_price, min_quantity, max_quantity)]
+
+
+@router.post(
+    "/{cart_id}/add/{item_id}",
+    status_code=HTTPStatus.CREATED,
+)
+async def post_item_to_cart(cart_id: int, item_id: int, response: Response) -> CartResponse:
+    entity = store.add_item_to_cart(cart_id, item_id)
+
+    if not entity:
+        raise HTTPException(
+            HTTPStatus.NOT_FOUND,
+            f"Request resource /{cart_id}/cart/{item_id} was not found",
+        )
+
+    # as REST states one should provide uri to newly created resource in location header
+    response.headers["location"] = f"/cart/{entity.id}"
+
+    return CartResponse.from_entity(entity)

hw2/hw/shop_api/api/item/__init__.py

@@ -0,0 +1 @@
+from .routes import router

hw2/hw/shop_api/api/item/contracts.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from pydantic import BaseModel, ConfigDict
+
+from hw2.hw.shop_api.store.models import (
+    ItemInfo,
+    ItemEntity,
+    PatchItemInfo
+)
+
+
+class ItemResponse(BaseModel):
+    id: int
+    name: str
+    price: float
+    deleted: bool
+
+    @staticmethod
+    def from_entity(entity: ItemEntity) -> ItemResponse:
+        return ItemResponse(
+            id=entity.id,
+            name=entity.info.name,
+            price=entity.info.price,
+            deleted=entity.info.deleted
+        )
+
+
+class ItemRequest(BaseModel):
+    name: str
+    price: float
+    deleted: bool = False
+
+    def as_item_info(self) -> ItemInfo:
+        return ItemInfo(name=self.name, price=self.price, deleted=self.deleted)
+
+
+class PatchItemRequest(BaseModel):
+    name: str | None = None
+    price: float | None = None
+
+    model_config = ConfigDict(extra="forbid")
+
+    def as_patch_item_info(self) -> PatchItemInfo:
+        return PatchItemInfo(name=self.name, price=self.price)