Integration

One-Ping Python Integration Guide

Send multi-channel notifications from your Python application with a single API call. Works with Django, Flask, FastAPI, scripts, and async applications using aiohttp.

Getting Started with One-Ping in Python

One-Ping's REST API integrates seamlessly with Python applications of any kind. Whether you are building a Django web application, a Flask microservice, a FastAPI backend, a data processing script, or a command-line tool, adding multi-channel notifications takes just a few lines of code. No proprietary SDK is required -- the standard requests library is all you need.

This guide walks you through everything from basic usage with the requests library to building a reusable helper class, integrating with Django and Flask frameworks, implementing async notifications with aiohttp, and following best practices for error handling and production deployments.

Quick Start

Install the Requests Library

If you do not already have it, install the requests library. It is the de facto standard for HTTP requests in Python. Run pip install requests in your terminal or add it to your requirements.txt file.

Get Your API Key

Sign up at app.one-ping.com and generate an API key from the dashboard. Store it as an environment variable called ONE_PING_API_KEY. Configure your notification channels (Telegram, Email, Slack, Discord) in the One-Ping settings.

Send Your First Notification

Make a POST request to https://api.one-ping.com/send with a JSON body containing your message, the list of channels, and the recipient. See the code examples below for the exact implementation.

Basic Example with Requests

Here is the simplest way to send a multi-channel notification from Python using the requests library:

import os
import requests

def send_notification(message, channels, recipient):
    response = requests.post(
        'https://api.one-ping.com/send',
        headers={
            'Authorization': f'Bearer {os.environ["ONE_PING_API_KEY"]}',
            'Content-Type': 'application/json'
        },
        json={
            'message': message,
            'channels': channels,
            'recipient': recipient
        },
        timeout=10
    )
    response.raise_for_status()
    return response.json()

# Usage
result = send_notification(
    message='New order #1234 from John Doe - $99.00',
    channels=['telegram', 'email', 'slack'],
    recipient='[email protected]'
)
print(f'Notification sent: {result}')

Reusable One-Ping Helper Class

For production applications, wrap the API in a reusable class with built-in retry logic, timeout handling, and convenience methods for common channel combinations:

# one_ping.py
import os
import time
import logging
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logger = logging.getLogger(__name__)

class OnePing:
    def __init__(self, api_key=None, base_url='https://api.one-ping.com', timeout=10, max_retries=2):
        self.api_key = api_key or os.environ.get('ONE_PING_API_KEY')
        self.base_url = base_url
        self.timeout = timeout

        # Configure session with retry logic
        self.session = requests.Session()
        retry_strategy = Retry(
            total=max_retries,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount('https://', adapter)
        self.session.headers.update({
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        })

    def send(self, message, channels, recipient, metadata=None):
        payload = {
            'message': message,
            'channels': channels,
            'recipient': recipient
        }
        if metadata:
            payload['metadata'] = metadata

        try:
            response = self.session.post(
                f'{self.base_url}/send',
                json=payload,
                timeout=self.timeout
            )
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            logger.error(f'One-Ping notification failed: {e}')
            raise

    # Convenience methods
    def to_telegram(self, message, recipient):
        return self.send(message, ['telegram'], recipient)

    def to_slack(self, message, recipient):
        return self.send(message, ['slack'], recipient)

    def to_all(self, message, recipient):
        return self.send(message, ['telegram', 'email', 'slack', 'discord'], recipient)

# Usage
ping = OnePing()
ping.send(
    'Deployment complete: v2.1.0 deployed to production',
    ['telegram', 'slack'],
    '[email protected]'
)

Django Integration

Integrating One-Ping into a Django application is straightforward. You can create a utility module and call it from views, signals, or management commands. Here is a complete example using Django signals to send a notification when a new order is created:

# notifications/utils.py
from django.conf import settings
from one_ping import OnePing

# Initialize once, reuse across the app
ping = OnePing(api_key=settings.ONE_PING_API_KEY)

def notify_new_order(order):
    ping.send(
        message=f'New order #{order.id} from {order.customer_name} - ${order.total:.2f}',
        channels=['telegram', 'email', 'slack'],
        recipient=order.customer_email,
        metadata={'order_id': str(order.id), 'source': 'django'}
    )

# orders/signals.py
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order
from notifications.utils import notify_new_order

@receiver(post_save, sender=Order)
def order_created_handler(sender, instance, created, **kwargs):
    if created:
        try:
            notify_new_order(instance)
        except Exception as e:
            # Log but do not break the order flow
            import logging
            logging.getLogger(__name__).error(f'Notification failed: {e}')

Django settings tip: Add ONE_PING_API_KEY to your Django settings file, loaded from environment variables: ONE_PING_API_KEY = os.environ.get('ONE_PING_API_KEY'). For production, use django-environ or a similar package to manage secrets securely.

Flask Integration

In Flask applications, you can use One-Ping in route handlers, error handlers, or background tasks with Celery. Here is a Flask example with an error notification handler:

# app.py
from flask import Flask, request, jsonify
from one_ping import OnePing

app = Flask(__name__)
ping = OnePing()

@app.route('/api/contact', methods=['POST'])
def handle_contact_form():
    data = request.get_json()

    # Save to database...

    # Notify team about new contact form submission
    try:
        ping.send(
            message=f'New contact form:\nName: {data["name"]}\nEmail: {data["email"]}\nMessage: {data["message"]}',
            channels=['telegram', 'slack'],
            recipient='[email protected]'
        )
    except Exception:
        pass  # Non-critical, do not fail the request

    return jsonify({'success': True})

@app.errorhandler(500)
def handle_500(error):
    # Notify dev team about server errors
    ping.send(
        message=f'500 Error in {request.method} {request.path}:\n{str(error)}',
        channels=['telegram', 'discord'],
        recipient='[email protected]'
    )
    return jsonify({'error': 'Internal server error'}), 500

Async Support with aiohttp

For high-performance applications using asyncio (such as FastAPI, aiohttp servers, or async scripts), here is an async version of the One-Ping client that does not block the event loop:

# one_ping_async.py
import os
import aiohttp
import logging

logger = logging.getLogger(__name__)

class AsyncOnePing:
    def __init__(self, api_key=None, base_url='https://api.one-ping.com', timeout=10):
        self.api_key = api_key or os.environ.get('ONE_PING_API_KEY')
        self.base_url = base_url
        self.timeout = aiohttp.ClientTimeout(total=timeout)
        self._session = None

    async def _get_session(self):
        if self._session is None or self._session.closed:
            self._session = aiohttp.ClientSession(
                headers={
                    'Authorization': f'Bearer {self.api_key}',
                    'Content-Type': 'application/json'
                },
                timeout=self.timeout
            )
        return self._session

    async def send(self, message, channels, recipient, metadata=None):
        session = await self._get_session()
        payload = {
            'message': message,
            'channels': channels,
            'recipient': recipient
        }
        if metadata:
            payload['metadata'] = metadata

        async with session.post(f'{self.base_url}/send', json=payload) as response:
            if response.status != 200:
                error = await response.json()
                raise Exception(f'One-Ping error: {error.get("message", response.status)}')
            return await response.json()

    async def close(self):
        if self._session and not self._session.closed:
            await self._session.close()

# Usage with FastAPI
from fastapi import FastAPI

app = FastAPI()
ping = AsyncOnePing()

@app.post('/orders')
async def create_order(order: dict):
    # Save order to database...

    await ping.send(
        message=f'New order from {order["customer_name"]} - ${order["total"]}',
        channels=['telegram', 'email', 'slack'],
        recipient=order['customer_email']
    )
    return {'success': True}

Integration Capabilities

Sync and Async

Choose between synchronous requests for simple scripts and Django, or async aiohttp for FastAPI and high-concurrency applications.

Automatic Retries

The helper class uses urllib3's retry adapter with exponential backoff, handling transient errors and rate limits automatically.

Framework Agnostic

Works with Django, Flask, FastAPI, Pyramid, Bottle, or standalone scripts. No framework-specific dependencies required.

Celery Compatible

Offload notification sending to Celery workers for background processing. The helper class is pickle-safe and thread-safe.

Background Tasks with Celery

For high-traffic applications, send notifications asynchronously using Celery to avoid blocking your web request handlers:

# tasks.py
from celery import shared_task
from one_ping import OnePing

@shared_task(bind=True, max_retries=3)
def send_notification_task(self, message, channels, recipient, metadata=None):
    try:
        ping = OnePing()
        return ping.send(message, channels, recipient, metadata)
    except Exception as exc:
        self.retry(exc=exc, countdown=60 * self.request.retries)

# Usage in a view
send_notification_task.delay(
    message='Welcome to our platform!',
    channels=['email', 'telegram'],
    recipient='[email protected]'
)

Python version compatibility: The examples in this guide work with Python 3.7 and above. The async examples require Python 3.8+ for proper async/await support. Install aiohttp with pip install aiohttp for the async client.

Common Use Cases

Ready to simplify your notifications?

Start free with 100 messages/month. No credit card required.

Get started free