Source code for pubsub.core.topicargspec

"""
Definitions related to message data specification.

:copyright: Copyright since 2006 by Oliver Schoenborn, all rights reserved.
:license: BSD, see LICENSE_BSD_Simple.txt for details.
"""

import weakref
from typing import Tuple, List, Sequence as Seq, Mapping, Dict, Callable, Any, Optional, Union

from .topicutils import stringize, WeakNone
from .annotations import annotationType
from .topicexc import MessageDataSpecError
from .listener import getArgs as getListenerArgs, UserListener


ArgsDocs = Dict[str, str]
MsgData = Mapping[str, Any]


def verifyArgsDifferent(allArgs, allParentArgs, topicName):
    """
    Verify that allArgs does not contain any of allParentArgs. Raise
    MessageDataSpecError if fail.
    """
    extra = set(allArgs).intersection(allParentArgs)
    if extra:
        msg = 'Args %%s already used in parent of "%s"' % topicName
        raise MessageDataSpecError(msg, tuple(extra))


def verifySubset(all, sub, topicName, extraMsg=''):
    """
    Verify that sub is a subset of all for topicName. Raise
    MessageDataSpecError if fail.
    """
    notInAll = set(sub).difference(all)
    if notInAll:
        args = ','.join(all)
        msg = 'Params [%s] missing inherited [%%s] for topic "%s"%s' % (args, topicName, extraMsg)
        raise MessageDataSpecError(msg, tuple(notInAll))


def topicArgsFromCallable(_callable: UserListener, ignoreArgs: Seq[str] = ()) -> Tuple[ArgsDocs, List[str]]:
    """
    Get the topic message data names and list of those that are required,
    by introspecting given callable. Returns a pair, (args, required)
    where args is a dictionary of allowed message data names vs docstring,
    and required states which ones are required rather than optional.
    """
    argsInfo = getListenerArgs(_callable, ignoreArgs=ignoreArgs)
    required = argsInfo.getRequiredArgs()
    defaultDoc = 'UNDOCUMENTED'
    args = dict.fromkeys(argsInfo.allParams, defaultDoc)
    return args, required


class ArgSpecGiven:
    """
    The message data specification (MDS) for a topic.

    This consists of each argument name that listener should have in its
    signature, plus which ones are required in any sendMessage(), and a
    documentation string for each argument. This instance will be transformed
    into an ArgsInfo object which is basically a superset of that information,
    needed to ensure that the arguments specifications satisfy
    pubsub policies for chosen API version.
    """

    SPEC_GIVEN_NONE = 1  # specification not given
    SPEC_GIVEN_ALL = 3  # all args specified

    def __init__(self, argsDocs: ArgsDocs = None, reqdArgs: Seq[str] = None):
        self.reqdArgs = tuple(reqdArgs or ())

        if argsDocs is None:
            self.argsSpecType = ArgSpecGiven.SPEC_GIVEN_NONE
            self.argsDocs = {}
        else:
            self.argsSpecType = ArgSpecGiven.SPEC_GIVEN_ALL
            self.argsDocs = argsDocs

            # check that all args marked as required are in argsDocs
            missingArgs = set(self.reqdArgs).difference(self.argsDocs.keys())  # py3: iter keys ok
            if missingArgs:
                msg = 'Params [%s] missing inherited required args [%%s]' % ','.join(argsDocs.keys())  # iter keys ok
                raise MessageDataSpecError(msg, missingArgs)

    def setAll(self, allArgsDocs: ArgsDocs, reqdArgs: Seq[str] = None):
        self.argsDocs = allArgsDocs
        self.reqdArgs = reqdArgs or ()
        self.argsSpecType = ArgSpecGiven.SPEC_GIVEN_ALL

    def isComplete(self) -> bool:
        """Returns True if the definition is usable, false otherwise."""
        return self.argsSpecType == ArgSpecGiven.SPEC_GIVEN_ALL

    def getOptional(self) -> List[str]:
        """Get the list of optional arguments"""
        return tuple(set(self.argsDocs.keys()).difference(self.reqdArgs))

    def __str__(self):
        return "%s, %s, %s" % (self.argsDocs, self.reqdArgs, self.argsSpecType)


[docs] class SenderMissingReqdMsgDataError(RuntimeError): """ Raised when a sendMessage() is missing arguments tagged as 'required' by pubsub topic of message. """ def __init__(self, topicName: str, argNames: Seq[str], missing: Seq[str]): argsStr = ','.join(argNames) missStr = ','.join(missing) msg = "Some required args missing in call to sendMessage('%s', %s): %s" \ % (stringize(topicName), argsStr, missStr) RuntimeError.__init__(self, msg)
[docs] class SenderUnknownMsgDataError(RuntimeError): """ Raised when a sendMessage() has arguments not listed among the topic's message data specification (MDS). """ def __init__(self, topicName: str, argNames: Seq[str], extra: Seq[str]): argsStr = ','.join(argNames) extraStr = ','.join(extra) msg = "Some optional args unknown in call to sendMessage('%s', %s): %s" \ % (topicName, argsStr, extraStr) RuntimeError.__init__(self, msg)
@annotationType class ArgsInfo: pass class ArgsInfo: """ Encode the Message Data Specification (MDS) for a given topic. ArgsInfos form a tree identical to that of Topics in that ArgInfos have a reference to their parent and children ArgInfos, created for the parent and children topics. The only difference between an ArgsInfo and an ArgSpecGiven is that the latter is what "user thinks is ok" whereas former has been validated: the specification for this topic is a strict superset of the specification of its parent, and a strict subset of the specification of each of its children. Also, the instance can be used to check validity and filter arguments. The MDS can be created "empty", ie "incomplete", meaning it cannot yet be used to validate listener subscriptions to topics. """ SPEC_MISSING = 10 # no args given SPEC_COMPLETE = 12 # all args, but not confirmed via user spec def __init__(self, topicNameTuple: Seq[str], specGiven: ArgSpecGiven, parentArgsInfo: ArgsInfo): self.topicNameTuple = topicNameTuple self.allOptional = () # topic message optional arg names self.allDocs = {} # doc for each arg self.allRequired = () # topic message required arg names self.argsSpecType = self.SPEC_MISSING self.parentAI = WeakNone() if parentArgsInfo is not None: self.parentAI = weakref.ref(parentArgsInfo) parentArgsInfo.__addChildAI(self) self.childrenAI = [] if specGiven.isComplete(): self.__setAllArgs(specGiven) if parentArgsInfo is None: assert self.argsAddedToParent is not None else: while not parentArgsInfo.isComplete(): parentArgsInfo = parentArgsInfo.parentAI() self.argsAddedToParent = set(self.getArgs()).difference(parentArgsInfo.getArgs()) def isComplete(self) -> bool: return self.argsSpecType == self.SPEC_COMPLETE def getArgs(self) -> List[str]: return self.allOptional + self.allRequired def numArgs(self) -> int: return len(self.allOptional) + len(self.allRequired) def getReqdArgs(self) -> List[str]: return self.allRequired def getOptArgs(self) -> List[str]: return self.allOptional def getArgsDocs(self) -> ArgsDocs: return self.allDocs.copy() def setArgsDocs(self, docs: ArgsDocs): """docs is a mapping from arg names to their documentation""" if not self.isComplete(): raise RuntimeError('Topic MDS is not complete, cannot set docs!') for arg, doc in docs.items(): self.allDocs[arg] = doc def check(self, msgData: MsgData): """ Check that the message arguments given satisfy the topic message data specification (MDS). :param msgData: the topic message data to check for validity :raise SenderMissingReqdMsgDataError: if some required args are missing or not known :raise SenderUnknownMsgDataError: if some optional args are unknown. """ all = set(msgData) # check that it has all required args needReqd = set(self.allRequired) hasReqd = (needReqd <= all) if not hasReqd: raise SenderMissingReqdMsgDataError( self.topicNameTuple, list(msgData.keys()), needReqd - all) # check that all other args are among the optional spec optional = all - needReqd ok = (optional <= set(self.allOptional)) if not ok: raise SenderUnknownMsgDataError(self.topicNameTuple, list(msgData.keys()), optional - set(self.allOptional)) def filterArgs(self, msgData: MsgData) -> MsgData: """ Returns a dict which contains only those items of msgData which are defined for topic. E.g. if msgData is {a:1, b:'b'} and topic arg spec is ('a',) then return {a:1}. The returned dict is valid only if check(msgData) was called (or check(superset of msgData) was called). :param msgData: the topic message data to filter """ assert self.isComplete() if len(msgData) == self.numArgs(): return msgData # only keep the keys from msgData that are also in topic's kwargs # method 1: SLOWEST # newKwargs = dict( (k,msgData[k]) for k in self.__msgArgs.allOptional if k in msgData ) # newKwargs.update( (k,msgData[k]) for k in self.__msgArgs.allRequired ) # method 2: FAST: # argNames = self.__msgArgs.getArgs() # newKwargs = dict( (key, val) for (key, val) in msgData.iteritems() if key in argNames ) # method 3: FASTEST: argNames = set(self.getArgs()).intersection(msgData) newKwargs = dict((k, msgData[k]) for k in argNames) return newKwargs def hasSameArgs(self, *argNames: Seq[str]) -> bool: """ Returns true if self has all the message arguments given, no more and no less. Order does not matter. So if getArgs() returns ('arg1', 'arg2') then self.hasSameArgs('arg2', 'arg1') will return true. """ return set(argNames) == set(self.getArgs()) def hasParent(self, argsInfo: ArgsInfo) -> bool: """return True if self has argsInfo object as parent""" return self.parentAI() is argsInfo def getCompleteAI(self) -> ArgsInfo: """ Get the closest arg spec, starting from self and moving to parent, that is complete. So if self.isComplete() is True, then returns self, otherwise returns parent (if parent.isComplete()), etc. """ AI = self while AI is not None: if AI.isComplete(): return AI AI = AI.parentAI() # dereference weakref return None def updateAllArgsFinal(self, topicDefn: ArgSpecGiven): """ This can only be called once, if the construction was done with ArgSpecGiven.SPEC_GIVEN_NONE """ assert not self.isComplete() assert topicDefn.isComplete() self.__setAllArgs(topicDefn) def __addChildAI(self, childAI: ArgsInfo): assert childAI not in self.childrenAI self.childrenAI.append(childAI) def __notifyParentCompleted(self): """Parent should call this when parent ArgsInfo has been completed""" assert self.parentAI().isComplete() if self.isComplete(): # verify that our spec is compatible with parent's self.__validateArgsToParent() self.argsAddedToParent = set(self.getArgs()).difference(self.parentAI().getArgs()) else: for argsInfo in self.childrenAI: argsInfo.__notifyAncestorCompleted(self.parentAI()) def __notifyAncestorCompleted(self, parentAI): if self.isComplete(): # verify that our spec is compatible with parent's self.__validateArgsToParent() self.argsAddedToParent = set(self.getArgs()).difference(parentAI.getArgs()) else: for argsInfo in self.childrenAI: argsInfo.__notifyAncestorCompleted(parentAI) def __validateArgsToParent(self): # validate relative to parent arg spec closestParentAI = self.parentAI().getCompleteAI() if closestParentAI is not None: # verify that parent args is a subset of spec given: topicName = stringize(self.topicNameTuple) verifySubset(self.getArgs(), closestParentAI.getArgs(), topicName) verifySubset(self.allRequired, closestParentAI.getReqdArgs(), topicName, ' required args') def __setAllArgs(self, specGiven: ArgSpecGiven): assert specGiven.isComplete() self.allOptional = tuple(specGiven.getOptional()) self.allRequired = specGiven.reqdArgs self.allDocs = specGiven.argsDocs.copy() # doc for each arg self.argsSpecType = self.SPEC_COMPLETE parentArgsInfo = self.parentAI() if parentArgsInfo is None: self.argsAddedToParent = [] else: self.__validateArgsToParent() while not parentArgsInfo.isComplete(): parentArgsInfo = parentArgsInfo.parentAI() self.argsAddedToParent = set(self.getArgs()).difference(parentArgsInfo.getArgs()) # notify our children for childAI in self.childrenAI: childAI.__notifyParentCompleted()