Simple Backend homework

simple_backend/src/task_tracker/.gitignore

@@ -0,0 +1,4 @@
+*.idea/
+*.json
+*.env
+*.test_cloudflare_ai.py

simple_backend/src/task_tracker/cloud_storage.py

@@ -0,0 +1,31 @@
+import os
+from cloudflare_ai import BaseHTTPClient
+from typing import List
+
+
+class JSONBinStorage(BaseHTTPClient):
+    def __init__(self, bin_id: str, api_key: str):
+        self.bin_id = bin_id
+        self.api_key = api_key
+        self.validate_credentials()  # ← Вызов абстрактного метода
+        base_url = f'https://api.jsonbin.io/v3/b/{self.bin_id}'
+        headers = {
+            'X-Master-Key': self.api_key,
+            'Content-Type': 'application/json'
+        }
+        super().__init__(base_url, headers)
+
+    def validate_credentials(self):
+        if not self.bin_id or not self.api_key:
+            raise ValueError("JSON_BIN_ID или JSON_API_KEY не найдены")
+
+    def get_service_name(self) -> str:
+        return "JSONBin Storage"
+
+    def load_data(self) -> List[dict]:
+        result = self._make_request("GET")
+        return result.get('record', [])
+
+    def save_data(self, data: List[dict]) -> bool:
+        self._make_request("PUT", json=data)
+        return True

simple_backend/src/task_tracker/cloudflare_ai.py

@@ -0,0 +1,81 @@
+import requests
+import os
+from dotenv import load_dotenv
+from pydantic import BaseModel
+from abc import ABC, abstractmethod
+from typing import Dict, Any
+
+load_dotenv('passwords.env')
+
+
+class BaseHTTPClient(ABC):
+    def __init__(self, base_url: str, headers: Dict[str, str], timeout: int = 30):
+        self.base_url = base_url
+        self.headers = headers
+        self.timeout = timeout
+
+    def _make_request(self, method: str, endpoint: str = "", **kwargs) -> Dict[str, Any]:
+        """Общий метод для выполнения HTTP запросов"""
+        url = f"{self.base_url}{endpoint}"
+        try:
+            response = requests.request(
+                method=method,
+                url=url,
+                headers=self.headers,
+                timeout=self.timeout,
+                **kwargs
+            )
+            if response.status_code == 200:
+                return response.json()
+            else:
+                raise Exception(f"HTTP Error {response.status_code}: {response.text}")
+        except requests.RequestException as e:
+            raise Exception(f"Network error: {e}")
+
+    @abstractmethod
+    def validate_credentials(self):
+        pass
+
+    @abstractmethod
+    def get_service_name(self) -> str:
+        pass
+
+class AIRequest(BaseModel):
+    messages: list
+
+class AIResponse(BaseModel):
+    response: str
+
+class CloudflareAI(BaseHTTPClient):
+    def __init__(self):
+        self.api_token = os.getenv("CF_API_TOKEN")
+        self.account_id = os.getenv("CF_ACCOUNT_ID")
+        self.validate_credentials()
+        base_url = f"https://api.cloudflare.com/client/v4/accounts/{self.account_id}/ai/run/"
+        headers = {
+            "Authorization": f"Bearer {self.api_token}",
+            "Content-Type": "application/json"
+        }
+        super().__init__(base_url, headers)
+
+    def validate_credentials(self):
+        if not self.api_token or not self.account_id:
+            raise ValueError("CF_API_TOKEN или CF_ACCOUNT_ID не найдены в .env")
+
+    def get_service_name(self) -> str:
+        return "Cloudflare AI"
+
+    def generate_solution(self, task_text: str) -> str:
+        payload = AIRequest(
+            messages=[
+                {"role": "system",
+                 "content": "Ты — опытный проектный менеджер. Предложи 3–5 конкретных шагов для решения задачи. Формат: нумерованный список."
+                 },
+                {"role": "user",
+                 "content": task_text
+                 }
+            ]
+        )
+        result = self._make_request("POST", "@cf/meta/llama-3-8b-instruct", json=payload.model_dump())
+        ai_response = AIResponse(**result["result"])
+        return ai_response.response

simple_backend/src/task_tracker/main.py

@@ -1,19 +1,44 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, HTTPException
+from task_manager import TaskManager
+from pydantic import BaseModel
+from typing import Optional
+
 
 app = FastAPI()
+task_manager = TaskManager()
+
+class TaskModel(BaseModel):
+    title: str
+    description: Optional[str] = None
+    status: bool = False
 
-@app.get("/tasks")
+@app.get('/tasks')
 def get_tasks():
-    pass
+    return task_manager.get_all()
 
-@app.post("/tasks")
-def create_task(task):
-    pass
+@app.post('/tasks')
+def create_task(task: TaskModel):
+    try:
+        task_data = task.model_dump()
+        return task_manager.create(
+            title=task_data.get('title', ''),
+            description=task_data.get('description', ''),
+            status=task_data.get('status', False)
+            )
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
 
 @app.put("/tasks/{task_id}")
-def update_task(task_id: int):
-    pass
+def update_task(task_id: int, task_update: TaskModel):
+    update_data = task_update.model_dump(exclude_unset=True)
+    update = task_manager.update(task_id, **update_data)
+    if not update:
+        raise HTTPException(status_code=404, detail='Task not found')
+    return update
 
-@app.delete("/tasks/{task_id}")
+@app.delete('/tasks/{task_id}')
 def delete_task(task_id: int):
-    pass
+    result = task_manager.delete(task_id)
+    if not result:
+        raise HTTPException(status_code=404, detail='Task not found')
+    return {'message': 'Task deleted'}

simple_backend/src/task_tracker/readme.md

@@ -0,0 +1,28 @@
+    Задание №2
+1. Проблема хранения данных в оперативной памяти:
+- данные теряются при перезапуске приложения;
+- невозможно масштабировать приложение;
+- утечки памяти и ограничение объемом оперативной памяти.
+
+2. Улучшения после того, как список из оперативной памяти изменился на файл проекта:
+- задачи сохраняются между перезапусками приложения;
+- данные хранятся на диске;
+- можно сделать копию JSON файла;
+- данные можно просмотреть и редактировать вручную.
+Избавились ли мы таким способом от хранения состояния или нет?
+- нет, состояние просто перенесено из оперативной памяти на диск.
+Где еще можно хранить задачи и какие есть преимущества и недостатки этих подходов?
+- реляционные базы данных;
+- NoSQL базы данных;
+- In-memory базы данных. 
+Состояние гонки - это когда несколько потоков или процессов обращаются к одним и тем же данным, и происходит изменение одним из них.
+Проблемы в бекенде на данном этапе:
+- несколько запросов на обновление одной задачи могут перезаписать друг друга;
+- невозможно проверить, не изменилась ли задача с момента её загрузки клиентом;
+- если один метод упадёт после изменения данных, система останется в несогласованном состоянии;
+- клиент не получает информации, если его изменения были перезаписаны другим запросом.
+Решения:
+- реализовать очередь задач (асинхронный подход), то есть все изменения ставить в очередь, обрабатывать последовательно;
+- источники событий, вместо хранения текущего состояния сохраняются все изменения (события). Текущее состояние вычисляется как результат применения цепочки событий;
+- эмуляция транзакций через многошаговые HTTP‑запросы;
+- сериализация операций через единую очередь (например, RabbitMQ), то есть обрабатываются строго последовательно.

simple_backend/src/task_tracker/task_manager.py

@@ -0,0 +1,82 @@
+import os
+from cloud_storage import JSONBinStorage
+from cloudflare_ai import CloudflareAI
+from dotenv import load_dotenv
+from pydantic import BaseModel
+from typing import Optional, List
+
+load_dotenv('passwords.env')
+
+class TaskModel(BaseModel):
+    title: str
+    description: Optional[str] = None
+    status: bool = False
+
+class TaskResponse(TaskModel):
+    id: int
+    ai_solution: Optional[str] = None
+
+class TaskManager:
+    def __init__(self):
+        self.bin_id = os.getenv('JSON_BIN_ID')
+        self.api_key = os.getenv('JSON_API_KEY')
+        if not self.bin_id or not self.api_key:
+            raise ValueError('Не найдены JSON_BIN_ID или JSON_API_KEY')
+        self.storage = JSONBinStorage(self.bin_id, self.api_key)
+        self.ai = CloudflareAI()
+
+    def _tasks(self) -> List[dict]:
+        return self.storage.load_data()
+
+    def get_all(self) -> List[TaskResponse]:
+        tasks = self._tasks()
+        return [TaskResponse(**task) for task in tasks]
+
+    def _save(self, tasks):
+        self.storage.save_data(tasks)
+
+    def create(self, title: str, description: str = '', status: bool = False) -> TaskResponse:
+        task_data = TaskModel(title=title, description=description, status=status)
+        tasks = self._tasks()
+        ids = [task['id'] for task in tasks] if tasks else []
+        task_id = max(ids) + 1 if ids else 1
+        ai_solution = ''
+        if task_data.description:
+            try:
+                ai_solution = self.ai.generate_solution(f'{task_data.title}. {task_data.description}')
+            except Exception as e:
+                ai_solution = f'{e}. Не удалось сгенерировать решение'
+
+        new_task = {'id': task_id,
+                    'title': task_data.title,
+                    'description': task_data.description,
+                    'status': task_data.status,
+                    'ai_solution': ai_solution,
+                    }
+        tasks.append(new_task)
+        self._save(tasks)
+        return TaskResponse(**new_task)
+
+    def update(self, task_id: int, **updates) -> Optional[TaskResponse]:
+        update_data = TaskModel(**updates)
+        valid_updates = update_data.model_dump(exclude_unset=True)
+        if not valid_updates:
+            return None
+        tasks = self._tasks()
+        for task in tasks:
+            if task['id'] == task_id:
+                for k, v in valid_updates.items():
+                    if v is not None:
+                        task[k] = v
+                self._save(tasks)
+                return TaskResponse(**task)
+        return None
+
+    def delete(self, task_id: int) -> bool:
+        tasks = self._tasks()
+        initial_count = len(tasks)
+        tasks = [task for task in tasks if task.get('id') != task_id]
+        if len(tasks) == initial_count:
+            return False
+        self._save(tasks)
+        return True