###############################################################################
#
# The MIT License (MIT)
#
# Copyright (c) Crossbar.io Technologies GmbH
#
# 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 typing import Optional, Union
from autobahn.util import public
from autobahn.wamp.types import RegisterOptions, SubscribeOptions
__all__ = (
'Pattern',
'register',
'subscribe',
'error',
'convert_starred_uri'
)
[docs]def convert_starred_uri(uri: str):
"""
Convert a starred URI to a standard WAMP URI and a detected matching
policy. A starred URI is one that may contain the character '*' used
to mark URI wildcard components or URI prefixes. Starred URIs are
more comfortable / intuitive to use at the user/API level, but need
to be converted for use on the wire (WAMP protocol level).
This function takes a possibly starred URI, detects the matching policy
implied by stars, and returns a pair (uri, match) with any stars
removed from the URI and the detected matching policy.
An URI like 'com.example.topic1' (without any stars in it) is
detected as an exact-matching URI.
An URI like 'com.example.*' (with exactly one star at the very end)
is detected as a prefix-matching URI on 'com.example.'.
An URI like 'com.*.foobar.*' (with more than one star anywhere) is
detected as a wildcard-matching URI on 'com..foobar.' (in this example,
there are two wildcard URI components).
Note that an URI like 'com.example.*' is always detected as
a prefix-matching URI 'com.example.'. You cannot express a wildcard-matching
URI 'com.example.' using the starred URI notation! A wildcard matching on
'com.example.' is different from prefix-matching on 'com.example.' (which
matches a strict superset of the former!). This is one reason we don't use
starred URIs for WAMP at the protocol level.
"""
assert(type(uri) == str)
cnt_stars = uri.count('*')
if cnt_stars == 0:
match = 'exact'
elif cnt_stars == 1 and uri[-1] == '*':
match = 'prefix'
uri = uri[:-1]
else:
match = 'wildcard'
uri = uri.replace('*', '')
return uri, match
[docs]@public
class Pattern(object):
"""
A WAMP URI Pattern.
.. todo::
* suffix matches
* args + kwargs
* uuid converter
* multiple URI patterns per decorated object
* classes: Pattern, EndpointPattern, ..
"""
URI_TARGET_ENDPOINT = 1
URI_TARGET_HANDLER = 2
URI_TARGET_EXCEPTION = 3
URI_TYPE_EXACT = 1
URI_TYPE_PREFIX = 2
URI_TYPE_WILDCARD = 3
_URI_COMPONENT = re.compile(r"^[a-z0-9][a-z0-9_\-]*$")
"""
Compiled regular expression for a WAMP URI component.
"""
_URI_NAMED_COMPONENT = re.compile(r"^<([a-z][a-z0-9_]*)>$")
"""
Compiled regular expression for a named WAMP URI component.
.. note::
This pattern is stricter than a general WAMP URI component since a valid Python identifier is required.
"""
_URI_NAMED_CONVERTED_COMPONENT = re.compile(r"^<([a-z][a-z0-9_]*):([a-z]*)>$")
"""
Compiled regular expression for a named and type-converted WAMP URI component.
.. note::
This pattern is stricter than a general WAMP URI component since a valid Python identifier is required.
"""
def __init__(self, uri: str, target: int, options: Optional[Union[SubscribeOptions, RegisterOptions]] = None,
check_types: Optional[bool] = None):
"""
:param uri: The URI or URI pattern, e.g. ``"com.myapp.product.<product:int>.update"``.
:type uri: str
:param target: The target for this pattern: a procedure endpoint (a callable),
an event handler (a callable) or an exception (a class).
:type target: callable or obj
:param options: An optional options object
:type options: None or RegisterOptions or SubscribeOptions
:param check_types: Enable automatic type checking against (Python 3.5+) type hints
specified on the ``endpoint`` callable. Types are checked at run-time on each
invocation of the ``endpoint`` callable. When a type mismatch occurs, the error
is forwarded to the callee code in ``onUserError`` override method of
:class:`autobahn.wamp.protocol.ApplicationSession`. An error
of type :class:`autobahn.wamp.exception.TypeCheckError` is also raised and
returned to the caller (via the router).
:type check_types: bool
"""
assert (type(uri) == str)
assert (len(uri) > 0)
assert (target in [Pattern.URI_TARGET_ENDPOINT,
Pattern.URI_TARGET_HANDLER,
Pattern.URI_TARGET_EXCEPTION])
if target == Pattern.URI_TARGET_ENDPOINT:
assert (options is None or type(options) == RegisterOptions)
elif target == Pattern.URI_TARGET_HANDLER:
assert (options is None or type(options) == SubscribeOptions)
else:
options = None
components = uri.split('.')
_URI_COMP_CHARS = r'[^\s\.#]+'
# _URI_COMP_CHARS = r'[\da-z_]+'
# _URI_COMP_CHARS = r'[a-z0-9][a-z0-9_\-]*'
pl = []
nc = {}
group_count = 0
for i in range(len(components)):
component = components[i]
match = Pattern._URI_NAMED_CONVERTED_COMPONENT.match(component)
if match:
name, comp_type = match.groups()
if comp_type not in ['str', 'string', 'int', 'suffix']:
raise TypeError("invalid URI")
if comp_type == 'suffix' and i != len(components) - 1:
raise TypeError("invalid URI")
if name in nc:
raise TypeError("invalid URI")
if comp_type in ['str', 'string', 'suffix']:
nc[name] = str
elif comp_type == 'int':
nc[name] = int
else:
# should not arrive here
raise TypeError("logic error")
pl.append("(?P<{}>{})".format(name, _URI_COMP_CHARS))
group_count += 1
continue
match = Pattern._URI_NAMED_COMPONENT.match(component)
if match:
name = match.groups()[0]
if name in nc:
raise TypeError("invalid URI")
nc[name] = str
pl.append("(?P<{}>{})".format(name, _URI_COMP_CHARS))
group_count += 1
continue
match = Pattern._URI_COMPONENT.match(component)
if match:
pl.append(component)
continue
if component == '':
group_count += 1
pl.append(r"({})".format(_URI_COMP_CHARS))
nc[group_count] = str
continue
raise TypeError("invalid URI")
if nc:
# URI pattern
self._type = Pattern.URI_TYPE_WILDCARD
p = "^" + r"\.".join(pl) + "$"
self._pattern = re.compile(p)
self._names = nc
else:
# exact URI
self._type = Pattern.URI_TYPE_EXACT
self._pattern = None
self._names = None
self._uri = uri
self._target = target
self._options = options
self._check_types = check_types
@public
@property
def options(self):
"""
Returns the Options instance (if present) for this pattern.
:return: None or the Options instance
:rtype: None or RegisterOptions or SubscribeOptions
"""
return self._options
@public
@property
def uri_type(self):
"""
Returns the URI type of this pattern
:return:
:rtype: Pattern.URI_TYPE_EXACT, Pattern.URI_TYPE_PREFIX or Pattern.URI_TYPE_WILDCARD
"""
return self._type
[docs] @public
def uri(self):
"""
Returns the original URI (pattern) for this pattern.
:returns: The URI (pattern), e.g. ``"com.myapp.product.<product:int>.update"``.
:rtype: str
"""
return self._uri
[docs] def match(self, uri):
"""
Match the given (fully qualified) URI according to this pattern
and return extracted args and kwargs.
:param uri: The URI to match, e.g. ``"com.myapp.product.123456.update"``.
:type uri: str
:returns: A tuple ``(args, kwargs)``
:rtype: tuple
"""
args = []
kwargs = {}
if self._type == Pattern.URI_TYPE_EXACT:
return args, kwargs
elif self._type == Pattern.URI_TYPE_WILDCARD:
match = self._pattern.match(uri)
if match:
for key in self._names:
val = match.group(key)
val = self._names[key](val)
kwargs[key] = val
return args, kwargs
else:
raise ValueError('no match')
[docs] @public
def is_endpoint(self):
"""
Check if this pattern is for a procedure endpoint.
:returns: ``True``, iff this pattern is for a procedure endpoint.
:rtype: bool
"""
return self._target == Pattern.URI_TARGET_ENDPOINT
[docs] @public
def is_handler(self):
"""
Check if this pattern is for an event handler.
:returns: ``True``, iff this pattern is for an event handler.
:rtype: bool
"""
return self._target == Pattern.URI_TARGET_HANDLER
[docs] @public
def is_exception(self):
"""
Check if this pattern is for an exception.
:returns: ``True``, iff this pattern is for an exception.
:rtype: bool
"""
return self._target == Pattern.URI_TARGET_EXCEPTION
[docs]@public
def register(uri: Optional[str], options: Optional[RegisterOptions] = None, check_types: Optional[bool] = None):
"""
Decorator for WAMP procedure endpoints.
:param uri:
:type uri: str
:param options:
:type options: None or RegisterOptions
:param check_types: Enable automatic type checking against (Python 3.5+) type hints
specified on the ``endpoint`` callable. Types are checked at run-time on each
invocation of the ``endpoint`` callable. When a type mismatch occurs, the error
is forwarded to the callee code in ``onUserError`` override method of
:class:`autobahn.wamp.protocol.ApplicationSession`. An error
of type :class:`autobahn.wamp.exception.TypeCheckError` is also raised and
returned to the caller (via the router).
:type check_types: bool
"""
def decorate(f):
assert(callable(f))
if uri is None:
real_uri = '{}'.format(f.__name__)
else:
real_uri = uri
if not hasattr(f, '_wampuris'):
f._wampuris = []
f._wampuris.append(Pattern(real_uri, Pattern.URI_TARGET_ENDPOINT, options, check_types))
return f
return decorate
[docs]@public
def subscribe(uri: Optional[str], options: Optional[SubscribeOptions] = None, check_types: Optional[bool] = None):
"""
Decorator for WAMP event handlers.
:param uri:
:type uri: str
:param options:
:type options: None or SubscribeOptions
:param check_types: Enable automatic type checking against (Python 3.5+) type hints
specified on the ``endpoint`` callable. Types are checked at run-time on each
invocation of the ``endpoint`` callable. When a type mismatch occurs, the error
is forwarded to the callee code in ``onUserError`` override method of
:class:`autobahn.wamp.protocol.ApplicationSession`. An error
of type :class:`autobahn.wamp.exception.TypeCheckError` is also raised and
returned to the caller (via the router).
:type check_types: bool
"""
def decorate(f):
assert(callable(f))
if not hasattr(f, '_wampuris'):
f._wampuris = []
f._wampuris.append(Pattern(uri, Pattern.URI_TARGET_HANDLER, options, check_types))
return f
return decorate
[docs]@public
def error(uri: str):
"""
Decorator for WAMP error classes.
"""
def decorate(cls):
assert(issubclass(cls, Exception))
if not hasattr(cls, '_wampuris'):
cls._wampuris = []
cls._wampuris.append(Pattern(uri, Pattern.URI_TARGET_EXCEPTION))
return cls
return decorate