Стварэнне СААС-сервіса для чату з AI на базе Next.js, FastAPI, Credits і SSE

Создаць інтэрфейс для чату на адміністрацыі API модэлі — гэта проста задача; аднак продукт таксама павінен захоўваць ключы занятымі на боку сервера, контролюваць выкарыстоўвання ресурсаў, запобегаць зласовам, чыста аддаўаць токены для стрімування дадзеных і забезпечваць можлівасць повторнага стварэння продукту. У цым кераванні гэтыя элементы ствараюцца за дапамою Next.js на базе React 19, FastAPI, PostgreSQL з падтрымкай pgvector і Redis. Спрацавляйце з кодам як з наброском: кожны раздзіл пазначае, што трэба спачатку забезпечыць надзею.

Бэкенд — гэта контролер

Браузер ніколі не выклекае прадастальніка модэлі. Ваш API аутентыфікуе корыстувальніка, пераканаецца ў наяўнасці лімітов і кредытаў, выклекае модэль і фіксуе вартасць ўжытку.

┌─────────────────────────────────────────────────────────────┐
│                        React 19 App                          │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │   Server     │  │   Client     │  │   Chat Interface │  │
│  │  Components  │  │  Components  │  │  (Streaming)     │  │
│  │  (Initial    │  │  (Interactivity│  │                  │  │
│  │   data fetch)│  │   + State)   │  │                  │  │
│  └──────┬───────┘  └──────┬───────┘  └────────┬─────────┘  │
│         │                 │                    │              │
│         └─────────────────┴────────────────────┘              │
│                           │                                  │
│                    TanStack Query                            │
│                    (Auth + Sync)                             │
└───────────────────────────┬─────────────────────────────────┘
                            │ HTTPS / JSON
┌───────────────────────────┴─────────────────────────────────┐
│                      FastAPI Backend                         │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │   Auth       │  │   Rate       │  │   OpenAI         │  │
│  │  Middleware  │  │  Limiter     │  │  Service         │  │
│  │  (JWT +      │  │  (Redis)     │  │  (Structured     │  │
│  │   Credits)   │  │              │  │   Output)        │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │  Background  │  │  PostgreSQL  │  │  pgvector        │  │
│  │  Worker      │  │  (Users +    │  │  (Embeddings)    │  │
│  │  (Celery)    │  │  Conversations│  │                  │  │
│  └──────────────┘  └──────────────┘  └──────────────────┘  │
└─────────────────────────────────────────────────────────────┘

Компаненты сервера завантажаюць історыю без дадатковага JavaScript, асінхронны FastAPI обрабоцвае повольныя запыты да модэлі без блакання потокаў, pgvector залишае можлівасць для выкарыстоўвання дадзеных, Redis храніць калькуляторы лімітав і кэшаваныя эмбеддынгі, а працоўнік бере на сябе задачы парсавання PDF і стварэння эмбеддынгаў, адводзячы іх з маршруту запыту.

Модэль дадзеных: дапісванне да журналу

Шымат паведамляе працоўнікоў, размовы, паведамленні і рэчыстку кредытавання.

# backend/models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text, Numeric, Index
from sqlalchemy.dialects.postgresql import UUID, JSONB, VECTOR
from sqlalchemy.orm import relationship
from datetime import datetime
import uuid
Base = declarative_base()
class User(Base):
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    email = Column(String, unique=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    credits_balance = Column(Numeric(10, 2), default=0.00)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, default=datetime.utcnow)

    conversations = relationship("Conversation", back_populates="user")

    __table_args__ = (
        Index('ix_users_email', 'email'),
    )
class Conversation(Base):
    __tablename__ = "conversations"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
    title = Column(String, nullable=True)
    model = Column(String, default="gpt-4o")
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    user = relationship("User", back_populates="conversations")
    messages = relationship("Message", back_populates="conversation", order_by="Message.created_at")
class Message(Base):
    __tablename__ = "messages"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    conversation_id = Column(UUID(as_uuid=True), ForeignKey("conversations.id"), nullable=False)
    role = Column(String, nullable=False)  # 'user', 'assistant', 'system'
    content = Column(Text, nullable=False)
    token_count = Column(Integer, default=0)
    metadata = Column(JSONB, default={})  # Store model, latency, finish_reason
    created_at = Column(DateTime, default=datetime.utcnow)

    conversation = relationship("Conversation", back_populates="messages")
class CreditLedger(Base):
    __tablename__ = "credit_ledger"

    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    user_id = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=False)
    amount = Column(Numeric(10, 2), nullable=False)  # Negative for usage
    description = Column(String, nullable=False)
    reference_id = Column(String, nullable=True)  # conversation_id or message_id
    created_at = Column(DateTime, default=datetime.utcnow)

У зяместа змены залішку грошэй кожная пакупка чыста выкананая дзеянне прыўязываецца, і залішак становіць суму ўсіх такіх дзеянняў, што забезпечвае летапіс аудыту і не дазволяе пакрыць якія-небудзь апошнія змены. Перад выкарыстоўванням трэба паспрабаваць выправіць калькі прынтамоў: Message.metadata перакрываецца з атрыбутам, які зазначаны ў декларатывнай базе SQLAlchemy, таму яго трэба пераявіць (напрыклад, meta); не хватает такіх імпортаў, як Boolean і declarative_base; а таксама тип столбца pgvector выйшоў з пакета pgvector, а не з дыялекта PostgreSQL у SQLAlchemy.

Сэрвісы і маршруты FastAPI

Традыцыйныя структуры раздзеляюць аутэнтыкацыю, логіку AI, маршруты і проміжныя компоненты:

backend/
├── main.py
├── config.py
├── database.py
├── models.py
├── schemas.py
├── auth/
│   ├── utils.py
│   ├── dependencies.py
│   └── router.py
├── ai/
│   ├── service.py
│   └── schemas.py
├── api/
│   ├── conversations.py
│   └── credits.py
├── middleware/
│   └── rate_limit.py
└── worker.py

Сэрвіс стрімавання, які ведае колькасць выкарыстоўвання

Этыя службы дазваляюць 20 запыткаў на хвіліна на адного пользователя через Redis, пераканальваюць баланс, трансляюць Server-Sent Events, а пасля выконваюць запіс прыемленае звесці асистента і негатыўнага запису ў бухгалтарскім рэўюце разам.

# backend/ai/service.py
from openai import AsyncOpenAI
from typing import AsyncGenerator
import json
from sqlalchemy.ext.asyncio import AsyncSession
from models import Message, CreditLedger
from schemas import ChatCompletionRequest, ChatStreamChunk
import redis.asyncio as redis
client = AsyncOpenAI(api_key=settings.OPENAI_API_KEY)
redis_client = redis.from_url(settings.REDIS_URL)
class AIService:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def stream_chat(
        self,
        user_id: str,
        conversation_id: str,
        messages: list[dict],
        model: str = "gpt-4o"
    ) -> AsyncGenerator[str, None]:
        """
        Streams chat completion and deducts credits.
        Yields SSE-formatted strings.
        """
        # 1. Check rate limit
        key = f"ratelimit:chat:{user_id}"
        current = await redis_client.incr(key)
        if current == 1:
            await redis_client.expire(key, 60)
        if current > 20:  # 20 requests per minute
            yield f"data: {json.dumps({'error': 'Rate limit exceeded'})}\n\n"
            return

        # 2. Check credits (simplified; use ledger sum in production)
        balance = await self.get_credit_balance(user_id)
        if balance < 0.01:
            yield f"data: {json.dumps({'error': 'Insufficient credits'})}\n\n"
            return

        # 3. Stream from OpenAI
        try:
            stream = await client.chat.completions.create(
                model=model,
                messages=messages,
                stream=True,
                max_tokens=4096,
            )

            full_content = ""
            total_tokens = 0

            async for chunk in stream:
                delta = chunk.choices[0].delta.content or ""
                full_content += delta
                total_tokens += 1  # Approximate; use tiktoken for accuracy

                yield f"data: {json.dumps({'chunk': delta, 'done': False})}\n\n"

            # 4. Persist assistant message
            assistant_msg = Message(
                conversation_id=conversation_id,
                role="assistant",
                content=full_content,
                token_count=total_tokens,
                metadata={"model": model, "finish_reason": "stop"}
            )
            self.db.add(assistant_msg)

            # 5. Deduct credits (approximate $0.002 per 1K tokens for GPT-4o)
            cost = (total_tokens / 1000) * 0.002
            ledger_entry = CreditLedger(
                user_id=user_id,
                amount=-cost,
                description=f"Chat completion: {conversation_id}",
                reference_id=str(conversation_id)
            )
            self.db.add(ledger_entry)
            await self.db.commit()

            yield f"data: {json.dumps({'done': True, 'tokens': total_tokens})}\n\n"

        except Exception as e:
            yield f"data: {json.dumps({'error': str(e)})}\n\n"

    async def get_credit_balance(self, user_id: str) -> float:
        # In production, use a SQL SUM query
        result = await self.db.execute(
            select(func.sum(CreditLedger.amount)).where(CreditLedger.user_id == user_id)
        )
        return result.scalar() or 0.00

Увага: адна токэн на частку — гэта апрыксима, таму выкарыстоўваеце данні пра выкарыстанне, якія падаюць апрантар, або токенайзер. Зафіксаваная цэна є ілюстратывай, а не актуальной; цены трэба загружаць з налашчанняў. Кроме таго, пераканальванне баланса да трансляцыі і вызьманне сумы пасля неё не є атамарным; трэба заздалегідь заставіць кредыты або заблокаваць рядок пользователя.

Канец падачы звесцей

Рутэр пераканальвае права власніка (404, а не 403, каб не адкрываць ID), зберагае звесцю пользователя і вяртае StreamingResponse:

# backend/api/conversations.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from sqlalchemy.ext.asyncio import AsyncSession
from auth.dependencies import get_current_user
from database import get_db
from ai.service import AIService
from models import Conversation, Message
from schemas import CreateMessageRequest
router = APIRouter(prefix="/conversations", tags=["conversations"])
@router.post("/{conversation_id}/messages")
async def create_message(
    conversation_id: str,
    request: CreateMessageRequest,
    current_user = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    # Verify ownership
    conversation = await db.get(Conversation, conversation_id)
    if not conversation or conversation.user_id != current_user.id:
        raise HTTPException(status_code=404, detail="Conversation not found")

    # Persist user message
    user_msg = Message(
        conversation_id=conversation_id,
        role="user",
        content=request.content
    )
    db.add(user_msg)
    await db.commit()

    # Build message history
    messages = [
        {"role": m.role, "content": m.content}
        for m in conversation.messages
    ]
    messages.append({"role": "user", "content": request.content})

    # Stream response
    ai_service = AIService(db)

    return StreamingResponse(
        ai_service.stream_chat(
            user_id=str(current_user.id),
            conversation_id=conversation_id,
            messages=messages,
            model=conversation.model
        ),
        media_type="text/event-stream"
    )

З асінхронным SQLAlchemy, калі працаваць з conversation.messages у ледзівы спосаб, выйшваея памылка, таму асоціяцыю трэба загружаць якомога раней.

Глобальны ліміт частоты

Другія, абсалютнейшыя ліміты у розмаху 100 запросоў на хвілі працуюць як мідлвэр:

# backend/middleware/rate_limit.py
from fastapi import Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
import redis.asyncio as redis
redis_client = redis.from_url("redis://localhost:6379")
class RateLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Skip rate limiting for auth routes
        if request.url.path.startswith("/auth"):
            return await call_next(request)

        # Identify user (from JWT cookie or header)
        user_id = request.state.user_id if hasattr(request.state, "user_id") else "anonymous"

        key = f"ratelimit:global:{user_id}"
        current = await redis_client.incr(key)

        if current == 1:
            await redis_client.expire(key, 60)

        if current > 100:  # 100 requests per minute globally
            raise HTTPException(status_code=429, detail="Too many requests")

        response = await call_next(request)
        response.headers["X-RateLimit-Remaining"] = str(100 - current)
        return response

Эксцепцыя HTTPException, якая вызываецца ўнутрь BaseHTTPMiddleware, не ператвараецца на стандартны код 429 праз працэўнікі FastAPI, таму замест таго вяртаецца JSONResponse з статусам 429. Фіксаваныя вакна таксама дазволяюць періоды падвышанага навантажэння.

Фронтэнд

Дрэва App Router спаўнае стороннія компаненты сервера з кліентскімі компанетамі для жывага чату:

src/
├── app/
│   ├── layout.tsx
│   ├── page.tsx
│   ├── (dashboard)/
│   │   ├── layout.tsx
│   │   ├── conversations/
│   │   │   ├── page.tsx           # Server Component: list conversations
│   │   │   └── [id]/
│   │   │       ├── page.tsx       # Server Component: load initial messages
│   │   │       └── ChatClient.tsx # Client Component: streaming + input
│   │   └── settings/
│   │       └── page.tsx
│   └── api/                       # Route handlers if needed
├── components/
│   ├── ui/                        # shadcn/ui components
│   └── chat/
│       ├── MessageList.tsx
│       ├── MessageItem.tsx
│       └── StreamingText.tsx
├── lib/
│   ├── api.ts                     # Axios instance with interceptors
│   └── auth.ts
├── hooks/
│   ├── useChatStream.ts           # Custom hook for SSE
│   └── useConversations.ts
└── providers.tsx

Зачытанне історыі на сервере

Старонка чытае кукі і паралельна запрашае співрозмову і паведамленні, тады першыя даны, якія праказваюцца, ўжо маюць інфармацыю чату:

// app/(dashboard)/conversations/[id]/page.tsx
import { notFound } from "next/navigation";
import { cookies } from "next/headers";
import { ChatClient } from "./ChatClient";

async function getConversation(id: string, token: string) {
  const res = await fetch(`${process.env.API_URL}/conversations/${id}`, {
    headers: { Authorization: `Bearer ${token}` },
    next: { revalidate: 0 } // Dynamic data
  });

  if (!res.ok) return null;
  return res.json();
}async function getMessages(id: string, token: string) {
  const res = await fetch(`${process.env.API_URL}/conversations/${id}/messages`, {
    headers: { Authorization: `Bearer ${token}` },
  });

  if (!res.ok) return [];
  return res.json();
}export default async function ConversationPage({
  params
}: {
  params: { id: string }
}) {
  const cookieStore = cookies();
  const token = cookieStore.get("access_token")?.value;

  if (!token) return notFound();

  // Parallel data fetching on the server
  const [conversation, messages] = await Promise.all([
    getConversation(params.id, token),
    getMessages(params.id, token)
  ]);

  if (!conversation) return notFound();

  return (
    <div className="flex h-[calc(100vh-4rem)] flex-col">
      <header className="border-b px-6 py-4">
        <h1 className="text-lg font-semibold">{conversation.title || "New Chat"}</h1>
      </header>

      {/* Client component handles interactivity and streaming */}
      <ChatClient
        conversationId={params.id}
        initialMessages={messages}
      />
    </div>
  );
}

У Next.js 15 і новейшых версіях cookies() і params ўжо асінхронныя; трэба на яны чакаць за дапамогою await.

Стрімаванне дадзенняў у інтэрфейс

Кліент оптымістычна дадае паведамленне корыстніка і порожню альтэрнатыўную ячэйку для асістента, выканае запыт типу POST і заполняе гэту ячэйку на адпаведным стрыме:

// app/(dashboard)/conversations/[id]/ChatClient.tsx
"use client";import { useState, useRef, useCallback } from "react";
import { useChatStream } from "@/hooks/useChatStream";
import { MessageList } from "@/components/chat/MessageList";
import { ChatInput } from "@/components/chat/ChatInput";
interface Message {
  id: string;
  role: "user" | "assistant";
  content: string;
  isStreaming?: boolean;
}
export function ChatClient({
  conversationId,
  initialMessages
}: {
  conversationId: string;
  initialMessages: Message[];
}) {
  const [messages, setMessages] = useState<Message[]>(initialMessages);
  const [input, setInput] = useState("");
  const [isLoading, setIsLoading] = useState(false);

  const handleSubmit = useCallback(async (content: string) => {
    if (!content.trim() || isLoading) return;

    // 1. Optimistic UI: Add user message immediately
    const userMessage: Message = {
      id: `temp-${Date.now()}`,
      role: "user",
      content
    };

    const assistantMessage: Message = {
      id: `temp-assistant-${Date.now()}`,
      role: "assistant",
      content: "",
      isStreaming: true
    };

    setMessages(prev => [...prev, userMessage, assistantMessage]);
    setInput("");
    setIsLoading(true);

    // 2. Open SSE connection
    const response = await fetch(`/api/conversations/${conversationId}/messages`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ content })
    });

    if (!response.body) {
      setIsLoading(false);
      return;
    }

    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let assistantContent = "";

    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        const chunk = decoder.decode(value);
        const lines = chunk.split("\n");

        for (const line of lines) {
          if (line.startsWith("data: ")) {
            const data = JSON.parse(line.slice(6));

            if (data.error) {
              throw new Error(data.error);
            }

            if (data.chunk) {
              assistantContent += data.chunk;
              setMessages(prev =>
                prev.map(msg =>
                  msg.id === assistantMessage.id
                    ? { ...msg, content: assistantContent }
                    : msg
                )
              );
            }

            if (data.done) {
              setMessages(prev =>
                prev.map(msg =>
                  msg.id === assistantMessage.id
                    ? { ...msg, isStreaming: false, tokenCount: data.tokens }
                    : msg
                )
              );
            }
          }
        }
      }
    } catch (error) {
      setMessages(prev =>
        prev.map(msg =>
          msg.id === assistantMessage.id
            ? { ...msg, content: "Error: Failed to generate response", isStreaming: false }
            : msg
        )
      );
    } finally {
      setIsLoading(false);
      reader.releaseLock();
    }
  }, [conversationId, isLoading]);

  return (
    <div className="flex flex-1 flex-col">
      <MessageList messages={messages} />
      <ChatInput
        value={input}
        onChange={setInput}
        onSubmit={handleSubmit}
        disabled={isLoading}
      />
    </div>
  );
}

Незважаючы на прымэтку, тут выкарыстоўваецца fetch з чытальнікам стрыму, а не EventSource, який не можа адправляць тэлы запытавання типу POST. Неабяжна зберагаць часткі ліній, калі фрагменты можаць іх разлучыць, а таксама выкарыстоўваць AbortController для абрання запыту пад час знікнення элемента. Абсалютны адрес /api/... таксама предпакуе наявнасць працоўніка маршрута або перапісву, які праекранавае запыты да FastAPI.

Axios-кліент для всіх іншых задач

За дапамогою httpOnly-кукі кліент адправляе аутантыфікацыйныя даны, дадае ID для трэйсінгу і прабуе зноў пасля адновлення стораніцы у разе коду 401:

// lib/api.ts
import axios from "axios";
export const api = axios.create({
  baseURL: process.env.NEXT_PUBLIC_API_URL,
  withCredentials: true, // Send cookies
});
// Request interceptor: add auth token
api.interceptors.request.use((config) => {
  // Token is in httpOnly cookie, sent automatically
  // Add request ID for tracing
  config.headers["X-Request-ID"] = crypto.randomUUID();
  return config;
});
// Response interceptor: handle 401 and token refresh
api.interceptors.response.use(
  (response) => response,
  async (error) => {
    const originalRequest = error.config;

    if (error.response?.status === 401 && !originalRequest._retry) {
      originalRequest._retry = true;

      try {
        await axios.post(
          `${process.env.NEXT_PUBLIC_API_URL}/auth/refresh`,
          {},
          { withCredentials: true }
        );

        return api(originalRequest);
      } catch (refreshError) {
        window.location.href = "/login";
        return Promise.reject(refreshError);
      }
    }

    return Promise.reject(error);
  }
);

Кожны з паралельных кодаў 401 спрычынаюць апшуранне; для версіі без проблем, связаных з адночасным выконанням, пагляньце на рашэнне проблемы апшурання.

Застосавленне кредытавання як залежнасці

Залежнасць FastAPI сумуе данні з рэўістара і адхоўвае запита з кодам 402 Payment Required, калі залишак занадта малы, прычаму яна выконваецца раней за будзь-які вызов модэлю:

# backend/api/deps.py
from fastapi import Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from auth.dependencies import get_current_user
from database import get_db
from models import CreditLedger
from sqlalchemy import func, select
async def require_credits(
    min_credits: float = 0.01,
    current_user = Depends(get_current_user),
    db: AsyncSession = Depends(get_db)
):
    result = await db.execute(
        select(func.sum(CreditLedger.amount)).where(
            CreditLedger.user_id == current_user.id
        )
    )
    balance = result.scalar() or 0.00

    if balance < min_credits:
        raise HTTPException(
            status_code=402,  # Payment Required
            detail=f"Insufficient credits. Balance: ${balance:.4f}"
        )

    return current_user
# Usage
@router.post("/conversations/{id}/messages")
async def create_message(
    ...,
    user = Depends(require_credits)  # Enforces payment before AI call
):
    ...

Ў ёй таксама існуе вышэўзгаданая проблема неатомічнасці.

Тэставанне без вызову модэлю

MSW 2.x вяртае фальшывы струмень супарадкавання, а тэст з Testing Library пераказвае оптымістичны і стрімаваны текст:

// tests/mocks/openai.ts
import { http, HttpResponse } from "msw";
export const openaiHandlers = [
  http.post("https://api.openai.com/v1/chat/completions", async () => {
    const encoder = new TextEncoder();

    const stream = new ReadableStream({
      start(controller) {
        const chunks = [
          'data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n',
          'data: {"choices":[{"delta":{"content":" world"}}]}\n\n',
          'data: {"choices":[{"delta":{}}],"done":true}\n\n',
        ];

        chunks.forEach(chunk => {
          controller.enqueue(encoder.encode(chunk));
        });

        controller.close();
      }
    });

    return new HttpResponse(stream, {
      headers: {
        "Content-Type": "text/event-stream",
      },
    });
  }),
];
// tests/ChatClient.test.tsx
import { render, screen, waitFor } from "@testing-library/react";
import userEvent from "@testing-library/user-event";
import { ChatClient } from "@/app/(dashboard)/conversations/[id]/ChatClient";
it("streams assistant response", async () => {
  const user = userEvent.setup();

  render(
    <ChatClient
      conversationId="test-id"
      initialMessages={[]}
    />
  );

  await user.type(screen.getByPlaceholderText("Ask anything..."), "Hello");
  await user.click(screen.getByRole("button", { name: /send/i }));

  // Optimistic user message appears immediately
  expect(screen.getByText("Hello")).toBeInTheDocument();

  // Streaming response appears
  await waitFor(() => {
    expect(screen.getByText("Hello world")).toBeInTheDocument();
  });
});

Хэндлер імітуе URL-адреса прадастоўніка, але компонент вызывае ваш API, таму імітуйце сам адпаведны канцэнтры. Тэст бэкенда падмешчае кліента і пераказвае наявнасць адной строчкі ў рэўістарах:

# tests/test_chat.py
import pytest
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
async def test_chat_deducts_credits(client, auth_headers, db_session):
    with patch("ai.service.client.chat.completions.create") as mock_create:
        mock_create.return_value = AsyncMock()
        mock_create.return_value.__aiter__.return_value = [
            MockChunk(delta=MockContent("Hi"))
        ]

        response = await client.post(
            "/conversations/test-id/messages",
            json={"content": "Hello"},
            headers=auth_headers
        )

        assert response.status_code == 200

        # Verify credit deduction
        ledger = await db_session.execute(
            select(CreditLedger).where(CreditLedger.user_id == test_user_id)
        )
        assert len(ledger.scalars().all()) == 1

Развяртанне з контэйнерамі

Багатаступенчаты Dockerfile стварае Next.js у режыме самастоятельнай працы і запускае яго як пользователя, не ў ролі root:

# frontend/Dockerfile
FROM node:20-alpine AS builder
WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build
FROM node:20-alpine AS runner
WORKDIR /app
ENV NODE_ENV=production
RUN addgroup --system --gid 1001 nodejs
RUN adduser --system --uid 1001 nextjs
COPY --from=builder /app/public ./public
COPY --from=builder --chown=nextjs:nodejs /app/.next/standalone ./
COPY --from=builder --chown=nextjs:nodejs /app/.next/static ./.next/static
USER nextjs
EXPOSE 3000
ENV PORT=3000
CMD ["node", "server.js"]

Compose запускае PostgreSQL з аблікаваннем стану – pgvector, а таксама Redis, API, фронтэнд і nginx:

version: "3.8"
services:
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: ${DB_PASSWORD}
      POSTGRES_DB: saas_db
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U app"]
      interval: 5s
      timeout: 5s
      retries: 5
redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
  backend:
    build: ./backend
    environment:
      DATABASE_URL: postgresql+asyncpg://app:${DB_PASSWORD}@postgres/saas_db
      REDIS_URL: redis://redis:6379/0
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      SECRET_KEY: ${SECRET_KEY}
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_started
    command: uvicorn main:app --host 0.0.0.0 --port 8000
  frontend:
    build: ./frontend
    environment:
      NEXT_PUBLIC_API_URL: http://backend:8000
    depends_on:
      - backend
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - frontend
      - backend
volumes:
  postgres_data:
  redis_data:

NEXT_PUBLIC_ значэнняя ўтвараюцца пад час будовы для браузера, і ён не можа разгледзець backend:8000; выкорыстаюце публічны маршрут nginx.

Галоўныя выводы

• Зберагаюце ключы, цены і ліміты на стороне сервера; модель ёсць залежнасцю, якая знаходзіцца за вашым API.
• Запісвайце викорыстоўванне ў кніге рахункоў, дзе можна толькі дадаваць даны, і зробіце крок аблікавання і вычысцэння атомным.
• Ведзіце стрімы з явным анулюванням і буфераваннем лінакоў.
• У тэстах імітавайце саўны контракт вашага API, а не контракт постачальніка.