Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, *alternatives):
self.alternatives = [IConstraint(a) for a in alternatives]
self.alternatives = tuple(self.alternatives)
# TODO: taster/opentypes should be a union of the alternatives'
def adapt_obj_to_iconstraint(iface, t):
if iface is not IConstraint:
return None
assert not IConstraint.providedBy(t) # not sure about this
c = constraintMap.get(t, None)
if c:
return c
for (typ, constraintMaker) in constraintTypeMap:
if isinstance(t, typ):
c = constraintMaker(t)
if c:
return c
# RIFoo means accept either a Referenceable that implements RIFoo, or a
# RemoteReference that points to just such a Referenceable. This is
# hooked in by remoteinterface.py, when it calls addToConstraintTypeMap
# we are the only way to make constraints
# TODO: relax this, use schema=Any for the args that don't have
# default values. This would make:
# def foo(a, b=int): return None
# equivalent to:
# def foo(a=Any, b=int): return None
why = "RemoteInterface methods must have default values for all their arguments"
raise InvalidRemoteInterface(why)
self.argumentNames = names
self.argConstraints = {}
self.required = []
for i in range(len(names)):
argname = names[i]
constraint = typeList[i]
if not isinstance(constraint, Optional):
self.required.append(argname)
self.argConstraints[argname] = IConstraint(constraint)
# call the method, its 'return' value is the return constraint
self.responseConstraint = IConstraint(method())
self.options = {} # return, wait, reliable, etc
try:
methodSchema.checkAllArgs(args, kwargs, False)
except Violation as v:
v.setLocation("%s.%s(%s)" % (interfaceName, methodName,
v.getLocation()))
raise
# the Interface gets to constraint the return value too, so
# make a note of it to use later
req.setConstraint(methodSchema.getResponseConstraint())
# if the caller specified a _resultConstraint, that overrides
# the schema's one
if resultConstraint != "none":
# overrides schema
req.setConstraint(IConstraint(resultConstraint))
clid = self.tracker.clid
slicer = call.CallSlicer(reqID, clid, methodName, args, kwargs)
# up to this point, we are not committed to sending anything to the
# far end. The various phases of commitment are:
# 1: once we tell our broker about the PendingRequest, we must
# promise to retire it eventually. Specifically, if we encounter an
# error before we give responsibility to the connection, we must
# retire it ourselves.
# 2: once we start sending the CallSlicer to the other end (in
# particular, once they receive the reqID), they might send us a
# response, so we must be prepared to handle that. Giving the
# PendingRequest to the broker arranges for this to happen.
def __init__(self, keyConstraint, valueConstraint, maxKeys=None):
self.keyConstraint = IConstraint(keyConstraint)
self.valueConstraint = IConstraint(valueConstraint)
self.maxKeys = maxKeys
def checkObject(self, obj, inbound):
# mangled into _RemoteMethodSchema__response or something. When I
# change it to use _response instead, it works.
if _response:
self.responseConstraint = IConstraint(_response)
self.options = {} # return, wait, reliable, etc
if kwargs.has_key("__ignoreUnknown__"):
self.ignoreUnknown = kwargs["__ignoreUnknown__"]
del kwargs["__ignoreUnknown__"]
if kwargs.has_key("__acceptUnknown__"):
self.acceptUnknown = kwargs["__acceptUnknown__"]
del kwargs["__acceptUnknown__"]
for argname, constraint in kwargs.items():
self.argumentNames.append(argname)
constraint = IConstraint(constraint)
self.argConstraints[argname] = constraint
if not isinstance(constraint, Optional):
self.required.append(argname)
def _parseRemoteInterface(self, iname, attrs):
remote_attrs = {}
remote_name = attrs.get("__remote_name__", iname)
# and see if there is a __remote_name__ . We delete it because
# InterfaceClass doesn't like arbitrary attributes
if attrs.has_key("__remote_name__"):
del attrs["__remote_name__"]
# determine all remotely-callable methods
names = [name for name in attrs.keys()
if ((type(attrs[name]) == types.FunctionType and
not name.startswith("_")) or
IConstraint.providedBy(attrs[name]))]
# turn them into constraints. Tag each of them with their name and
# the RemoteInterface they came from.
for name in names:
m = attrs[name]
if not IConstraint.providedBy(m):
m = RemoteMethodSchema(method=m)
m.name = name
m.interface = self
remote_attrs[name] = m
# delete the methods, so zope's InterfaceClass doesn't see them.
# Particularly necessary for things defined with IConstraints.
del attrs[name]
return remote_name, remote_attrs
def __init__(self, keyConstraint, valueConstraint, maxKeys=None):
self.keyConstraint = IConstraint(keyConstraint)
self.valueConstraint = IConstraint(valueConstraint)
self.maxKeys = maxKeys
def checkObject(self, obj, inbound):
# equivalent to:
# def foo(a=Any, b=int): return None
why = "RemoteInterface methods must have default values for all their arguments"
raise InvalidRemoteInterface(why)
self.argumentNames = names
self.argConstraints = {}
self.required = []
for i in range(len(names)):
argname = names[i]
constraint = typeList[i]
if not isinstance(constraint, Optional):
self.required.append(argname)
self.argConstraints[argname] = IConstraint(constraint)
# call the method, its 'return' value is the return constraint
self.responseConstraint = IConstraint(method())
self.options = {} # return, wait, reliable, etc
def __init__(self, *attrTuples, **kwargs):
self.ignoreUnknown = kwargs.get('ignoreUnknown', False)
self.acceptUnknown = kwargs.get('acceptUnknown', False)
self.keys = {}
for name, constraint in (list(attrTuples) +
kwargs.get('attributes', {}).items()):
assert name not in self.keys.keys()
self.keys[name] = IConstraint(constraint)