Source code for glados.request

import json
import logging
from datetime import datetime
from typing import TYPE_CHECKING, NoReturn, Optional, Union

from .datastore import DataStore, DataStoreInteraction
from .route_type import BOT_ROUTES, RouteType
from .utils import PyJSON

    from sqlalchemy.orm import Session

[docs]class SlackVerification: """An object to hold slack verification data Parameters ---------- data raw request body. This is used to verify the message is from slack. timestamp The X-Slack-Request-Timestamp from the headers of the request. This is used to verify the message is from slack. signature: The X-Slack-Signature from the headers of the request. This is used to verify the message is from slack. """ def __init__(self, data: str, timestamp: str = None, signature: str = None): = data self.timestamp = timestamp self.signature = signature @property def json(self) -> dict: """Returns the dict of the SlackVerification""" return { "data":, "timestamp": self.timestamp, "signature": self.signature, }
[docs]class GladosRequest: """GLaDOS Request Object. This holds all the data required to process the request. Parameters ---------- route_type what type of route is this route what is the route to be called slack_verify slack data used for verifying the request came from Slack bot_name The name of the bot to send the request to. This is used for select RouteTypes json the json paylod of the request data data to send with the request. This should be from a database kwargs Examples -------- >>> request = GladosRequest(RouteType.SendMessage, "send_mock", json={"message":"my message"}) >>> print(request.json.message) my message >>> try: ... print(request.json.other_param) ... except AttributeError: ... print("ERROR") ERROR """ def __init__( self, route_type: RouteType, route: str = None, slack_verify: SlackVerification = None, bot_name: str = None, json: Union[str, dict] = None, data: dict = None, **kwargs, ): if not json: json = dict() self.json = PyJSON(json) self.route_type = route_type self.bot_name = bot_name self._route = route self.slack_verify = slack_verify self.response_url = None self.trigger_id = None self._data = data self._datastore = None # type: Optional[DataStore] self._session = ( None ) # type: Optional[Session] # This is the datastore session for this request. self._interaction = None # type: Optional[DataStoreInteraction] # This is the interaction database object for this request. self.auto_link = False # if this is true, then it will expect the response from the plugin to want to try to autolink to the interaction datastore. if route_type is RouteType.Interaction: self.response_url = self.json.get("response_url") self.trigger_id = self.json.get("trigger_id") if route_type is RouteType.Menu: self._route = self.json.action_id if route_type is RouteType.Interaction: self._route = self.json.actions[0].action_id if route_type is RouteType.Events: self._route = self.json.event.type self.new_interaction = None @property def route(self) -> str: """the actual route If the route automatically prefixed the route with the bot name, it will return the route with the prefix """ return ( f"{self.bot_name}_{self._route}" if self.route_type in BOT_ROUTES else self._route ) @route.setter def route(self, value): self._route = value
[docs] def set_session(self, session: "Session") -> NoReturn: """Set the session for this request. Parameters ---------- session session to use for this request. Raises ------ :obj:`ConnectionError` If the session is not active raise a ConnectionError """ self._session = session if not self._session.is_active: raise ConnectionError("request session is not active")
[docs] def set_datastore(self, datastore: "DataStore") -> NoReturn: """Set the Datastore and session for the request. Parameters ---------- datastore Datastore to use. This datastore will be used to create the session. """ self._datastore = datastore self.set_session(self._datastore.create_session())
[docs] def set_interaction_from_datastore(self) -> NoReturn: """Get the interaction object from the datastore.""" # Get the interaction object from the datastore # see if there is channel and message ts in the payload. if not self._session: raise ConnectionError("session not set for request") container_payload = self.json.get("container") if not container_payload: logging.warning(f"no container block in body: {self.json.to_dict()}") self._interaction = None return self.interaction channel = container_payload.get("channel_id") message_ts = container_payload.get("message_ts") try: massage_ts = datetime.fromtimestamp(float(message_ts)) except Exception as e: logging.warning(f"error parsing message ts: {message_ts} : {e}") if None in [channel, message_ts]: logging.warning( f"missing channel_id or message_ts in container: {container_payload}" ) self._interaction = None return self.interaction interaction = self._datastore.find_interaction_by_channel_ts( channel, message_ts, self._session ) self._interaction = interaction
[docs] def add_interaction_to_datastore( self, interaction: DataStoreInteraction ) -> Optional[DataStoreInteraction]: """Add an interaction to the datastore and return the updated interaction. Notes ----- The interaction_id can be retrieved by doing interaction.interaction_id Parameters ---------- interaction the interaction to be added """ if not self._datastore: logging.warning("datastore not set for request") return if not self._session: raise ConnectionError("session not set for request") return self._datastore.insert_interaction(interaction, self._session)
[docs] def close_session(self) -> NoReturn: """Close session for request""" if not self._session or not self._session.is_active: self._session.close()
[docs] def rollback_session(self) -> NoReturn: """Rollback the session.""" if self._session.is_active: self._session.rollback()
[docs] def has_interaction(self) -> bool: """Check if request has interaction. """ return True if self._interaction else False
[docs] def has_new_interaction(self) -> bool: """check if request has a new interaction object.""" return True if self.new_interaction else False
[docs] def gen_new_interaction( self, *, followup_action=None, followup_ts=None, ttl=None, data=None, auto_link: bool = True, auto_set: bool = True, ) -> DataStoreInteraction: """Generate a new interaction object and set it as new_interaction. Parameters ---------- followup_action followup_ts ttl data auto_link set this request to auto-link using the return payload. The return payload must be the response from sending a slack message. auto_set set this new interaction object as the request new_interaction """ if not data: data = dict() self.auto_link = auto_link new_interaction = DataStoreInteraction( bot=self.bot_name, followup_action=followup_action, followup_ts=followup_ts, ttl=ttl, data=data, ) if auto_set: self.new_interaction = new_interaction return self.new_interaction return new_interaction
@property def interaction_id(self) -> Optional[str]: """Returns the interaction_id of request.interaction""" if not self.has_interaction(): return None return self.interaction.interaction_id @property def interaction(self) -> Optional[DataStoreInteraction]: """Returns the interaction for the request""" if not self.has_interaction(): return None return self._interaction @property def data(self) -> PyJSON: """Returns the data object of the request""" return PyJSON(self._data) @property def data_blob(self) -> dict: """Returns the raw dict of the data object""" return self._data @data.setter def data(self, value): if type(value) is str: try: self._data = json.loads(value) except json.JSONDecodeError: logging.error(f"JSONDecodeError on string {value}") except Exception as e: logging.error(f"{e} on parsing JSON from: {value}") if type(value) is dict: self._data = value