Skip to content

Best Practices

This guide provides best practices for integrating with the VibeMobi Favorite Number Management API. Following these guidelines will help ensure reliable, secure, and efficient API usage.

Authentication Best Practices

Token Management

Secure Token Storage

import os
from cryptography.fernet import Fernet

# Store tokens securely using environment variables
API_TOKEN = os.getenv('VIBEMOBI_API_TOKEN')

# For additional security, encrypt tokens at rest
def encrypt_token(token):
    key = Fernet.generate_key()
    f = Fernet(key)
    encrypted_token = f.encrypt(token.encode())
    return encrypted_token, key

def decrypt_token(encrypted_token, key):
    f = Fernet(key)
    decrypted_token = f.decrypt(encrypted_token)
    return decrypted_token.decode()

Token Refresh Strategy

import requests
import time
from datetime import datetime, timedelta

class TokenManager:
    def __init__(self, username, password):
        self.username = username
        self.password = password
        self.token = None
        self.expires_at = None

    def get_valid_token(self):
        if not self.token or self.is_token_expired():
            self.refresh_token()
        return self.token

    def is_token_expired(self):
        if not self.expires_at:
            return True
        return datetime.now() >= self.expires_at

    def refresh_token(self):
        response = requests.post(
            'https://fav3.vibemobi.com/v1/login/access-token',
            data={
                'username': self.username,
                'password': self.password
            }
        )

        if response.status_code == 200:
            data = response.json()
            self.token = data['access_token']
            # Assume token expires in 1 hour (adjust based on actual expiry)
            self.expires_at = datetime.now() + timedelta(hours=1)
        else:
            raise Exception('Failed to refresh token')

Security Guidelines

Security Checklist

  • ✅ Never hardcode API credentials in source code
  • ✅ Use HTTPS for all API communications
  • ✅ Store tokens securely using environment variables or secret management
  • ✅ Implement token refresh mechanisms
  • ✅ Use different credentials for different environments (dev, staging, prod)
  • ✅ Regularly rotate API credentials
  • ✅ Monitor for unauthorized API usage
  • ✅ Implement proper logging without exposing sensitive data

Error Handling

Robust Error Handling Pattern

import requests
import time
import logging
from typing import Optional, Dict, Any

class VibeMobiAPIClient:
    def __init__(self, base_url: str, token_manager: TokenManager):
        self.base_url = base_url
        self.token_manager = token_manager
        self.logger = logging.getLogger(__name__)

    def make_request(self, method: str, endpoint: str, data: Optional[Dict] = None, 
                    max_retries: int = 3) -> Dict[Any, Any]:
        """Make API request with retry logic and proper error handling"""

        url = f"{self.base_url}{endpoint}"
        headers = {
            'Authorization': f'Bearer {self.token_manager.get_valid_token()}',
            'Content-Type': 'application/json'
        }

        for attempt in range(max_retries):
            try:
                response = requests.request(method, url, json=data, headers=headers)

                # Handle different status codes
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 401:
                    # Token expired, refresh and retry
                    self.token_manager.refresh_token()
                    headers['Authorization'] = f'Bearer {self.token_manager.get_valid_token()}'
                    continue
                elif response.status_code == 429:
                    # Rate limited, wait and retry
                    wait_time = 2 ** attempt
                    self.logger.warning(f"Rate limited, waiting {wait_time} seconds")
                    time.sleep(wait_time)
                    continue
                elif response.status_code == 422:
                    # Validation error, don't retry
                    error_detail = response.json().get('detail', 'Validation error')
                    raise ValueError(f"Validation error: {error_detail}")
                else:
                    response.raise_for_status()

            except requests.exceptions.ConnectionError as e:
                if attempt == max_retries - 1:
                    raise ConnectionError(f"Failed to connect after {max_retries} attempts: {e}")
                wait_time = 2 ** attempt
                time.sleep(wait_time)

            except requests.exceptions.Timeout as e:
                if attempt == max_retries - 1:
                    raise TimeoutError(f"Request timed out after {max_retries} attempts: {e}")
                wait_time = 2 ** attempt
                time.sleep(wait_time)

        raise Exception(f"Request failed after {max_retries} attempts")

Error Response Handling

def handle_api_error(response):
    """Handle different types of API errors"""

    if response.status_code == 400:
        return "Bad request - check your request parameters"
    elif response.status_code == 401:
        return "Unauthorized - check your authentication token"
    elif response.status_code == 403:
        return "Forbidden - insufficient permissions"
    elif response.status_code == 404:
        return "Resource not found"
    elif response.status_code == 409:
        return "Conflict - resource already exists or constraint violation"
    elif response.status_code == 422:
        # Parse validation errors
        try:
            error_detail = response.json().get('detail', [])
            if isinstance(error_detail, list):
                errors = []
                for error in error_detail:
                    field = '.'.join(str(loc) for loc in error.get('loc', []))
                    message = error.get('msg', 'Unknown error')
                    errors.append(f"{field}: {message}")
                return "Validation errors: " + "; ".join(errors)
            else:
                return f"Validation error: {error_detail}"
        except:
            return "Validation error occurred"
    elif response.status_code == 429:
        return "Rate limit exceeded - please slow down your requests"
    elif response.status_code >= 500:
        return "Server error - please try again later"
    else:
        return f"Unexpected error: {response.status_code}"

Rate Limiting and Performance

Implementing Rate Limiting

import time
from collections import deque
from threading import Lock

class RateLimiter:
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = time_window
        self.requests = deque()
        self.lock = Lock()

    def wait_if_needed(self):
        with self.lock:
            now = time.time()

            # Remove old requests outside the time window
            while self.requests and self.requests[0] <= now - self.time_window:
                self.requests.popleft()

            # If we're at the limit, wait
            if len(self.requests) >= self.max_requests:
                sleep_time = self.time_window - (now - self.requests[0])
                if sleep_time > 0:
                    time.sleep(sleep_time)
                    # Remove the old request after waiting
                    self.requests.popleft()

            # Record this request
            self.requests.append(now)

# Usage
rate_limiter = RateLimiter(max_requests=100, time_window=60)  # 100 requests per minute

def make_api_call():
    rate_limiter.wait_if_needed()
    # Make your API call here
    pass

Batch Operations

def batch_add_favorites(client, customer_phone, favorites_list, batch_size=10):
    """Add multiple favorites in batches to avoid overwhelming the API"""

    results = []
    errors = []

    for i in range(0, len(favorites_list), batch_size):
        batch = favorites_list[i:i + batch_size]

        for favorite in batch:
            try:
                result = client.make_request('POST', '/v1/favorites/', {
                    'customer_phone_number': customer_phone,
                    'phone_number': favorite['phone_number'],
                    'label': favorite['label']
                })
                results.append(result)

                # Small delay between requests in the same batch
                time.sleep(0.1)

            except Exception as e:
                errors.append({
                    'favorite': favorite,
                    'error': str(e)
                })

        # Longer delay between batches
        if i + batch_size < len(favorites_list):
            time.sleep(1)

    return results, errors

Data Validation

Input Validation

import re
from typing import Dict, List, Any

class DataValidator:
    @staticmethod
    def validate_phone_number(phone_number: str) -> bool:
        """Validate phone number format"""
        if not phone_number or len(phone_number) < 8 or len(phone_number) > 20:
            return False

        # Basic phone number pattern (adjust based on your requirements)
        pattern = r'^[\d\-\+\(\)\s]+$'
        return bool(re.match(pattern, phone_number))

    @staticmethod
    def validate_email(email: str) -> bool:
        """Validate email format"""
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        return bool(re.match(pattern, email))

    @staticmethod
    def validate_favorite_label(label: str) -> bool:
        """Validate favorite label"""
        return bool(label and 1 <= len(label.strip()) <= 50)

    @staticmethod
    def sanitize_input(data: Dict[str, Any]) -> Dict[str, Any]:
        """Sanitize input data"""
        sanitized = {}

        for key, value in data.items():
            if isinstance(value, str):
                # Remove leading/trailing whitespace
                value = value.strip()
                # Remove potentially harmful characters
                value = re.sub(r'[<>"\']', '', value)

            sanitized[key] = value

        return sanitized

# Usage example
def add_favorite_with_validation(client, customer_phone, phone_number, label):
    validator = DataValidator()

    # Validate inputs
    if not validator.validate_phone_number(customer_phone):
        raise ValueError("Invalid customer phone number")

    if not validator.validate_phone_number(phone_number):
        raise ValueError("Invalid favorite phone number")

    if not validator.validate_favorite_label(label):
        raise ValueError("Invalid label (must be 1-50 characters)")

    # Sanitize data
    data = validator.sanitize_input({
        'customer_phone_number': customer_phone,
        'phone_number': phone_number,
        'label': label
    })

    return client.make_request('POST', '/v1/favorites/', data)

Logging and Monitoring

Comprehensive Logging

import logging
import json
from datetime import datetime

class APILogger:
    def __init__(self, name: str):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)

        # Create formatter
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )

        # Console handler
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)

        # File handler
        file_handler = logging.FileHandler('api_client.log')
        file_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)

    def log_request(self, method: str, endpoint: str, data: dict = None):
        """Log API request"""
        log_data = {
            'type': 'request',
            'method': method,
            'endpoint': endpoint,
            'timestamp': datetime.now().isoformat()
        }

        if data:
            # Don't log sensitive data
            safe_data = {k: v for k, v in data.items() 
                        if k not in ['password', 'token', 'secret']}
            log_data['data'] = safe_data

        self.logger.info(f"API Request: {json.dumps(log_data)}")

    def log_response(self, status_code: int, response_time: float, error: str = None):
        """Log API response"""
        log_data = {
            'type': 'response',
            'status_code': status_code,
            'response_time_ms': round(response_time * 1000, 2),
            'timestamp': datetime.now().isoformat()
        }

        if error:
            log_data['error'] = error
            self.logger.error(f"API Error: {json.dumps(log_data)}")
        else:
            self.logger.info(f"API Response: {json.dumps(log_data)}")

Performance Monitoring

import time
from contextlib import contextmanager

@contextmanager
def monitor_api_call(logger, operation_name):
    """Context manager to monitor API call performance"""
    start_time = time.time()
    try:
        yield
        end_time = time.time()
        duration = end_time - start_time
        logger.info(f"{operation_name} completed in {duration:.2f} seconds")
    except Exception as e:
        end_time = time.time()
        duration = end_time - start_time
        logger.error(f"{operation_name} failed after {duration:.2f} seconds: {e}")
        raise

# Usage
with monitor_api_call(logger, "Add Favorite"):
    result = client.make_request('POST', '/v1/favorites/', data)

Testing Strategies

Unit Testing

import unittest
from unittest.mock import Mock, patch
import requests

class TestVibeMobiAPI(unittest.TestCase):
    def setUp(self):
        self.client = VibeMobiAPIClient(
            base_url='https://fav3.vibemobi.com',
            token_manager=Mock()
        )
        self.client.token_manager.get_valid_token.return_value = 'test_token'

    @patch('requests.request')
    def test_add_favorite_success(self, mock_request):
        # Mock successful response
        mock_response = Mock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            'id': '123',
            'phone_number': '1234567890',
            'label': 'Test'
        }
        mock_request.return_value = mock_response

        result = self.client.make_request('POST', '/v1/favorites/', {
            'customer_phone_number': '0987654321',
            'phone_number': '1234567890',
            'label': 'Test'
        })

        self.assertEqual(result['id'], '123')
        mock_request.assert_called_once()

    @patch('requests.request')
    def test_add_favorite_validation_error(self, mock_request):
        # Mock validation error response
        mock_response = Mock()
        mock_response.status_code = 422
        mock_response.json.return_value = {
            'detail': [
                {
                    'loc': ['body', 'phone_number'],
                    'msg': 'field required',
                    'type': 'value_error.missing'
                }
            ]
        }
        mock_request.return_value = mock_response

        with self.assertRaises(ValueError):
            self.client.make_request('POST', '/v1/favorites/', {})

if __name__ == '__main__':
    unittest.main()

Integration Testing

import pytest
import os

class TestVibeMobiIntegration:
    @pytest.fixture
    def client(self):
        # Use test credentials
        token_manager = TokenManager(
            username=os.getenv('TEST_USERNAME'),
            password=os.getenv('TEST_PASSWORD')
        )
        return VibeMobiAPIClient(
            base_url=os.getenv('TEST_API_URL', 'https://test-api.vibemobi.com'),
            token_manager=token_manager
        )

    def test_full_favorite_lifecycle(self, client):
        """Test complete favorite lifecycle: add, update, check, remove"""

        customer_phone = '1234567890'
        favorite_phone = '0987654321'

        # Add favorite
        add_result = client.make_request('POST', '/v1/favorites/', {
            'customer_phone_number': customer_phone,
            'phone_number': favorite_phone,
            'label': 'Test Contact'
        })

        assert add_result['phone_number'] == favorite_phone
        assert add_result['label'] == 'Test Contact'

        # Check favorite
        check_result = client.make_request('GET', '/v1/favorites/check', {
            'customer_phone': customer_phone,
            'target_phone': favorite_phone
        })

        assert check_result['is_favorite'] is True

        # Update label
        update_result = client.make_request('PUT', '/v1/favorites/', {
            'customer_phone_number': customer_phone,
            'phone_number': favorite_phone,
            'new_label': 'Updated Contact'
        })

        assert update_result['label'] == 'Updated Contact'

        # Remove favorite
        client.make_request('DELETE', '/v1/favorites/', {
            'customer_phone_number': customer_phone,
            'favorite_phone': favorite_phone
        })

        # Verify removal
        check_result = client.make_request('GET', '/v1/favorites/check', {
            'customer_phone': customer_phone,
            'target_phone': favorite_phone
        })

        assert check_result['is_favorite'] is False

Environment Management

Configuration Management

import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class APIConfig:
    base_url: str
    username: str
    password: str
    timeout: int = 30
    max_retries: int = 3
    rate_limit_per_minute: int = 100

    @classmethod
    def from_environment(cls, env_prefix: str = 'VIBEMOBI'):
        """Load configuration from environment variables"""
        return cls(
            base_url=os.getenv(f'{env_prefix}_API_URL', 'https://fav3.vibemobi.com'),
            username=os.getenv(f'{env_prefix}_USERNAME'),
            password=os.getenv(f'{env_prefix}_PASSWORD'),
            timeout=int(os.getenv(f'{env_prefix}_TIMEOUT', '30')),
            max_retries=int(os.getenv(f'{env_prefix}_MAX_RETRIES', '3')),
            rate_limit_per_minute=int(os.getenv(f'{env_prefix}_RATE_LIMIT', '100'))
        )

    def validate(self):
        """Validate configuration"""
        if not self.base_url:
            raise ValueError("API base URL is required")
        if not self.username:
            raise ValueError("Username is required")
        if not self.password:
            raise ValueError("Password is required")

# Usage
config = APIConfig.from_environment()
config.validate()

Environment-Specific Settings

# Development environment (.env.dev)
VIBEMOBI_API_URL=https://dev-api.vibemobi.com
VIBEMOBI_USERNAME=dev_user
VIBEMOBI_PASSWORD=dev_password
VIBEMOBI_RATE_LIMIT=50

# Staging environment (.env.staging)
VIBEMOBI_API_URL=https://staging-api.vibemobi.com
VIBEMOBI_USERNAME=staging_user
VIBEMOBI_PASSWORD=staging_password
VIBEMOBI_RATE_LIMIT=75

# Production environment (.env.prod)
VIBEMOBI_API_URL=https://fav3.vibemobi.com
VIBEMOBI_USERNAME=prod_user
VIBEMOBI_PASSWORD=prod_password
VIBEMOBI_RATE_LIMIT=100

Summary

Following these best practices will help you build robust, secure, and efficient integrations with the VibeMobi API:

  1. Security First: Always use secure authentication and never expose credentials
  2. Handle Errors Gracefully: Implement comprehensive error handling and retry logic
  3. Respect Rate Limits: Use rate limiting and batch operations appropriately
  4. Validate Data: Always validate and sanitize input data
  5. Monitor and Log: Implement comprehensive logging and monitoring
  6. Test Thoroughly: Use both unit and integration testing
  7. Manage Environments: Use proper configuration management for different environments

These practices will ensure your integration is production-ready and maintainable.