Louis Vézina
5 years ago
parent
1967582020
commit
777913bd40
Binary file not shown.
@ -1,175 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# Copyright (C) 2019 Chris Caron <lead2gold@gmail.com>
|
||||
# All rights reserved.
|
||||
#
|
||||
# This code is licensed under the MIT License.
|
||||
#
|
||||
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
# of this software and associated documentation files(the "Software"), to deal
|
||||
# in the Software without restriction, including without limitation the rights
|
||||
# to use, copy, modify, merge, publish, distribute, sublicense, and / or sell
|
||||
# copies of the Software, and to permit persons to whom the Software is
|
||||
# furnished to do so, subject to the following conditions :
|
||||
#
|
||||
# The above copyright notice and this permission notice shall be included in
|
||||
# all copies or substantial portions of the Software.
|
||||
#
|
||||
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.IN NO EVENT SHALL THE
|
||||
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||
# THE SOFTWARE.
|
||||
|
||||
import re
|
||||
from . import pushjet
|
||||
|
||||
from ..NotifyBase import NotifyBase
|
||||
from ...common import NotifyType
|
||||
from ...AppriseLocale import gettext_lazy as _
|
||||
|
||||
PUBLIC_KEY_RE = re.compile(
|
||||
r'^[a-z0-9]{4}-[a-z0-9]{6}-[a-z0-9]{12}-[a-z0-9]{5}-[a-z0-9]{9}$', re.I)
|
||||
|
||||
SECRET_KEY_RE = re.compile(r'^[a-z0-9]{32}$', re.I)
|
||||
|
||||
|
||||
class NotifyPushjet(NotifyBase):
|
||||
"""
|
||||
A wrapper for Pushjet Notifications
|
||||
"""
|
||||
|
||||
# The default descriptive name associated with the Notification
|
||||
service_name = 'Pushjet'
|
||||
|
||||
# The default protocol
|
||||
protocol = 'pjet'
|
||||
|
||||
# The default secure protocol
|
||||
secure_protocol = 'pjets'
|
||||
|
||||
# A URL that takes you to the setup/help of the specific protocol
|
||||
setup_url = 'https://github.com/caronc/apprise/wiki/Notify_pushjet'
|
||||
|
||||
# Disable throttle rate for Pushjet requests since they are normally
|
||||
# local anyway (the remote/online service is no more)
|
||||
request_rate_per_sec = 0
|
||||
|
||||
# Define object templates
|
||||
templates = (
|
||||
'{schema}://{secret_key}@{host}',
|
||||
'{schema}://{secret_key}@{host}:{port}',
|
||||
)
|
||||
|
||||
# Define our tokens
|
||||
template_tokens = dict(NotifyBase.template_tokens, **{
|
||||
'host': {
|
||||
'name': _('Hostname'),
|
||||
'type': 'string',
|
||||
'required': True,
|
||||
},
|
||||
'port': {
|
||||
'name': _('Port'),
|
||||
'type': 'int',
|
||||
'min': 1,
|
||||
'max': 65535,
|
||||
},
|
||||
'secret_key': {
|
||||
'name': _('Secret Key'),
|
||||
'type': 'string',
|
||||
'required': True,
|
||||
'private': True,
|
||||
},
|
||||
})
|
||||
|
||||
def __init__(self, secret_key, **kwargs):
|
||||
"""
|
||||
Initialize Pushjet Object
|
||||
"""
|
||||
super(NotifyPushjet, self).__init__(**kwargs)
|
||||
|
||||
if not secret_key:
|
||||
# You must provide a Pushjet key to work with
|
||||
msg = 'You must specify a Pushjet Secret Key.'
|
||||
self.logger.warning(msg)
|
||||
raise TypeError(msg)
|
||||
|
||||
# store our key
|
||||
self.secret_key = secret_key
|
||||
|
||||
def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs):
|
||||
"""
|
||||
Perform Pushjet Notification
|
||||
"""
|
||||
# Always call throttle before any remote server i/o is made
|
||||
self.throttle()
|
||||
|
||||
server = "https://" if self.secure else "http://"
|
||||
|
||||
server += self.host
|
||||
if self.port:
|
||||
server += ":" + str(self.port)
|
||||
|
||||
try:
|
||||
api = pushjet.pushjet.Api(server)
|
||||
service = api.Service(secret_key=self.secret_key)
|
||||
|
||||
service.send(body, title)
|
||||
self.logger.info('Sent Pushjet notification.')
|
||||
|
||||
except (pushjet.errors.PushjetError, ValueError) as e:
|
||||
self.logger.warning('Failed to send Pushjet notification.')
|
||||
self.logger.debug('Pushjet Exception: %s' % str(e))
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def url(self):
|
||||
"""
|
||||
Returns the URL built dynamically based on specified arguments.
|
||||
"""
|
||||
|
||||
# Define any arguments set
|
||||
args = {
|
||||
'format': self.notify_format,
|
||||
'overflow': self.overflow_mode,
|
||||
'verify': 'yes' if self.verify_certificate else 'no',
|
||||
}
|
||||
|
||||
default_port = 443 if self.secure else 80
|
||||
|
||||
return '{schema}://{secret_key}@{hostname}{port}/?{args}'.format(
|
||||
schema=self.secure_protocol if self.secure else self.protocol,
|
||||
secret_key=NotifyPushjet.quote(self.secret_key, safe=''),
|
||||
hostname=NotifyPushjet.quote(self.host, safe=''),
|
||||
port='' if self.port is None or self.port == default_port
|
||||
else ':{}'.format(self.port),
|
||||
args=NotifyPushjet.urlencode(args),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def parse_url(url):
|
||||
"""
|
||||
Parses the URL and returns enough arguments that can allow
|
||||
us to substantiate this object.
|
||||
|
||||
Syntax:
|
||||
pjet://secret_key@hostname
|
||||
pjet://secret_key@hostname:port
|
||||
pjets://secret_key@hostname
|
||||
pjets://secret_key@hostname:port
|
||||
|
||||
"""
|
||||
results = NotifyBase.parse_url(url)
|
||||
|
||||
if not results:
|
||||
# We're done early as we couldn't load the results
|
||||
return results
|
||||
|
||||
# Store it as it's value
|
||||
results['secret_key'] = \
|
||||
NotifyPushjet.unquote(results.get('user'))
|
||||
|
||||
return results
|
||||
@ -1,6 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
"""A Python API for Pushjet. Send notifications to your phone from Python scripts!"""
|
||||
|
||||
from .pushjet import Service, Device, Subscription, Message, Api
|
||||
from .errors import PushjetError, AccessError, NonexistentError, SubscriptionError, RequestError, ServerError
|
||||
@ -1,48 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
|
||||
from requests import RequestException
|
||||
|
||||
import sys
|
||||
if sys.version_info[0] < 3:
|
||||
# This is built into Python 3.
|
||||
class ConnectionError(Exception):
|
||||
pass
|
||||
|
||||
class PushjetError(Exception):
|
||||
"""All the errors inherit from this. Therefore, ``except PushjetError`` catches all errors."""
|
||||
|
||||
class AccessError(PushjetError):
|
||||
"""Raised when a secret key is missing for a service method that needs one."""
|
||||
|
||||
class NonexistentError(PushjetError):
|
||||
"""Raised when an attempt to access a nonexistent service is made."""
|
||||
|
||||
class SubscriptionError(PushjetError):
|
||||
"""Raised when an attempt to subscribe to a service that's already subscribed to,
|
||||
or to unsubscribe from a service that isn't subscribed to, is made."""
|
||||
|
||||
class RequestError(PushjetError, ConnectionError):
|
||||
"""Raised if something goes wrong in the connection to the API server.
|
||||
Inherits from ``ConnectionError`` on Python 3, and can therefore be caught
|
||||
with ``except ConnectionError`` there.
|
||||
|
||||
:ivar requests_exception: The underlying `requests <http://docs.python-requests.org>`__
|
||||
exception. Access this if you want to handle different HTTP request errors in different ways.
|
||||
"""
|
||||
|
||||
def __str__(self):
|
||||
return "requests.{error}: {description}".format(
|
||||
error=self.requests_exception.__class__.__name__,
|
||||
description=str(self.requests_exception)
|
||||
)
|
||||
|
||||
def __init__(self, requests_exception):
|
||||
self.requests_exception = requests_exception
|
||||
|
||||
class ServerError(PushjetError):
|
||||
"""Raised if the API server has an error while processing your request.
|
||||
This getting raised means there's a bug in the server! If you manage to
|
||||
track down what caused it, you can `open an issue on Pushjet's GitHub page
|
||||
<https://github.com/Pushjet/Pushjet-Server-Api/issues>`__.
|
||||
"""
|
||||
@ -1,313 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import sys
|
||||
import requests
|
||||
from functools import partial
|
||||
|
||||
from six import text_type
|
||||
from six.moves.urllib.parse import urljoin
|
||||
|
||||
from .utilities import (
|
||||
NoNoneDict,
|
||||
requires_secret_key, with_api_bound,
|
||||
is_valid_uuid, is_valid_public_key, is_valid_secret_key, repr_format
|
||||
)
|
||||
from .errors import NonexistentError, SubscriptionError, RequestError, ServerError
|
||||
|
||||
DEFAULT_API_URL = 'https://api.pushjet.io/'
|
||||
|
||||
class PushjetModel(object):
|
||||
_api = None # This is filled in later.
|
||||
|
||||
class Service(PushjetModel):
|
||||
"""A Pushjet service to send messages through. To receive messages, devices
|
||||
subscribe to these.
|
||||
|
||||
:param secret_key: The service's API key for write access. If provided,
|
||||
:func:`~pushjet.Service.send`, :func:`~pushjet.Service.edit`, and
|
||||
:func:`~pushjet.Service.delete` become available.
|
||||
Either this or the public key parameter must be present.
|
||||
:param public_key: The service's public API key for read access only.
|
||||
Either this or the secret key parameter must be present.
|
||||
|
||||
:ivar name: The name of the service.
|
||||
:ivar icon_url: The URL to the service's icon. May be ``None``.
|
||||
:ivar created: When the service was created, as seconds from epoch.
|
||||
:ivar secret_key: The service's secret API key, or ``None`` if the service is read-only.
|
||||
:ivar public_key: The service's public API key, to be used when subscribing to the service.
|
||||
"""
|
||||
|
||||
def __repr__(self):
|
||||
return "<Pushjet Service: \"{}\">".format(repr_format(self.name))
|
||||
|
||||
def __init__(self, secret_key=None, public_key=None):
|
||||
if secret_key is None and public_key is None:
|
||||
raise ValueError("Either a secret key or public key "
|
||||
"must be provided.")
|
||||
elif secret_key and not is_valid_secret_key(secret_key):
|
||||
raise ValueError("Invalid secret key provided.")
|
||||
elif public_key and not is_valid_public_key(public_key):
|
||||
raise ValueError("Invalid public key provided.")
|
||||
self.secret_key = text_type(secret_key) if secret_key else None
|
||||
self.public_key = text_type(public_key) if public_key else None
|
||||
self.refresh()
|
||||
|
||||
def _request(self, endpoint, method, is_secret, params=None, data=None):
|
||||
params = params or {}
|
||||
if is_secret:
|
||||
params['secret'] = self.secret_key
|
||||
else:
|
||||
params['service'] = self.public_key
|
||||
return self._api._request(endpoint, method, params, data)
|
||||
|
||||
@requires_secret_key
|
||||
def send(self, message, title=None, link=None, importance=None):
|
||||
"""Send a message to the service's subscribers.
|
||||
|
||||
:param message: The message body to be sent.
|
||||
:param title: (optional) The message's title. Messages can be without title.
|
||||
:param link: (optional) An URL to be sent with the message.
|
||||
:param importance: (optional) The priority level of the message. May be
|
||||
a number between 1 and 5, where 1 is least important and 5 is most.
|
||||
"""
|
||||
data = NoNoneDict({
|
||||
'message': message,
|
||||
'title': title,
|
||||
'link': link,
|
||||
'level': importance
|
||||
})
|
||||
self._request('message', 'POST', is_secret=True, data=data)
|
||||
|
||||
@requires_secret_key
|
||||
def edit(self, name=None, icon_url=None):
|
||||
"""Edit the service's attributes.
|
||||
|
||||
:param name: (optional) A new name to give the service.
|
||||
:param icon_url: (optional) A new URL to use as the service's icon URL.
|
||||
Set to an empty string to remove the service's icon entirely.
|
||||
"""
|
||||
data = NoNoneDict({
|
||||
'name': name,
|
||||
'icon': icon_url
|
||||
})
|
||||
if not data:
|
||||
return
|
||||
self._request('service', 'PATCH', is_secret=True, data=data)
|
||||
self.name = text_type(name)
|
||||
self.icon_url = text_type(icon_url)
|
||||
|
||||
@requires_secret_key
|
||||
def delete(self):
|
||||
"""Delete the service. Irreversible."""
|
||||
self._request('service', 'DELETE', is_secret=True)
|
||||
|
||||
def _update_from_data(self, data):
|
||||
self.name = data['name']
|
||||
self.icon_url = data['icon'] or None
|
||||
self.created = data['created']
|
||||
self.public_key = data['public']
|
||||
self.secret_key = data.get('secret', getattr(self, 'secret_key', None))
|
||||
|
||||
def refresh(self):
|
||||
"""Refresh the server's information, in case it could be edited from elsewhere.
|
||||
|
||||
:raises: :exc:`~pushjet.NonexistentError` if the service was deleted before refreshing.
|
||||
"""
|
||||
key_name = 'public'
|
||||
secret = False
|
||||
if self.secret_key is not None:
|
||||
key_name = 'secret'
|
||||
secret = True
|
||||
|
||||
status, response = self._request('service', 'GET', is_secret=secret)
|
||||
if status == requests.codes.NOT_FOUND:
|
||||
raise NonexistentError("A service with the provided {} key "
|
||||
"does not exist (anymore, at least).".format(key_name))
|
||||
self._update_from_data(response['service'])
|
||||
|
||||
@classmethod
|
||||
def _from_data(cls, data):
|
||||
# This might be a no-no, but I see little alternative if
|
||||
# different constructors with different parameters are needed,
|
||||
# *and* a default __init__ constructor should be present.
|
||||
# This, along with the subclassing for custom API URLs, may
|
||||
# very well be one of those pieces of code you look back at
|
||||
# years down the line - or maybe just a couple of weeks - and say
|
||||
# "what the heck was I thinking"? I assure you, though, future me.
|
||||
# This was the most reasonable thing to get the API + argspecs I wanted.
|
||||
obj = cls.__new__(cls)
|
||||
obj._update_from_data(data)
|
||||
return obj
|
||||
|
||||
@classmethod
|
||||
def create(cls, name, icon_url=None):
|
||||
"""Create a new service.
|
||||
|
||||
:param name: The name of the new service.
|
||||
:param icon_url: (optional) An URL to an image to be used as the service's icon.
|
||||
:return: The newly-created :class:`~pushjet.Service`.
|
||||
"""
|
||||
data = NoNoneDict({
|
||||
'name': name,
|
||||
'icon': icon_url
|
||||
})
|
||||
_, response = cls._api._request('service', 'POST', data=data)
|
||||
return cls._from_data(response['service'])
|
||||
|
||||
class Device(PushjetModel):
|
||||
"""The "receiver" for messages. Subscribes to services and receives any
|
||||
messages they send.
|
||||
|
||||
:param uuid: The device's unique ID as a UUID. Does not need to be registered
|
||||
before using it. A UUID can be generated with ``uuid.uuid4()``, for example.
|
||||
:ivar uuid: The UUID the device was initialized with.
|
||||
"""
|
||||
|
||||
def __repr__(self):
|
||||
return "<Pushjet Device: {}>".format(self.uuid)
|
||||
|
||||
def __init__(self, uuid):
|
||||
uuid = text_type(uuid)
|
||||
if not is_valid_uuid(uuid):
|
||||
raise ValueError("Invalid UUID provided. Try uuid.uuid4().")
|
||||
self.uuid = text_type(uuid)
|
||||
|
||||
def _request(self, endpoint, method, params=None, data=None):
|
||||
params = (params or {})
|
||||
params['uuid'] = self.uuid
|
||||
return self._api._request(endpoint, method, params, data)
|
||||
|
||||
def subscribe(self, service):
|
||||
"""Subscribe the device to a service.
|
||||
|
||||
:param service: The service to subscribe to. May be a public key or a :class:`~pushjet.Service`.
|
||||
:return: The :class:`~pushjet.Service` subscribed to.
|
||||
|
||||
:raises: :exc:`~pushjet.NonexistentError` if the provided service does not exist.
|
||||
:raises: :exc:`~pushjet.SubscriptionError` if the provided service is already subscribed to.
|
||||
"""
|
||||
data = {}
|
||||
data['service'] = service.public_key if isinstance(service, Service) else service
|
||||
status, response = self._request('subscription', 'POST', data=data)
|
||||
if status == requests.codes.CONFLICT:
|
||||
raise SubscriptionError("The device is already subscribed to that service.")
|
||||
elif status == requests.codes.NOT_FOUND:
|
||||
raise NonexistentError("A service with the provided public key "
|
||||
"does not exist (anymore, at least).")
|
||||
return self._api.Service._from_data(response['service'])
|
||||
|
||||
def unsubscribe(self, service):
|
||||
"""Unsubscribe the device from a service.
|
||||
|
||||
:param service: The service to unsubscribe from. May be a public key or a :class:`~pushjet.Service`.
|
||||
:raises: :exc:`~pushjet.NonexistentError` if the provided service does not exist.
|
||||
:raises: :exc:`~pushjet.SubscriptionError` if the provided service isn't subscribed to.
|
||||
"""
|
||||
data = {}
|
||||
data['service'] = service.public_key if isinstance(service, Service) else service
|
||||
status, _ = self._request('subscription', 'DELETE', data=data)
|
||||
if status == requests.codes.CONFLICT:
|
||||
raise SubscriptionError("The device is not subscribed to that service.")
|
||||
elif status == requests.codes.NOT_FOUND:
|
||||
raise NonexistentError("A service with the provided public key "
|
||||
"does not exist (anymore, at least).")
|
||||
|
||||
def get_subscriptions(self):
|
||||
"""Get all the subscriptions the device has.
|
||||
|
||||
:return: A list of :class:`~pushjet.Subscription`.
|
||||
"""
|
||||
_, response = self._request('subscription', 'GET')
|
||||
subscriptions = []
|
||||
for subscription_dict in response['subscriptions']:
|
||||
subscriptions.append(Subscription(subscription_dict))
|
||||
return subscriptions
|
||||
|
||||
def get_messages(self):
|
||||
"""Get all new (that is, as of yet unretrieved) messages.
|
||||
|
||||
:return: A list of :class:`~pushjet.Message`.
|
||||
"""
|
||||
_, response = self._request('message', 'GET')
|
||||
messages = []
|
||||
for message_dict in response['messages']:
|
||||
messages.append(Message(message_dict))
|
||||
return messages
|
||||
|
||||
class Subscription(object):
|
||||
"""A subscription to a service, with the metadata that entails.
|
||||
|
||||
:ivar service: The service the subscription is to, as a :class:`~pushjet.Service`.
|
||||
:ivar time_subscribed: When the subscription was made, as seconds from epoch.
|
||||
:ivar last_checked: When the device last retrieved messages from the subscription,
|
||||
as seconds from epoch.
|
||||
:ivar device_uuid: The UUID of the device that owns the subscription.
|
||||
"""
|
||||
|
||||
def __repr__(self):
|
||||
return "<Pushjet Subscription to service \"{}\">".format(repr_format(self.service.name))
|
||||
|
||||
def __init__(self, subscription_dict):
|
||||
self.service = Service._from_data(subscription_dict['service'])
|
||||
self.time_subscribed = subscription_dict['timestamp']
|
||||
self.last_checked = subscription_dict['timestamp_checked']
|
||||
self.device_uuid = subscription_dict['uuid'] # Not sure this is needed, but...
|
||||
|
||||
class Message(object):
|
||||
"""A message received from a service.
|
||||
|
||||
:ivar message: The message body.
|
||||
:ivar title: The message title. May be ``None``.
|
||||
:ivar link: The URL the message links to. May be ``None``.
|
||||
:ivar time_sent: When the message was sent, as seconds from epoch.
|
||||
:ivar importance: The message's priority level between 1 and 5, where 1 is
|
||||
least important and 5 is most.
|
||||
:ivar service: The :class:`~pushjet.Service` that sent the message.
|
||||
"""
|
||||
|
||||
def __repr__(self):
|
||||
return "<Pushjet Message: \"{}\">".format(repr_format(self.title or self.message))
|
||||
|
||||
def __init__(self, message_dict):
|
||||
self.message = message_dict['message']
|
||||
self.title = message_dict['title'] or None
|
||||
self.link = message_dict['link'] or None
|
||||
self.time_sent = message_dict['timestamp']
|
||||
self.importance = message_dict['level']
|
||||
self.service = Service._from_data(message_dict['service'])
|
||||
|
||||
class Api(object):
|
||||
"""An API with a custom URL. Use this if you're connecting to a self-hosted
|
||||
Pushjet API instance, or a non-standard one in general.
|
||||
|
||||
:param url: The URL to the API instance.
|
||||
:ivar url: The URL to the API instance, as supplied.
|
||||
"""
|
||||
|
||||
def __repr__(self):
|
||||
return "<Pushjet Api: {}>".format(self.url).encode(sys.stdout.encoding, errors='replace')
|
||||
|
||||
def __init__(self, url):
|
||||
self.url = text_type(url)
|
||||
self.Service = with_api_bound(Service, self)
|
||||
self.Device = with_api_bound(Device, self)
|
||||
|
||||
def _request(self, endpoint, method, params=None, data=None):
|
||||
url = urljoin(self.url, endpoint)
|
||||
try:
|
||||
r = requests.request(method, url, params=params, data=data)
|
||||
except requests.RequestException as e:
|
||||
raise RequestError(e)
|
||||
status = r.status_code
|
||||
if status == requests.codes.INTERNAL_SERVER_ERROR:
|
||||
raise ServerError(
|
||||
"An error occurred in the server while processing your request. "
|
||||
"This should probably be reported to: "
|
||||
"https://github.com/Pushjet/Pushjet-Server-Api/issues"
|
||||
)
|
||||
try:
|
||||
response = r.json()
|
||||
except ValueError:
|
||||
response = {}
|
||||
return status, response
|
||||
|
||||
@ -1,64 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
import re
|
||||
import sys
|
||||
from decorator import decorator
|
||||
from .errors import AccessError
|
||||
|
||||
# Help class(...es? Nah. Just singular for now.)
|
||||
|
||||
class NoNoneDict(dict):
|
||||
"""A dict that ignores values that are None. Not completely API-compatible
|
||||
with dict, but contains all that's needed.
|
||||
"""
|
||||
def __repr__(self):
|
||||
return "NoNoneDict({dict})".format(dict=dict.__repr__(self))
|
||||
|
||||
def __init__(self, initial={}):
|
||||
self.update(initial)
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
if value is not None:
|
||||
dict.__setitem__(self, key, value)
|
||||
|
||||
def update(self, data):
|
||||
for key, value in data.items():
|
||||
self[key] = value
|
||||
|
||||
# Decorators / factories
|
||||
|
||||
@decorator
|
||||
def requires_secret_key(func, self, *args, **kwargs):
|
||||
"""Raise an error if the method is called without a secret key."""
|
||||
if self.secret_key is None:
|
||||
raise AccessError("The Service doesn't have a secret "
|
||||
"key provided, and therefore lacks write permission.")
|
||||
return func(self, *args, **kwargs)
|
||||
|
||||
def with_api_bound(cls, api):
|
||||
new_cls = type(cls.__name__, (cls,), {
|
||||
'_api': api,
|
||||
'__doc__': (
|
||||
"Create a :class:`~pushjet.{name}` bound to the API. "
|
||||
"See :class:`pushjet.{name}` for documentation."
|
||||
).format(name=cls.__name__)
|
||||
})
|
||||
return new_cls
|
||||
|
||||
# Helper functions
|
||||
|
||||
UUID_RE = re.compile(r'^[A-Fa-f0-9]{8}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{4}-[A-Fa-f0-9]{12}$')
|
||||
PUBLIC_KEY_RE = re.compile(r'^[A-Za-z0-9]{4}-[A-Za-z0-9]{6}-[A-Za-z0-9]{12}-[A-Za-z0-9]{5}-[A-Za-z0-9]{9}$')
|
||||
SECRET_KEY_RE = re.compile(r'^[A-Za-z0-9]{32}$')
|
||||
|
||||
is_valid_uuid = lambda s: UUID_RE.match(s) is not None
|
||||
is_valid_public_key = lambda s: PUBLIC_KEY_RE.match(s) is not None
|
||||
is_valid_secret_key = lambda s: SECRET_KEY_RE.match(s) is not None
|
||||
|
||||
def repr_format(s):
|
||||
s = s.replace('\n', ' ').replace('\r', '')
|
||||
original_length = len(s)
|
||||
s = s[:30]
|
||||
s += '...' if len(s) != original_length else ''
|
||||
s = s.encode(sys.stdout.encoding, errors='replace')
|
||||
return s
|
||||
Loading…
Reference in new issue