Django Integration

Complete example of MONCRENEAU API integration with Django 4.x and Django REST Framework.

Project Structure

moncreneau-django/
├── requirements.txt
├── .env.example
├── manage.py
├── moncreneau_project/
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── appointments/
│   ├── __init__.py
│   ├── apps.py
│   ├── models.py
│   ├── views.py
│   ├── serializers.py
│   ├── urls.py
│   └── services/
│       ├── __init__.py
│       ├── moncreneau_client.py
│       └── webhook_handler.py
├── webhooks/
│   ├── __init__.py
│   ├── middleware.py
│   ├── views.py
│   └── urls.py
└── tests/
  ├── __init__.py
  ├── test_appointments.py
  └── test_webhooks.py

Configuration

requirements.txt

Django==4.2.8
djangorestframework==3.14.0
requests==2.31.0
python-dotenv==1.0.0
celery==5.3.4
redis==5.0.1
django-cors-headers==4.3.1

# Testing
pytest==7.4.3
pytest-django==4.7.0
pytest-cov==4.1.0

.env.example

# Django
SECRET_KEY=your-secret-key-here
DEBUG=True
ALLOWED_HOSTS=localhost,127.0.0.1

# MONCRENEAU API
MONCRENEAU_API_URL=https://mc-prd.duckdns.org/api/v1
MONCRENEAU_API_KEY=your_api_key_here
MONCRENEAU_WEBHOOK_SECRET=your_webhook_secret_here

# Celery
CELERY_BROKER_URL=redis://localhost:6379/0
CELERY_RESULT_BACKEND=redis://localhost:6379/0

Settings

moncreneau_project/settings.py (excerpt)

import os
from pathlib import Path
from dotenv import load_dotenv

load_dotenv()

BASE_DIR = Path(__file__).resolve().parent.parent

SECRET_KEY = os.getenv('SECRET_KEY')
DEBUG = os.getenv('DEBUG', 'False') == 'True'
ALLOWED_HOSTS = os.getenv('ALLOWED_HOSTS', '').split(',')

INSTALLED_APPS = [
  'django.contrib.admin',
  'django.contrib.auth',
  'django.contrib.contenttypes',
  'django.contrib.sessions',
  'django.contrib.messages',
  'django.contrib.staticfiles',
  'rest_framework',
  'corsheaders',
  'appointments',
  'webhooks',
]

MIDDLEWARE = [
  'django.middleware.security.SecurityMiddleware',
  'django.contrib.sessions.middleware.SessionMiddleware',
  'corsheaders.middleware.CorsMiddleware',
  'django.middleware.common.CommonMiddleware',
  # Note: CSRF disabled for webhooks, handle differently in production
  'django.contrib.auth.middleware.AuthenticationMiddleware',
  'django.contrib.messages.middleware.MessageMiddleware',
  'django.middleware.clickjacking.XFrameOptionsMiddleware',
]

ROOT_URLCONF = 'moncreneau_project.urls'

# MONCRENEAU Configuration
MONCRENEAU = {
  'API_URL': os.getenv('MONCRENEAU_API_URL'),
  'API_KEY': os.getenv('MONCRENEAU_API_KEY'),
  'WEBHOOK_SECRET': os.getenv('MONCRENEAU_WEBHOOK_SECRET'),
  'TIMEOUT': 30,
}

# Celery Configuration
CELERY_BROKER_URL = os.getenv('CELERY_BROKER_URL')
CELERY_RESULT_BACKEND = os.getenv('CELERY_RESULT_BACKEND')
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'UTC'

# Logging
LOGGING = {
  'version': 1,
  'disable_existing_loggers': False,
  'formatters': {
      'verbose': {
          'format': '{levelname} {asctime} {module} {message}',
          'style': '{',
      },
  },
  'handlers': {
      'console': {
          'class': 'logging.StreamHandler',
          'formatter': 'verbose',
      },
      'file': {
          'class': 'logging.FileHandler',
          'filename': 'moncreneau.log',
          'formatter': 'verbose',
      },
  },
  'loggers': {
      'appointments': {
          'handlers': ['console', 'file'],
          'level': 'INFO',
      },
      'webhooks': {
          'handlers': ['console', 'file'],
          'level': 'INFO',
      },
  },
}

Services

appointments/services/moncreneau_client.py

import logging
import requests
from typing import Dict, Any
from django.conf import settings

logger = logging.getLogger(__name__)

class MonCreneauClient:
  def __init__(self):
      self.api_url = settings.MONCRENEAU['API_URL']
      self.api_key = settings.MONCRENEAU['API_KEY']
      self.timeout = settings.MONCRENEAU['TIMEOUT']
      
      self.session = requests.Session()
      self.session.headers.update({
          'Authorization': f'Bearer {self.api_key}',
          'Content-Type': 'application/json',
      })
  
  def create_appointment(self, data: Dict[str, Any]) -> Dict[str, Any]:
      """Create a new appointment."""
      try:
          logger.info(
              f"Creating appointment for organization {data.get('organizationId')}"
          )
          
          response = self.session.post(
              f'{self.api_url}/appointments',
              json=data,
              timeout=self.timeout,
          )
          response.raise_for_status()
          
          appointment = response.json()
          logger.info(f"Appointment created: {appointment['id']}")
          
          return appointment
          
      except requests.exceptions.HTTPError as e:
          return self._handle_http_error(e)
      except requests.exceptions.RequestException as e:
          logger.error(f"Request failed: {str(e)}")
          raise
  
  def get_appointment(self, appointment_id: int) -> Dict[str, Any]:
      """Get appointment by ID."""
      try:
          response = self.session.get(
              f'{self.api_url}/appointments/{appointment_id}',
              timeout=self.timeout,
          )
          response.raise_for_status()
          return response.json()
          
      except requests.exceptions.RequestException as e:
          logger.error(f"Failed to get appointment {appointment_id}: {str(e)}")
          raise
  
  def cancel_appointment(self, appointment_id: int) -> None:
      """Cancel an appointment."""
      try:
          logger.info(f"Cancelling appointment {appointment_id}")
          
          response = self.session.delete(
              f'{self.api_url}/appointments/{appointment_id}',
              timeout=self.timeout,
          )
          response.raise_for_status()
          
          logger.info(f"Appointment {appointment_id} cancelled")
          
      except requests.exceptions.RequestException as e:
          logger.error(f"Failed to cancel appointment {appointment_id}: {str(e)}")
          raise
  
  def _handle_http_error(self, error: requests.exceptions.HTTPError):
      """Handle HTTP errors from the API."""
      status_code = error.response.status_code
      
      if status_code == 402:
          logger.error("Insufficient credits")
          raise ValueError("Insufficient credits")
      elif status_code == 400:
          error_data = error.response.json()
          logger.error(f"Validation error: {error_data}")
          raise ValueError(f"Invalid data: {error_data}")
      elif status_code == 429:
          logger.error("Rate limit exceeded")
          raise ValueError("Rate limit exceeded")
      else:
          logger.error(f"API error: {error.response.text}")
          raise

appointments/services/webhook_handler.py

import logging
from typing import Dict, Any
from celery import shared_task

logger = logging.getLogger(__name__)

class WebhookEventHandler:
  @staticmethod
  def process_event(event_type: str, event_data: Dict[str, Any]) -> None:
      """Route webhook events to appropriate handlers."""
      logger.info(f"Processing webhook event: {event_type}")
      
      handlers = {
          'appointment.created': WebhookEventHandler._handle_appointment_created,
          'appointment.updated': WebhookEventHandler._handle_appointment_updated,
          'appointment.cancelled': WebhookEventHandler._handle_appointment_cancelled,
          'appointment.validated': WebhookEventHandler._handle_appointment_validated,
          'appointment.completed': WebhookEventHandler._handle_appointment_completed,
      }
      
      handler = handlers.get(event_type)
      if handler:
          handler(event_data)
      else:
          logger.warning(f"Unknown event type: {event_type}")
  
  @staticmethod
  def _handle_appointment_created(event_data: Dict[str, Any]) -> None:
      logger.info(f"Handling appointment.created: {event_data['object']['id']}")
      # Send confirmation email
      # Send SMS
      # Update local database
  
  @staticmethod
  def _handle_appointment_updated(event_data: Dict[str, Any]) -> None:
      logger.info(f"Handling appointment.updated: {event_data['object']['id']}")
      # Notify user of changes
  
  @staticmethod
  def _handle_appointment_cancelled(event_data: Dict[str, Any]) -> None:
      logger.info(f"Handling appointment.cancelled: {event_data['object']['id']}")
      # Send cancellation email
      # Release resources
  
  @staticmethod
  def _handle_appointment_validated(event_data: Dict[str, Any]) -> None:
      logger.info(f"Handling appointment.validated: {event_data['object']['id']}")
      # Confirm with staff
  
  @staticmethod
  def _handle_appointment_completed(event_data: Dict[str, Any]) -> None:
      logger.info(f"Handling appointment.completed: {event_data['object']['id']}")
      # Send satisfaction survey


@shared_task
def process_webhook_event(event_type: str, event_data: Dict[str, Any]):
  """Celery task for async webhook processing."""
  WebhookEventHandler.process_event(event_type, event_data)

Views

appointments/views.py

import logging
from rest_framework import status
from rest_framework.decorators import api_view
from rest_framework.response import Response
from .services.moncreneau_client import MonCreneauClient

logger = logging.getLogger(__name__)
client = MonCreneauClient()

@api_view(['POST'])
def create_appointment(request):
  """Create a new appointment."""
  try:
      appointment = client.create_appointment(request.data)
      return Response(appointment, status=status.HTTP_201_CREATED)
  except ValueError as e:
      return Response(
          {'error': str(e)},
          status=status.HTTP_400_BAD_REQUEST
      )
  except Exception as e:
      logger.error(f"Error creating appointment: {str(e)}")
      return Response(
          {'error': 'Internal server error'},
          status=status.HTTP_500_INTERNAL_SERVER_ERROR
      )

@api_view(['GET'])
def get_appointment(request, appointment_id):
  """Get appointment by ID."""
  try:
      appointment = client.get_appointment(appointment_id)
      return Response(appointment)
  except Exception as e:
      logger.error(f"Error getting appointment: {str(e)}")
      return Response(
          {'error': 'Appointment not found'},
          status=status.HTTP_404_NOT_FOUND
      )

@api_view(['DELETE'])
def cancel_appointment(request, appointment_id):
  """Cancel an appointment."""
  try:
      client.cancel_appointment(appointment_id)
      return Response({'message': 'Appointment cancelled'})
  except Exception as e:
      logger.error(f"Error cancelling appointment: {str(e)}")
      return Response(
          {'error': str(e)},
          status=status.HTTP_500_INTERNAL_SERVER_ERROR
      )

webhooks/views.py

import logging
import hmac
import hashlib
import json
from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_http_methods
from django.conf import settings
from appointments.services.webhook_handler import process_webhook_event

logger = logging.getLogger(__name__)

def verify_webhook_signature(payload: bytes, signature: str) -> bool:
  """Verify HMAC-SHA256 signature."""
  if not signature or not signature.startswith('sha256='):
      logger.warning('Invalid signature format')
      return False
  
  received_hash = signature[7:]
  secret = settings.MONCRENEAU['WEBHOOK_SECRET']
  
  expected_hash = hmac.new(
      secret.encode('utf-8'),
      payload,
      hashlib.sha256
  ).hexdigest()
  
  # Timing-safe comparison
  return hmac.compare_digest(expected_hash, received_hash)

@csrf_exempt
@require_http_methods(["POST"])
def handle_webhook(request):
  """Handle incoming webhooks from MONCRENEAU."""
  signature = request.headers.get('X-Moncreneau-Signature')
  payload = request.body
  
  # Verify signature
  if not verify_webhook_signature(payload, signature):
      logger.warning('Invalid webhook signature')
      return HttpResponse('Invalid signature', status=401)
  
  try:
      # Parse event
      event = json.loads(payload.decode('utf-8'))
      event_type = event.get('type')
      event_data = event.get('data')
      
      # Process asynchronously with Celery
      process_webhook_event.delay(event_type, event_data)
      
      return HttpResponse('OK', status=200)
      
  except json.JSONDecodeError as e:
      logger.error(f"Invalid JSON: {str(e)}")
      return HttpResponse('Invalid JSON', status=400)
  except Exception as e:
      logger.error(f"Error processing webhook: {str(e)}")
      return HttpResponse('Internal server error', status=500)

URLs

appointments/urls.py

from django.urls import path
from . import views

urlpatterns = [
  path('appointments/', views.create_appointment, name='create_appointment'),
  path('appointments/<int:appointment_id>/', views.get_appointment, name='get_appointment'),
  path('appointments/<int:appointment_id>/cancel/', views.cancel_appointment, name='cancel_appointment'),
]

webhooks/urls.py

from django.urls import path
from . import views

urlpatterns = [
  path('moncreneau/', views.handle_webhook, name='webhook'),
]

moncreneau_project/urls.py

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
  path('admin/', admin.site.urls),
  path('api/', include('appointments.urls')),
  path('webhooks/', include('webhooks.urls')),
]

Tests

tests/test_webhooks.py

import json
import hmac
import hashlib
from django.test import TestCase, Client
from django.conf import settings

class WebhookTestCase(TestCase):
  def setUp(self):
      self.client = Client()
      self.webhook_url = '/webhooks/moncreneau/'
      self.secret = settings.MONCRENEAU['WEBHOOK_SECRET']
  
  def sign_payload(self, payload: str) -> str:
      """Generate HMAC signature."""
      signature = hmac.new(
          self.secret.encode('utf-8'),
          payload.encode('utf-8'),
          hashlib.sha256
      ).hexdigest()
      return f'sha256={signature}'
  
  def test_webhook_accepts_valid_signature(self):
      """Test webhook accepts valid signature."""
      payload = json.dumps({
          'id': 'evt_test_123',
          'type': 'appointment.created',
          'data': {'object': {'id': 123}}
      })
      signature = self.sign_payload(payload)
      
      response = self.client.post(
          self.webhook_url,
          data=payload,
          content_type='application/json',
          HTTP_X_MONCRENEAU_SIGNATURE=signature
      )
      
      self.assertEqual(response.status_code, 200)
  
  def test_webhook_rejects_invalid_signature(self):
      """Test webhook rejects invalid signature."""
      payload = json.dumps({'id': 'evt_test'})
      
      response = self.client.post(
          self.webhook_url,
          data=payload,
          content_type='application/json',
          HTTP_X_MONCRENEAU_SIGNATURE='sha256=invalid'
      )
      
      self.assertEqual(response.status_code, 401)

Getting Started

# 1. Installation
pip install -r requirements.txt

# 2. Configuration
cp .env.example .env
# Edit .env with your API keys

# 3. Migrations
python manage.py migrate

# 4. Development
python manage.py runserver

# 5. Celery worker (separate terminal)
celery -A moncreneau_project worker -l info

Testing

# Run tests
pytest

# With coverage
pytest --cov=. --cov-report=html

# Specific tests
pytest tests/test_webhooks.py

Celery Configuration

moncreneau_project/celery.py

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'moncreneau_project.settings')

app = Celery('moncreneau_project')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

Docker (optional)

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["gunicorn", "moncreneau_project.wsgi:application", "--bind", "0.0.0.0:8000"]

Resources

• MONCRENEAU API Documentation
• Webhooks Guide
• Django Documentation
• Django REST Framework