Differential D2636 Diff 9462 swh/core/api/negotiation.py

Changeset View

Standalone View

swh/core/api/negotiation.py

Show All 21 Lines
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE		# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.		# SOFTWARE.
#		#

from collections import defaultdict		from collections import defaultdict
from decorator import decorator		from decorator import decorator
from inspect import getcallargs		from inspect import getcallargs

from typing import Any, List, Optional		from typing import Any, List, Optional, Callable, \
		Type, NoReturn, DefaultDict

		from requests import Response


class FormatterNotFound(Exception):		class FormatterNotFound(Exception):
pass		pass


class Formatter:		class Formatter:
format = None # type: Optional[str]		format: Optional[str] = None
mimetypes = [] # type: List[Any]		mimetypes: List[str] = []

def __init__(self, request_mimetype=None):		def __init__(self, request_mimetype: Optional[str] = None) -> None:
if request_mimetype is None or request_mimetype not in self.mimetypes:		if request_mimetype is None or request_mimetype not in self.mimetypes:
try:		try:
self.response_mimetype = self.mimetypes[0]		self.response_mimetype = self.mimetypes[0]
except IndexError:		except IndexError:
raise NotImplementedError(		raise NotImplementedError(
"%s.mimetypes should be a non-empty list" %		"%s.mimetypes should be a non-empty list" %
self.__class__.__name__)		self.__class__.__name__)
else:		else:
self.response_mimetype = request_mimetype		self.response_mimetype = request_mimetype

def configure(self):		def configure(self) -> None:
pass		pass

def render(self, obj):		def render(self, obj: Any) -> bytes:
		vlorentzUnsubmitted Done Inline Actions `bytes` vlorentz: `bytes`
raise NotImplementedError(		raise NotImplementedError(
"render() should be implemented by Formatter subclasses")		"render() should be implemented by Formatter subclasses")

def __call__(self, obj):		def __call__(self, obj: Any) -> Response:
return self._make_response(		return self._make_response(
self.render(obj), content_type=self.response_mimetype)		self.render(obj), content_type=self.response_mimetype)

def _make_response(self, body, content_type):		def _make_response(self, body: bytes, content_type: str) -> Response:
		vlorentzUnsubmitted Done Inline Actions `bytes` vlorentz: `bytes`
raise NotImplementedError(		raise NotImplementedError(
"_make_response() should be implemented by "		"_make_response() should be implemented by "
"framework-specific subclasses of Formatter"		"framework-specific subclasses of Formatter"
)		)


class Negotiator:		class Negotiator:

def __init__(self, func):		def __init__(self, func: Callable[..., Any]) -> None:
self.func = func		self.func = func
self._formatters = []		self._formatters: List[Type[Formatter]] = []
self._formatters_by_format = defaultdict(list)		self._formatters_by_format: DefaultDict = defaultdict(list)
self._formatters_by_mimetype = defaultdict(list)		self._formatters_by_mimetype: DefaultDict = defaultdict(list)

def __call__(self, args, *kwargs):		def __call__(self, args, *kwargs) -> Response:
result = self.func(args, *kwargs)		result = self.func(args, *kwargs)
format = getcallargs(self.func, args, *kwargs).get('format')		format = getcallargs(self.func, args, *kwargs).get('format')
mimetype = self.best_mimetype()		mimetype = self.best_mimetype()

try:		try:
formatter = self.get_formatter(format, mimetype)		formatter = self.get_formatter(format, mimetype)
except FormatterNotFound as e:		except FormatterNotFound as e:
return self._abort(404, str(e))		return self._abort(404, str(e))

return formatter(result)		return formatter(result)

def register_formatter(self, formatter, args, *kwargs):		def register_formatter(self, formatter: Type[Formatter],
		args, *kwargs) -> None:
self._formatters.append(formatter)		self._formatters.append(formatter)
self._formatters_by_format[formatter.format].append(		self._formatters_by_format[formatter.format].append(
(formatter, args, kwargs))		(formatter, args, kwargs))
for mimetype in formatter.mimetypes:		for mimetype in formatter.mimetypes:
self._formatters_by_mimetype[mimetype].append(		self._formatters_by_mimetype[mimetype].append(
(formatter, args, kwargs))		(formatter, args, kwargs))

def get_formatter(self, format=None, mimetype=None):		def get_formatter(self, format: Optional[str] = None,
		mimetype: Optional[str] = None) -> Formatter:
if format is None and mimetype is None:		if format is None and mimetype is None:
raise TypeError(		raise TypeError(
"get_formatter expects one of the 'format' or 'mimetype' "		"get_formatter expects one of the 'format' or 'mimetype' "
"kwargs to be set")		"kwargs to be set")

if format is not None:		if format is not None:
try:		try:
# the first added will be the most specific		# the first added will be the most specific
Show All 11 Lines	def get_formatter(self, format: Optional[str] = None,
raise FormatterNotFound(		raise FormatterNotFound(
"Formatter for mimetype '%s' not found!" % mimetype)		"Formatter for mimetype '%s' not found!" % mimetype)

formatter = formatter_cls(request_mimetype=mimetype)		formatter = formatter_cls(request_mimetype=mimetype)
formatter.configure(args, *kwargs)		formatter.configure(args, *kwargs)
return formatter		return formatter

@property		@property
def accept_mimetypes(self):		def accept_mimetypes(self) -> List[str]:
return [m for f in self._formatters for m in f.mimetypes]		return [m for f in self._formatters for m in f.mimetypes]

def best_mimetype(self):		def best_mimetype(self) -> str:
raise NotImplementedError(		raise NotImplementedError(
"best_mimetype() should be implemented in "		"best_mimetype() should be implemented in "
"framework-specific subclasses of Negotiator"		"framework-specific subclasses of Negotiator"
)		)

def _abort(self, status_code, err=None):		def _abort(self, status_code: int, err: Optional[str] = None) -> NoReturn:
		vlorentzUnsubmitted Done Inline Actions missing type of `err` vlorentz: missing type of `err`
		vlorentzUnsubmitted Done Inline Actions `err` is optional vlorentz: `err` is optional
		legauAuthorUnsubmitted Done Inline Actions If this err should be annoted Optional, should all other parameters with default values be Optional as well ? legau: If this err should be annoted Optional, should all other parameters with default values be…
		vlorentzUnsubmitted Done Inline Actions No, `Optional` means they can be None ( https://docs.python.org/3/library/typing.html#typing.Optional ) vlorentz: No, `Optional` means they can be None ( https://docs.python.org/3/library/typing.html#typing.
		legauAuthorUnsubmitted Done Inline Actions Yes I meant for the other parameters where None is the default parameters. For example def get_formatter(self, format: str = None, mimetype: str = None) -> Formatter: line 102. legau: Yes I meant for the other parameters where None is the default parameters. For example ```…
		vlorentzUnsubmitted Done Inline Actions then yes vlorentz: then yes
raise NotImplementedError(		raise NotImplementedError(
"_abort() should be implemented in framework-specific "		"_abort() should be implemented in framework-specific "
"subclasses of Negotiator"		"subclasses of Negotiator"
)		)


def negotiate(negotiator_cls, formatter_cls, args, *kwargs):		def negotiate(negotiator_cls: Type[Negotiator], formatter_cls: Type[Formatter],
		args, *kwargs) -> Callable:
def _negotiate(f, args, *kwargs):		def _negotiate(f, args, *kwargs):
return f.negotiator(args, *kwargs)		return f.negotiator(args, *kwargs)

def decorate(f):		def decorate(f):
if not hasattr(f, 'negotiator'):		if not hasattr(f, 'negotiator'):
f.negotiator = negotiator_cls(f)		f.negotiator = negotiator_cls(f)

f.negotiator.register_formatter(formatter_cls, args, *kwargs)		f.negotiator.register_formatter(formatter_cls, args, *kwargs)
return decorator(_negotiate, f)		return decorator(_negotiate, f)

return decorate		return decorate