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":
100 if name
and uri_tail
:
102 return f
"/{name}{uri_tail}"
104 def scan_files(self
, base_dir
):
105 stack
= [base_dir
.iterdir()]
108 path
= next(stack
[-1])
109 except StopIteration:
113 if self
.SCAN_FILES_RECURSIVELY
:
114 stack
.append(path
.iterdir())
115 elif not self
.is_ignored_filename(path
):
116 if path
.suffix
== StaticTemplateFile
.TEMPLATE_SUFFIX
:
117 static_file
= StaticTemplateFile(path
, self
)
119 static_file
= StaticFile(path
)
123 def is_ignored_filename(path
):
124 return path
.name
.startswith(".")
126 async def on_get(self
, req
, resp
):
127 resource
= self
.files_per_uris
[req
.path
]
128 resp
.content_type
= resource
.media_type
129 resp
.data
= resource
.get().encode()
130 resp
.status
= HTTP_OK
133 class HubApp(BaseHubApp
):
134 def __init__(self
, app
, base_dir
):
135 super().__init
__(app
, base_dir
)
136 self
.wsh
= WebSocketHub(self
)
137 self
.ws_uris
= {self.uri_from(f"ws_{value}
"): value for value in ("client
", "master
")}
138 for uri, suffix in self.ws_uris.items():
139 app.add_route(uri, self.wsh, suffix=suffix)
142 class RootApp(BaseHubApp):
143 SCAN_FILES_RECURSIVELY = False
145 def __init__(self, app, base_dir):
146 super().__init__(app, base_dir, "")
149 def is_master_uri(uri_tail):