1 from pathlib
import Path
3 from falcon
.constants
import MEDIA_HTML
, MEDIA_JS
, MEDIA_TEXT
4 from falcon
.status_codes
import HTTP_OK
6 from .websocket
import WebSocketHub
7 from .utils
import scramble
13 media_types_per_suffix
= {
20 def __init__(self
, path
):
22 self
.media_type
= self
.get_media_type(path
)
27 def get_media_type(cls
, path
):
28 return cls
.media_types_per_suffix
[path
.suffix
]
31 mtime
= self
.path
.stat().st_mtime
32 if mtime
!= self
.mtime
:
33 with open(self
.path
) as fh
:
34 self
.content
= fh
.read()
39 return f
"<{type(self).__name__} {self.path}>"
42 class StaticTemplateFile(StaticFile
):
43 TEMPLATE_SUFFIX
= ".j2"
45 def __init__(self
, path
, hubapp
):
46 super().__init
__(path
)
48 self
.context
= {"hubapp": self.hubapp}
49 self
.template
= hubapp
.app
.env
.get_template(
50 str(path
.relative_to(hubapp
.app
.env
.loader
.searchpath
[0]))
54 return self
.template
.render(self
.context
)
57 def get_media_type(cls
, path
):
58 assert path
.suffix
== cls
.TEMPLATE_SUFFIX
59 return cls
.media_types_per_suffix
[path
.suffixes
[-2]]
63 SCAN_FILES_RECURSIVELY
= True
65 def __init__(self
, app
, base_dir
, name
=None):
67 self
.base_dir
= base_dir
68 self
.name
= name
if name
is not None else base_dir
.name
69 self
.app
.hubapps
[self
.name
] = self
70 self
.files_per_uris
= {
71 self
.uri_from(file.path
): file for file in self
.scan_files(base_dir
)
73 for uri
in self
.files_per_uris
:
74 app
.add_route(uri
, self
)
77 def is_master_uri(uri_tail
):
79 start
= uri_tail
.rfind(slash
)
80 pos
= uri_tail
.find(".", start
if start
>= 0 else 0)
82 uri_tail
[start
+ len(slash
):pos
] == "master"
83 or "master" in uri_tail
[:start
].split(slash
)
86 def uri_from(self
, path
):
87 if isinstance(path
, Path
) and path
.is_absolute():
88 uri_tail
= str(path
.relative_to(self
.base_dir
))
91 suffix
= StaticTemplateFile
.TEMPLATE_SUFFIX
92 if uri_tail
.endswith(suffix
):
93 uri_tail
= uri_tail
[:-len(suffix
)]
95 if self
.is_master_uri(uri_tail
):
96 uri_tail
= scramble(self
.app
.secret
, uri_tail
)
97 elif uri_tail
== "index.html":
102 if name
and uri_tail
:
104 return f
"/{name}{uri_tail}"
106 def scan_files(self
, base_dir
):
107 stack
= [base_dir
.iterdir()]
110 path
= next(stack
[-1])
111 except StopIteration:
115 if self
.SCAN_FILES_RECURSIVELY
:
116 stack
.append(path
.iterdir())
117 elif not self
.is_ignored_filename(path
):
118 if path
.suffix
== StaticTemplateFile
.TEMPLATE_SUFFIX
:
119 static_file
= StaticTemplateFile(path
, self
)
121 static_file
= StaticFile(path
)
125 def is_ignored_filename(path
):
126 return path
.name
.startswith(".")
128 async def on_get(self
, req
, resp
):
129 resource
= self
.files_per_uris
[req
.path
]
130 resp
.content_type
= resource
.media_type
131 resp
.data
= resource
.get().encode()
132 resp
.status
= HTTP_OK
135 class HubApp(BaseHubApp
):
136 def __init__(self
, app
, base_dir
):
137 super().__init
__(app
, base_dir
)
138 self
.wsh
= WebSocketHub(self
)
139 self
.ws_uris
= {self.uri_from(f"ws_{value}
"): value for value in ("client
", "master
")}
140 for uri, suffix in self.ws_uris.items():
141 app.add_route(uri, self.wsh, suffix=suffix)
144 class RootApp(BaseHubApp):
145 SCAN_FILES_RECURSIVELY = False
147 def __init__(self, app, base_dir):
148 super().__init__(app, base_dir, "root
")
151 def is_master_uri(uri_tail):