Задача 2

simple_backend/src/task_tracker/README.md

@@ -0,0 +1,39 @@
+## Минусы хранить данные в ОЗУ
+1. При перезапуске всё теряется
+2. Отсутствие масштабируемости
+3. Сложность интеграции с внешними сервисами
+4. Ограничение объёма данных
+
+## что улучшилось после того, как список из оперативной памяти изменился на файл проекта?
+* файл не теряется при перезапуске
+* легкая интеграция 
+* можно создать резервные копии
+
+## избавились ли мы таким способом от хранения состояния или нет?
+* Нет, мы не избавились от хранения состояния, мы просто перенесли его из оперативной памяти на файл.
+
+## где еще можно хранить задачи и какие есть преимущества и недостатки этих подходов?
+### реляционные бд
+**Плюсы:**
+- Высокая масштабируемость и доступность.
+- Простая интеграция с фронтендом и API.
+- Автоматическое резервное копирование в облаке.
+- Можно работать с несколькими клиентами одновременно.
+
+**Минусы:**
+- Может быть дороже (облачные сервисы).
+- Требует сетевого подключения.
+- Сложнее отлаживать локально.
+### nosql
+**Плюсы:**
+- Высокая масштабируемость и доступность.
+- Простая интеграция с фронтендом и API.
+- Автоматическое резервное копирование в облаке.
+- Можно работать с несколькими клиентами одновременно.
+
+**Минусы:**
+- Может быть дороже (облачные сервисы).
+- Требует сетевого подключения.
+- Сложнее отлаживать локально.
+
+### Проблема гонки пока есть, её надо решать, если приложение будет использоваться несколькими клиентами одновременно.

simple_backend/src/task_tracker/basehttp.py

@@ -0,0 +1,26 @@
+import requests
+
+class BaseHTTPClient:
+    def __init__(self, base_url: str, headers: dict | None = None):
+        self.base_url = base_url.rstrip("/")
+        self.headers = headers or {}
+
+    def get(self, endpoint: str = "", **kwargs):
+        url = f"{self.base_url}/{endpoint.lstrip('/')}"
+        res = requests.get(url, headers=self.headers, **kwargs)
+        res.raise_for_status()
+        return res.json()
+
+    def post(self, endpoint: str = "", json: dict | None = None, **kwargs):
+        url = f"{self.base_url}/{endpoint.lstrip('/')}"
+        res = requests.post(url, headers=self.headers, json=json, **kwargs)
+        res.raise_for_status()
+        return res.json()
+
+    def put(self, endpoint: str = "", json: dict | None = None, **kwargs):
+        url = f"{self.base_url}/{endpoint.lstrip('/')}"
+        res = requests.put(url, headers=self.headers, json=json, **kwargs)
+        res.raise_for_status()
+        return res.json()
+
+

simple_backend/src/task_tracker/clients.py

@@ -0,0 +1,31 @@
+from basehttp import BaseHTTPClient
+
+class CloudFlareClient(BaseHTTPClient):
+    inputs = [
+        {"role": "system", "content":
+            "Отвечай по-русски, кратко (1–2 предложения),"
+            "как лучше выполнить задачу из списка дел. Без приветствий и воды."
+            "Отвечай естественно, без вводных фраз вроде «Для выполнения задачи…», «Чтобы сделать…», «Следует…»."
+            "Сразу переходи к сути,"
+            "Если в задаче есть действие, пиши как"
+            " будто даёшь понятную инструкцию или совет, а не академическое объяснение"}
+    ]
+
+    def __init__(self, api_token: str, account_id: str):
+        super().__init__(
+            base_url=f"https://api.cloudflare.com/client/v4/accounts/{account_id}/ai/run",
+            headers={
+                "Authorization": f"Bearer {api_token}",
+                "Content-Type": "application/json"
+            }
+        )
+
+    def generate_answer(self, task: str) -> str:
+        data = {"messages": self.inputs + [{"role": "user", "content": task}]}
+        result = self.post("@cf/meta/llama-3-8b-instruct", json=data)
+        response_text = result["result"]["response"]
+
+        if not response_text:
+            raise ValueError("Пустой или некорректный ответ от CloudFlare AI")
+
+        return response_text

simple_backend/src/task_tracker/config.py

@@ -0,0 +1,16 @@
+from pydantic_settings import BaseSettings
+from starlette.config import Config
+
+config = Config(".env")
+
+class Settings(BaseSettings):
+    BIN_ID: str = config("BIN_ID", cast=str)
+    MASTER_KEY: str = config("MASTER_KEY", cast=str)
+    API_TOKEN_AI: str = config("API_TOKEN_AI", cast=str)
+    ACCOUNT_ID_AI: str = config("ACCOUNT_ID_AI", cast=str)
+
+    class Config:
+        env_file = ".env"
+        env_file_encoding = "utf-8"
+
+settings = Settings()

simple_backend/src/task_tracker/main.py

@@ -1,19 +1,50 @@
 from fastapi import FastAPI
+from fastapi.responses import JSONResponse
+from pydantic import ValidationError
+from storage import JSONStorage, CloudJSONStorage
+from models import Task
+from config import settings
+from clients import CloudFlareClient
+from typing import List
 
 app = FastAPI()
+storage = CloudJSONStorage(bin_id=settings.BIN_ID, master_key=settings.MASTER_KEY)
+client_ai = CloudFlareClient(settings.API_TOKEN_AI, settings.ACCOUNT_ID_AI)
 
-@app.get("/tasks")
+
+@app.exception_handler(ValueError)
+async def value_error_handler(request, exc: ValueError):
+    return JSONResponse(status_code=404, content={"detail": str(exc)})
+
+@app.exception_handler(ValidationError)
+async def validation_error_handler(request, exc: ValidationError):
+    return JSONResponse(status_code=422, content={"detail": exc.errors()})
+
+@app.exception_handler(Exception)
+async def global_exception_handler(request, exc: Exception):
+    return JSONResponse(status_code=500, content={"detail": "Internal Server Error"})
+
+
+
+@app.get("/tasks", response_model=List[Task])
 def get_tasks():
-    pass
+    return storage.list_tasks()
 
-@app.post("/tasks")
-def create_task(task):
-    pass
+@app.post("/tasks", response_model=Task)
+def create_task(task_data: Task):
+    ai_reply = client_ai.generate_answer(task_data.title)
+    task_data.title = f"{task_data.title} — {ai_reply}"
+    task = storage.create_task(task_data)
+    return task
 
-@app.put("/tasks/{task_id}")
-def update_task(task_id: int):
-    pass
+@app.put("/tasks/{task_id}", response_model=Task)
+def update_task(task_id: int, task_data: Task):
+    ai_reply = client_ai.generate_answer(task_data.title)
+    task_data.title = f"{task_data.title} — {ai_reply}"
+    task = storage.update_task(task_id, task_data)
+    return task
 
-@app.delete("/tasks/{task_id}")
+@app.delete("/tasks/{task_id}", response_model=Task)
 def delete_task(task_id: int):
-    pass
+    task = storage.delete_task(task_id)
+    return task

simple_backend/src/task_tracker/models.py

@@ -0,0 +1,8 @@
+from pydantic import BaseModel, Field
+
+class Task(BaseModel):
+    title: str = Field(..., min_length=5, max_length=300, description="Название задачи")
+
+    status: bool = False
+    id: int
+

simple_backend/src/task_tracker/storage.py

@@ -0,0 +1,112 @@
+import json
+from pathlib import Path
+from models import Task
+from basehttp import BaseHTTPClient
+
+class IsMemoryStorage:
+
+    def __init__(self):
+        self.tasks = []
+        self.next_task = 1
+
+
+    def list_tasks(self) -> list:
+        return self.tasks
+
+    def create_task(self, task_data: dict) -> Task:
+        task = Task(id=self.next_task, **task_data)
+        self.tasks.append(task)
+        self.next_task += 1
+
+        return task
+
+    def get_task(self, task_id)->Task:
+        for task in self.tasks:
+            if task['id'] == task_id:
+                return task
+        return None
+
+    def update_task(self, task_id:int, task_data):
+        for task in self.tasks:
+            if task['id'] == task_id:
+                task.update(task_data)
+                return task
+        return None
+
+
+    def delete_task(self, task_id):
+        task = self.get_task(task_id)
+        if not task: return False
+        self.tasks.remove(task)
+        return True
+
+
+class JSONStorage:
+    def __init__(self, filepath: str | None = "tasks.json"):
+        self.file = Path(filepath)
+        if not self.file.exists():
+            self.file.write_text("[]", encoding="utf-8")
+
+    def _load(self) -> list[Task]:
+        data = json.loads(self.file.read_text(encoding="utf-8"))
+        return [Task(**t) for t in data]
+
+    def _save(self, tasks: list[Task]):
+        data = [t.model_dump() for t in tasks]
+        self.file.write_text(json.dumps(data, indent=4, ensure_ascii=False), encoding="utf-8")
+
+    def list_tasks(self) -> list[Task]:
+        return self._load()
+
+    def get_task(self, task_id: int) -> Task:
+        for task in self._load():
+            if task.id == task_id:
+                return task
+        raise ValueError(f"Task with id={task_id} not found")
+
+
+    def create_task(self, task_data: Task) -> Task:
+        tasks = self._load()
+        new_id = max([t.id for t in tasks], default=0) + 1
+        task = Task(id=new_id, **task_data.model_dump(exclude={'id'}))
+        tasks.append(task)
+        self._save(tasks)
+        return task
+
+    def update_task(self, task_id: int, task_data: Task) -> Task:
+        tasks = self._load()
+        task = next((t for t in tasks if t.id == task_id), None)
+        if not task:
+            raise ValueError(f"Task with id={task_id} not found")
+        updated_task = task.model_copy(update=task_data.model_dump(exclude={'id'}))
+        tasks[tasks.index(task)] = updated_task
+        self._save(tasks)
+        return updated_task
+
+    def delete_task(self, task_id: int) -> Task:
+        tasks = self._load()
+        task = next((t for t in tasks if t.id == task_id), None)
+        if not task:
+            raise ValueError(f"Task with id={task_id} not found")
+        tasks.remove(task)
+        self._save(tasks)
+        return task
+
+class CloudJSONStorage(BaseHTTPClient, JSONStorage):
+    def __init__(self, bin_id: str, master_key: str):
+        super().__init__(
+            base_url=f"https://api.jsonbin.io/v3/b/{bin_id}",
+            headers={
+                "X-Master-Key": master_key,
+                "Content-Type": "application/json",
+            }
+        )
+
+    def _load(self) -> list[Task]:
+        record = self.get("latest").get("record", {})
+        tasks_data = record.get("tasks", [])
+        return [Task(**task) for task in tasks_data]
+
+    def _save(self, tasks: list[Task]) -> dict:
+        payload = {"tasks": [task.model_dump() for task in tasks]}
+        return self.put("", json=payload)