from builtins import object
import logging
import json
# noinspection PyMethodMayBeStatic,PyUnusedLocal
[docs]class OnAWBase(object):
""" Base class to be extended to extend ActingWeb functionality """
def __init__(self):
self.config = None
self.myself = None
self.webobj = None
self.auth = None
[docs] def aw_init(self, auth=None, webobj=None):
self.config = auth.config
self.myself = auth.actor
self.webobj = webobj
self.auth = auth
[docs] def bot_post(self, path):
"""Called on POSTs to /bot.
Note, there will not be any actor initialised.
"""
# Safety valve to make sure we don't do anything if bot is not
# configured.
if not self.config.bot['token'] or len(self.config.bot['token']) == 0:
return False
if path == 'something':
pass
# try:
# body = json.loads(req.request.body.decode('utf-8', 'ignore'))
# logging.debug('Bot callback: ' + req.request.body.decode('utf-8', 'ignore'))
# except:
# return 400
#
# This is how actor can be initialised if the bot request
# contains a value that has been stored as an actor property.
# This value must be a primary key for the external oauth identity
# that the actor is representing.
# Here, oauthId (from oauth service) has earlier been stored as a property
# myself = actor.Actor()
# myself.get_from_property(name='oauthId', value=<PROPERTY-VALUE>)
# if myself.id:
# logging.debug('Found Actor(' + myself.id + ')')
#
# If we havent''
# if not myself.id:
# myself.create(url=self.config.root, creator= <EMAIL>,
# passphrase=self.config.new_token())
# Now store the oauthId propery
# myself.store.oauthId = <ID-VALUE>
# Send confirmation message that actor has been created
# return True
# Do something
return True
[docs] def get_properties(self, path: list, data: dict) -> dict or None:
""" Called on GET to properties for transformations to be done
:param path: Target path requested
:type path: list[str]
:param data: Data retrieved from data store to be returned
:type data: dict
:return: The transformed data to return to requestor or None if 404 should be returned
:rtype: dict or None
"""
return data
[docs] def delete_properties(self, path: list, old: dict, new: dict) -> bool:
""" Called on DELETE to properties
:param path: Target path to be deleted
:type path: list[str]
:param old: Property value that will be deleted (or changed)
:type old: dict
:param new: Property value after path has been deleted
:type new: dict
:return: True if DELETE is allowed, False if 403 should be returned
:rtype: bool
"""
return True
[docs] def put_properties(self, path: list, old: dict, new: dict) -> dict or None:
""" Called on PUT to properties for transformations to be done before save
:param path: Target path requested to be updated
:type path: list[str]
:param old: Old data from database
:type old: dict
:param new:
:type new: New data from PUT request (after merge)
:return: The dict that should be stored or None if 400 should be returned and nothing stored
:rtype: dict or None
"""
return new
[docs] def post_properties(self, prop: str, data: dict) -> dict or None:
""" Called on POST to properties, once for each property
:param prop: Property to be created
:type prop: str
:param data: The data to be stored in prop
:type data: dict
:return: The transformed data to store in prop or None if that property should be skipped and not stored
:rtype: dict or None
"""
return data
[docs] def get_callbacks(self, name):
"""Customizible function to handle GET /callbacks"""
# return True if callback has been processed
# THE BELOW IS SAMPLE CODE
# my_oauth=oauth.oauth(token = myself.store.oauth_token)
if name == 'something':
return
# END OF SAMPLE CODE
return False
[docs] def delete_callbacks(self, name):
"""Customizible function to handle DELETE /callbacks"""
# return True if callback has been processed
return False
[docs] def post_callbacks(self, name):
"""Customizible function to handle POST /callbacks"""
# return True if callback has been processed
# THE BELOW IS SAMPLE CODE
# logging.debug("Callback body: "+req.request.body.decode('utf-8', 'ignore'))
# non-json POSTs to be handled first
# if name == 'somethingelse':
# return True
# Handle json POSTs below
# body = json.loads(req.request.body.decode('utf-8', 'ignore'))
# data = body['data']
# if name == 'somethingmore':
# callback_id = req.request.get('id')
# req.response.set_status(204)
# return True
# req.response.set_status(403, "Callback not found.")
# END OF SAMPLE CODE
return False
[docs] def post_subscriptions(self, sub, peerid, data):
"""Customizible function to process incoming callbacks/subscriptions/ callback with json body, return True if
processed, False if not."""
logging.debug("Got callback and processed " + sub["subscriptionid"] +
" subscription from peer " + peerid + " with json blob: " + json.dumps(data))
return True
[docs] def delete_actor(self):
# THIS METHOD IS CALLED WHEN AN ACTOR IS REQUESTED TO BE DELETED.
# THE BELOW IS SAMPLE CODE
# Clean up anything associated with this actor before it is deleted.
# END OF SAMPLE CODE
return
[docs] def check_on_oauth_success(self, token=None):
# THIS METHOD IS CALLED WHEN AN OAUTH AUTHORIZATION HAS BEEN SUCCESSFULLY MADE AND BEFORE APPROVAL
return True
[docs] def actions_on_oauth_success(self):
# THIS METHOD IS CALLED WHEN AN OAUTH AUTHORIZATION HAS BEEN SUCCESSFULLY MADE
return True
[docs] def get_resources(self, name):
""" Called on GET to resources. Return struct for json out.
Returning {} will give a 404 response back to requestor.
"""
return {}
[docs] def delete_resources(self, name):
""" Called on DELETE to resources. Return struct for json out.
Returning {} will give a 404 response back to requestor.
"""
return {}
[docs] def put_resources(self, name, params):
""" Called on PUT to resources. Return struct for json out.
Returning {} will give a 404 response back to requestor.
Returning an error code after setting the response will not change
the error code.
"""
return {}
[docs] def post_resources(self, name, params):
""" Called on POST to resources. Return struct for json out.
Returning {} will give a 404 response back to requestor.
Returning an error code after setting the response will not change
the error code.
"""
return {}
[docs] def www_paths(self, path=''):
# THIS METHOD IS CALLED WHEN AN actorid/www/* PATH IS CALLED (AND AFTER ACTINGWEB DEFAULT PATHS HAVE BEEN
# HANDLED)
# THE BELOW IS SAMPLE CODE
# if path == '' or not myself:
# logging.info('Got an on_www_paths without proper parameters.')
# return False
# if path == 'something':
# return True
# END OF SAMPLE CODE
return False