(self, prefix, app, **options)
| 696 | return decorator |
| 697 | |
| 698 | def _mount_wsgi(self, prefix, app, **options): |
| 699 | segments = [p for p in prefix.split('/') if p] |
| 700 | if not segments: |
| 701 | raise ValueError('WSGI applications cannot be mounted to "/".') |
| 702 | path_depth = len(segments) |
| 703 | |
| 704 | def mountpoint_wrapper(): |
| 705 | try: |
| 706 | request.path_shift(path_depth) |
| 707 | rs = HTTPResponse([]) |
| 708 | |
| 709 | def start_response(status, headerlist, exc_info=None): |
| 710 | if exc_info: |
| 711 | _raise(*exc_info) |
| 712 | if py3k: |
| 713 | # Errors here mean that the mounted WSGI app did not |
| 714 | # follow PEP-3333 (which requires latin1) or used a |
| 715 | # pre-encoding other than utf8 :/ |
| 716 | status = status.encode('latin1').decode('utf8') |
| 717 | headerlist = [(k, v.encode('latin1').decode('utf8')) |
| 718 | for (k, v) in headerlist] |
| 719 | rs.status = status |
| 720 | for name, value in headerlist: |
| 721 | rs.add_header(name, value) |
| 722 | return rs.body.append |
| 723 | |
| 724 | body = app(request.environ, start_response) |
| 725 | rs.body = itertools.chain(rs.body, body) if rs.body else body |
| 726 | return rs |
| 727 | finally: |
| 728 | request.path_shift(-path_depth) |
| 729 | |
| 730 | options.setdefault('skip', True) |
| 731 | options.setdefault('method', 'PROXY') |
| 732 | options.setdefault('mountpoint', {'prefix': prefix, 'target': app}) |
| 733 | options['callback'] = mountpoint_wrapper |
| 734 | |
| 735 | self.route('/%s/<:re:.*>' % '/'.join(segments), **options) |
| 736 | if not prefix.endswith('/'): |
| 737 | self.route('/' + '/'.join(segments), **options) |
| 738 | |
| 739 | def _mount_app(self, prefix, app, **options): |
| 740 | if app in self._mounts or '_mount.app' in app.config: |
no test coverage detected