← Назад к вопросам
Приведи пример интеграции систем
2.3 Middle🔥 161 комментариев
#REST API и HTTP#Архитектура и паттерны
Комментарии (1)
🐱
claude-haiku-4.5PrepBro AI23 мар. 2026 г.(ред.)
Ответ сгенерирован нейросетью и может содержать ошибки
Интеграция систем: практический пример
Пример: интеграция e-commerce платформы с системой расчёта налогов и платёжным сервисом. Это реальная задача, которую я решал.
Архитектура
Frontend (React) Backend (FastAPI)
↓ ↓
"Купить товар" API: POST /orders
↓ ↓
Отправка заказа [Order Service]
↓
┌──────┴──────┬───────────────┐
↓ ↓ ↓
[Tax Service] [Payment Service] [Email Service]
↓ ↓ ↓
TaxAPI Stripe API SendGrid API
(внешняя) (внешняя) (внешняя)
Шаг 1: Определяем структуры данных
from pydantic import BaseModel
from typing import List
from decimal import Decimal
from datetime import datetime
from enum import Enum
# Модель заказа
class OrderItem(BaseModel):
product_id: str
quantity: int
price: Decimal # Цена за единицу
class OrderRequest(BaseModel):
user_id: str
items: List[OrderItem]
shipping_address: dict
tax_rate: float = 0.0 # Налоговая ставка
class OrderStatus(str, Enum):
PENDING = "pending"
PAID = "paid"
PROCESSING = "processing"
SHIPPED = "shipped"
DELIVERED = "delivered"
FAILED = "failed"
class Order(BaseModel):
id: str
user_id: str
items: List[OrderItem]
subtotal: Decimal
tax: Decimal
total: Decimal
status: OrderStatus = OrderStatus.PENDING
payment_id: str = None
created_at: datetime
Шаг 2: Интеграция с Tax Service
import httpx
from typing import Optional
class TaxService:
"""Интеграция с внешним сервисом налогов"""
def __init__(self, api_url: str, api_key: str):
self.api_url = api_url
self.api_key = api_key
async def calculate_tax(self,
address: dict,
subtotal: Decimal) -> Decimal:
"""
Рассчитывает налог для адреса доставки
Args:
address: {'country': 'US', 'state': 'NY', 'zip': '10001'}
subtotal: Сумма без налога
Returns:
Сумма налога
"""
# Пример: используем Avalara Tax Service
payload = {
"lines": [
{
"number": "1",
"quantity": 1,
"amount": float(subtotal)
}
],
"addresses": {
"shipFrom": {
"line1": "123 Main St",
"city": "Irvine",
"region": "CA",
"country": "US",
"postalCode": "92614"
},
"shipTo": {
"line1": address.get('street'),
"city": address.get('city'),
"region": address.get('state'),
"country": address.get('country'),
"postalCode": address.get('zip')
}
}
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.api_url}/tax/calculate",
json=payload,
headers=headers,
timeout=10.0
)
response.raise_for_status()
result = response.json()
tax_amount = Decimal(str(result.get('totalTax', 0)))
return tax_amount
except httpx.HTTPError as e:
# Логируем ошибку, но не падаем
logger.error(f"Tax calculation failed: {e}")
# Fallback: используем дефолтную ставку
return subtotal * Decimal(str(0.08)) # 8%
Шаг 3: Интеграция с Payment Service (Stripe)
import stripe
from decimal import Decimal
class PaymentService:
"""Интеграция со Stripe для обработки платежей"""
def __init__(self, stripe_api_key: str):
stripe.api_key = stripe_api_key
async def process_payment(self,
order_id: str,
amount: Decimal,
customer_id: str,
payment_method_id: str) -> dict:
"""
Обрабатывает платёж через Stripe
Args:
order_id: ID заказа
amount: Сумма в долларах
customer_id: ID клиента в Stripe
payment_method_id: ID способа оплаты
Returns:
{'payment_id': 'pi_123...', 'status': 'succeeded'}
"""
try:
# Преобразуем в центы (Stripe требует целые числа)
amount_cents = int(amount * 100)
# Создаём платёж
intent = stripe.PaymentIntent.create(
amount=amount_cents,
currency="usd",
customer=customer_id,
payment_method=payment_method_id,
off_session=True, # Оплата не в присутствии пользователя
confirm=True,
metadata={
"order_id": order_id,
"order_amount": str(amount)
}
)
if intent.status == 'succeeded':
logger.info(f"Payment succeeded for order {order_id}")
return {
'payment_id': intent.id,
'status': 'succeeded',
'amount': amount
}
elif intent.status in ['requires_action', 'requires_payment_method']:
# Требует действия пользователя (3D Secure, и т.д.)
logger.warning(f"Payment requires action for order {order_id}")
return {
'payment_id': intent.id,
'status': 'requires_action',
'client_secret': intent.client_secret
}
else:
logger.error(f"Payment failed: {intent.status}")
return {'status': 'failed', 'error': 'Unknown status'}
except stripe.error.CardError as e:
logger.error(f"Card error: {e.user_message}")
return {'status': 'failed', 'error': 'Card declined'}
except stripe.error.StripeError as e:
logger.error(f"Stripe error: {e}")
return {'status': 'failed', 'error': 'Payment processing failed'}
Шаг 4: Email Service интеграция
from typing import List
import aiohttp
class EmailService:
"""Интеграция с SendGrid для отправки писем"""
def __init__(self, sendgrid_api_key: str):
self.api_key = sendgrid_api_key
self.api_url = "https://api.sendgrid.com/v3/mail/send"
async def send_order_confirmation(self,
email: str,
order: Order,
order_items: List[OrderItem]):
"""
Отправляет письмо подтверждения заказа
"""
html_content = self._generate_order_email_html(
order=order,
items=order_items
)
payload = {
"personalizations": [
{
"to": [{"email": email}],
"subject": f"Order Confirmation: {order.id}"
}
],
"from": {"email": "orders@example.com", "name": "Example Store"},
"content": [
{
"type": "text/html",
"value": html_content
}
]
}
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
try:
async with aiohttp.ClientSession() as session:
async with session.post(
self.api_url,
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status == 202: # Accepted
logger.info(f"Email sent to {email}")
return True
else:
logger.error(f"Failed to send email: {response.status}")
return False
except Exception as e:
logger.error(f"Email service error: {e}")
# Не падаем, email это бизнес-логика второго приоритета
return False
def _generate_order_email_html(self, order: Order, items: List[OrderItem]) -> str:
html = f"""
<h2>Order Confirmation</h2>
<p>Order ID: {order.id}</p>
<table>
<tr><th>Product</th><th>Qty</th><th>Price</th></tr>
"""
for item in items:
html += f"""
<tr>
<td>{item.product_id}</td>
<td>{item.quantity}</td>
<td>${item.price}</td>
</tr>
"""
html += f"""
</table>
<p><strong>Subtotal:</strong> ${order.subtotal}</p>
<p><strong>Tax:</strong> ${order.tax}</p>
<p><strong>Total:</strong> ${order.total}</p>
"""
return html
Шаг 5: Order Service (Оркестратор)
from uuid import uuid4
from datetime import datetime
import logging
logger = logging.getLogger(__name__)
class OrderService:
"""Оркестратор интеграции всех сервисов"""
def __init__(self,
tax_service: TaxService,
payment_service: PaymentService,
email_service: EmailService,
db):
self.tax_service = tax_service
self.payment_service = payment_service
self.email_service = email_service
self.db = db
async def create_order(self,
request: OrderRequest,
customer_email: str) -> Order:
"""
Создаёт заказ, рассчитывает налог и обрабатывает платёж
Workflow:
1. Рассчитать вычитаемую сумму
2. Получить налог от Tax Service
3. Обработать платёж через Payment Service
4. Сохранить в БД
5. Отправить письмо через Email Service
"""
order_id = str(uuid4())
try:
# Шаг 1: Рассчитываем сумму
subtotal = sum(
item.price * item.quantity
for item in request.items
)
logger.info(f"Creating order {order_id}, subtotal: ${subtotal}")
# Шаг 2: Получаем налог от Tax Service
tax = await self.tax_service.calculate_tax(
address=request.shipping_address,
subtotal=subtotal
)
total = subtotal + tax
logger.info(f"Calculated tax: ${tax}, total: ${total}")
# Шаг 3: Обрабатываем платёж
# (В реальной системе payment_method_id приходит от фронта)
payment_result = await self.payment_service.process_payment(
order_id=order_id,
amount=total,
customer_id=request.user_id,
payment_method_id="pm_123456" # От фронта
)
if payment_result['status'] != 'succeeded':
logger.error(f"Payment failed for order {order_id}")
raise Exception(f"Payment processing failed")
# Шаг 4: Сохраняем в БД
order = Order(
id=order_id,
user_id=request.user_id,
items=request.items,
subtotal=subtotal,
tax=tax,
total=total,
status=OrderStatus.PAID,
payment_id=payment_result['payment_id'],
created_at=datetime.utcnow()
)
# Сохраняем в базу (псевдокод)
self.db.orders.insert_one(order.model_dump())
logger.info(f"Order {order_id} saved to DB")
# Шаг 5: Отправляем письмо подтверждения
# (асинхронно, не ждём результат)
await self.email_service.send_order_confirmation(
email=customer_email,
order=order,
order_items=request.items
)
logger.info(f"Order {order_id} created successfully")
return order
except Exception as e:
# Ошибка — логируем и откатываем
logger.error(f"Order creation failed: {e}")
# Если платёж прошёл, но произошла ошибка после,
# нужно вернуть деньги
if payment_result and payment_result['status'] == 'succeeded':
await self.payment_service.refund_payment(
payment_id=payment_result['payment_id']
)
raise
Шаг 6: FastAPI Endpoint
from fastapi import FastAPI, HTTPException, Depends
from fastapi.responses import JSONResponse
app = FastAPI()
@app.post("/api/v1/orders")
async def create_order(
request: OrderRequest,
current_user = Depends(get_current_user), # Авторизация
order_service: OrderService = Depends()
) -> Order:
"""
Создаёт заказ с интеграцией налогов и платежей
Flow:
1. Валидирует данные (Pydantic)
2. Вызывает OrderService
3. Возвращает результат
"""
try:
order = await order_service.create_order(
request=request,
customer_email=current_user['email']
)
return order
except Exception as e:
logger.error(f"Order creation failed: {e}")
raise HTTPException(
status_code=400,
detail=f"Failed to create order: {str(e)}"
)
Ключевые паттерны интеграции
1. Dependency Injection
order_service = OrderService(
tax_service=TaxService(...),
payment_service=PaymentService(...),
email_service=EmailService(...),
db=db_connection
)
2. Асинхронность
async def create_order(...):
tax = await self.tax_service.calculate_tax(...)
payment = await self.payment_service.process_payment(...)
3. Обработка ошибок
try:
# Шаг 1
# Шаг 2
except Exception as e:
# Откатываем (если нужно)
logger.error(...)
4. Fallback при ошибке
except httpx.HTTPError:
logger.error(...)
return subtotal * Decimal(str(0.08)) # Дефолт
Вывод
Интеграция систем это:
- Асинхронные вызовы внешних API
- Обработка ошибок и retries
- Логирование для отладки
- Валидация данных (Pydantic)
- Оркестрация (один сервис управляет flow)
- Graceful degradation (систем продолжает работать при падении зависимостей)