File: routing.py

package info (click to toggle)
dask.distributed 2022.12.1%2Bds.1-3
  • links: PTS, VCS
  • area: main
  • in suites: bookworm
  • size: 10,164 kB
  • sloc: python: 81,938; javascript: 1,549; makefile: 228; sh: 100
file content (69 lines) | stat: -rw-r--r-- 2,548 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
from __future__ import annotations

import os

import tornado.httputil
import tornado.routing
from tornado import web


def _descend_routes(router, routers=None, out=None):
    if routers is None:
        routers = set()
    if out is None:
        out = set()
    if router in routers:
        return
    routers.add(router)
    for rule in list(router.named_rules.values()) + router.rules:
        if isinstance(rule.matcher, tornado.routing.PathMatches):
            if issubclass(rule.target, tornado.web.StaticFileHandler):
                prefix = rule.matcher.regex.pattern.rstrip("(.*)$").rstrip("/")
                path = rule.target_kwargs["path"]
                for d, _, files in os.walk(path):
                    for fn in files:
                        fullpath = d + "/" + fn
                        ourpath = fullpath.replace(path, prefix).replace("\\", "/")
                        out.add(ourpath)
            else:
                out.add(rule.matcher.regex.pattern.rstrip("$"))
        if isinstance(rule.target, tornado.routing.RuleRouter):
            _descend_routes(rule.target, routers, out)


class DirectoryHandler(web.RequestHandler):
    """Crawls the HTTP application to find all routes"""

    def get(self):
        out = set()
        routers = set()
        for app in self.application.applications + [self.application]:
            if "bokeh" in str(app):
                out.update(set(app.app_paths))
            else:
                _descend_routes(app.default_router, routers, out)
                _descend_routes(app.wildcard_router, routers, out)
        self.write({"paths": sorted(out)})


class RoutingApplication(web.Application):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.applications = []
        self.add_handlers(".*$", [(r"/sitemap.json", DirectoryHandler)])

    def find_handler(  # type: ignore[no-untyped-def]
        self, request: tornado.httputil.HTTPServerRequest, **kwargs
    ):
        handler = super().find_handler(request, **kwargs)
        if handler and not issubclass(handler.handler_class, web.ErrorHandler):
            return handler
        else:
            for app in self.applications:
                handler = app.find_handler(request, **kwargs) or handler
                if handler and not issubclass(handler.handler_class, web.ErrorHandler):
                    break
            return handler

    def add_application(self, application: web.Application) -> None:
        self.applications.append(application)