HW4 + HW5 Семчук Екатерина

.github/workflows/hw5-tests.yml

@@ -0,0 +1,53 @@
+name: "HW5 Tests"
+
+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"]
+
+    services:
+      postgres:
+        image: postgres:15
+        env:
+          POSTGRES_DB: hw5_db
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: password
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+
+    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 -vv --cov=shop_api ./tests

.gitignore

@@ -132,3 +132,4 @@ dmypy.json
 
 # macOS
 .DS_Store
+/hw4/settings/

hw1/app.py

@@ -1,5 +1,139 @@
+import math
+import json
+from http import HTTPStatus
 from typing import Any, Awaitable, Callable
 
+async def read_body(receive):
+    body = b''
+    more_body = True
+    while more_body:
+        message = await receive()
+        body += message.get('body', b'')
+        more_body = message.get('more_body', False)
+    return body
+
+
+def parse_query_string(scope: dict[str, Any]):
+    result = {}
+    query_string = scope.get('query_string').decode()
+    if len(query_string) == 0:
+        return result
+    for entry in query_string.split('&'):
+        key, value = entry.split('=')
+        result[key] = value
+    return result
+
+
+def not_found_response(value = ""):
+    return {
+        'value': str(value),
+        'code': HTTPStatus.NOT_FOUND
+    }
+
+def bad_request_response(value = ""):
+    return {
+        'value': str(value),
+        'code': HTTPStatus.BAD_REQUEST
+    }
+
+def unprocessable_entity_response(value = ""):
+    return {
+        'value': str(value),
+        'code': HTTPStatus.UNPROCESSABLE_ENTITY
+    }
+
+def ok_response(value = ""):
+    return {
+        'value': str(value),
+        'code': HTTPStatus.OK
+    }
+
+
+def get_factorial(query_params: dict[str, Any], path_parameters: list[str], body: bytes) -> dict[str, Any]:
+    n = int(query_params.get('n'))
+    if n < 0:
+        return bad_request_response("Invalid value for n, must be a non-negative")
+    result = math.factorial(n)
+    return ok_response(result)
+
+
+def get_fibonacci(query_params: dict[str, Any], path_parameters: list[str], body: bytes) -> dict[str, Any]:
+    n = int(path_parameters[0])
+    if n < 0:
+        return bad_request_response("Invalid value for n, must be a non-negative")
+    a, b = 0, 1
+    for _ in range(n):
+        a, b = b, a + b
+    return ok_response(a)
+
+
+def get_mean(query_params: dict[str, Any], path_parameters: list[str], body: bytes) -> dict[str, Any]:
+    data = json.loads(body.decode())
+    if len(data) == 0:
+        return bad_request_response("Invalid value for body, must be non-empty array of floats")
+    result = sum(data) / len(data)
+    return ok_response(result)
+
+
+def route_request(scope: dict[str, Any]) -> Callable:
+    method = scope.get('method')
+    path = scope.get('path')
+
+    function = None
+    routing_params = []
+
+    if method == "GET":
+        if path.startswith("/factorial"):
+            function = get_factorial
+            routing_params = path[10:].split("/")
+        elif path.startswith("/fibonacci"): 
+            function = get_fibonacci
+            routing_params = path[11:].split("/")
+        elif path.startswith("/mean"): 
+            function = get_mean
+            routing_params = path[6:].split("/")
+
+    return [function, routing_params]
+
+
+async def handle_http(scope: dict[str, Any], receive: Callable, send: Callable):
+
+    function, path_parameters = route_request(scope)
+    if function is None:
+        response = not_found_response()
+
+    else:
+        try:
+            body = await read_body(receive)
+            query_params = parse_query_string(scope)
+            response = function(query_params, path_parameters, body)
+        except Exception:
+            response = unprocessable_entity_response()
+
+    code = response['code']
+    body = json.dumps({"result": response['value']})
+    await send({
+        'type': 'http.response.start',
+        'status': code,
+        'headers': [
+            [b'content-type', b'text/plain'],
+        ]
+    })
+    await send({
+        'type': 'http.response.body',
+        'body': body.encode()
+    })
+
+
+async def handle_lifespan(scope: dict[str, Any], receive: Callable, send: Callable):
+    while True:
+        message = await receive()
+        if message['type'] == 'lifespan.startup':
+            await send({'type': 'lifespan.startup.complete'})
+        elif message['type'] == 'lifespan.shutdown':
+            await send({'type': 'lifespan.shutdown.complete'})
+            return
+
 
 async def application(
     scope: dict[str, Any],
@@ -12,7 +146,12 @@ async def application(
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    scope_type = scope['type']
+    if scope_type == 'lifespan':
+        await handle_lifespan(scope, receive, send)
+    elif scope_type == 'http':
+        await handle_http(scope, receive, send)
+
 
 if __name__ == "__main__":
     import uvicorn

hw2/hw/shop_api/main.py

@@ -1,3 +1,183 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, Response, HTTPException, status
+from pydantic import BaseModel, conint, confloat
 
 app = FastAPI(title="Shop API")
+cart_id_counter = 0
+item_id_counter = 0
+
+carts = {}
+items = {}
+
+posint = conint(gt=0)
+uint = conint(ge=0)
+ufloat = confloat(ge=0)
+
+class ModifiedItem(BaseModel):
+    name: str | None = None
+    price: float | None = None
+    model_config = {
+        "extra": "forbid"
+    }
+
+class Item(BaseModel):
+    id: uint = 0
+    name: str
+    price: float
+    deleted: bool = False
+
+class CartItem(BaseModel):
+    id: uint = 0
+    name: str
+    price: float
+    quantity: int = 0
+    deleted: bool = False
+
+class Cart(BaseModel):
+    id: int
+    price: float = 0.0
+    items: list[CartItem] = []
+
+
+"""
+cart
+"""
+
+# POST cart - создание, работает как RPC, не принимает тело, возвращает идентификатор
+@app.post("/cart", status_code=status.HTTP_201_CREATED)
+async def create_cart(response: Response):
+    global cart_id_counter
+    cart_id_counter += 1
+    new_cart = Cart(
+        id=cart_id_counter
+    )
+    carts[new_cart.id] = new_cart
+    response.headers["Location"] = f"/cart/{new_cart.id}"
+    return {"id": new_cart.id}
+
+
+# GET /cart/{id} - получение корзины по id
+@app.get("/cart/{cart_id}", status_code=status.HTTP_200_OK)
+async def get_cart(cart_id: int):
+    return carts[cart_id]
+
+
+# GET /cart - получение списка корзин с query-параметрами
+#    offset - неотрицательное целое число, смещение по списку (опционально, по-умолчанию 0)
+#    limit - положительное целое число, ограничение на количество (опционально, по-умолчанию 10)
+#    min_price - число с плавающей запятой, минимальная цена включительно (опционально, если нет, не учитывает в фильтре)
+#    max_price - число с плавающей запятой, максимальная цена включительно (опционально, если нет, не учитывает в фильтре)
+#    min_quantity - неотрицательное целое число, минимальное общее число товаров включительно (опционально, если нет, не учитывается в фильтре)
+#    max_quantity - неотрицательное целое число, максимальное общее число товаров включительно (опционально, если нет, не учитывается в фильтре)
+@app.get("/cart", status_code=status.HTTP_200_OK)
+async def get_cart_list(offset: uint = 0, limit: posint = 10,
+                        min_price: ufloat = None, max_price: ufloat = None,
+                        min_quantity: uint = None, max_quantity: uint = None):
+    filtered_carts = []
+    for cart in list(carts.values())[offset:]:
+        if len(filtered_carts) == limit:
+            break
+
+        min_price_ok = cart.price >= min_price if min_price else True
+        max_price_ok = cart.price <= max_price if max_price else True
+        min_quantity_ok = sum(item.quantity for item in cart.items) >= min_quantity if not min_quantity is None else True
+        max_quantity_ok = sum(item.quantity for item in cart.items) <= max_quantity if not max_quantity is None else True
+
+        if min_price_ok and max_price_ok and min_quantity_ok and max_quantity_ok:
+            filtered_carts.append(cart)
+    return filtered_carts
+
+
+# POST /cart/{cart_id}/add/{item_id} - добавление в корзину с cart_id предмета с item_id,
+# если товар уже есть, то увеличивается его количество
+@app.post("/cart/{cart_id}/add/{item_id}", status_code=status.HTTP_200_OK)
+async def add_to_cart(cart_id: int, item_id: int):
+    cart = carts[cart_id]
+    item = items[item_id]
+
+    item_exists = False
+    for i in range(len(cart.items)):
+        if cart.items[i].id == item_id:
+            cart.items[i].quantity += 1
+            item_exists = True
+            break
+
+    if not item_exists:
+        new_item = CartItem(
+            **item.model_dump(),
+            quantity=1
+        )
+        cart.items.append(new_item)
+
+    cart.price += item.price
+
+
+"""
+item
+"""
+
+# POST /item - добавление нового товара
+@app.post("/item", status_code=status.HTTP_201_CREATED)
+async def create_item(item: Item):
+    global item_id_counter
+    item_id_counter += 1
+    new_item = Item(id=item_id_counter, name=item.name, price=item.price)
+    items[new_item.id] = new_item
+    return new_item
+
+# GET /item/{id} - получение товара по id
+@app.get("/item/{item_id}", status_code=status.HTTP_200_OK)
+async def get_item(item_id: int):
+    if items[item_id].deleted:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND
+        )
+    return items[item_id]
+
+
+# GET /item - получение списка товаров с query-параметрами
+#    offset - неотрицательное целое число, смещение по списку (опционально, по-умолчанию 0)
+#    limit - положительное целое число, ограничение на количество (опционально, по-умолчанию 10)
+#    min_price - число с плавающей запятой, минимальная цена (опционально, если нет, не учитывает в фильтре)
+#    max_price - число с плавающей запятой, максимальная цена (опционально, если нет, не учитывает в фильтре)
+#    show_deleted - булевая переменная, показывать ли удаленные товары (по умолчанию False)
+@app.get("/item", status_code=status.HTTP_200_OK)
+async def get_item_list(offset: uint = 0, limit: posint = 10,
+                        min_price: ufloat = None, max_price: ufloat = None,
+                        show_deleted: bool = False):
+    filtered_items = []
+    for item in list(items.values())[offset:]:
+        if len(filtered_items) == limit:
+            break
+
+        min_price_ok = item.price >= min_price if min_price else True
+        max_price_ok = item.price <= max_price if max_price else True
+        show_deleted_ok = (not item.deleted) or show_deleted
+
+        if min_price_ok and max_price_ok and show_deleted_ok:
+            filtered_items.append(item)
+
+    return filtered_items
+
+# PUT /item/{id} - замена товара по id (создание запрещено, только замена существующего)
+@app.put("/item/{item_id}", status_code=status.HTTP_200_OK)
+async def put_item(item_id: int, new_item: Item):
+    new_item.id = item_id
+    items[item_id] = new_item
+    return new_item
+
+# PATCH /item/{id} - частичное обновление товара по id (разрешено менять все поля, кроме deleted)
+@app.patch("/item/{item_id}", status_code=status.HTTP_200_OK)
+async def patch_item(item_id: int, new_item: ModifiedItem):
+    item = items[item_id]
+    if item.deleted:
+        raise HTTPException(status_code=status.HTTP_304_NOT_MODIFIED)
+    if new_item.name and item.name != new_item.name:
+        items[item_id].name = new_item.name
+    if new_item.price and item.price != new_item.price:
+        items[item_id].price = new_item.price
+    return items[item_id]
+
+# DELETE /item/{id} - удаление товара по id (товар помечается как удаленный)
+@app.delete("/item/{item_id}", status_code=status.HTTP_200_OK)
+async def delete_item(item_id: int):
+    items[item_id].deleted = True