lab 7

students_folder/Nikulina/lab7/.gitignore

@@ -0,0 +1,7 @@
+logs/
+*.log
+pycache/
+*.pyc
+dag_processor_manager/
+scheduler/
+pgdata/

students_folder/Nikulina/lab7/app/main.py

@@ -0,0 +1,15 @@
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from app.operations.database_migrations import create_table, delete_table
+from app.router_habits import router as habits_router
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    await create_table()
+    print("Таблицы созданы")
+    yield
+    await delete_table()
+    print("Таблицы удалены")
+
+app = FastAPI(lifespan=lifespan)
+app.include_router(habits_router)

students_folder/Nikulina/lab7/app/models/database_models.py

@@ -0,0 +1,27 @@
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+from sqlalchemy import DateTime, BIGINT, Integer, Text
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+def utc_now():
+    return datetime.now(ZoneInfo("UTC"))
+
+class TableModel(DeclarativeBase):
+    pass
+
+class HabitOrm(TableModel):
+    __tablename__ = "habit"
+    __table_args__ = {"schema": "public"}
+
+    id: Mapped[int] = mapped_column(BIGINT, primary_key=True, autoincrement=True)
+    user_id: Mapped[int]
+    name: Mapped[str]
+    description: Mapped[str | None]
+
+class CompletionOrm(TableModel):
+    __tablename__ = "completion"
+    __table_args__ = {"schema": "public"}
+
+    id: Mapped[int] = mapped_column(BIGINT, primary_key=True, autoincrement=True)
+    habit_id: Mapped[int]
+    date: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=utc_now)

students_folder/Nikulina/lab7/app/models/store_models.py

@@ -0,0 +1,43 @@
+from pydantic import BaseModel, AfterValidator, Field, ConfigDict, RootModel
+from typing import Annotated, List, Optional
+from datetime import datetime
+from zoneinfo import ZoneInfo
+
+def validate_aware(dt: datetime) -> datetime:
+    if dt.tzinfo is None:
+        raise ValueError("datetime must include timezone information")
+    return dt
+
+AwareDatetime = Annotated[datetime, AfterValidator(validate_aware)]
+
+def utc_now():
+    return datetime.now(ZoneInfo("UTC"))
+
+class HabitBase(BaseModel):
+    user_id: int
+    name: str
+    description: Optional[str] = None
+
+class HabitAdd(HabitBase):
+    pass
+
+class HabitGet(HabitBase):
+    id: int
+    model_config = ConfigDict(from_attributes=True)
+
+class HabitListGet(RootModel[List[HabitGet]]):
+    model_config = ConfigDict(from_attributes=True)
+
+class CompletionBase(BaseModel):
+    habit_id: int
+    date: AwareDatetime = Field(default_factory=utc_now)
+
+class CompletionAdd(CompletionBase):
+    pass
+
+class CompletionGet(CompletionBase):
+    id: int
+    model_config = ConfigDict(from_attributes=True)
+
+class CompletionListGet(RootModel[List[CompletionGet]]):
+    model_config = ConfigDict(from_attributes=True)

students_folder/Nikulina/lab7/app/operations/database_migrations.py

@@ -0,0 +1,14 @@
+from sqlalchemy.ext.asyncio import create_async_engine
+from app.models.database_models import TableModel
+
+engine = create_async_engine(
+    "postgresql+asyncpg://postgres:postgres@localhost:5433/postgres"
+)
+
+async def create_table():
+    async with engine.begin() as conn:
+        await conn.run_sync(TableModel.metadata.create_all)
+
+async def delete_table():
+    async with engine.begin() as conn:
+        await conn.run_sync(TableModel.metadata.drop_all)

students_folder/Nikulina/lab7/app/operations/database_operations.py

@@ -0,0 +1,79 @@
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+from sqlalchemy import select, func
+from typing import List
+from datetime import timedelta, datetime, timezone
+
+from app.models.database_models import HabitOrm, CompletionOrm
+from app.models.store_models import (
+    HabitAdd, HabitListGet, HabitGet,
+    CompletionAdd, CompletionListGet, CompletionGet
+)
+
+engine = create_async_engine(
+    "postgresql+asyncpg://postgres:postgres@localhost:5433/postgres"
+)
+new_session = async_sessionmaker(engine, expire_on_commit=False)
+
+class HabitWorkflow:
+
+    @classmethod
+    async def add_habit(cls, habit: HabitAdd) -> int:
+        async with new_session() as session:
+            data = habit.model_dump()
+            model = HabitOrm(**data)
+            session.add(model)
+            await session.flush()
+            await session.commit()
+            return model.id
+
+    @classmethod
+    async def get_habits(cls) -> HabitListGet:
+        async with new_session() as session:
+            result = await session.execute(select(HabitOrm))
+            models = result.scalars().all()
+            return HabitListGet.model_validate(models).root
+
+class CompletionWorkflow:
+
+    @classmethod
+    async def add_completion(cls, comp: CompletionAdd) -> int:
+        async with new_session() as session:
+            data = comp.model_dump()
+            model = CompletionOrm(**data)
+            session.add(model)
+            await session.flush()
+            await session.commit()
+            return model.id
+
+    @classmethod
+    async def get_completions(cls, habit_id: int) -> CompletionListGet:
+        async with new_session() as session:
+            result = await session.execute(
+                select(CompletionOrm).where(CompletionOrm.habit_id == habit_id)
+            )
+            models = result.scalars().all()
+            return CompletionListGet.model_validate(models).root
+
+    @classmethod
+    async def get_streak(cls, habit_id: int) -> int:
+        async with new_session() as session:
+            result = await session.execute(
+                select(CompletionOrm.date).where(
+                    CompletionOrm.habit_id == habit_id
+                ).order_by(CompletionOrm.date.desc())
+            )
+            dates = [row for row, in result.all()]
+
+            if not dates:
+                return 0
+
+            streak = 0
+            today = datetime.now(timezone.utc).date()
+
+            for i, day in enumerate(dates):
+                if day.date() == today - timedelta(days=streak):
+                    streak += 1
+                else:
+                    break
+
+            return streak

students_folder/Nikulina/lab7/app/router_habits.py

@@ -0,0 +1,35 @@
+from fastapi import APIRouter, Depends
+from typing import List
+
+from app.operations.database_operations import HabitWorkflow, CompletionWorkflow
+from app.models.store_models import (
+    HabitAdd, HabitGet, CompletionAdd, CompletionGet
+)
+
+router = APIRouter(
+    prefix="/habits",
+    tags=["Habits"],
+)
+
+@router.post("/add", summary="Создание привычки")
+async def add_habit(habit: HabitAdd = Depends(HabitAdd)) -> dict:
+    new_id = await HabitWorkflow.add_habit(habit)
+    return {"id": new_id}
+
+@router.get("/all", summary="Список привычек")
+async def get_habits() -> List[HabitGet]:
+    return await HabitWorkflow.get_habits()
+
+@router.post("/complete", summary="Отметка выполнения привычки")
+async def add_completion(c: CompletionAdd = Depends(CompletionAdd)) -> dict:
+    new_id = await CompletionWorkflow.add_completion(c)
+    return {"id": new_id}
+
+@router.get("/{habit_id}/completions", summary="Получение всех выполнений привычки")
+async def get_completions(habit_id: int) -> List[CompletionGet]:
+    return await CompletionWorkflow.get_completions(habit_id)
+
+@router.get("/{habit_id}/streak", summary="Серия подряд выполненных дней")
+async def get_streak(habit_id: int) -> dict:
+    streak = await CompletionWorkflow.get_streak(habit_id)
+    return {"streak": streak}

students_folder/Nikulina/lab7/docker-compose.yaml

@@ -0,0 +1,35 @@
+version: '3.9'
+
+services:
+  postgres:
+    image: postgres:13
+    container_name: postgres_container
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: habit_db
+      PGDATA: /var/lib/postgresql/data/pgdata
+    ports:
+      - "5433:5432"
+    volumes:
+      - ./pgdata:/var/lib/postgresql/data/pgdata
+    command: >
+      postgres -c max_connections=1000
+               -c shared_buffers=256MB
+               -c effective_cache_size=768MB
+               -c maintenance_work_mem=64MB
+               -c checkpoint_completion_target=0.7
+               -c wal_buffers=16MB
+               -c default_statistics_target=100
+    healthcheck:
+      test: [ "CMD-SHELL", "pg_isready -U postgres_user -d postgres_db" ]
+      interval: 30s
+      timeout: 10s
+      retries: 5
+    restart: unless-stopped
+    tty: true
+    stdin_open: true
+
+volumes:
+  pgdata:
+    driver: local

students_folder/Nikulina/lab7/requirements.txt

@@ -0,0 +1,9 @@
+fastapi==0.110.2
+pydantic==2.7.0
+SQLAlchemy==2.0.30
+uvicorn==0.29.0
+asyncpg==0.29.0
+pytest
+pytest-asyncio
+httpx
+pyyaml

students_folder/Nikulina/lab7/tests/expected_responses.yaml

@@ -0,0 +1,19 @@
+add_habit_response:
+  id: 1
+
+get_habits_response:
+  - id: 1
+    user_id: 1
+    name: "Drink water"
+    description: "2 liters per day"
+
+add_completion_response:
+  id: 1
+
+get_completions_response:
+  - id: 1
+    habit_id: 1
+    date: ANY
+
+streak_response:
+  streak: 1

students_folder/Nikulina/lab7/tests/test_api.py

@@ -0,0 +1,71 @@
+import pytest
+import yaml
+import httpx
+from datetime import datetime, UTC
+
+BASE_URL = "http://127.0.0.1:8000"
+
+with open("tests/expected_responses.yaml", "r", encoding="utf-8") as f:
+    expected = yaml.safe_load(f)
+
+@pytest.mark.asyncio
+async def test_habits_api():
+
+    async with httpx.AsyncClient() as client:
+
+        add_payload = {
+            "user_id": 1,
+            "name": "Drink water",
+            "description": "2 liters per day"
+        }
+
+        response = await client.post(f"{BASE_URL}/habits/add", params=add_payload)
+        assert response.status_code == 200
+        assert "id" in response.json()
+
+        expected["add_habit_response"]["id"] = response.json()["id"]
+
+        response = await client.get(f"{BASE_URL}/habits/all")
+        assert response.status_code == 200
+
+        habits = response.json()
+        assert len(habits) > 0
+
+        expected["get_habits_response"][0]["id"] = habits[0]["id"]
+
+        assert habits[0]["user_id"] == expected["get_habits_response"][0]["user_id"]
+        assert habits[0]["name"] == expected["get_habits_response"][0]["name"]
+        assert habits[0]["description"] == expected["get_habits_response"][0]["description"]
+
+        habit_id = habits[0]["id"]
+
+        completion_payload = {
+            "habit_id": habit_id,
+            "date": datetime.now(UTC).isoformat().replace("+00:00", "Z")
+        }
+
+        response = await client.post(
+            f"{BASE_URL}/habits/complete",
+            params=completion_payload
+        )
+
+        assert response.status_code == 200
+
+        completion_id = response.json()["id"]
+        expected["add_completion_response"]["id"] = completion_id
+
+        response = await client.get(f"{BASE_URL}/habits/{habit_id}/completions")
+        assert response.status_code == 200
+
+        completions = response.json()
+        assert len(completions) > 0
+
+        assert completions[0]["habit_id"] == habit_id
+        assert "date" in completions[0]
+
+        response = await client.get(f"{BASE_URL}/habits/{habit_id}/streak")
+        assert response.status_code == 200
+
+        streak = response.json()
+        assert "streak" in streak
+        assert isinstance(streak["streak"], int)