-
Notifications
You must be signed in to change notification settings - Fork 46
/
labextension-server.py
106 lines (87 loc) · 3.04 KB
/
labextension-server.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
97
98
99
100
101
102
103
104
105
106
from starlette.applications import Starlette
from starlette.responses import (
JSONResponse,
RedirectResponse,
StreamingResponse,
)
from starlette.staticfiles import StaticFiles
from starlette.routing import Route
import os
import zipfile
import io
def get_labextensions_dir(app_name):
home_dir = os.path.expanduser("~")
return os.path.join(
home_dir,
".sepal-data",
"sepal",
"jupyter",
"current-kernels",
app_name,
"venv",
"share",
"jupyter",
"labextensions",
)
class DynamicStaticFiles:
def __init__(self):
pass
async def __call__(self, scope, receive, send):
request = scope["path_params"]
app_name = request["app_name"]
labextensions_dir = get_labextensions_dir(app_name)
# Adjust the path in scope to be relative to the labextensions directory
original_path = scope["path"]
relative_path = original_path.split(f"/app/{app_name}/labextensions")[-1] or "/"
# Create a new scope with the updated path
scope = dict(scope)
scope["path"] = relative_path
static_app = StaticFiles(directory=labextensions_dir)
await static_app(scope, receive, send)
async def generate_manifest(request):
app_name = request.path_params["app_name"]
labextensions_dir = get_labextensions_dir(app_name)
extensions = []
for root, dirs, files in os.walk(labextensions_dir):
if "package.json" in files:
rel_dir = os.path.relpath(root, labextensions_dir)
extensions.append(rel_dir)
return JSONResponse(extensions)
async def download_all_extensions(request):
app_name = request.path_params["app_name"]
labextensions_dir = get_labextensions_dir(app_name)
zip_stream = io.BytesIO()
with zipfile.ZipFile(zip_stream, "w", zipfile.ZIP_DEFLATED) as zip_file:
for root, dirs, files in os.walk(labextensions_dir):
for file in files:
file_path = os.path.join(root, file)
arcname = os.path.relpath(file_path, labextensions_dir)
zip_file.write(file_path, arcname)
zip_stream.seek(0)
return StreamingResponse(
zip_stream,
media_type="application/zip",
headers={"Content-Disposition": "attachment; filename=labextensions.zip"},
)
# Homepage redirection
async def homepage(request):
app_name = request.path_params["app_name"]
return RedirectResponse(url=f"/app/{app_name}/labextensions/")
# Define the routes
routes = [
Route(
"/app/{app_name}/labextensions/download_all_extensions", download_all_extensions
),
Route("/app/{app_name}/manifest.json", generate_manifest),
Route("/app/{app_name}/labextensions/{path:path}", DynamicStaticFiles()),
Route("/app/{app_name}/", homepage),
]
# Create the Starlette application
app = Starlette(
debug=True,
routes=routes,
)
# If you want to run the app using 'python server.py'
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)