hw1 + hw2 + hw3 + hw4 + hw5

.github/workflows/hw2-tests-custom.yml

@@ -0,0 +1,69 @@
+name: "HW2 Tests Custom (with Postgres)"
+
+# Запускаем тесты при изменении файлов в hw2/hw/
+on:
+  pull_request:
+    branches: [ main ]
+    paths: [ 'hw2/hw/**' ]
+  push:
+    branches: [ main ]
+    paths: [ 'hw2/hw/**' ]
+
+jobs:
+  test-hw2-custom:
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: hw2/hw
+
+    strategy:
+      matrix:
+        python-version: ["3.12", "3.13"]
+
+    services:
+      postgres:
+        image: postgres:latest
+        env:
+          POSTGRES_USER: user
+          POSTGRES_PASSWORD: password
+          POSTGRES_DB: shop_api
+        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
+      run: |
+        python -m pip install --upgrade pip
+        pip install -r requirements.txt
+
+    - name: Run tests with in-memory database
+      env:
+        PYTHONPATH: ${{ github.workspace }}/hw2/hw
+        SHOP_API_DB_TYPE: in-memory
+      run: |
+        pytest test_homework2.py -v
+
+    - name: Run tests with Postgres database
+      env:
+        PYTHONPATH: ${{ github.workspace }}/hw2/hw
+        SHOP_API_DB_TYPE: postgres
+        POSTGRES_ADDRESS: localhost
+        POSTGRES_USER: user
+        POSTGRES_PASSWORD: password
+        POSTGRES_DB: shop_api
+        POSTGRES_PORT: 5432
+      run: |
+        pytest test_homework2.py -v

hw1/app.py

@@ -1,19 +1,197 @@
 from typing import Any, Awaitable, Callable
+from urllib.parse import parse_qs
+import json
+import math
+import re
+
+
+HTTP_OK = 200
+HTTP_BAD_REQUEST = 400
+HTTP_NOT_FOUND = 404
+HTTP_UNPROCESSABLE_ENTITY = 422
+
+
+def fibonacci(n: int) -> int:
+    if n == 0:
+        return 0
+    elif n == 1:
+        return 1
+
+    a, b = 0, 1
+    for _ in range(2, n + 1):
+        a, b = b, a + b
+    return b
+
+
+def factorial(n: int) -> int:
+    return math.factorial(n)
+
+
+def mean(numbers: list[int | float]) -> float:
+    if not numbers:
+        raise ValueError('Empty list')
+    return sum(numbers) / len(numbers)
+
+
+async def read_body(receive: Callable[[], Awaitable[dict[str, Any]]]) -> bytes:
+    body_parts = []
+    while True:
+        message = await receive()
+        if message['type'] != 'http.request':
+            continue
+
+        body_parts.append(message.get('body', b''))
+        if not message.get('more_body', False):
+            break
+
+    return b''.join(body_parts)
+
+
+async def send_response(
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+    status: int,
+    body: dict[str, Any] | None = None,
+):
+    await send({
+        'type': 'http.response.start',
+        'status': status,
+        'headers': [[b"content-type", b"application/json"]],
+    })
+
+    response_body = json.dumps(body if body else {}).encode('utf-8')
+    await send({
+        'type': 'http.response.body',
+        'body': response_body,
+    })
+
+
+async def handle_lifespan(
+    receive: Callable[[], Awaitable[dict[str, Any]]],
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+):
+    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 handle_fibonacci(
+    path: str,
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+):
+    match = re.match(r'^/fibonacci/(.+)$', path)
+    if not match:
+        await send_response(send, HTTP_NOT_FOUND)
+        return
+
+    try:
+        n = int(match.group(1))
+        if n < 0:
+            await send_response(send, HTTP_BAD_REQUEST)
+            return
+
+        result = fibonacci(n)
+        await send_response(send, HTTP_OK, {'result': result})
+    except ValueError:
+        await send_response(send, HTTP_UNPROCESSABLE_ENTITY)
+
+
+async def handle_factorial(
+    query_string: bytes,
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+):
+    if not query_string:
+        await send_response(send, HTTP_UNPROCESSABLE_ENTITY)
+        return
+
+    params = parse_qs(query_string.decode('utf-8'))
+
+    if 'n' not in params or len(params['n']) != 1:
+        await send_response(send, HTTP_UNPROCESSABLE_ENTITY)
+        return
+
+    try:
+        n = int(params['n'][0])
+        if n < 0:
+            await send_response(send, HTTP_BAD_REQUEST)
+            return
+
+        result = factorial(n)
+        await send_response(send, HTTP_OK, {'result': result})
+    except ValueError:
+        await send_response(send, HTTP_UNPROCESSABLE_ENTITY)
+
+
+async def handle_mean(
+    receive: Callable[[], Awaitable[dict[str, Any]]],
+    send: Callable[[dict[str, Any]], Awaitable[None]],
+):
+    body_bytes = await read_body(receive)
+
+    if not body_bytes:
+        await send_response(send, HTTP_UNPROCESSABLE_ENTITY)
+        return
+
+    try:
+        data = json.loads(body_bytes.decode('utf-8'))
+
+        if not isinstance(data, list):
+            await send_response(send, HTTP_UNPROCESSABLE_ENTITY)
+            return
+
+        if len(data) == 0:
+            await send_response(send, HTTP_BAD_REQUEST)
+            return
+
+        if not all(isinstance(x, (int, float)) for x in data):
+            await send_response(send, HTTP_UNPROCESSABLE_ENTITY)
+            return
+
+        result = mean(data)
+        await send_response(send, HTTP_OK, {'result': result})
+    except (json.JSONDecodeError, ValueError):
+        await send_response(send, HTTP_UNPROCESSABLE_ENTITY)
 
 
 async def application(
     scope: dict[str, Any],
     receive: Callable[[], Awaitable[dict[str, Any]]],
     send: Callable[[dict[str, Any]], Awaitable[None]],
 ):
-    """
+    '''
     Args:
         scope: Словарь с информацией о запросе
         receive: Корутина для получения сообщений от клиента
         send: Корутина для отправки сообщений клиенту
-    """
-    # TODO: Ваша реализация здесь
+    '''
+    if scope['type'] == 'lifespan':
+        await handle_lifespan(receive, send)
+        return
+
+    if scope['type'] != 'http':
+        return
+
+    method = scope['method']
+    path = scope['path']
+
+    if method != 'GET':
+        await send_response(send, HTTP_NOT_FOUND)
+        return
+
+    if path.startswith('/fibonacci/'):
+        await handle_fibonacci(path, send)
+    elif path == '/factorial':
+        query_string = scope.get('query_string', b'')
+        await handle_factorial(query_string, send)
+    elif path == '/mean':
+        await handle_mean(receive, send)
+    else:
+        await send_response(send, HTTP_NOT_FOUND)
+
 
-if __name__ == "__main__":
+if __name__ == '__main__':
     import uvicorn
-    uvicorn.run("app:application", host="0.0.0.0", port=8000, reload=True)
+    uvicorn.run('app:application', host='0.0.0.0', port=8000, reload=True)

hw2/hw/Dockerfile

@@ -0,0 +1,23 @@
+FROM python:3.12 AS base
+
+ARG PYTHONFAULTHANDLER=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONHASHSEED=random \
+    PIP_NO_CACHE_DIR=on \
+    PIP_DISABLE_PIP_VERSION_CHECK=on \
+    PIP_DEFAULT_TIMEOUT=500
+
+RUN apt-get update && apt-get install -y gcc
+RUN python -m pip install --upgrade pip
+
+WORKDIR $APP_ROOT/src
+COPY . ./
+
+ENV VIRTUAL_ENV=$APP_ROOT/src/.venv \
+    PATH=$APP_ROOT/src/.venv/bin:$PATH
+
+RUN pip install -r requirements.txt
+
+FROM base as local
+
+CMD ["uvicorn", "shop_api.main:app", "--port", "8080", "--host", "0.0.0.0"]

hw2/hw/DockerfilePytest

@@ -0,0 +1,25 @@
+FROM python:3.12 AS base
+
+ARG PYTHONFAULTHANDLER=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONHASHSEED=random \
+    PIP_NO_CACHE_DIR=on \
+    PIP_DISABLE_PIP_VERSION_CHECK=on \
+    PIP_DEFAULT_TIMEOUT=500
+
+RUN apt-get update && apt-get install -y gcc
+RUN python -m pip install --upgrade pip
+
+WORKDIR $APP_ROOT/src
+COPY . ./
+
+ENV VIRTUAL_ENV=$APP_ROOT/src/.venv \
+    PATH=$APP_ROOT/src/.venv/bin:$PATH
+
+RUN pip install -r requirements.txt
+
+FROM base as local
+
+CMD ["bash", "-c", "\
+    SHOP_API_DB_TYPE=in-memory pytest -vv --cov=. --cov-report= --showlocals --strict ./test_homework2.py && \
+    SHOP_API_DB_TYPE=postgres pytest -vv --cov=. --cov-append --cov-report=term-missing --showlocals --strict ./test_homework2.py"]

hw2/hw/docker-compose-pytest.yml

@@ -0,0 +1,60 @@
+version: "3"
+
+services:
+
+  local:
+    build:
+      context: .
+      dockerfile: ./DockerfilePytest
+      target: local
+    restart: no
+    ports:
+      - 8080:8080
+    environment:
+      PYTHONPATH: .
+      POSTGRES_ADDRESS: postgres
+      POSTGRES_PORT: ${POSTGRES_PORT}
+      POSTGRES_USER: ${POSTGRES_USER}
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
+    depends_on:
+      postgres:
+        condition: service_healthy
+
+  postgres:
+    image: postgres:latest
+    ports:
+      - "${POSTGRES_PORT}:5432"
+    restart: always
+    environment:
+      POSTGRES_USER: ${POSTGRES_USER}
+      POSTGRES_DB: ${POSTGRES_DB}
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
+    healthcheck:
+      test: [ "CMD-SHELL", "pg_isready" ]
+      interval: 1s
+      timeout: 5s
+      retries: 10
+
+  grafana:
+    image: grafana/grafana:latest
+    ports:
+      - 3000:3000
+    restart: always
+    volumes:
+      - grafana-storage:/var/lib/grafana
+
+  prometheus:
+    image: prom/prometheus
+    volumes:
+      - ./settings/prometheus/:/etc/prometheus/
+    command:
+      - "--config.file=/etc/prometheus/prometheus.yml"
+      - "--storage.tsdb.path=/prometheus"
+      - "--web.console.libraries=/usr/share/prometheus/console_libraries"
+      - "--web.console.templates=/usr/share/prometheus/consoles"
+    ports:
+      - 9090:9090
+    restart: always
+
+volumes:
+  grafana-storage: {}