git + простые бэкэнды

.github/workflows/ci.yml

@@ -0,0 +1,34 @@
+name: Python CI - Ruff
+
+on:
+  push:
+    branches: [ main, second-branch ]
+  pull_request:
+    branches: [ main ]
+
+jobs:
+  lint:
+    runs-on: ubuntu-22.04
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: 3.13
+
+      - name: Cache pip packages
+        uses: actions/cache@v3
+        with:
+          path: ~/.cache/pip
+          key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
+          restore-keys: |
+            ${{ runner.os }}-pip-
+
+      - name: Install Ruff
+        run: pip install ruff==0.13.3
+
+      - name: Run Ruff Check
+        run: ruff check . --line-length 88 --select E,F,W --show-files

simple_backend/src/task_tracker/.gitignore

@@ -0,0 +1,2 @@
+.env
+tasks.json

simple_backend/src/task_tracker/base_http_client.py

@@ -0,0 +1,27 @@
+from abc import ABC
+from typing import Any
+import requests
+
+class BaseHTTPClient(ABC):
+    def __init__(self, token: str, url: str):
+        self.token = token
+        self.url = url
+        self.headers = {
+            "Authorization": f"Bearer {self.token}",
+            "Content-Type": "application/json"
+        }
+
+    def get(self) -> dict[str, Any]:
+        response = requests.get(self.url, headers=self.headers)
+        response.raise_for_status()
+        return response.json()
+
+    def post(self, payload: dict[str, Any]) -> dict[str, Any]:
+        response = requests.post(self.url, headers=self.headers, json=payload)
+        response.raise_for_status()
+        return response.json()
+
+    def patch(self, payload: dict[str, Any]) -> dict[str, Any]:
+        response = requests.patch(self.url, headers=self.headers, json=payload)
+        response.raise_for_status()
+        return response.json()

simple_backend/src/task_tracker/cloudflare_llm.py

@@ -0,0 +1,17 @@
+from .base_http_client import BaseHTTPClient
+from .config import settings
+
+class CloudflareLLM(BaseHTTPClient):
+    def __init__(self):
+        url = f"https://api.cloudflare.com/client/v4/accounts/{settings.CLOUDFLARE_ACCOUNT_ID}/ai/run/{settings.CLOUDFLARE_MODEL}"
+        super().__init__(token=settings.CLOUDFLARE_AUTH_TOKEN, url=url)
+
+    def get_solution(self, task_text: str) -> str:
+        payload = {
+            "messages": [
+                {"role": "system", "content": "Ты помощник, объясни, как решить задачу"},
+                {"role": "user", "content": task_text}
+            ]
+        }
+        response = self.post(payload)
+        return response["result"]["response"]

simple_backend/src/task_tracker/config.py

@@ -0,0 +1,18 @@
+from pydantic_settings import BaseSettings
+from dotenv import load_dotenv
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent
+
+class Settings(BaseSettings):
+    CLOUDFLARE_AUTH_TOKEN: str
+    CLOUDFLARE_ACCOUNT_ID: str
+    CLOUDFLARE_MODEL: str
+    GITHUB_TOKEN: str
+    GIST_ID: str
+
+    class Config:
+        env_file = BASE_DIR / ".env"
+        env_file_encoding = "utf-8"
+
+settings = Settings()

simple_backend/src/task_tracker/gist_storage.py

@@ -0,0 +1,29 @@
+import json
+from typing import Any
+from .base_http_client import BaseHTTPClient
+from .config import settings
+from fastapi import HTTPException
+
+class GistStorage(BaseHTTPClient):
+    def __init__(self):
+        url = f"https://api.github.com/gists/{settings.GIST_ID}"
+        super().__init__(token=settings.GITHUB_TOKEN, url=url)
+
+    def load_data(self):
+        gist_data = self.get()
+        content = gist_data["files"]["tasks.json"]["content"]
+        return json.loads(content)
+
+    def save_data(self, tasks):
+        updated_content = json.dumps(tasks, ensure_ascii=False, indent=2)
+        payload = {"files": {"tasks.json": {"content": updated_content}}}
+        return self.patch(payload)
+
+    def delete_task_by_id(self,task_id: int) -> None:
+        tasks = self.load_data()
+        for i, task in enumerate(tasks):
+            if task["id"] == task_id:
+                tasks.pop(i)
+                self.save_data(tasks)
+                return
+        raise HTTPException(status_code=404, detail="Task not found")

simple_backend/src/task_tracker/main.py

@@ -1,19 +1,77 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+from typing import Optional
+from task_tracker.gist_storage import GistStorage
+from task_tracker.cloudflare_llm import CloudflareLLM
 
 app = FastAPI()
 
-@app.get("/tasks")
+storage = GistStorage()
+llm = CloudflareLLM()
+
+class TaskCreate(BaseModel):
+    title: str
+    status: Optional[str] = None
+
+class TaskUpdate(BaseModel):
+    title: Optional[str] = None
+    status: Optional[str] = None
+
+class Task(BaseModel):
+    id: int
+    title: str
+    status: str
+    solution: str
+
+
+@app.exception_handler(Exception)
+def global_exception_handler(request: Request, exc: Exception):
+    return JSONResponse(
+        status_code=500,
+        content={"detail": str(exc)}
+    )
+
+
+@app.get("/tasks", response_model=list[Task])
 def get_tasks():
-    pass
+    return storage.load_data()
+
+
+@app.post("/tasks", response_model=Task)
+def create_task(task: TaskCreate):
+    tasks = storage.load_data()
+    new_id = max([t["id"] for t in tasks], default=0) + 1
+    solution = llm.get_solution(task.title)
+    new_task = Task(
+        id=new_id,
+        title=task.title,
+        status=task.status or "in work",
+        solution=solution
+    )
+    tasks.append(new_task.dict())
+    storage.save_data(tasks)
+    return new_task
+
 
-@app.post("/tasks")
-def create_task(task):
-    pass
+@app.put("/tasks/{task_id}", response_model=Task)
+def update_task(task_id: int, task_update: TaskUpdate):
+    tasks = storage.load_data()
+    for i, task_dict in enumerate(tasks):
+        if task_dict["id"] == task_id:
+            updated_task = Task(
+                id=task_id,
+                title=task_update.title or task_dict["title"],
+                status=task_update.status or task_dict["status"],
+                solution=task_dict["solution"]
+            )
+            tasks[i] = updated_task.dict()
+            storage.save_data(tasks)
+            return updated_task
+    raise HTTPException(status_code=404, detail="Task not found")
 
-@app.put("/tasks/{task_id}")
-def update_task(task_id: int):
-    pass
 
-@app.delete("/tasks/{task_id}")
+@app.delete("/tasks/{task_id}", status_code=200)
 def delete_task(task_id: int):
-    pass
+    storage.delete_task_by_id(task_id)
+    return

simple_backend/src/task_tracker/readme.md

@@ -0,0 +1,23 @@
+1) Минусы подхода с хранением задач в оперативной памяти(списке python):
+	- при каждом включении(перезагрузке) сервера - список пуст, т.е. исчезают уже занесённые в него задачи.
+	- каждый сервер(даже с одинаковым названием) - это отдельный список со своим состоянием.
+2) Что улучшилось после того, как список из оперативной памяти изменился на файл проекта?
+	- теперь при перезагрузке сервера задачи не исчезают, а хранятся и подгружаются, т.е. становятся
+	  постоянными.
+	- файл можно отредактировать даже без поднятия сервера. удобство работы с данными. масштабируемость?
+3) Избавились ли мы таким способом от хранения состояния или нет?
+	- нет, не избавились, но приложение стало stateless. мы всё также продолжаем читать/изменять/сохранять данные, которые меняют это
+	  состояние только уже не в оперативной памяти, а на диске в файле.
+4) Где еще можно хранить задачи и какие есть преимущества и недостатки этих подходов?
+	- Базы данных. Минусы: нужно развернуть базу, настроить и уметь работать/обслуживать её.
+		           Плюсы: структурированность, одновременно могут работать много пользователей.
+	- Облачные сервисы. Минусы: зависимость от данного ресурса, предоставляющего сервис.
+				    медленнее, чем локальное подключение.
+			    Плюсы: данные не хранятся локально, удалённый доступ, не нужно самостоятельно
+				   разворачивать свою базу данных.
+5) Какие проблемы остались на этапе "backend-stateless" и какие есть решения?
+	Проблемы: - если два пользователя одновременно пытаются делать операции, данные могут потеряться.
+		  - не реализована атомарная запись.
+	Решения: - сделать так, чтобы одновременно выполнялась только 1 запись.
+		 - использовать базы данных, где обрабатываются одновременные операции.
+

simple_backend/src/task_tracker/requirements.txt

@@ -1,2 +1,2 @@
-fastapi
-uvicorn[standard]
+fastapi==0.118.0
+uvicorn[standard]==0.37.0

simple_backend/src/task_tracker/storage_client.py

@@ -0,0 +1,14 @@
+# не используется, т.к. локальное хранилище не используется, а используется
+# gist_storage.py 
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+class StorageClient(ABC):
+    @abstractmethod
+    def load_data(self) -> Any:
+	pass
+
+    @abstractmethod
+    def save_data(self, data: Any):
+	pass

simple_backend/src/task_tracker/task_storage.py

@@ -0,0 +1,20 @@
+# не используется. локальное хранение задач заменено на gist_storage.py
+
+from .storage_client import StorageClient
+from pathlib import Path
+from typing import Any
+import json
+
+class TaskStorage(StorageClient):
+    def __init__(self, file_path: str):
+	self.file_path = Path(file_path)
+	if not self.file_path.exists():
+	    self.save_data([])
+
+    def load_data(self) -> list[dict[str, Any]]:
+        with self.file_path.open("r", encoding="utf-8") as f:
+            return json.load(f)
+
+    def save_data(self, data: Any):
+        with self.file_path.open("w", encoding="utf-8") as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)

simple_backend/src/task_tracker/tasks.json

@@ -0,0 +1,12 @@
+[
+  {
+    "id": 1,
+    "title": "а",
+    "status": "in work"
+  },
+  {
+    "id": 2,
+    "title": "б",
+    "status": "in work"
+  }
+]