/
handlers.py
96 lines (79 loc) · 3.27 KB
/
handlers.py
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
"""Tornado handlers for kernel specifications.
Preliminary documentation at https://github.com/ipython/ipython/wiki/IPEP-25%3A-Registry-of-installed-kernels#rest-api
"""
# Copyright (c) Jupyter Development Team.
# Distributed under the terms of the Modified BSD License.
import glob
import json
import os
pjoin = os.path.join
from tornado import web
from ...base.handlers import APIHandler
from ...utils import ensure_async, url_path_join, url_unescape
def kernelspec_model(handler, name, spec_dict, resource_dir):
"""Load a KernelSpec by name and return the REST API model"""
d = {"name": name, "spec": spec_dict, "resources": {}}
# Add resource files if they exist
resource_dir = resource_dir
for resource in ["kernel.js", "kernel.css"]:
if os.path.exists(pjoin(resource_dir, resource)):
d["resources"][resource] = url_path_join(
handler.base_url, "kernelspecs", name, resource
)
for logo_file in glob.glob(pjoin(resource_dir, "logo-*")):
fname = os.path.basename(logo_file)
no_ext, _ = os.path.splitext(fname)
d["resources"][no_ext] = url_path_join(handler.base_url, "kernelspecs", name, fname)
return d
def is_kernelspec_model(spec_dict):
"""Returns True if spec_dict is already in proper form. This will occur when using a gateway."""
return (
isinstance(spec_dict, dict)
and "name" in spec_dict
and "spec" in spec_dict
and "resources" in spec_dict
)
class MainKernelSpecHandler(APIHandler):
@web.authenticated
async def get(self):
ksm = self.kernel_spec_manager
km = self.kernel_manager
model = {}
model["default"] = km.default_kernel_name
model["kernelspecs"] = specs = {}
kspecs = await ensure_async(ksm.get_all_specs())
for kernel_name, kernel_info in kspecs.items():
try:
if is_kernelspec_model(kernel_info):
d = kernel_info
else:
d = kernelspec_model(
self, kernel_name, kernel_info["spec"], kernel_info["resource_dir"]
)
except Exception:
self.log.error("Failed to load kernel spec: '%s'", kernel_name, exc_info=True)
continue
specs[kernel_name] = d
self.set_header("Content-Type", "application/json")
self.finish(json.dumps(model))
class KernelSpecHandler(APIHandler):
@web.authenticated
async def get(self, kernel_name):
ksm = self.kernel_spec_manager
kernel_name = url_unescape(kernel_name)
try:
spec = await ensure_async(ksm.get_kernel_spec(kernel_name))
except KeyError as e:
raise web.HTTPError(404, u"Kernel spec %s not found" % kernel_name) from e
if is_kernelspec_model(spec):
model = spec
else:
model = kernelspec_model(self, kernel_name, spec.to_dict(), spec.resource_dir)
self.set_header("Content-Type", "application/json")
self.finish(json.dumps(model))
# URL to handler mappings
kernel_name_regex = r"(?P<kernel_name>[\w\.\-%]+)"
default_handlers = [
(r"/api/kernelspecs", MainKernelSpecHandler),
(r"/api/kernelspecs/%s" % kernel_name_regex, KernelSpecHandler),
]