diff --git a/README.md b/README.md
index 99a87bc..3ef4606 100644
--- a/README.md
+++ b/README.md
@@ -1,45 +1,44 @@
# Python Status.io
A Python wrapper around the Status.io API.
[](https://pypi.python.org/pypi/statusio-python/)
-[](https://travis-ci.org/statusio/statusio-python)
## Introduction
This library provides a pure Python interface for the [Status.io API](http://docs.statusio.apiary.io/). It works with Python versions from 2.6+.
[Status.io](https://status.io) provides hosted system status pages.
## Installing
You can install statusio-python using::
$ pip install statusio-python
## Documentation
View the last release API documentation at: http://developers.status.io/
## Using
The library provides a Python wrapper around the Status.io API.
### API
The API is exposed via the `statusio.Api` class.
To create an instance of the `statusio.Api` with yout credentials:
>>> import statusio
>>> api = statusio.Api(api_id='api_id',
api_key='api_key')
-To your status page summary:
+To retrieve your status page summary:
>>> summary = api.StatusSummary('status_page_id')
>>> print(summary)
-There are many more API methods, to read the full API documentation::
+View the full API documentation::
$ pydoc statusio.Api
\ No newline at end of file
diff --git a/README.rst b/README.rst
index 237b179..2d19003 100644
--- a/README.rst
+++ b/README.rst
@@ -1,58 +1,56 @@
Python Status.io
================
A Python wrapper around the Status.io API.
-|Downloads| |Travis CI|
+|Downloads|
Introduction
------------
This library provides a pure Python interface for the `Status.io
API `__. It works with Python versions
from 2.6+.
-`Status.io `__ provides hosted system status pages.
+`Status.io `__ provides hosted system status pages.
Installing
----------
You can install statusio-python using::
$ pip install statusio-python
Documentation
-------------
View the last release API documentation at:
http://developers.status.io
Using
-----
The library provides a Python wrapper around the Status.io API.
API
~~~
The API is exposed via the ``statusio.Api`` class.
To create an instance of the ``statusio.Api`` with yout credentials:
>>> import statusio
>>> api = statusio.Api(api_id='api_id',
api_key='api_key')
-To your status page summary:
+To retrieve your status page summary:
>>> summary = api.StatusSummary('status_page_id')
>>> print(summary)
-There are many more API methods, to read the full API documentation::
+View the full API documentation::
$ pydoc statusio.Api
.. |Downloads| image:: https://img.shields.io/pypi/v/statusio-python.svg
:target: https://pypi.python.org/pypi/statusio-python/
-.. |Travis CI| image:: https://travis-ci.org/statusio/statusio-python.svg
- :target: https://travis-ci.org/statusio/statusio-python
diff --git a/setup.py b/setup.py
index d0573c0..59417e1 100644
--- a/setup.py
+++ b/setup.py
@@ -1,38 +1,38 @@
#!/usr/bin/env python
'''The setup and build script for the statusio-python library.'''
import os
from setuptools import setup, find_packages
def read(*paths):
"""Build a file path from *paths* and return the contents."""
with open(os.path.join(*paths), 'r') as f:
return f.read()
setup(
name='statusio-python',
- version='0.4',
+ version='0.5',
author='Status.io',
- author_email='support@status.io',
+ author_email='hello@status.io',
license='Apache License 2.0',
url='https://github.com/statusio/statusio-python',
keywords='status.io api statusio',
description='A Python wrapper around the Status.io API',
long_description=(read('README.rst')),
packages=find_packages(exclude=['tests*']),
install_requires=['future', 'requests'],
classifiers=[
'Development Status :: 5 - Production/Stable',
'Intended Audience :: Developers',
'License :: OSI Approved :: Apache Software License',
'Operating System :: OS Independent',
'Topic :: Software Development :: Libraries :: Python Modules',
'Topic :: Internet',
'Programming Language :: Python',
'Programming Language :: Python :: 2',
'Programming Language :: Python :: 2.7',
],
)
diff --git a/statusio/api.py b/statusio/api.py
index b06251e..39c2f14 100644
--- a/statusio/api.py
+++ b/statusio/api.py
@@ -1,1007 +1,1003 @@
#!/usr/bin/env python
"""A library that provides a Python interface to the Status.io API"""
from __future__ import division
from __future__ import print_function
import sys
import gzip
import time
import types
import base64
import re
import datetime
from calendar import timegm
import requests
import io
from past.utils import old_div
try:
# python 3
from urllib.parse import urlparse, urlunparse, urlencode
from urllib.request import urlopen
from urllib.request import __version__ as urllib_version
except ImportError:
from urlparse import urlparse, urlunparse
from urllib2 import urlopen
from urllib import urlencode
from urllib import __version__ as urllib_version
from statusio import (__version__, json)
class Api(object):
"""A python interface into the Status.io API
Example usage:
Simples example
>>> import statusio
>>> api = statusio.Api(API_ID, API_KEY)
>>> result = api.StatusSummary(STATUSPAGE_ID)
>>> print(result)
There are many other methods, including:
>>> api.ComponentList(statuspage_id)
>>> api.ComponentStatusUpdate(statuspage_id, components, containers, details, current_status)
>>> api.IncidentList(statuspage_id)
>>> api.IncidentMessage(statuspage_id, message_id)
>>> api.IncidentCreate(statuspage_id, components, containers, incident_name, incident_details, current_status, current_state, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0, all_infrastructure_affected=0)
>>> api.IncidentUpdate(statuspage_id, incident_id, incident_details, current_status, current_state, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0)
>>> api.IncidentResolve(statuspage_id, incident_id, incident_details, current_status, current_state, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0)
>>> api.IncidentDelete(statuspage_id, incident_id)
>>> api.MaintenanceList(statuspage_id)
>>> api.MaintenanceMessage(statuspage_id, message_id)
- >>> api.MaintenanceSchedule(statuspage_id, components, containers, maintenance_name, maintenance_details, date_planned_start, time_planned_start, date_planned_end, time_planned_end, automation=0, all_infrastructure_affected=0, maintenance_notify_now=0, maintenance_notify_1_hr=0, maintenance_notify_24_hr=0, maintenance_notify_72_hr=0)
+ >>> api.MaintenanceSchedule(statuspage_id, infrastructure_affected, maintenance_name, maintenance_details, date_planned_start, time_planned_start, date_planned_end, time_planned_end, automation=0, all_infrastructure_affected=0, maintenance_notify_now=0, maintenance_notify_1_hr=0, maintenance_notify_24_hr=0, maintenance_notify_72_hr=0)
>>> api.MaintenanceStart(statuspage_id, maintenance_id, maintenance_details, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0)
>>> api.MaintenanceUpdate(statuspage_id, maintenance_id, maintenance_details, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0)
>>> api.MaintenanceFinish(statuspage_id, maintenance_id, maintenance_details, notify_email=0, notify_sms=0, notify_webhook=0, social=0, irc=0, hipchat=0, slack=0)
>>> api.MaintenanceDelete(statuspage_id, maintenance_id)
>>> api.MetricUpdate(statuspage_id, metric_id, day_avg, day_start, day_dates, day_values, week_avg, week_start, week_dates, week_values, month_avg, month_start, month_dates, month_values)
>>> api.StatusSummary(statuspage_id)
>>> api.SubscriberList(statuspage_id)
>>> api.SubscriberAdd(statuspage_id, method, address, silent=1, granular='')
>>> api.SubscriberUpdate(statuspage_id, subscriber_id, address, granular='')
>>> api.SubscriberRemove(statuspage_id, subscriber_id)
"""
def __init__(self,
api_id,
api_key,
version=2,
base_url='https://api.status.io'
):
"""Instantiate a new statusio.Api object.
Args:
api_id:
Your Status.io API ID.
api_key:
Your Status.io API KEY.
version:
API version number. [Optional]
base_url:
API base URL. [Optional]
"""
self._api_id = api_id
self._api_key = api_key
self.base_url = '%s/v%d' % (base_url, version)
def ComponentList(self, statuspage_id):
"""List all components.
Args:
statuspage_id:
Status page ID
Returns:
A JSON object.
"""
url = '%s/component/list/%s' % (self.base_url, statuspage_id)
resp = self._RequestUrl(url, 'GET')
data = json.loads(resp.content.decode('utf-8'))
return data
def ComponentStatusUpdate(self,
statuspage_id,
components,
containers,
details,
current_status):
"""Update the status of a component on the fly without creating an incident or maintenance.
Args:
statuspage_id:
Status page ID
components:
ID of each affected component
containers:
ID of each affected container
details:
A brief message describing this update
current_status:
Any numeric status code.
Returns:
A JSON object.
"""
url = '%s/component/status/update' % self.base_url
resp = self._RequestUrl(url, 'POST', data={
'statuspage_id': statuspage_id,
'components': components,
'containers': containers,
'details': details,
'current_status': current_status
})
data = json.loads(resp.content.decode('utf-8'))
return data
def IncidentList(self, statuspage_id):
"""List all active and resolved incidents.
Args:
statuspage_id:
Status page ID
Returns:
A JSON object.
"""
url = '%s/incident/list/%s' % (self.base_url, statuspage_id)
resp = self._RequestUrl(url, 'GET')
data = json.loads(resp.content.decode('utf-8'))
return data
def IncidentMessage(self,
statuspage_id,
message_id):
"""Display incident message.
Args:
statuspage_id:
Status page ID
message_id:
Message ID
Returns:
A JSON object.
"""
url = '%s/incident/message/%s/%s' % (self.base_url,
statuspage_id, message_id)
resp = self._RequestUrl(url, 'GET')
data = json.loads(resp.content.decode('utf-8'))
return data
def IncidentCreate(self,
statuspage_id,
components,
containers,
incident_name,
incident_details,
current_status,
current_state,
notify_email=0,
notify_sms=0,
notify_webhook=0,
social=0,
irc=0,
hipchat=0,
slack=0,
all_infrastructure_affected=0):
"""Create a new incident.
Args:
statuspage_id:
Status page ID
components:
ID of each affected component
containers:
ID of each affected container
incident_name:
A descriptive title for the incident
incident_details:
Message describing this incident
current_status:
The status of the components and containers affected by this incident
current_state:
The state of this incident
notify_email:
Notify email subscribers (1 = Send notification)
notify_sms:
Notify SMS subscribers (1 = Send notification)
notify_webhook:
Notify webhook subscribers (1 = Send notification)
social:
Automatically Tweet this update. (1 = Send Tweet)
irc:
Notify IRC channel (1 = Send notification)
hipchat:
Notify HipChat room (1 = Send notification)
slack:
Notify Slack channel (1 = Send notification)
Returns:
A JSON object.
"""
url = '%s/incident/create' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'components': components,
'containers': containers,
'incident_name': incident_name,
'incident_details': incident_details,
'current_status': current_status,
'current_state': current_state,
'notify_email': notify_email,
'notify_sms': notify_sms,
'notify_webhook': notify_webhook,
'social': social,
'irc': irc,
'hipchat': hipchat,
'slack': slack,
'all_infrastructure_affected': all_infrastructure_affected
})
data = json.loads(resp.content.decode('utf-8'))
return data
def IncidentUpdate(self,
statuspage_id,
incident_id,
incident_details,
current_status,
current_state,
notify_email=0,
notify_sms=0,
notify_webhook=0,
social=0,
irc=0,
hipchat=0,
slack=0):
"""Update an existing incident
Args:
statuspage_id:
Status page ID
incident_id:
Incident ID
incident_details:
Message describing this incident
current_status:
The status of the components and containers affected by this incident
current_state:
The state of this incident
notify_email:
Notify email subscribers (1 = Send notification)
notify_sms:
Notify SMS subscribers (1 = Send notification)
notify_webhook:
Notify webhook subscribers (1 = Send notification)
social:
Automatically Tweet this update. (1 = Send Tweet)
irc:
Notify IRC channel (1 = Send notification)
hipchat:
Notify HipChat room (1 = Send notification)
slack:
Notify Slack channel (1 = Send notification)
Returns:
A JSON object.
"""
url = '%s/incident/update' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'incident_id': incident_id,
'incident_details': incident_details,
'current_status': current_status,
'current_state': current_state,
'notify_email': notify_email,
'notify_sms': notify_sms,
'notify_webhook': notify_webhook,
'social': social,
'irc': irc,
'hipchat': hipchat,
'slack': slack
})
data = json.loads(resp.content.decode('utf-8'))
return data
def IncidentResolve(self,
statuspage_id,
incident_id,
incident_details,
current_status,
current_state,
notify_email=0,
notify_sms=0,
notify_webhook=0,
social=0,
irc=0,
hipchat=0,
slack=0):
"""Resolve an existing incident. The incident will be shown in the history instead of on the main page.
Args:
statuspage_id:
Status page ID
incident_id:
Incident ID
incident_details:
Message describing this incident
current_status:
The status of the components and containers affected by this incident
current_state:
The state of this incident
notify_email:
Notify email subscribers (1 = Send notification)
notify_sms:
Notify SMS subscribers (1 = Send notification)
notify_webhook:
Notify webhook subscribers (1 = Send notification)
social:
Automatically Tweet this update. (1 = Send Tweet)
irc:
Notify IRC channel (1 = Send notification)
hipchat:
Notify HipChat room (1 = Send notification)
slack:
Notify Slack channel (1 = Send notification)
Returns:
A JSON object.
"""
url = '%s/incident/resolve' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'incident_id': incident_id,
'incident_details': incident_details,
'current_status': current_status,
'current_state': current_state,
'notify_email': notify_email,
'notify_sms': notify_sms,
'notify_webhook': notify_webhook,
'social': social,
'irc': irc,
'hipchat': hipchat,
'slack': slack
})
data = json.loads(resp.content.decode('utf-8'))
return data
def IncidentDelete(self,
statuspage_id,
incident_id):
"""Delete an existing incident. The incident will be deleted forever and cannot be recovered.
Args:
statuspage_id:
Status page ID
incident_id:
Incident ID
Returns:
A JSON object.
"""
url = '%s/incident/delete' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'incident_id': incident_id
})
data = json.loads(resp.content.decode('utf-8'))
return data
def MaintenanceList(self, statuspage_id):
"""List all active, resolved and upcoming maintenances
Args:
statuspage_id:
Status page ID
Returns:
A JSON object.
"""
url = '%s/maintenance/list/%s' % (self.base_url, statuspage_id)
resp = self._RequestUrl(url, 'GET')
data = json.loads(resp.content.decode('utf-8'))
return data
def MaintenanceMessage(self,
statuspage_id,
message_id):
"""Display maintenance message
Args:
statuspage_id:
Status page ID
incident_id:
Message ID
Returns:
A JSON object.
"""
url = '%s/maintenance/message/%s/%s' % (
self.base_url, statuspage_id, message_id)
resp = self._RequestUrl(url, 'GET')
data = json.loads(resp.content.decode('utf-8'))
return data
def MaintenanceSchedule(self,
statuspage_id,
- components,
- containers,
+ infrastructure_affected,
maintenance_name,
maintenance_details,
date_planned_start,
time_planned_start,
date_planned_end,
time_planned_end,
automation=0,
all_infrastructure_affected=0,
maintenance_notify_now=0,
maintenance_notify_1_hr=0,
maintenance_notify_24_hr=0,
maintenance_notify_72_hr=0):
"""Schedule a new maintenance
Args:
statuspage_id:
Status page ID
- components:
- ID of each affected component
- containers:
- ID of each affected container
+ infrastructure_affected:
+ ID of each affected component and container combo
maintenance_name:
A descriptive title for this maintenance
maintenance_details:
Message describing this maintenance
date_planned_start:
Date maintenance is expected to start
time_planned_start:
Time maintenance is expected to start
date_planned_end:
Date maintenance is expected to end
time_planned_end:
Time maintenance is expected to end
automation:
Automatically start and end the maintenance (default = 0)
all_infrastructure_affected:
Affect all components and containers (default = 0)
maintenance_notify_now:
Notify subscribers now (1 = Send notification)
maintenance_notify_1_hr:
Notify subscribers 1 hour before scheduled maintenance start time (1 = Send notification)
maintenance_notify_24_hr:
Notify subscribers 24 hours before scheduled maintenance start time (1 = Send notification)
maintenance_notify_72_hr:
Notify subscribers 72 hours before scheduled maintenance start time (1 = Send notification)
Returns:
A JSON object.
"""
url = '%s/maintenance/schedule' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
- 'components': components,
- 'containers': containers,
+ 'infrastructure_affected': infrastructure_affected,
'maintenance_name': maintenance_name,
'maintenance_details': maintenance_details,
'date_planned_start': date_planned_start,
'time_planned_start': time_planned_start,
'date_planned_end': date_planned_end,
'time_planned_end': time_planned_end,
'automation': automation,
'all_infrastructure_affected': all_infrastructure_affected,
'maintenance_notify_now': maintenance_notify_now,
'maintenance_notify_1_hr': maintenance_notify_1_hr,
'maintenance_notify_24_hr': maintenance_notify_24_hr,
'maintenance_notify_72_hr': maintenance_notify_72_hr
})
data = json.loads(resp.content.decode('utf-8'))
return data
def MaintenanceStart(self,
statuspage_id,
maintenance_id,
maintenance_details,
notify_email=0,
notify_sms=0,
notify_webhook=0,
social=0,
irc=0,
hipchat=0,
slack=0):
"""Begin a scheduled maintenance now
Args:
statuspage_id:
Status page ID
maintenance_id:
Maintenance ID
maintenance_details:
Message describing this maintenance update
notify_email:
Notify email subscribers (1 = Send notification)
notify_sms:
Notify SMS subscribers (1 = Send notification)
notify_webhook:
Notify webhook subscribers (1 = Send notification)
social:
Automatically Tweet this update. (1 = Send Tweet)
irc:
Notify IRC channel (1 = Send notification)
hipchat:
Notify HipChat room (1 = Send notification)
slack:
Notify Slack channel (1 = Send notification)
Returns:
A JSON object.
"""
url = '%s/maintenance/start' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'maintenance_id': maintenance_id,
'maintenance_details': maintenance_details,
'notify_email': notify_email,
'notify_sms': notify_sms,
'notify_webhook': notify_webhook,
'social': social,
'irc': irc,
'hipchat': hipchat,
'slack': slack
})
data = json.loads(resp.content.decode('utf-8'))
return data
def MaintenanceUpdate(self,
statuspage_id,
maintenance_id,
maintenance_details,
notify_email=0,
notify_sms=0,
notify_webhook=0,
social=0,
irc=0,
hipchat=0,
slack=0):
"""Update an active maintenance
Args:
statuspage_id:
Status page ID
maintenance_id:
Maintenance ID
maintenance_details:
Message describing this maintenance update
notify_email:
Notify email subscribers (1 = Send notification)
notify_sms:
Notify SMS subscribers (1 = Send notification)
notify_webhook:
Notify webhook subscribers (1 = Send notification)
social:
Automatically Tweet this update. (1 = Send Tweet)
irc:
Notify IRC channel (1 = Send notification)
hipchat:
Notify HipChat room (1 = Send notification)
slack:
Notify Slack channel (1 = Send notification)
Returns:
A JSON object.
"""
url = '%s/maintenance/update' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'maintenance_id': maintenance_id,
'maintenance_details': maintenance_details,
'notify_email': notify_email,
'notify_sms': notify_sms,
'notify_webhook': notify_webhook,
'social': social,
'irc': irc,
'hipchat': hipchat,
'slack': slack
})
data = json.loads(resp.content.decode('utf-8'))
return data
def MaintenanceFinish(self,
statuspage_id,
maintenance_id,
maintenance_details,
notify_email=0,
notify_sms=0,
notify_webhook=0,
social=0,
irc=0,
hipchat=0,
slack=0):
"""Close an active maintenance. The maintenance will be moved to the history.
Args:
statuspage_id:
Status page ID
maintenance_id:
Maintenance ID
maintenance_details:
Message describing this maintenance update
notify_email:
Notify email subscribers (1 = Send notification)
notify_sms:
Notify SMS subscribers (1 = Send notification)
notify_webhook:
Notify webhook subscribers (1 = Send notification)
social:
Automatically Tweet this update. (1 = Send Tweet)
irc:
Notify IRC channel (1 = Send notification)
hipchat:
Notify HipChat room (1 = Send notification)
slack:
Notify Slack channel (1 = Send notification)
Returns:
A JSON object.
"""
url = '%s/maintenance/finish' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'maintenance_id': maintenance_id,
'maintenance_details': maintenance_details,
'notify_email': notify_email,
'notify_sms': notify_sms,
'notify_webhook': notify_webhook,
'social': social,
'irc': irc,
'hipchat': hipchat,
'slack': slack
})
data = json.loads(resp.content.decode('utf-8'))
return data
def MaintenanceDelete(self,
statuspage_id,
maintenance_id):
"""Delete an existing maintenance. The maintenance will be deleted forever and cannot be recovered.
Args:
statuspage_id:
Status page ID
maintenance_id:
Maintenance ID
Returns:
A JSON object.
"""
url = '%s/maintenance/delete' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'maintenance_id': maintenance_id
})
data = json.loads(resp.content.decode('utf-8'))
return data
def MetricUpdate(self,
statuspage_id,
metric_id,
day_avg,
day_start,
day_dates,
day_values,
week_avg,
week_start,
week_dates,
week_values,
month_avg,
month_start,
month_dates,
month_values):
"""Update custom metric data
Args:
statuspage_id:
Status page ID
metric_id:
Metric ID
day_avg:
Average value for past 24 hours
day_start:
UNIX timestamp for start of metric timeframe
day_dates:
An array of timestamps for the past 24 hours (2014-03-28T05:43:00+00:00)
day_values:
An array of values matching the timestamps (Must be 24 values)
week_avg:
Average value for past 7 days
week_start:
UNIX timestamp for start of metric timeframe
week_dates:
An array of timestamps for the past 7 days (2014-03-28T05:43:00+00:00)
week_values:
An array of values matching the timestamps (Must be 7 values)
month_avg:
Average value for past 30 days
month_start:
UNIX timestamp for start of metric timeframe
month_dates:
An array of timestamps for the past 30 days (2014-03-28T05:43:00+00:00)
month_values:
An array of values matching the timestamps (Must be 30 values)
Returns:
A JSON object.
"""
url = '%s/metric/update' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'metric_id': metric_id,
'day_avg': day_avg,
'day_start': day_start,
'day_dates': day_dates,
'day_values': day_values,
'week_avg': week_avg,
'week_start': week_start,
'week_dates': week_dates,
'week_values': week_values,
'month_avg': month_avg,
'month_start': month_start,
'month_dates': month_dates,
'month_values': month_values
})
data = json.loads(resp.content.decode('utf-8'))
return data
def StatusSummary(self, statuspage_id):
"""Show the summary status for all components and containers
Args:
statuspage_id:
Status page ID
Returns:
A JSON object.
"""
url = '%s/status/summary/%s' % (self.base_url, statuspage_id)
resp = self._RequestUrl(url, 'GET')
data = json.loads(resp.content.decode('utf-8'))
return data
def SubscriberList(self, statuspage_id):
"""List all subscribers
Args:
statuspage_id:
Status page ID
Returns:
A JSON object.
"""
url = '%s/subscriber/list/%s' % (self.base_url, statuspage_id)
resp = self._RequestUrl(url, 'GET')
data = json.loads(resp.content.decode('utf-8'))
return data
def SubscriberAdd(self,
statuspage_id,
method,
address,
silent=1,
granular=''):
"""Add a new subscriber
Args:
statuspage_id:
Status page ID
method:
Communication method of subscriber. Valid methods are `email`, `sms` or `webhook`
address:
Subscriber address (SMS number must include country code ie. +1)
silent:
Supress the welcome message (1 = Do not send notification)
granular:
List of component_container combos
Returns:
A JSON object.
"""
url = '%s/subscriber/add' % self.base_url
resp = self._RequestUrl(url, 'POST', {
'statuspage_id': statuspage_id,
'method': method,
'address': address,
'silent': silent,
'granular': granular,
})
data = json.loads(resp.content.decode('utf-8'))
return data
def SubscriberUpdate(self,
statuspage_id,
subscriber_id,
address,
granular=''):
"""Update existing subscriber
Args:
statuspage_id:
Status page ID
subscriber_id:
SubscriberAdd ID
address:
Subscriber address (SMS number must include country code ie. +1)
granular:
List of component_container combos
Returns:
A JSON object.
"""
url = '%s/subscriber/update' % self.base_url
resp = self._RequestUrl(url, 'PATCH', {
'statuspage_id': statuspage_id,
'subscriber_id': subscriber_id,
'address': address,
'granular': granular,
})
data = json.loads(resp.content.decode('utf-8'))
return data
def SubscriberRemove(self,
statuspage_id,
subscriber_id):
"""Delete subscriber
Args:
statuspage_id:
Status page ID
subscriber_id:
Subscriber ID
Returns:
A JSON object.
"""
url = '%s/subscriber/remove/%s/%s' % (
self.base_url, statuspage_id, subscriber_id)
resp = self._RequestUrl(url, 'DELETE')
data = json.loads(resp.content.decode('utf-8'))
return data
def _BuildUrl(self, url, path_elements=None, extra_params=None):
# Break url into constituent parts
(scheme, netloc, path, params, query, fragment) = urlparse(url)
# Add any additional path elements to the path
if path_elements:
# Filter out the path elements that have a value of None
p = [i for i in path_elements if i]
if not path.endswith('/'):
path += '/'
path += '/'.join(p)
# Add any additional query parameters to the query string
if extra_params and len(extra_params) > 0:
extra_query = self._EncodeParameters(extra_params)
# Add it to the existing query
if query:
query += '&' + extra_query
else:
query = extra_query
# Return the rebuilt URL
return urlunparse((scheme, netloc, path, params, query, fragment))
def _Encode(self, s):
if self._input_encoding:
return str(s, self._input_encoding).encode('utf-8')
else:
return str(s).encode('utf-8')
def _EncodeParameters(self, parameters):
"""Return a string in key=value&key=value form.
Values of None are not included in the output string.
Args:
parameters:
A dict of (key, value) tuples, where value is encoded as
specified by self._encoding
Returns:
A URL-encoded string in "key=value&key=value" form
"""
if parameters is None:
return None
else:
return urlencode(dict([(k, self._Encode(v)) for k, v in list(
parameters.items()) if v is not None]))
def _EncodePostData(self, post_data):
"""Return a string in key=value&key=value form.
Values are assumed to be encoded in the format specified by self._encoding,
and are subsequently URL encoded.
Args:
post_data:
A dict of (key, value) tuples, where value is encoded as
specified by self._encoding
Returns:
A URL-encoded string in "key=value&key=value" form
"""
if post_data is None:
return None
else:
return urlencode(dict([(k, self._Encode(v))
for k, v in list(post_data.items())]))
def _RequestUrl(self, url, verb, data=None):
"""Request a url.
Args:
url:
The web location we want to retrieve.
verb:
Either POST or GET.
data:
A dict of (str, unicode) key/value pairs.
Returns:
A JSON object.
"""
if verb == 'POST':
try:
return requests.post(
url,
data=json.dumps(data),
headers={
'x-api-id': self._api_id,
'x-api-key': self._api_key,
'content-type': 'application/json'
}
)
except requests.RequestException as e:
print('Error: ' + str(e))
elif verb == 'GET':
url = self._BuildUrl(url, extra_params=data)
try:
return requests.get(
url,
headers={
'x-api-id': self._api_id,
'x-api-key': self._api_key
}
)
except requests.RequestException as e:
print('Error: ' + str(e))
elif verb == 'PATCH':
try:
return requests.patch(
url,
data=json.dumps(data),
headers={
'x-api-id': self._api_id,
'x-api-key': self._api_key,
'content-type': 'application/json'
}
)
except requests.RequestException as e:
print('Error: ' + str(e))
elif verb == 'DELETE':
url = self._BuildUrl(url, extra_params=data)
try:
return requests.delete(
url,
headers={
'x-api-id': self._api_id,
'x-api-key': self._api_key,
'content-type': 'application/json'
}
)
except requests.RequestException as e:
print('Error: ' + str(e))
return 0
diff --git a/tests/test_api.py b/tests/test_api.py
index 49439c6..a8e19a9 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -1,340 +1,340 @@
# encoding: utf-8
import os
import time
import unittest
import statusio
-API_ID = os.environ.get('API_ID')
-API_KEY = os.environ.get('API_KEY')
-STATUSPAGE_ID = os.environ.get('STATUSPAGE_ID')
-COMPONENTS = [os.environ.get('COMPONENT')]
-CONTAINERS = [os.environ.get('CONTAINER')]
-METRIC_ID = os.environ.get('METRIC_ID')
+API_ID = ''
+API_KEY = ''
+STATUSPAGE_ID = '568d8a3e3cada8c2490000dd'
+COMPONENTS = '568d8a3e3cada8c2490000ed'
+CONTAINERS = '568d8a3e3cada8c2490000ec'
+COMPONENT_CONTAINER_COMBO = '568d8a3e3cada8c2490000ed-568d8a3e3cada8c2490000ec'
+METRIC_ID = '568d8ab5efe35d412f0006f8'
ID1 = ''
ID2 = ''
class ApiTest(unittest.TestCase):
def setUp(self):
self._api = statusio.Api(API_ID, API_KEY)
# STATUS
def testStatus1Summary(self):
# Test the statusio.StatusSummary method
print('Testing StatusSummary')
data = self._api.StatusSummary(STATUSPAGE_ID)
self.assertEqual(data['status']['error'], 'no')
# SUBSCRIBER
def testSubscriber1Add(self):
# Test the statusio.SubscriberAdd method
print('Testing SubscriberAdd')
global ID1, ID2
data = self._api.SubscriberAdd(
- STATUSPAGE_ID, "email", "phillip.j.fry@planet-express12.com")
+ STATUSPAGE_ID, "email", "test@example.com")
self.assertEqual(data['status']['error'], 'no')
ID1 = data['subscriber_id']
print(data['subscriber_id'])
print(ID1)
def testSubscriber2List(self):
# Test the statusio.SubscriberList method
print('Testing SubscriberList')
global ID1, ID2
data = self._api.SubscriberList(STATUSPAGE_ID)
self.assertEqual(data['status']['error'], 'no')
self.assertEqual(data['result']['email'][0]['_id'], ID1)
def testSubscriber3Update(self):
# Test the statusio.SubscriberUpdate method
print('Testing SubscriberUpdate')
global ID1, ID2
data = self._api.SubscriberUpdate(
- STATUSPAGE_ID, ID1, "phillip.j.fry.1@planet-express.com")
+ STATUSPAGE_ID, ID1, "test@example.com")
self.assertEqual(data['status']['error'], 'no')
def testSubscriber4Remove(self):
# Test the statusio.SubscriberRemove method
print('Testing SubscriberRemove')
global ID1, ID2
data = self._api.SubscriberRemove(STATUSPAGE_ID, ID1)
self.assertEqual(data['status']['error'], 'no')
# MAINTENANCE
def testMaintenance1Schedule(self):
# Test the statusio.MaintenanceSchedule method
print('Testing MaintenanceSchedule')
global ID1, ID2
data = self._api.MaintenanceSchedule(
STATUSPAGE_ID,
- COMPONENTS,
- CONTAINERS,
+ COMPONENT_CONTAINER_COMBO,
'Autotest',
'Autotest Description',
'2018/12/31',
'23:59',
'2019/01/01',
'23:59')
self.assertEqual(data['status']['error'], 'no')
ID1 = data['result']
def testMaintenance2List(self):
# Test the statusio.MaintenanceList method
print('Testing MaintenanceList')
global ID1, ID2
data = self._api.MaintenanceList(STATUSPAGE_ID)
self.assertEqual(data['status']['error'], 'no')
self.assertEqual(
data['result']['upcoming_maintenances'][0]['_id'], ID1)
ID2 = data['result']['upcoming_maintenances'][0]['messages'][0]['_id']
def testMaintenance3Message(self):
# Test the statusio.MaintenanceMessage method
print('Testing MaintenanceMessage')
global ID1, ID2
data = self._api.MaintenanceMessage(STATUSPAGE_ID, ID2)
self.assertEqual(data['status']['error'], 'no')
def testMaintenance4Start(self):
# Test the statusio.MaintenanceStart method
print('Testing MaintenanceStart')
global ID1, ID2
data = self._api.MaintenanceStart(
STATUSPAGE_ID, ID1, 'Autotest details')
self.assertEqual(data['status']['error'], 'no')
def testMaintenance5Update(self):
# Test the statusio.MaintenanceUpdate method
print('Testing MaintenanceUpdate')
global ID1, ID2
data = self._api.MaintenanceUpdate(
STATUSPAGE_ID, ID1, 'Autotest details update')
self.assertEqual(data['status']['error'], 'no')
def testMaintenance7Finish(self):
# Test the statusio.MaintenanceFinish method
print('Testing MaintenanceFinish')
global ID1, ID2
data = self._api.MaintenanceFinish(
STATUSPAGE_ID, ID1, 'Autotest details finish')
self.assertEqual(data['status']['error'], 'no')
def testMaintenance8Delete(self):
# Test the statusio.MaintenanceDelete method
print('Testing MaintenanceDelete')
global ID1, ID2
data = self._api.MaintenanceDelete(STATUSPAGE_ID, ID1)
self.assertEqual(data['status']['error'], 'no')
# INCIDENT
def testIncident1Create(self):
# Test the statusio.IncidentCreate method
print('Testing IncidentCreate')
global ID1, ID2
data = self._api.IncidentCreate(
STATUSPAGE_ID,
COMPONENTS,
CONTAINERS,
'Autotest',
'Autotest details',
300,
100)
self.assertEqual(data['status']['error'], 'no')
ID1 = data['result']
def testIncident2List(self):
# Test the statusio.IncidentList method
print('Testing IncidentList')
global ID1, ID2
data = self._api.IncidentList(STATUSPAGE_ID)
self.assertEqual(data['status']['error'], 'no')
self.assertEqual(data['result']['active_incidents'][0]['_id'], ID1)
ID2 = data['result']['active_incidents'][0]['messages'][0]['_id']
def testIncident3Message(self):
# Test the statusio.IncidentMessage method
print('Testing IncidentMessage')
global ID1, ID2
data = self._api.IncidentMessage(STATUSPAGE_ID, ID2)
self.assertEqual(data['status']['error'], 'no')
def testIncident5Update(self):
# Test the statusio.IncidentUpdate method
print('Testing IncidentUpdate')
global ID1, ID2
data = self._api.IncidentUpdate(
STATUSPAGE_ID, ID1, 'Autotest details update', 300, 100)
self.assertEqual(data['status']['error'], 'no')
def testIncident7Resolve(self):
# Test the statusio.IncidentResolve method
print('Testing IncidentResolve')
global ID1, ID2
data = self._api.IncidentResolve(
STATUSPAGE_ID, ID1, 'Autotest details resolve', 300, 100)
self.assertEqual(data['status']['error'], 'no')
def testIncident8Delete(self):
# Test the statusio.IncidentDelete method
print('Testing IncidentDelete')
global ID1, ID2
data = self._api.IncidentDelete(STATUSPAGE_ID, ID1)
self.assertEqual(data['status']['error'], 'no')
# METRIC
def testMetric1Update(self):
# Test the statusio.MetricUpdate method
print('Testing MetricUpdate')
global ID1, ID2
data = self._api.MetricUpdate(STATUSPAGE_ID,
METRIC_ID,
20.69,
1395981878000,
["2014-03-28T05:43:00+00:00",
"2014-03-28T06:43:00+00:00",
"2014-03-28T07:43:00+00:00",
"2014-03-28T08:43:00+00:00",
"2014-03-28T09:43:00+00:00",
"2014-03-28T10:43:00+00:00",
"2014-03-28T11:43:00+00:00",
"2014-03-28T12:43:00+00:00",
"2014-03-28T13:43:00+00:00",
"2014-03-28T14:43:00+00:00",
"2014-03-28T15:43:00+00:00",
"2014-03-28T16:43:00+00:00",
"2014-03-28T17:43:00+00:00",
"2014-03-28T18:43:00+00:00",
"2014-03-28T19:43:00+00:00",
"2014-03-28T20:43:00+00:00",
"2014-03-28T21:43:00+00:00",
"2014-03-28T22:43:00+00:00",
"2014-03-28T23:43:00+00:00",
"2014-03-29T00:43:00+00:00",
"2014-03-29T01:43:00+00:00",
"2014-03-29T02:43:00+00:00",
"2014-03-29T03:43:00+00:00"],
[20.70,
20.00,
19.20,
19.80,
19.90,
20.10,
21.40,
23.00,
27.40,
28.70,
27.50,
29.30,
28.50,
27.20,
28.60,
28.70,
25.90,
23.40,
22.40,
21.40,
19.80,
19.50,
20.00],
20.07,
1395463478000,
["2014-03-22T04:43:00+00:00",
"2014-03-23T04:43:00+00:00",
"2014-03-24T04:43:00+00:00",
"2014-03-25T04:43:00+00:00",
"2014-03-26T04:43:00+00:00",
"2014-03-27T04:43:00+00:00",
"2014-03-28T04:43:00+00:00"],
[23.10,
22.10,
22.20,
22.30,
22.10,
18.70,
17.00],
10.63,
1393476280000,
["2014-02-28T04:43:00+00:00",
"2014-03-01T04:43:00+00:00",
"2014-03-02T04:43:00+00:00",
"2014-03-03T04:43:00+00:00",
"2014-03-04T04:43:00+00:00",
"2014-03-05T04:43:00+00:00",
"2014-03-06T04:43:00+00:00",
"2014-03-07T04:43:00+00:00",
"2014-03-08T04:43:00+00:00",
"2014-03-09T04:43:00+00:00",
"2014-03-10T04:43:00+00:00",
"2014-03-11T04:43:00+00:00",
"2014-03-12T04:43:00+00:00",
"2014-03-13T04:43:00+00:00",
"2014-03-14T04:43:00+00:00",
"2014-03-15T04:43:00+00:00",
"2014-03-16T04:43:00+00:00",
"2014-03-17T04:43:00+00:00",
"2014-03-18T04:43:00+00:00",
"2014-03-19T04:43:00+00:00",
"2014-03-20T04:43:00+00:00",
"2014-03-21T04:43:00+00:00",
"2014-03-22T04:43:00+00:00",
"2014-03-23T04:43:00+00:00",
"2014-03-24T04:43:00+00:00",
"2014-03-25T04:43:00+00:00",
"2014-03-26T04:43:00+00:00",
"2014-03-27T04:43:00+00:00",
"2014-03-28T04:43:00+00:00"],
[0.00,
0.00,
0.00,
0.00,
0.00,
0.00,
0.00,
0.00,
0.00,
0.00,
0.00,
0.00,
18.50,
18.60,
18.40,
16.60,
16.80,
17.90,
19.90,
21.30,
22.80,
20.00,
17.30,
19.10,
21.50,
22.40,
22.50,
22.00,
21.80])
self.assertEqual(data['status']['error'], 'no')
# COMPONENT
def testComponent1List(self):
# Test the statusio.ComponentList method
print('Testing ComponentList')
global ID1, ID2
data = self._api.ComponentList(STATUSPAGE_ID)
self.assertEqual(data['status']['error'], 'no')
def testComponent2StatusUpdate(self):
# Test the statusio.ComponentStatusUpdate method
print('Testing ComponentStatusUpdate')
global ID1, ID2
data = self._api.ComponentStatusUpdate(
STATUSPAGE_ID, COMPONENTS, CONTAINERS, 'Test status', 300)
self.assertEqual(data['status']['error'], 'no')