Roman Kharkovskoy HW1

.github/workflows/hw5-ci.yml

@@ -0,0 +1,57 @@
+name: Python CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        python-version: [3.11, 3.13]
+
+    defaults:
+      run:
+        working-directory: hw5
+
+    steps:
+    - name: Checkout repository
+      uses: actions/checkout@v3
+
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      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-cov fastapi uvicorn httpx
+
+    - name: Set PYTHONPATH
+      run: echo "PYTHONPATH=$PWD" >> $GITHUB_ENV
+
+    - name: Run tests with coverage
+      run: |
+        pytest --cov=shop_api \
+               --cov-report=term-missing \
+               --cov-report=xml \
+               --cov-report=html \
+               tests/
+
+    - name: Upload coverage XML
+      uses: actions/upload-artifact@v4
+      with:
+        name: coverage-xml
+        path: coverage.xml
+
+    - name: Upload coverage HTML
+      uses: actions/upload-artifact@v4
+      with:
+        name: coverage-html
+        path: htmlcov/

hw1/app.py

@@ -1,5 +1,8 @@
 from typing import Any, Awaitable, Callable
+from urllib.parse import parse_qs
+import json
 
+from utils import fibonacci, factorial, mean
 
 async def application(
     scope: dict[str, Any],
@@ -12,7 +15,92 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    if scope.get("type") == "lifespan":
+        while True:
+            message = await receive()
+            if message.get("type") == "lifespan.startup":
+                await send({"type": "lifespan.startup.complete"})
+            elif message.get("type") == "lifespan.shutdown":
+                await send({"type": "lifespan.shutdown.complete"})
+                return
+
+    path = scope["path"]
+    query = parse_qs(scope["query_string"].decode())
+    method = scope.get("method", "GET")
+
+    def json_response(data: dict, status_code: int = 200):
+        return json.dumps(data).encode(), b"application/json", status_code
+
+    if path == "/factorial" and method == "GET":
+        n_raw = query.get("n", [None])[0]
+        if not n_raw:
+            body, content_type, status = json_response({"detail": "Missing or empty 'n' parameter"}, 422)
+        else:
+            try:
+                n = int(n_raw)
+            except Exception:
+                body, content_type, status = json_response({"detail": "'n' must be an integer"}, 422)
+            else:
+                if n < 0:
+                    body, content_type, status = json_response({"detail": "'n' must be >= 0"}, 400)
+                else:
+                    body, content_type, status = json_response({"result": factorial(n)})
+
+    elif path.startswith("/fibonacci") and method == "GET":
+        parts = path.split("/")
+        n_raw = parts[2] if len(parts) > 2 else ""
+        if not n_raw:
+            body, content_type, status = json_response({"detail": "Missing or empty n parameter"}, 422)
+        else:
+            try:
+                n = int(n_raw)
+            except Exception:
+                body, content_type, status = json_response({"detail": "n must be integer"}, 422)
+            else:
+                if n < 0:
+                    body, content_type, status = json_response({"detail": "n must be >= 0"}, 400)
+                else:
+                    body, content_type, status = json_response({"result": fibonacci(n)})
+
+    elif path == "/mean" and method == "GET":
+        raw_body = scope.get("body", b"")
+        while True:
+            message = await receive()
+            if message["type"] == "http.request":
+                raw_body += message.get("body", b"")
+                if not message.get("more_body", False):
+                    break
+        try:
+            body_data = json.loads(raw_body.decode()) if raw_body else None
+        except Exception:
+            body_data = None
+
+        if body_data is None:
+            body, content_type, status = json_response({"detail": "Missing or invalid body"}, 422)
+        elif not isinstance(body_data, list) or not body_data:
+            body, content_type, status = json_response({"detail": "Body must be non-empty list"}, 400)
+        else:
+            try:
+                numbers = [float(x) for x in body_data]
+            except Exception:
+                body, content_type, status = json_response({"detail": "All elements must be numbers"}, 400)
+            else:
+                body, content_type, status = json_response({"result": mean(",".join(str(x) for x in numbers))})
+
+    else:
+        body, content_type, status = json_response({"detail": "Not Found"}, 404)
+
+    await send({
+        "type": "http.response.start",
+        "status": status,
+        "headers": [[b"content-type", content_type]]
+    })
+
+    await send({
+        "type": "http.response.body",
+        "body": body
+    })
+
 
 if __name__ == "__main__":
     import uvicorn

hw1/utils.py

@@ -0,0 +1,19 @@
+def fibonacci(n):
+    n = int(n)
+    if n <= 0:
+        return []
+    seq = [0, 1]
+    for _ in range(2, n):
+        seq.append(seq[-1] + seq[-2])
+    return seq[:n]
+
+def factorial(n):
+    n = int(n)
+    result = 1
+    for i in range(2, n+1):
+        result *= i
+    return result
+
+def mean(numbers):
+    numbers = [float(x) for x in numbers.split(",")]
+    return sum(numbers) / len(numbers)

hw2/hw/shop_api/main.py

@@ -1,3 +1,9 @@
 from fastapi import FastAPI
+from shop_api.routers import items, carts, chat
+
 
 app = FastAPI(title="Shop API")
+
+app.include_router(items.router)
+app.include_router(carts.router)
+app.include_router(chat.router)

hw2/hw/shop_api/routers/carts.py

@@ -0,0 +1,58 @@
+from fastapi import APIRouter, HTTPException, Query, Path, Response
+from shop_api.schemas.cart import Cart
+from shop_api.utils.cart_utils import compute_cart
+from shop_api.storage.memory import _carts, _items, _lock, _next_cart_id
+from typing import Optional, List
+
+
+router = APIRouter(prefix="/cart", tags=["carts"])
+
+@router.post("/", status_code=201)
+def create_cart(response: Response):
+    global _next_cart_id
+    with _lock:
+        cid = _next_cart_id
+        _next_cart_id += 1
+        _carts[cid] = {}
+    response.headers["Location"] = f"/cart/{cid}"
+    return {"id": cid}
+
+@router.get("/{id}", response_model=Cart)
+def get_cart(id: int = Path(..., gt=0)):
+    try:
+        return compute_cart(id)
+    except KeyError:
+        raise HTTPException(status_code=404, detail="cart not found")
+
+@router.get("/", response_model=List[Cart])
+def list_carts(
+    offset: int = Query(0, ge=0),
+    limit: int = Query(10, gt=0),
+    min_price: Optional[float] = Query(None, ge=0),
+    max_price: Optional[float] = Query(None, ge=0),
+    min_quantity: Optional[int] = Query(None, ge=0),
+    max_quantity: Optional[int] = Query(None, ge=0),
+):
+    carts = []
+    for cid in sorted(_carts.keys()):
+        cart = compute_cart(cid)
+        total_qty = sum(i.quantity for i in cart.items)
+        if min_quantity is not None and total_qty < min_quantity:
+            continue
+        if max_quantity is not None and total_qty > max_quantity:
+            continue
+        if min_price is not None and cart.price < min_price:
+            continue
+        if max_price is not None and cart.price > max_price:
+            continue
+        carts.append(cart)
+    return carts[offset: offset + limit]
+
+@router.post("/{cart_id}/add/{item_id}", response_model=Cart)
+def add_item(cart_id: int, item_id: int):
+    if cart_id not in _carts:
+        raise HTTPException(status_code=404, detail="cart not found")
+    if item_id not in _items:
+        raise HTTPException(status_code=404, detail="item not found")
+    _carts[cart_id][item_id] = _carts[cart_id].get(item_id, 0) + 1
+    return compute_cart(cart_id)

hw2/hw/shop_api/routers/chat.py

@@ -0,0 +1,30 @@
+from fastapi import APIRouter, WebSocket, WebSocketDisconnect
+from collections import defaultdict
+import random
+import string
+from typing import Dict, List
+
+router = APIRouter(prefix="/chat", tags=["chat"])
+
+chat_rooms: Dict[str, List[WebSocket]] = defaultdict(list)
+usernames: Dict[WebSocket, str] = {}
+
+def random_username() -> str:
+    return ''.join(random.choices(string.ascii_letters + string.digits, k=8))
+
+@router.websocket("/{chat_name}")
+async def websocket_chat(websocket: WebSocket, chat_name: str):
+    await websocket.accept()
+    username = random_username()
+    usernames[websocket] = username
+    chat_rooms[chat_name].append(websocket)
+    try:
+        while True:
+            data = await websocket.receive_text()
+            message = f"{username} :: {data}"
+            for ws in chat_rooms[chat_name]:
+                if ws != websocket:
+                    await ws.send_text(message)
+    except WebSocketDisconnect:
+        chat_rooms[chat_name].remove(websocket)
+        del usernames[websocket]

hw2/hw/shop_api/routers/items.py

@@ -0,0 +1,81 @@
+from fastapi import APIRouter, HTTPException, Query, Response
+from shop_api.schemas.item import Item, ItemCreate
+from shop_api.storage.memory import _items, _lock, _next_item_id
+from typing import Optional, List
+
+
+router = APIRouter(prefix="/item", tags=["items"])
+
+@router.post("/", response_model=Item, status_code=201)
+def create_item(item: ItemCreate):
+    global _next_item_id
+    with _lock:
+        iid = _next_item_id
+        _next_item_id += 1
+        new_item = Item(id=iid, name=item.name, price=item.price)
+        _items[iid] = new_item
+    return new_item
+
+@router.get("/{id}", response_model=Item)
+def get_item(id: int):
+    item = _items.get(id)
+    if not item or item.deleted:
+        raise HTTPException(status_code=404, detail="item not found")
+    return item
+
+@router.get("/", response_model=List[Item])
+def list_items(
+    offset: int = Query(0, ge=0),
+    limit: int = Query(10, gt=0),
+    min_price: Optional[float] = Query(None, ge=0),
+    max_price: Optional[float] = Query(None, ge=0),
+    show_deleted: bool = Query(False),
+):
+    items = []
+    for it in _items.values():
+        if not show_deleted and it.deleted:
+            continue
+        if min_price is not None and it.price < min_price:
+            continue
+        if max_price is not None and it.price > max_price:
+            continue
+        items.append(it)
+    return items[offset: offset + limit]
+
+@router.put("/{id}", response_model=Item)
+def replace_item(id: int, item: ItemCreate):
+    if id not in _items:
+        raise HTTPException(status_code=404, detail="item not found")
+    existing = _items[id]
+    existing.name = item.name
+    existing.price = item.price
+    _items[id] = existing
+    return existing
+
+@router.patch("/{id}", response_model=Item)
+def patch_item(id: int, patch: dict):
+    if id not in _items:
+        raise HTTPException(status_code=404, detail="item not found")
+    item = _items[id]
+    if item.deleted:
+        return Response(status_code=304)
+    allowed_keys = {"name", "price"}
+    if not set(patch.keys()).issubset(allowed_keys):
+        raise HTTPException(status_code=422)
+    if "price" in patch:
+        price = patch["price"]
+        if price is not None and price < 0:
+            raise HTTPException(status_code=422)
+        item.price = price
+    if "name" in patch:
+        item.name = patch["name"]
+    _items[id] = item
+    return item
+
+@router.delete("/{id}")
+def delete_item(id: int):
+    item = _items.get(id)
+    if not item:
+        return {"status": "ok"}
+    item.deleted = True
+    return {"status": "ok"}

hw2/hw/shop_api/schemas/cart.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+from typing import List
+
+
+class CartItem(BaseModel):
+    id: int
+    name: str
+    quantity: int
+    available: bool
+
+class Cart(BaseModel):
+    id: int
+    items: List[CartItem]
+    price: float