EQP HTTP API Endpoints

HTTP API Endpoints Module - Technical Manual

Overview

The HTTP API Endpoints module is a powerful tool for Odoo that enables seamless integration with external APIs. It provides a centralized, user-friendly interface for managing HTTP requests, handling authentication, monitoring API calls, and implementing robust error handling.

What is it?

This module allows you to:

• Configure HTTP endpoints for external API integrations
• Manage authentication (Basic Auth, Bearer Token, API Key, Custom Headers)
• Execute HTTP requests with automatic retry mechanisms
• Monitor and log all API interactions
• Handle errors gracefully with user-friendly messages
• Use a simple mixin to integrate HTTP calls into any Odoo model

Why use it?

• Centralized Management: All API endpoints in one place
• Consistent Interface: Same methods for all HTTP operations
• Error Handling: Built-in retry mechanisms and error recovery
• Monitoring: Complete logging and success rate tracking
• Security: Secure credential management
• Testing: Built-in test connection functionality

Features

Core Features

• HTTP Methods Support: GET, POST, PUT, DELETE, PATCH
• Authentication Types: None, Basic Auth, Bearer Token, API Key, Custom Headers
• Environment Management: Test and Production URL support
• Retry Mechanisms: Configurable retry attempts and delays
• Request Logging: Complete request/response logging
• Success Monitoring: Success rate tracking and statistics
• Beautiful UI: Kanban interface for endpoint management

Security Features

• Secure Credential Storage: Passwords and tokens are stored securely
• Company Isolation: Endpoints are company-specific
• Access Control: Role-based permissions
• Audit Trail: Complete logging of all API interactions

 Monitoring Features

• Real-time Statistics: Success rates, request counts, response times
• Visual Indicators: Color-coded status in kanban view
• Detailed Logs: Request/response data with timestamps
• Error Tracking: Comprehensive error message logging
• Automatic Cleanup: Daily cleanup of old logs

Installation

Prerequisites

• Odoo 17.0 or later
• Python 3.8 or later
• Internet connectivity

Installation Steps

# Download the module  
# Clone or download the module to your Odoo addons directory  
cd /path/to/odoo/addons  
git clone <repository-url> http_api_endpoints  

Install in Odoo

1. Go to the Apps menu in Odoo
2. Search for "HTTP API Endpoints"
3. Click Install

Verify Installation

• Navigate to the HTTP API Endpoints menu
• You should see the main interface with demo endpoints

Configuration

Creating an Endpoint

• Go to HTTP API Endpoints → Endpoints
• Click Create
• Fill in the required fields

Field	Description	Example
Endpoint Name	Unique identifier for the endpoint	My API Integration
Production URL	Live API URL	https://api.example.com/v1/data
Test URL	Testing API URL	https://api-test.example.com/v1/data
HTTP Method	Request method	GET, POST, PUT, DELETE, PATCH
Authentication Type	Auth method	None, Basic Auth, Bearer Token, API Key

Authentication Configuration

Basic Authentication

Username: your_username  
Password: your_password  

Bearer Token

Token: your_bearer_token_here  

API Key

API Key Header Name: X-API-Key (default)  
Token: your_api_key_here  

Custom Headers

{  
    "Authorization": "Custom your_token_here",  
    "X-Custom-Header": "custom_value"  
}  

Advanced Configuration

Setting	Description	Default
Timeout	Request timeout in seconds	30
Retry Attempts	Number of retry attempts	3
Retry Delay	Delay between retries (seconds)	1,0
Enable Logging	Log all requests/responses	VERDADERO
Testing Mode	Use test URL instead of production	VERDADERO
Log Retention	Days to keep logs (automatic cleanup)	15

Test Configuration

Test Parameters (GET)

{  
    "userId": 1,  
    "status": "active"  
}  

Test Body (POST/PUT/PATCH)

{  
    "name": "Test Item",  
    "description": "Test description",  
    "active": true  
}  

Log Retention Configuration

• Go to Settings → General Settings
• Find the HTTP API Endpoints section
• Set Log Retention Days (default: 15 days)
• Old logs will be automatically deleted daily

Usage Guide

Using the Mixin in Your Models

The HttpEndpointMixin provides a simple interface for making HTTP requests from any Odoo model.

Basic Setup

from odoo import models, fields, api  
  
class MyModel(models.Model):  
    _name = 'my.model'  
    _inherit = ['http.endpoint.mixin']  # Add the mixin  
    _description = 'My Model with API Integration'  

Simple Methods (Recommended)

All simple methods return a consistent format: {'error': bool, 'content': data}

GET Request

def fetch_data(self):  
    result = self.http_get_simple('My API Integration', params={'id': 1})  
      
    if not result['error']:  
        data = result['content']  
        # Process successful response  
        return data  
    else:  
        # Handle error  
        self.message_post(  
            body=f"API Error: {result['content']}",  
            message_type='notification'  
        )  
        return False  

POST Request

def create_external_record(self, name, description):  
    data = {  
        'name': name,  
        'description': description,  
        'active': True  
    }  
      
    result = self.http_post_simple('My API Integration', data=data)  
      
    if not result['error']:  
        new_record = result['content']  
        self.external_id = new_record.get('id')  
        return new_record  
    else:  
        self.message_post(  
            body=f"Creation failed: {result['content']}",  
            message_type='notification'  
        )  
        return False  

PUT Request

def update_external_record(self, record_id, name, description):  
    data = {  
        'id': record_id,  
        'name': name,  
        'description': description,  
        'active': True  
    }  
      
    result = self.http_put_simple('My API Integration', data=data)  
      
    if not result['error']:  
        updated_record = result['content']  
        return updated_record  
    else:  
        self.message_post(  
            body=f"Update failed: {result['content']}",  
            message_type='notification'  
        )  
        return False  

PATCH Request

def partial_update_external_record(self, record_id, **kwargs):  
    result = self.http_patch_simple('My API Integration', data=kwargs)  
      
    if not result['error']:  
        updated_record = result['content']  
        return updated_record  
    else:  
        self.message_post(  
            body=f"Partial update failed: {result['content']}",  
            message_type='notification'  
        )  
        return False  

DELETE Request

def delete_external_record(self, record_id):  
    result = self.http_delete_simple('My API Integration')  
      
    if not result['error']:  
        self.message_post(  
            body=f"Record {record_id} deleted successfully",  
            message_type='notification'  
        )  
        return True  
    else:  
        self.message_post(  
            body=f"Deletion failed: {result['content']}",  
            message_type='notification'  
        )  
        return False  

Advanced Methods

For more control over the response, use the advanced methods:

def advanced_api_call(self):  
    try:  
        response = self.http_get('My API Integration', params={'id': 1})  
          
        # Full response object with status_code, headers, etc.  
        if response.get('success', False):  
            data = response.get('json', {})  
            status_code = response.get('status_code')  
            headers = response.get('headers', {})  
            return data  
        else:  
            error_msg = response.get('content', 'Unknown error')  
            raise UserError(f"API Error: {error_msg}")  
              
    except Exception as e:  
        self.message_post(  
            body=f"Request failed: {str(e)}",  
            message_type='notification'  
        )  
        return False  

Utility Methods

List Available Endpoints

def list_endpoints(self):  
    all_endpoints = self.get_available_endpoints()  
    get_endpoints = self.get_available_endpoints('GET')  
    return all_endpoints  

Safe Request

python
Copy
def safe_api_call(self):  
    result = self.http_request_safe(  
        'My API Integration',  
        default_response={'error': True, 'content': 'Fallback response'}  
    )  
    return result  

API Reference

HttpEndpointMixin Methods

Simple Methods (Recommended)

Method	Description	Parameters	Returns
http_get_simple()	GET request with simplified response	endpoint_identifier, params=None	{'error': bool, 'content': data}
http_post_simple()	POST request with simplified response	endpoint_identifier, data=None, params=None	{'error': bool, 'content': data}
http_put_simple()	PUT request with simplified response	endpoint_identifier, data=None, params=None	{'error': bool, 'content': data}
http_delete_simple()	DELETE request with simplified response	endpoint_identifier, params=None	{'error': bool, 'content': data}
http_patch_simple()	PATCH request with simplified response	endpoint_identifier, data=None, params=None	{'error': bool, 'content': data}

 Simple Method Response

json
Copy
{  
    "error": false,  
    "content": {  
        "id": 1,  
        "name": "Example",  
        "status": "active"  
    }  
}  

Advanced Method (Recommended)

Method	Description	Parameters	Returns
http_get()	GET request with full response	endpoint_identifier, params=None	Full response dict
http_post()	POST request with full response	endpoint_identifier, data=None, params=None	Full response dict
http_put()	PUT request with full response	endpoint_identifier, data=None, params=None	Full response dict
http_delete()	DELETE request with full response	endpoint_identifier, params=None	Full response dict
http_patch()	PATCH request with full response	endpoint_identifier, data=None, params=None	Full response dict

Advanced Method Response

{  
    "status_code": 200,  
    "headers": {...},  
    "content": "...",  
    "json": {...},  
    "duration": 0.123,  
    "success": true  
}  

Utility Methods

Method	Description	Parameters	Returns
get_available_endpoints()	List available endpoints	method=None	Recordset of endpoints
http_request_safe()	Safe request (no exceptions)	endpoint_identifier, data=None, params=None, default_response=None	Response or default

Response Formats

Simple Method Response

{
    'error': False,           # Boolean indicating success/failure
    'content': {              # Response data (dict, list, or string)
        'id': 1,
        'name': 'Example',
        'status': 'active'
    }
}
  

Advanced Method Response

{
    'status_code': 200,       # HTTP status code
    'headers': {...},         # Response headers
    'content': '...',         # Raw response content
    'json': {...},           # Parsed JSON (if applicable)
    'duration': 0.123,       # Request duration in seconds
    'success': True          # Boolean indicating success
}

Examples

Example 1: Customer Synchronization

class ResPartner(models.Model):  
    _name = 'res.partner'  
    _inherit = ['res.partner', 'http.endpoint.mixin']  
      
    external_customer_id = fields.Char(string="External Customer ID")  
    last_sync = fields.Datetime(string="Last Sync")  
      
    def sync_with_crm(self):  
        """Sync customer data with external CRM system"""  
        if not self.external_customer_id:  
            data = {  
                'name': self.name,  
                'email': self.email,  
                'phone': self.phone,  
                'street': self.street,  
                'city': self.city,  
                'zip': self.zip,  
                'country': self.country_id.name if self.country_id else None  
            }  
            result = self.http_post_simple('CRM Customer API', data=data)  
            if not result['error']:  
                customer_data = result['content']  
                self.external_customer_id = customer_data.get('id')  
                self.last_sync = fields.Datetime.now()  
                self.message_post(  
                    body=f"Customer synced successfully. External ID: {self.external_customer_id}",  
                    message_type='notification'  
                )  
                return True  
            else:  
                self.message_post(  
                    body=f"Sync failed: {result['content']}",  
                    message_type='notification'  
                )  
                return False  
        else:  
            data = {  
                'id': self.external_customer_id,  
                'name': self.name,  
                'email': self.email,  
                'phone': self.phone  
            }  
            result = self.http_put_simple('CRM Customer API', data=data)  
            if not result['error']:  
                self.last_sync = fields.Datetime.now()  
                self.message_post(  
                    body="Customer updated successfully",  
                    message_type='notification'  
                )  
                return True  
            else:  
                self.message_post(  
                    body=f"Update failed: {result['content']}",  
                    message_type='notification'  
                )  
                return False  

Example 2: Inventory Management

class StockPicking(models.Model):  
    _name = 'stock.picking'  
    _inherit = ['stock.picking', 'http.endpoint.mixin']  
      
    def sync_inventory(self):  
        """Sync inventory levels with external system"""  
        for move in self.move_ids_without_package:  
            product_data = {  
                'product_code': move.product_id.default_code,  
                'quantity': move.product_uom_qty,  
                'location': self.location_id.name,  
                'destination': self.location_dest_id.name,  
                'picking_type': self.picking_type_id.name  
            }  
            result = self.http_post_simple('Inventory API', data=product_data)  
            if not result['error']:  
                inventory_response = result['content']  
                move.message_post(  
                    body=f"Inventory synced. External ID: {inventory_response.get('id')}",  
                    message_type='notification'  
                )  
            else:  
                move.message_post(  
                    body=f"Inventory sync failed: {result['content']}",  
                    message_type='notification'  
                )  

Example 3: Order Processing

class SaleOrder(models.Model):  
    _name = 'sale.order'  
    _inherit = ['sale.order', 'http.endpoint.mixin']  
      
    external_order_id = fields.Char(string="External Order ID")  
      
    def process_order(self):  
        """Process order through external fulfillment system"""  
        order_data = {  
            'order_number': self.name,  
            'customer': {  
                'name': self.partner_id.name,  
                'email': self.partner_id.email,  
                'phone': self.partner_id.phone,  
                'address': {  
                    'street': self.partner_id.street,  
                    'city': self.partner_id.city,  
                    'zip': self.partner_id.zip,  
                    'country': self.partner_id.country_id.name  
                }  
            },  
            'items': []  
        }  
        for line in self.order_line:  
            order_data['items'].append({  
                'product_code': line.product_id.default_code,  
                'quantity': line.product_uom_qty,  
                'price': line.price_unit,  
                'description': line.name  
            })  
        result = self.http_post_simple('Fulfillment API', data=order_data)  
        if not result['error']:  
            fulfillment_response = result['content']  
            self.external_order_id = fulfillment_response.get('order_id')  
            self.state = 'sale'  
            self.message_post(  
                body=f"Order processed successfully. Tracking: {fulfillment_response.get('tracking_number')}",  
                message_type='notification'  
            )  
            return True  
        else:  
            self.message_post(  
                body=f"Order processing failed: {result['content']}",  
                message_type='notification'  
            )  
            return False  

Best Practices

Endpoint Naming

Use descriptive, unique names. Include the API or service name and indicate the operation (e.g., "CRM Create Customer", "Inventory Update Stock").

Error Handling

def robust_api_call(self):  
    """Example of robust error handling"""  
    try:  
        result = self.http_get_simple('My API', params={'id': 1})  
        if not result['error']:  
            data = result['content']  
            return self.process_successful_response(data)  
        else:  
            error_msg = result['content']  
            self.log_api_error(error_msg)  
            return self.handle_api_error(error_msg)  
    except UserError as e:  
        self.message_post(  
            body=f"Configuration error: {str(e)}",  
            message_type='notification'  
        )  
        return False  
    except Exception as e:  
        self.message_post(  
            body=f"Unexpected error: {str(e)}",  
            message_type='notification'  
        )  
        return False  

Data Validation

def validate_api_data(self, data):  
    """Validate data before sending to API"""  
    required_fields = ['name', 'email']  
    for field in required_fields:  
        if not data.get(field):  
            raise UserError(f"Field '{field}' is required")  
    if data.get('email') and '@' not in data['email']:  
        raise UserError("Invalid email format")  
    return True  

Logging and Monitoring

def log_api_interaction(self, endpoint_name, request_data, response_data):  
    """Log API interactions for debugging"""  
    _logger.info(  
        "API Call - Endpoint: %s, Request: %s, Response: %s",  
        endpoint_name,  
        request_data,  
        response_data  
    )  

Testing

def test_endpoint_connection(self):  
    """Test endpoint before using in production"""  
    try:  
        result = self.http_get_simple('My API', params={'test': True})  
        if not result['error']:  
            self.message_post(  
                body="Endpoint test successful",  
                message_type='notification'  
            )  
            return True  
        else:  
            self.message_post(  
                body=f"Endpoint test failed: {result['content']}",  
                message_type='notification'  
            )  
            return False  
    except Exception as e:  
        self.message_post(  
            body=f"Endpoint test error: {str(e)}",  
            message_type='notification'  
        )  
        return False  

Troubleshooting

"Endpoint not found" Error

Problem:UserError: Endpoint 'My API' not found

Solution:

• Check if the endpoint name is spelled correctly
• Verify the endpoint is active
• Ensure the endpoint exists in the current company

Authentication Errors

Problem: 401 Unauthorized or 403 Forbidden

Solution:

• Verify credentials are correct
• Check if the API key/token is valid
• Ensure the authentication type matches the API requirements

Timeout Errors

Problem: Request timeout errors

Solution:

• Increase the timeout value in endpoint configuration
• Check network connectivity
• Verify the API server is responding

JSON Decode Errors

Problem: 'list' object has no attribute 'get'

Solution:

• This is now fixed in the latest version
• The module properly handles both dict and list responses
• Update to the latest version if you encounter this

Method Mismatch Errors

Problem: Endpoint is not configured for GET requests

Solution:

• Check the HTTP method configuration in the endpoint
• Ensure you're using the correct method for the API endpoint
• Verify the API supports the method you're trying to use

Debugging Tips

1. Enable Logging: Set enable_logging = True in endpoint configuration
2. Check Logs: Review the HTTP Endpoint Logs for detailed request/response information
3. Test Connection: Use the "Test Connection" button to verify endpoint configuration
4. Use Simple Methods: Start with simple methods for easier error handling
5. Check Network: Verify internet connectivity and firewall settings

Performance Optimization

1. Use Appropriate Timeouts: Set reasonable timeout values
2. Implement Caching: Cache frequently accessed data
3. Batch Requests: Combine multiple operations when possible
4. Monitor Success Rates: Use the built-in statistics to identify issues
5. Retry Configuration: Adjust retry attempts and delays based on API behavior

Support

Documentation

• Module Documentation: This README file
• Demo Endpoints: Pre-configured examples using JSONPlaceholder API
• Usage Examples: Comprehensive code examples in the demo folder

Getting Help

1. Check the Logs: Review HTTP Endpoint Logs for detailed error information
2. Test Endpoints: Use the built-in test functionality
3. Review Examples: Check the demo endpoints and usage examples
4. Contact Support: Reach out to EQP Solutions for technical support

License

This module is licensed under the Odoo Proprietary License v1.0 (OPL-1).

Developed by EQP Solutions

Professional Odoo Development Services

For more information, visit: https://www.eqpsolutions.com