forked from reactive-python/reactpy-django
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathutils.py
384 lines (316 loc) · 13.3 KB
/
utils.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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
from __future__ import annotations
import contextlib
import inspect
import logging
import os
import re
from asyncio import iscoroutinefunction
from fnmatch import fnmatch
from importlib import import_module
from typing import Any, Callable, Sequence
from asgiref.sync import async_to_sync
from channels.db import database_sync_to_async
from django.db.models import ManyToManyField, ManyToOneRel, prefetch_related_objects
from django.db.models.base import Model
from django.db.models.query import QuerySet
from django.http import HttpRequest, HttpResponse
from django.template import engines
from django.utils.encoding import smart_str
from django.views import View
from reactpy.core.layout import Layout
from reactpy.types import ComponentConstructor
from reactpy_django.exceptions import (
ComponentDoesNotExistError,
ComponentParamError,
ViewDoesNotExistError,
)
_logger = logging.getLogger(__name__)
_component_tag = r"(?P<tag>component)"
_component_path = r"""(?P<path>"[^"'\s]+"|'[^"'\s]+')"""
_component_offline_kwarg = (
rf"""(\s*offline\s*=\s*{_component_path.replace(r"<path>", r"<offline_path>")})"""
)
_component_generic_kwarg = r"""(\s*.*?)"""
COMMENT_REGEX = re.compile(r"<!--[\s\S]*?-->")
COMPONENT_REGEX = re.compile(
r"{%\s*"
+ _component_tag
+ r"\s*"
+ _component_path
+ rf"({_component_offline_kwarg}|{_component_generic_kwarg})*?"
+ r"\s*%}"
)
async def render_view(
view: Callable | View,
request: HttpRequest,
args: Sequence,
kwargs: dict,
) -> HttpResponse:
"""Ingests a Django view (class or function) and returns an HTTP response object."""
# Convert class-based view to function-based view
if getattr(view, "as_view", None):
view = view.as_view() # type: ignore[union-attr]
# Async function view
if iscoroutinefunction(view):
response = await view(request, *args, **kwargs)
# Sync function view
else:
response = await database_sync_to_async(view)(request, *args, **kwargs)
# TemplateView
if getattr(response, "render", None):
response = await database_sync_to_async(response.render)()
return response
def register_component(component: ComponentConstructor | str):
"""Adds a component to the list of known registered components.
Args:
component: The component to register. Can be a component function or dotted path to a component.
"""
from reactpy_django.config import (
REACTPY_FAILED_COMPONENTS,
REACTPY_REGISTERED_COMPONENTS,
)
dotted_path = (
component if isinstance(component, str) else generate_obj_name(component)
)
try:
REACTPY_REGISTERED_COMPONENTS[dotted_path] = import_dotted_path(dotted_path)
except AttributeError as e:
REACTPY_FAILED_COMPONENTS.add(dotted_path)
raise ComponentDoesNotExistError(
f"Error while fetching '{dotted_path}'. {(str(e).capitalize())}."
) from e
def register_iframe(view: Callable | View | str):
"""Registers a view to be used as an iframe component.
Args:
view: The view to register. Can be a function or class based view, or a dotted path to a view.
"""
from reactpy_django.config import REACTPY_REGISTERED_IFRAME_VIEWS
if hasattr(view, "view_class"):
view = view.view_class
dotted_path = view if isinstance(view, str) else generate_obj_name(view)
try:
REACTPY_REGISTERED_IFRAME_VIEWS[dotted_path] = import_dotted_path(dotted_path)
except AttributeError as e:
raise ViewDoesNotExistError(
f"Error while fetching '{dotted_path}'. {(str(e).capitalize())}."
) from e
def import_dotted_path(dotted_path: str) -> Callable:
"""Imports a dotted path and returns the callable."""
module_name, component_name = dotted_path.rsplit(".", 1)
try:
module = import_module(module_name)
except ImportError as error:
raise RuntimeError(
f"Failed to import {module_name!r} while loading {component_name!r}"
) from error
return getattr(module, component_name)
class RootComponentFinder:
"""Searches Django templates to find and register all root components.
This should only be `run` once on startup to maintain synchronization during mulitprocessing.
"""
def run(self):
"""Registers all ReactPy components found within Django templates."""
# Get all template folder paths
paths = self.get_paths()
# Get all HTML template files
templates = self.get_templates(paths)
# Get all components
components = self.get_components(templates)
# Register all components
self.register_components(components)
def get_loaders(self):
"""Obtains currently configured template loaders."""
template_source_loaders = []
for e in engines.all():
if hasattr(e, "engine"):
template_source_loaders.extend(
e.engine.get_template_loaders(e.engine.loaders)
)
loaders = []
for loader in template_source_loaders:
if hasattr(loader, "loaders"):
loaders.extend(loader.loaders)
else:
loaders.append(loader)
return loaders
def get_paths(self) -> set[str]:
"""Obtains a set of all template directories."""
paths: set[str] = set()
for loader in self.get_loaders():
with contextlib.suppress(ImportError, AttributeError, TypeError):
module = import_module(loader.__module__)
get_template_sources = getattr(module, "get_template_sources", None)
if get_template_sources is None:
get_template_sources = loader.get_template_sources
paths.update(smart_str(origin) for origin in get_template_sources(""))
return paths
def get_templates(self, paths: set[str]) -> set[str]:
"""Obtains a set of all HTML template paths."""
extensions = [".html"]
templates: set[str] = set()
for path in paths:
for root, _, files in os.walk(path, followlinks=False):
templates.update(
os.path.join(root, name)
for name in files
if not name.startswith(".")
and any(fnmatch(name, f"*{glob}") for glob in extensions)
)
return templates
def get_components(self, templates: set[str]) -> set[str]:
"""Obtains a set of all ReactPy components by parsing HTML templates."""
components: set[str] = set()
for template in templates:
with contextlib.suppress(Exception):
with open(template, "r", encoding="utf-8") as template_file:
clean_template = COMMENT_REGEX.sub("", template_file.read())
regex_iterable = COMPONENT_REGEX.finditer(clean_template)
new_components: list[str] = []
for match in regex_iterable:
new_components.append(
match.group("path").replace('"', "").replace("'", "")
)
offline_path = match.group("offline_path")
if offline_path:
new_components.append(
offline_path.replace('"', "").replace("'", "")
)
components.update(new_components)
if not components:
_logger.warning(
"\033[93m"
"ReactPy did not find any components! "
"You are either not using any ReactPy components, "
"using the template tag incorrectly, "
"or your HTML templates are not registered with Django."
"\033[0m"
)
return components
def register_components(self, components: set[str]) -> None:
"""Registers all ReactPy components in an iterable."""
if components:
_logger.debug("Auto-detected ReactPy root components:")
for component in components:
try:
_logger.debug("\t+ %s", component)
register_component(component)
except Exception:
_logger.exception(
"\033[91m"
"ReactPy failed to register component '%s'!\n"
"This component path may not be valid, "
"or an exception may have occurred while importing.\n"
"See the traceback below for more information."
"\033[0m",
component,
)
def generate_obj_name(obj: Any) -> str:
"""Makes a best effort to create a name for an object.
Useful for JSON serialization of Python objects."""
# First attempt: Dunder methods
if hasattr(obj, "__module__"):
if hasattr(obj, "__name__"):
return f"{obj.__module__}.{obj.__name__}"
if hasattr(obj, "__class__") and hasattr(obj.__class__, "__name__"):
return f"{obj.__module__}.{obj.__class__.__name__}"
# Second attempt: String representation
with contextlib.suppress(Exception):
return str(obj)
# Fallback: Empty string
return ""
def django_query_postprocessor(
data: QuerySet | Model, many_to_many: bool = True, many_to_one: bool = True
) -> QuerySet | Model:
"""Recursively fetch all fields within a `Model` or `QuerySet` to ensure they are not performed lazily.
Behaviors can be modified through `QueryOptions` within your `use_query` hook.
Args:
data: The `Model` or `QuerySet` to recursively fetch fields from.
Keyword Args:
many_to_many: Whether or not to recursively fetch `ManyToManyField` relationships.
many_to_one: Whether or not to recursively fetch `ForeignKey` relationships.
Returns:
The `Model` or `QuerySet` with all fields fetched.
"""
# `QuerySet`, which is an iterable of `Model`/`QuerySet` instances
# https://github.com/typeddjango/django-stubs/issues/704
if isinstance(data, QuerySet):
for model in data:
django_query_postprocessor(
model,
many_to_many=many_to_many,
many_to_one=many_to_one,
)
# `Model` instances
elif isinstance(data, Model):
prefetch_fields: list[str] = []
for field in data._meta.get_fields():
# Force the query to execute
getattr(data, field.name, None)
if many_to_one and type(field) == ManyToOneRel: # noqa: E721
prefetch_fields.append(field.related_name or f"{field.name}_set")
elif many_to_many and isinstance(field, ManyToManyField):
prefetch_fields.append(field.name)
if prefetch_fields:
prefetch_related_objects([data], *prefetch_fields)
for field_str in prefetch_fields:
django_query_postprocessor(
getattr(data, field_str).get_queryset(),
many_to_many=many_to_many,
many_to_one=many_to_one,
)
# Unrecognized type
else:
raise TypeError(
f"Django query postprocessor expected a Model or QuerySet, got {data!r}.\n"
"One of the following may have occurred:\n"
" - You are using a non-Django ORM.\n"
" - You are attempting to use `use_query` to fetch non-ORM data.\n\n"
"If these situations seem correct, you may want to consider disabling the postprocessor via `QueryOptions`."
)
return data
def validate_component_args(func, *args, **kwargs):
"""
Validate whether a set of args/kwargs would work on the given component.
Raises `ComponentParamError` if the args/kwargs are invalid.
"""
signature = inspect.signature(func)
try:
signature.bind(*args, **kwargs)
except TypeError as e:
name = generate_obj_name(func)
raise ComponentParamError(
f"Invalid args for '{name}'. {str(e).capitalize()}."
) from e
def create_cache_key(*args):
"""Creates a cache key string that starts with `reactpy_django` contains
all *args separated by `:`."""
if not args:
raise ValueError("At least one argument is required to create a cache key.")
return f"reactpy_django:{':'.join(str(arg) for arg in args)}"
class SyncLayout(Layout):
"""Sync adapter for ReactPy's `Layout`. Allows it to be used in Django template tags.
This can be removed when Django supports async template tags.
"""
def __enter__(self):
async_to_sync(self.__aenter__)()
return self
def __exit__(self, *_):
async_to_sync(self.__aexit__)(*_)
def render(self):
return async_to_sync(super().render)()
def get_pk(model):
"""Returns the value of the primary key for a Django model."""
return getattr(model, model._meta.pk.name)
def strtobool(val):
"""Convert a string representation of truth to true (1) or false (0).
True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values
are 'n', 'no', 'f', 'false', 'off', and '0'. Raises ValueError if
'val' is anything else.
"""
val = val.lower()
if val in ("y", "yes", "t", "true", "on", "1"):
return 1
elif val in ("n", "no", "f", "false", "off", "0"):
return 0
else:
raise ValueError("invalid truth value %r" % (val,))