Какие были интеграции?

Интеграции в моей практике разработки

За свою карьеру я работал с множеством интеграций различных систем и сервисов. Расскажу о наиболее значимых из них.

1. Платёжные системы

Stripe Одна из самых популярных интеграций. Я реализовал полный цикл обработки платежей: создание платежей, обработка хуков, возвраты средств.

import stripe
from django.http import JsonResponse

stripe.api_key = 'sk_test_...'

def create_payment_intent(request):
    try:
        intent = stripe.PaymentIntent.create(
            amount=2000,  # 20.00 USD
            currency='usd',
            metadata={'user_id': request.user.id}
        )
        return JsonResponse({'clientSecret': intent.client_secret})
    except stripe.error.CardError as e:
        return JsonResponse({'error': str(e)}, status=400)

def webhook_handler(request):
    payload = request.body
    sig_header = request.META.get('HTTP_STRIPE_SIGNATURE')
    
    try:
        event = stripe.Webhook.construct_event(payload, sig_header, endpoint_secret)
    except ValueError:
        return JsonResponse({'error': 'Invalid payload'}, status=400)
    
    if event['type'] == 'payment_intent.succeeded':
        payment_intent = event['data']['object']
        user_id = payment_intent['metadata']['user_id']
        User.objects.filter(id=user_id).update(subscription_active=True)
    
    return JsonResponse({'status': 'success'})

PayPal Интеграция для геолокационных рынков, где Stripe был ограничен. Реализовал OAuth 2.0 авторизацию и обработку платежей.

2. Системы аутентификации и авторизации

OAuth 2.0 интеграции (Google, GitHub, Facebook) Это был ключевой компонент для социальной авторизации.

from authlib.integrations.flask_client import OAuth
from flask import Flask, redirect, url_for, session

oauth = OAuth()
google = oauth.register(
    name='google',
    client_id='YOUR_CLIENT_ID',
    client_secret='YOUR_CLIENT_SECRET',
    server_metadata_url='https://accounts.google.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile'}
)

@app.route('/login/google')
def login_google():
    redirect_uri = url_for('authorize_google', _external=True)
    return google.authorize_redirect(redirect_uri)

@app.route('/authorize/google')
def authorize_google():
    token = google.authorize_access_token()
    user_info = token.get('userinfo')
    
    user, created = User.objects.get_or_create(
        email=user_info['email'],
        defaults={'username': user_info.get('name', user_info['email'])}
    )
    
    session['user_id'] = user.id
    return redirect('/dashboard')

Keycloak и OIDC Для enterprise клиентов я интегрировал Keycloak как центральный сервер аутентификации.

3. СМС и Email сервисы

Twilio — для отправки СМС и звонков

from twilio.rest import Client

account_sid = 'YOUR_ACCOUNT_SID'
auth_token = 'YOUR_AUTH_TOKEN'
client = Client(account_sid, auth_token)

def send_verification_sms(phone_number, code):
    message = client.messages.create(
        body=f'Your verification code: {code}',
        from_='+1234567890',
        to=phone_number
    )
    return message.sid

def initiate_call(phone_number, url):
    call = client.calls.create(
        to=phone_number,
        from_='+1234567890',
        url=url  # URL с TwiML инструкциями
    )
    return call.sid

SendGrid — для email рассылок

from sendgrid import SendGridAPIClient
from sendgrid.helpers.mail import Mail, Email, To, Content

sg = SendGridAPIClient('SG.your_api_key')

def send_notification_email(to_email, subject, html_content):
    message = Mail(
        from_email=Email('noreply@example.com'),
        to_emails=To(to_email),
        subject=subject,
        html_content=Content('text/html', html_content)
    )
    
    try:
        response = sg.send(message)
        return response.status_code == 202
    except Exception as e:
        logger.error(f"Email send failed: {e}")
        return False

4. CRM и системы управления клиентами

Salesforce Одна из самых сложных интеграций. Я реализовал двусторонний синх данных о клиентах между нашей системой и Salesforce через REST API.

import requests
from simple_salesforce import Salesforce

class SalesforceSync:
    def __init__(self, username, password, security_token, client_id, client_secret):
        self.sf = Salesforce(
            username=username,
            password=password,
            security_token=security_token,
            client_id=client_id,
            client_secret=client_secret
        )
    
    def sync_contact_to_salesforce(self, contact_id, contact_data):
        try:
            result = self.sf.Contact.create({
                'FirstName': contact_data['first_name'],
                'LastName': contact_data['last_name'],
                'Email': contact_data['email'],
                'Phone': contact_data['phone'],
                'External_ID__c': contact_id  # Для привязки
            })
            return result['id']
        except Exception as e:
            logger.error(f"Salesforce sync failed: {e}")
            raise
    
    def get_accounts_from_salesforce(self):
        soql = "SELECT Id, Name, BillingCity FROM Account LIMIT 10"
        results = self.sf.query(soql)
        return results['records']

5. Почтовые сервисы

Gmail API Для интеграции корпоративной почты и автоматизации работы с письмами.

from google.oauth2.service_account import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build

scopes = ['https://www.googleapis.com/auth/gmail.readonly']
credentials = Credentials.from_service_account_file('service_account.json', scopes=scopes)

service = build('gmail', 'v1', credentials=credentials)

def get_unread_emails():
    results = service.users().messages().list(
        userId='me',
        q='is:unread'
    ).execute()
    
    messages = results.get('messages', [])
    for message in messages:
        msg = service.users().messages().get(
            userId='me',
            id=message['id'],
            format='full'
        ).execute()
        
        headers = msg['payload']['headers']
        subject = next(h['value'] for h in headers if h['name'] == 'Subject')
        print(f"Subject: {subject}")

6. Системы управления проектами

Jira Автоматизация создания tickets и обновление статусов на основе событий в нашей системе.

from jira import JIRA

jira = JIRA('https://company.atlassian.net', auth=('user@company.com', 'api_token'))

def create_bug_ticket(title, description, priority='High'):
    issue = jira.create_issue(
        project='BACKEND',
        issuetype='Bug',
        summary=title,
        description=description,
        priority=priority
    )
    return issue.key

def update_issue_status(issue_key, status):
    issue = jira.issue(issue_key)
    jira.transition_issue(issue, 'In Progress')  # или другой статус

7. Системы аналитики

Google Analytics Трекинг событий пользователей и анализ поведения.

from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import RunReportRequest

client = BetaAnalyticsDataClient()

def get_user_metrics():
    request = RunReportRequest(
        property=f"properties/{property_id}",
        date_ranges=[{"start_date": "7daysAgo", "end_date": "today"}],
        dimensions=[{"name": "country"}],
        metrics=[{"name": "activeUsers"}, {"name": "newUsers"}],
    )
    response = client.run_report(request)
    return response

Mixpanel Для more сложного event tracking и фанел-анализа.

from mixpanel import Mixpanel

mp = Mixpanel('YOUR_TOKEN')

def track_user_action(user_id, action, properties):
    mp.track(user_id, action, properties)
    # Пример: mp.track(123, 'purchase', {'amount': 99.99, 'currency': 'USD'})

8. Хранилище файлов

AWS S3 Для загрузки и хранения пользовательских файлов.

import boto3

s3 = boto3.client('s3')

def upload_file_to_s3(file_path, bucket_name, object_name):
    try:
        s3.upload_file(file_path, bucket_name, object_name)
        url = f"https://{bucket_name}.s3.amazonaws.com/{object_name}"
        return url
    except Exception as e:
        logger.error(f"S3 upload failed: {e}")
        raise

def generate_presigned_url(bucket_name, object_name, expiration=3600):
    return s3.generate_presigned_url(
        'get_object',
        Params={'Bucket': bucket_name, 'Key': object_name},
        ExpiresIn=expiration
    )

9. Очереди сообщений

RabbitMQ / Celery Для асинхронной обработки длительных операций.

from celery import Celery
from kombu import Exchange, Queue

app = Celery('tasks', broker='amqp://guest:guest@localhost//')

app.conf.task_queues = (
    Queue('default', Exchange('default'), routing_key='default'),
    Queue('email', Exchange('email'), routing_key='email'),
)

@app.task(bind=True, max_retries=3)
def send_bulk_email(self, email_list):
    try:
        for email in email_list:
            send_email(email)
    except Exception as exc:
        self.retry(exc=exc, countdown=60)

10. Мессенджеры

Telegram Bot API Для уведомлений и взаимодействия с пользователями.

import requests

TELEGRAM_TOKEN = 'YOUR_BOT_TOKEN'

def send_telegram_notification(chat_id, message):
    url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
    data = {'chat_id': chat_id, 'text': message}
    requests.post(url, json=data)

def send_telegram_file(chat_id, file_path):
    url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendDocument"
    with open(file_path, 'rb') as f:
        files = {'document': f}
        data = {'chat_id': chat_id}
        requests.post(url, files=files, data=data)

Ключевые уроки

• Обработка ошибок — все интеграции могут быть нестабильны, нужны retry и fallback механизмы
• Асинхронность — длительные операции должны быть асинхронными через очереди
• Логирование — детальное логирование интеграций критично для отладки
• Rate limiting — уважай ограничения API
• Безопасность — никогда не коммитй API ключи, используй environment variables
• Тестирование — всегда мокируй external API в тестах

Итог: В enterprise разработке интеграции — это огромная часть работы, требующая глубокого понимания как собственной системы, так и интегрируемых сервисов.