Hw1,2,3 + Hw4,5 (Postgresql operations + CI)

.github/workflows/hw45-tests.yml

@@ -0,0 +1,62 @@
+name: "HW45 Tests"
+
+on:
+  pull_request:
+    branches: [ main ]
+  push:
+    branches: [ main ]
+
+jobs:
+  test-hw45:
+    runs-on: ubuntu-latest
+
+    services:
+      postgres:
+        image: postgres:15
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+          POSTGRES_DB: db
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+
+    strategy:
+      matrix:
+        python-version: ["3.12"]
+
+    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: hw45
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+
+    - name: Wait for PostgreSQL starting
+      run: |
+        for i in {1..10}; do
+          pg_isready -h localhost -p 5432 -U postgres && echo "Ready" && break
+          echo "Still waiting to start db serice..."
+          sleep 2
+        done
+
+    - name: Run tests with check coverage
+      working-directory: hw45
+      env:
+        PYTHONPATH: ${{ github.workspace }}/hw45
+      run: |
+        export PYTHONPATH=$PYTHONPATH:$(pwd)
+        python -m coverage run -m pytest
+        python -m coverage report

hw1/README.md

@@ -2,57 +2,66 @@
 
 ## 🚀 Как начать работу
 
-
 ### 1. Установите зависимости
+
 ```bash
 pip install -r requirements.txt
 ```
 
 ### 2. Реализуйте ASGI приложение
+
 Откройте файл `app.py` и найдите функцию `application` с комментарием `TODO`.
 
 Вам необходимо реализовать обработку HTTP запросов внутри этой функции. Какие именно запросы вам нужно обработать?
 
 В лекции есть пример математического API с тремя ручками, их и надо обрабатывать
 
-
 ### 3. Запустите сервер для тестирования
+
 ```bash
 python app.py
 ```
 
 Сервер запустится на `http://localhost:8000`. Вы можете протестировать эндпоинты:
-- http://localhost:8000/fibonacci/10
-- http://localhost:8000/factorial?n=5
-- http://localhost:8000/mean?numbers=1,2,3
+
+- <http://localhost:8000/fibonacci/10>
+- <http://localhost:8000/factorial?n=5>
+- <http://localhost:8000/mean?numbers=1,2,3>
 
 ### 4. Запустите тесты локально, если необходимо
+
 ```bash
 pytest test_app.py -v
 ```
 
 ## 📤 Как сдать домашнее задание
 
 ### 1. Зафиксируйте изменения
+
 ```bash
 git add app.py
 git commit -m "Implement ASGI application with fibonacci, factorial, and mean endpoints"
 ```
 
 ### 2. Отправьте изменения в свой форк
+
 ```bash
 git push origin main
 ```
 
 ### 3. Создайте Pull Request
+
 1. Перейдите на страницу вашего форка на GitHub
 2. Нажмите кнопку "Compare & pull request"
 3. Убедитесь, что PR направлен в основной репозиторий в ветку `main`
 4. Добавьте описание ваших изменений
 5. Нажмите "Create pull request"
 
 ### 4. Проверьте статус тестов
-После создания PR автоматически запустятся тесты. Вы увидите статус проверки:
+
+После создания PR автоматически запустятся тесты. Вы увидите 
+статус проверки:
+
 - ✅ **Checks passing** - все тесты прошли успешно
 - ❌ **Checks failing** - есть ошибки, которые нужно исправить
 

hw1/app.py

@@ -1,18 +1,75 @@
-from typing import Any, Awaitable, Callable
+from http import HTTPStatus
+import re
+from urllib.parse import parse_qs
+
+from endpoints import send_error_response, factorial_endpoint, fibonacci_endpoint, mean_endpoint
+from utils import Receive, Scope, Send
 
 
 async def application(
-    scope: dict[str, Any],
-    receive: Callable[[], Awaitable[dict[str, Any]]],
-    send: Callable[[dict[str, Any]], Awaitable[None]],
+    scope: Scope,
+    receive: Receive,
+    send: Send,
 ):
     """
     Args:
         scope: Словарь с информацией о запросе
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
     """
-    # TODO: Ваша реализация здесь
+    print(f"Beginning connection. Scope: ", scope)
+
+    if scope["type"] == "lifespan":
+        await handle_lifetime(scope, receive, send)
+    elif scope["type"] == "http":
+        await handle_http(scope, receive, send)
+
+    print(f"Ending connection")
+
+
+async def handle_lifetime(scope: Scope, receive: Receive, send: Send):
+    assert scope["type"] == "lifespan"
+
+    while True:
+        message = await receive()
+        print(f"Got message:", message)
+
+        if message["type"] == "lifespan.startup":
+            await send({"type": "lifespan.startup.complete"})
+        elif message["type"] == "lifespan.shutdown":
+            await send({"type": "lifespan.shutdown.complete"})
+            break
+
+
+async def handle_http(scope: Scope, receive: Receive, send: Send):
+    assert scope["type"] == "http"
+
+    if scope["method"] != "GET":
+        await send_error_response(send, status=HTTPStatus.NOT_FOUND)
+    elif (m := re.match("/fibonacci/([^/\s]*)", scope["path"])):
+        await fibonacci_endpoint(scope, receive, send, value=m.group(1))
+    elif scope["path"] == "/factorial":
+        query_string = scope.get("query_string", b"").decode("utf-8")
+        query_params = parse_qs(query_string).get("n", [])
+        await factorial_endpoint(scope, receive, send, value=query_params)
+    elif scope["path"] == "/mean":
+        body = await read_body(receive)
+        await mean_endpoint(scope, receive, send, value=body)
+    else:
+        await send_error_response(send, status=HTTPStatus.NOT_FOUND)
+
+
+async def read_body(receive: Receive) -> str:
+    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.decode("utf-8")
+
 
 if __name__ == "__main__":
     import uvicorn

hw1/endpoints.py

@@ -0,0 +1,64 @@
+from http import HTTPStatus
+import json
+from utils import Receive, Scope, Send, calculate_fibonacci, calculate_factorial, calculate_mean
+
+
+async def fibonacci_endpoint(scope: Scope, receive: Receive, send: Send, value: str):
+    if value.startswith('-') and value[1:].isdigit():
+        await send_error_response(send, status=HTTPStatus.BAD_REQUEST)
+    elif value.isdigit():
+        response = str(calculate_fibonacci(int(value)))
+        await send_ok_response(send, response)
+    else:
+        await send_error_response(send, status=HTTPStatus.UNPROCESSABLE_ENTITY)
+
+
+async def factorial_endpoint(scope: Scope, receive: Receive, send: Send, value: list):
+    if len(value) and value[0].startswith('-') and value[0][1:].isdigit():
+        await send_error_response(send, status=HTTPStatus.BAD_REQUEST)
+    elif len(value) and value[0].isdigit():
+        response = str(calculate_factorial(int(value[0])))
+        await send_ok_response(send, response)
+    else:
+        await send_error_response(send, status=HTTPStatus.UNPROCESSABLE_ENTITY)
+
+
+async def mean_endpoint(scope: Scope, receive: Receive, send: Send, value: str):
+    try:
+        parsed = json.loads(value)
+
+        if not len(parsed) or not all(isinstance(x, (int, float)) for x in parsed):
+            await send_error_response(send, status=HTTPStatus.BAD_REQUEST)
+        else:
+            response = str(calculate_mean(parsed))
+            await send_ok_response(send, response)
+    except (json.JSONDecodeError, TypeError):
+        await send_error_response(send, status=HTTPStatus.UNPROCESSABLE_ENTITY)
+
+async def send_ok_response(send: Send, response: str):
+    json_response = json.dumps({"result": response}, ensure_ascii=False, indent=2).encode("utf-8")
+    await send_message(send=send, status=HTTPStatus.OK, body=json_response)
+
+
+async def send_error_response(send: Send, status=HTTPStatus.NOT_FOUND):
+    await send_message(send=send, status=status, body=b"")
+
+
+async def send_message(send: Send, status: HTTPStatus, body: bytes):
+    response_message = {
+        "type": "http.response.start",
+        "status": status,
+        "headers": [
+            [b"content-type", b"text/plain" if status == HTTPStatus.OK else b"application/json; charset=utf-8"]
+        ],
+    }
+    print("Sending response start:", response_message)
+    await send(response_message)
+
+    response_message = {
+        "type": "http.response.body",
+        "body": body,
+        "more_body": False,
+    }
+    print("Sending response body:", response_message)
+    await send(response_message)

hw1/utils.py

@@ -0,0 +1,31 @@
+from typing import Any, Awaitable, Callable
+from functools import lru_cache
+
+Scope = dict[str, Any]
+Receive = Callable[[], Awaitable[dict[str, Any]]]
+Send = Callable[[dict[str, Any]], Awaitable[None]]
+
+
+def calculate_fibonacci(n: int) -> int:
+    if n <= 1:
+        return n
+    a, b = 0, 1
+    for _ in range(2, n + 1):
+        a, b = b, a + b
+    return b
+
+
+@lru_cache(maxsize=128)
+def calculate_factorial(n: int) -> int:
+    if n < 0:
+        raise ValueError("Factorial is not defined for negative numbers")
+    if n == 0 or n == 1:
+        return 1
+    return n * calculate_factorial(n - 1)
+
+
+def calculate_mean(lst: list):
+    if not lst:
+        raise ValueError("Cannot calculate mean of empty list")
+
+    return sum(lst) / len(lst)

hw2/hw/shop_api/__init__.py

@@ -0,0 +1,4 @@
+from shop_api.data_storage import DataStorage
+
+
+data_storage = DataStorage("storage")