forked from frappe/frappe
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path__init__.py
More file actions
2416 lines (1813 loc) · 62.9 KB
/
Copy path__init__.py
File metadata and controls
2416 lines (1813 loc) · 62.9 KB
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
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# Copyright (c) 2022, Frappe Technologies Pvt. Ltd. and Contributors
# License: MIT. See LICENSE
"""
Frappe - Low Code Open Source Framework in Python and JS
Frappe, pronounced fra-pay, is a full stack, batteries-included, web
framework written in Python and Javascript with MariaDB as the database.
It is the framework which powers ERPNext. It is pretty generic and can
be used to build database driven apps.
Read the documentation: https://frappeframework.com/docs
"""
import functools
import gc
import importlib
import inspect
import json
import os
import re
import unicodedata
import warnings
from typing import TYPE_CHECKING, Any, Callable, Literal, Optional, TypeAlias, overload
import click
from werkzeug.local import Local, release_local
from frappe.query_builder import (
get_query,
get_query_builder,
patch_query_aggregation,
patch_query_execute,
)
from frappe.utils.caching import request_cache
from frappe.utils.data import cstr, sbool
# Local application imports
from .exceptions import *
from .utils.jinja import (
get_email_from_template,
get_jenv,
get_jloader,
get_template,
render_template,
)
from .utils.lazy_loader import lazy_import
__version__ = "15.0.0-dev"
__title__ = "Frappe Framework"
controllers = {}
local = Local()
cache = None
STANDARD_USERS = ("Guest", "Administrator")
_dev_server = int(sbool(os.environ.get("DEV_SERVER", False)))
_qb_patched = {}
re._MAXCACHE = (
50 # reduced from default 512 given we are already maintaining this on parent worker
)
_tune_gc = bool(sbool(os.environ.get("FRAPPE_TUNE_GC", True)))
if _dev_server:
warnings.simplefilter("always", DeprecationWarning)
warnings.simplefilter("always", PendingDeprecationWarning)
class _dict(dict):
"""dict like object that exposes keys as attributes"""
__slots__ = ()
__getattr__ = dict.get
__setattr__ = dict.__setitem__
__delattr__ = dict.__delitem__
__setstate__ = dict.update
def __getstate__(self):
return self
def update(self, *args, **kwargs):
"""update and return self -- the missing dict feature in python"""
super().update(*args, **kwargs)
return self
def copy(self):
return _dict(self)
def _(msg: str, lang: str | None = None, context: str | None = None) -> str:
"""Returns translated string in current lang, if exists.
Usage:
_('Change')
_('Change', context='Coins')
"""
from frappe.translate import get_all_translations
from frappe.utils import is_html, strip_html_tags
if not hasattr(local, "lang"):
local.lang = lang or "en"
if not lang:
lang = local.lang
non_translated_string = msg
if is_html(msg):
msg = strip_html_tags(msg)
# msg should always be unicode
msg = as_unicode(msg).strip()
translated_string = ""
all_translations = get_all_translations(lang)
if context:
string_key = f"{msg}:{context}"
translated_string = all_translations.get(string_key)
if not translated_string:
translated_string = all_translations.get(msg)
return translated_string or non_translated_string
def as_unicode(text: str, encoding: str = "utf-8") -> str:
"""Convert to unicode if required"""
if isinstance(text, str):
return text
elif text is None:
return ""
elif isinstance(text, bytes):
return str(text, encoding)
else:
return str(text)
def get_lang_dict(fortype: str, name: str | None = None) -> dict[str, str]:
"""Returns the translated language dict for the given type and name.
:param fortype: must be one of `doctype`, `page`, `report`, `include`, `jsfile`, `boot`
:param name: name of the document for which assets are to be returned."""
from frappe.translate import get_dict
return get_dict(fortype, name)
def set_user_lang(user: str, user_language: str | None = None) -> None:
"""Guess and set user language for the session. `frappe.local.lang`"""
from frappe.translate import get_user_lang
local.lang = get_user_lang(user) or user_language
# local-globals
db = local("db")
qb = local("qb")
conf = local("conf")
form = form_dict = local("form_dict")
request = local("request")
response = local("response")
session = local("session")
user = local("user")
flags = local("flags")
error_log = local("error_log")
debug_log = local("debug_log")
message_log = local("message_log")
lang = local("lang")
# This if block is never executed when running the code. It is only used for
# telling static code analyzer where to find dynamically defined attributes.
if TYPE_CHECKING:
from frappe.database.mariadb.database import MariaDBDatabase
from frappe.database.postgres.database import PostgresDatabase
from frappe.model.document import Document
from frappe.query_builder.builder import MariaDB, Postgres
from frappe.utils.redis_wrapper import RedisWrapper
db: MariaDBDatabase | PostgresDatabase
qb: MariaDB | Postgres
cache: RedisWrapper
# end: static analysis hack
def init(site: str, sites_path: str = ".", new_site: bool = False, force=False) -> None:
"""Initialize frappe for the current site. Reset thread locals `frappe.local`"""
if getattr(local, "initialised", None) and not force:
return
local.error_log = []
local.message_log = []
local.debug_log = []
local.flags = _dict(
{
"currently_saving": [],
"redirect_location": "",
"in_install_db": False,
"in_install_app": False,
"in_import": False,
"in_test": False,
"mute_messages": False,
"ignore_links": False,
"mute_emails": False,
"has_dataurl": False,
"new_site": new_site,
"read_only": False,
}
)
local.locked_documents = []
local.test_objects = {}
local.site = site
local.sites_path = sites_path
local.site_path = os.path.join(sites_path, site)
local.all_apps = None
local.request_ip = None
local.response = _dict({"docs": []})
local.task_id = None
local.conf = _dict(get_site_config())
local.lang = local.conf.lang or "en"
local.module_app = None
local.app_modules = None
local.user = None
local.user_perms = None
local.session = None
local.role_permissions = {}
local.valid_columns = {}
local.new_doc_templates = {}
local.jenv = None
local.jloader = None
local.cache = {}
local.form_dict = _dict()
local.preload_assets = {"style": [], "script": []}
local.session = _dict()
local.dev_server = _dev_server
local.qb = get_query_builder(local.conf.db_type or "mariadb")
local.qb.get_query = get_query
setup_redis_cache_connection()
setup_module_map()
if not _qb_patched.get(local.conf.db_type):
patch_query_execute()
patch_query_aggregation()
local.initialised = True
def connect(
site: str | None = None, db_name: str | None = None, set_admin_as_user: bool = True
) -> None:
"""Connect to site database instance.
:param site: If site is given, calls `frappe.init`.
:param db_name: Optional. Will use from `site_config.json`.
:param set_admin_as_user: Set Administrator as current user.
"""
from frappe.database import get_db
if site:
init(site)
local.db = get_db(user=db_name or local.conf.db_name)
if set_admin_as_user:
set_user("Administrator")
def connect_replica():
from frappe.database import get_db
user = local.conf.db_name
password = local.conf.db_password
port = local.conf.replica_db_port
if local.conf.different_credentials_for_replica:
user = local.conf.replica_db_name
password = local.conf.replica_db_password
local.replica_db = get_db(host=local.conf.replica_host, user=user, password=password, port=port)
# swap db connections
local.primary_db = local.db
local.db = local.replica_db
def get_site_config(sites_path: str | None = None, site_path: str | None = None) -> dict[str, Any]:
"""Returns `site_config.json` combined with `sites/common_site_config.json`.
`site_config` is a set of site wide settings like database name, password, email etc."""
config = {}
sites_path = sites_path or getattr(local, "sites_path", None)
site_path = site_path or getattr(local, "site_path", None)
if sites_path:
common_site_config = os.path.join(sites_path, "common_site_config.json")
if os.path.exists(common_site_config):
try:
config.update(get_file_json(common_site_config))
except Exception as error:
click.secho("common_site_config.json is invalid", fg="red")
print(error)
if site_path:
site_config = os.path.join(site_path, "site_config.json")
if os.path.exists(site_config):
try:
config.update(get_file_json(site_config))
except Exception as error:
click.secho(f"{local.site}/site_config.json is invalid", fg="red")
print(error)
elif local.site and not local.flags.new_site:
raise IncorrectSitePath(f"{local.site} does not exist")
return _dict(config)
def get_conf(site: str | None = None) -> dict[str, Any]:
if hasattr(local, "conf"):
return local.conf
# if no site, get from common_site_config.json
with init_site(site):
return local.conf
class init_site:
def __init__(self, site=None):
"""If site is None, initialize it for empty site ('') to load common_site_config.json"""
self.site = site or ""
def __enter__(self):
init(self.site)
return local
def __exit__(self, type, value, traceback):
destroy()
def destroy():
"""Closes connection and releases werkzeug local."""
if db:
db.close()
release_local(local)
def setup_redis_cache_connection():
"""Defines `frappe.cache` as `RedisWrapper` instance"""
global cache
if not cache:
from frappe.utils.redis_wrapper import RedisWrapper
cache = RedisWrapper.from_url(conf.get("redis_cache") or "redis://localhost:11311")
def get_traceback(with_context: bool = False) -> str:
"""Returns error traceback."""
from frappe.utils import get_traceback
return get_traceback(with_context=with_context)
def errprint(msg: str) -> None:
"""Log error. This is sent back as `exc` in response.
:param msg: Message."""
msg = as_unicode(msg)
if not request or (not "cmd" in local.form_dict) or conf.developer_mode:
print(msg)
error_log.append({"exc": msg})
def print_sql(enable: bool = True) -> None:
return cache.set_value("flag_print_sql", enable)
def log(msg: str) -> None:
"""Add to `debug_log`.
:param msg: Message."""
if not request:
if conf.get("logging") or False:
print(repr(msg))
debug_log.append(as_unicode(msg))
def msgprint(
msg: str,
title: str | None = None,
raise_exception: bool | type[Exception] = False,
as_table: bool = False,
as_list: bool = False,
indicator: Literal["blue", "green", "orange", "red", "yellow"] | None = None,
alert: bool = False,
primary_action: str = None,
is_minimizable: bool = False,
wide: bool = False,
) -> None:
"""Print a message to the user (via HTTP response).
Messages are sent in the `__server_messages` property in the
response JSON and shown in a pop-up / modal.
:param msg: Message.
:param title: [optional] Message title. Default: "Message".
:param raise_exception: [optional] Raise given exception and show message.
:param as_table: [optional] If `msg` is a list of lists, render as HTML table.
:param as_list: [optional] If `msg` is a list, render as un-ordered list.
:param primary_action: [optional] Bind a primary server/client side action.
:param is_minimizable: [optional] Allow users to minimize the modal
:param wide: [optional] Show wide modal
"""
import inspect
import sys
from frappe.utils import strip_html_tags
msg = safe_decode(msg)
out = _dict(message=msg)
@functools.lru_cache(maxsize=1024)
def _strip_html_tags(message):
return strip_html_tags(message)
def _raise_exception():
if raise_exception:
if inspect.isclass(raise_exception) and issubclass(raise_exception, Exception):
raise raise_exception(msg)
else:
raise ValidationError(msg)
if flags.mute_messages:
_raise_exception()
return
if as_table and type(msg) in (list, tuple):
out.as_table = 1
if as_list and type(msg) in (list, tuple):
out.as_list = 1
if sys.stdin and sys.stdin.isatty():
if out.as_list:
msg = [_strip_html_tags(msg) for msg in out.message]
else:
msg = _strip_html_tags(out.message)
if flags.print_messages and out.message:
print(f"Message: {_strip_html_tags(out.message)}")
out.title = title or _("Message", context="Default title of the message dialog")
if not indicator and raise_exception:
indicator = "red"
if indicator:
out.indicator = indicator
if is_minimizable:
out.is_minimizable = is_minimizable
if alert:
out.alert = 1
if raise_exception:
out.raise_exception = 1
if primary_action:
out.primary_action = primary_action
if wide:
out.wide = wide
message_log.append(json.dumps(out))
if raise_exception and hasattr(raise_exception, "__name__"):
local.response["exc_type"] = raise_exception.__name__
_raise_exception()
def clear_messages():
local.message_log = []
def get_message_log():
log = []
for msg_out in local.message_log:
log.append(json.loads(msg_out))
return log
def clear_last_message():
if len(local.message_log) > 0:
local.message_log = local.message_log[:-1]
def throw(
msg: str,
exc: type[Exception] = ValidationError,
title: str | None = None,
is_minimizable: bool = False,
wide: bool = False,
as_list: bool = False,
) -> None:
"""Throw execption and show message (`msgprint`).
:param msg: Message.
:param exc: Exception class. Default `frappe.ValidationError`"""
msgprint(
msg,
raise_exception=exc,
title=title,
indicator="red",
is_minimizable=is_minimizable,
wide=wide,
as_list=as_list,
)
def create_folder(path, with_init=False):
"""Create a folder in the given path and add an `__init__.py` file (optional).
:param path: Folder path.
:param with_init: Create `__init__.py` in the new folder."""
from frappe.utils import touch_file
if not os.path.exists(path):
os.makedirs(path)
if with_init:
touch_file(os.path.join(path, "__init__.py"))
def set_user(username: str):
"""Set current user.
:param username: **User** name to set as current user."""
local.session.user = username
local.session.sid = username
local.cache = {}
local.form_dict = _dict()
local.jenv = None
local.session.data = _dict()
local.role_permissions = {}
local.new_doc_templates = {}
local.user_perms = None
def get_user():
from frappe.utils.user import UserPermissions
if not local.user_perms:
local.user_perms = UserPermissions(local.session.user)
return local.user_perms
def get_roles(username=None) -> list[str]:
"""Returns roles of current user."""
if not local.session or not local.session.user:
return ["Guest"]
import frappe.permissions
return frappe.permissions.get_roles(username or local.session.user)
def get_request_header(key, default=None):
"""Return HTTP request header.
:param key: HTTP header key.
:param default: Default value."""
return request.headers.get(key, default)
def sendmail(
recipients=None,
sender="",
subject="No Subject",
message="No Message",
as_markdown=False,
delayed=True,
reference_doctype=None,
reference_name=None,
unsubscribe_method=None,
unsubscribe_params=None,
unsubscribe_message=None,
add_unsubscribe_link=1,
attachments=None,
content=None,
doctype=None,
name=None,
reply_to=None,
queue_separately=False,
cc=None,
bcc=None,
message_id=None,
in_reply_to=None,
send_after=None,
expose_recipients=None,
send_priority=1,
communication=None,
retry=1,
now=None,
read_receipt=None,
is_notification=False,
inline_images=None,
template=None,
args=None,
header=None,
print_letterhead=False,
with_container=False,
email_read_tracker_url=None,
):
"""Send email using user's default **Email Account** or global default **Email Account**.
:param recipients: List of recipients.
:param sender: Email sender. Default is current user or default outgoing account.
:param subject: Email Subject.
:param message: (or `content`) Email Content.
:param as_markdown: Convert content markdown to HTML.
:param delayed: Send via scheduled email sender **Email Queue**. Don't send immediately. Default is true
:param send_priority: Priority for Email Queue, default 1.
:param reference_doctype: (or `doctype`) Append as communication to this DocType.
:param reference_name: (or `name`) Append as communication to this document name.
:param unsubscribe_method: Unsubscribe url with options email, doctype, name. e.g. `/api/method/unsubscribe`
:param unsubscribe_params: Unsubscribe paramaters to be loaded on the unsubscribe_method [optional] (dict).
:param attachments: List of attachments.
:param reply_to: Reply-To Email Address.
:param message_id: Used for threading. If a reply is received to this email, Message-Id is sent back as In-Reply-To in received email.
:param in_reply_to: Used to send the Message-Id of a received email back as In-Reply-To.
:param send_after: Send after the given datetime.
:param expose_recipients: Display all recipients in the footer message - "This email was sent to"
:param communication: Communication link to be set in Email Queue record
:param inline_images: List of inline images as {"filename", "filecontent"}. All src properties will be replaced with random Content-Id
:param template: Name of html template from templates/emails folder
:param args: Arguments for rendering the template
:param header: Append header in email
:param with_container: Wraps email inside a styled container
"""
if recipients is None:
recipients = []
if cc is None:
cc = []
if bcc is None:
bcc = []
text_content = None
if template:
message, text_content = get_email_from_template(template, args)
message = content or message
if as_markdown:
from frappe.utils import md_to_html
message = md_to_html(message)
if not delayed:
now = True
from frappe.email.doctype.email_queue.email_queue import QueueBuilder
builder = QueueBuilder(
recipients=recipients,
sender=sender,
subject=subject,
message=message,
text_content=text_content,
reference_doctype=doctype or reference_doctype,
reference_name=name or reference_name,
add_unsubscribe_link=add_unsubscribe_link,
unsubscribe_method=unsubscribe_method,
unsubscribe_params=unsubscribe_params,
unsubscribe_message=unsubscribe_message,
attachments=attachments,
reply_to=reply_to,
cc=cc,
bcc=bcc,
message_id=message_id,
in_reply_to=in_reply_to,
send_after=send_after,
expose_recipients=expose_recipients,
send_priority=send_priority,
queue_separately=queue_separately,
communication=communication,
read_receipt=read_receipt,
is_notification=is_notification,
inline_images=inline_images,
header=header,
print_letterhead=print_letterhead,
with_container=with_container,
email_read_tracker_url=email_read_tracker_url,
)
# build email queue and send the email if send_now is True.
builder.process(send_now=now)
whitelisted = []
guest_methods = []
xss_safe_methods = []
allowed_http_methods_for_whitelisted_func = {}
def whitelist(allow_guest=False, xss_safe=False, methods=None):
"""
Decorator for whitelisting a function and making it accessible via HTTP.
Standard request will be `/api/method/[path.to.method]`
:param allow_guest: Allow non logged-in user to access this method.
:param methods: Allowed http method to access the method.
Use as:
@frappe.whitelist()
def myfunc(param1, param2):
pass
"""
if not methods:
methods = ["GET", "POST", "PUT", "DELETE"]
def innerfn(fn):
from frappe.utils.typing_validations import validate_argument_types
global whitelisted, guest_methods, xss_safe_methods, allowed_http_methods_for_whitelisted_func
# validate argument types only if request is present
in_request_or_test = lambda: getattr(local, "request", None) or local.flags.in_test # noqa: E731
# get function from the unbound / bound method
# this is needed because functions can be compared, but not methods
method = None
if hasattr(fn, "__func__"):
method = validate_argument_types(fn, apply_condition=in_request_or_test)
fn = method.__func__
else:
fn = validate_argument_types(fn, apply_condition=in_request_or_test)
whitelisted.append(fn)
allowed_http_methods_for_whitelisted_func[fn] = methods
if allow_guest:
guest_methods.append(fn)
if xss_safe:
xss_safe_methods.append(fn)
return method or fn
return innerfn
def is_whitelisted(method):
from frappe.utils import sanitize_html
is_guest = session["user"] == "Guest"
if method not in whitelisted or is_guest and method not in guest_methods:
summary = _("You are not permitted to access this resource.")
detail = _("Function {0} is not whitelisted.").format(
bold(f"{method.__module__}.{method.__name__}")
)
msg = f"<details><summary>{summary}</summary>{detail}</details>"
throw(msg, PermissionError, title="Method Not Allowed")
if is_guest and method not in xss_safe_methods:
# strictly sanitize form_dict
# escapes html characters like <> except for predefined tags like a, b, ul etc.
for key, value in form_dict.items():
if isinstance(value, str):
form_dict[key] = sanitize_html(value)
def read_only():
def innfn(fn):
def wrapper_fn(*args, **kwargs):
if conf.read_from_replica:
connect_replica()
try:
retval = fn(*args, **get_newargs(fn, kwargs))
finally:
if local and hasattr(local, "primary_db"):
local.db.close()
local.db = local.primary_db
return retval
return wrapper_fn
return innfn
def write_only():
# if replica connection exists, we have to replace it momentarily with the primary connection
def innfn(fn):
def wrapper_fn(*args, **kwargs):
primary_db = getattr(local, "primary_db", None)
replica_db = getattr(local, "replica_db", None)
in_read_only = getattr(local, "db", None) != primary_db
# switch to primary connection
if in_read_only and primary_db:
local.db = local.primary_db
try:
retval = fn(*args, **get_newargs(fn, kwargs))
finally:
# switch back to replica connection
if in_read_only and replica_db:
local.db = replica_db
return retval
return wrapper_fn
return innfn
def only_for(roles: list[str] | tuple[str] | str, message=False):
"""
Raises `frappe.PermissionError` if the user does not have any of the permitted roles.
:param roles: Permitted role(s)
"""
if local.flags.in_test or local.session.user == "Administrator":
return
if isinstance(roles, str):
roles = (roles,)
if set(roles).isdisjoint(get_roles()):
if not message:
raise PermissionError
throw(
_("This action is only allowed for {}").format(
", ".join(bold(_(role)) for role in roles),
),
PermissionError,
_("Not Permitted"),
)
def get_domain_data(module):
try:
domain_data = get_hooks("domains")
if module in domain_data:
return _dict(get_attr(get_hooks("domains")[module][0] + ".data"))
else:
return _dict()
except ImportError:
if local.flags.in_test:
return _dict()
else:
raise
def clear_cache(user: str | None = None, doctype: str | None = None):
"""Clear **User**, **DocType** or global cache.
:param user: If user is given, only user cache is cleared.
:param doctype: If doctype is given, only DocType cache is cleared."""
import frappe.cache_manager
import frappe.utils.caching
from frappe.website.router import clear_routing_cache
if doctype:
frappe.cache_manager.clear_doctype_cache(doctype)
reset_metadata_version()
elif user:
frappe.cache_manager.clear_user_cache(user)
else: # everything
from frappe import translate
frappe.cache_manager.clear_user_cache()
frappe.cache_manager.clear_domain_cache()
translate.clear_cache()
reset_metadata_version()
local.cache = {}
local.new_doc_templates = {}
for fn in get_hooks("clear_cache"):
get_attr(fn)()
frappe.utils.caching._SITE_CACHE.clear()
local.role_permissions = {}
if hasattr(local, "request_cache"):
local.request_cache.clear()
if hasattr(local, "system_settings"):
del local.system_settings
if hasattr(local, "website_settings"):
del local.website_settings
clear_routing_cache()
def only_has_select_perm(doctype, user=None, ignore_permissions=False):
if ignore_permissions:
return False
from frappe.permissions import get_role_permissions
user = user or local.session.user
permissions = get_role_permissions(doctype, user=user)
return permissions.get("select") and not permissions.get("read")
def has_permission(
doctype=None,
ptype="read",
doc=None,
user=None,
throw=False,
*,
parent_doctype=None,
):
"""
Returns True if the user has permission `ptype` for given `doctype` or `doc`
Raises `frappe.PermissionError` if user isn't permitted and `throw` is truthy
:param doctype: DocType for which permission is to be check.
:param ptype: Permission type (`read`, `write`, `create`, `submit`, `cancel`, `amend`). Default: `read`.
:param doc: [optional] Checks User permissions for given doc.
:param user: [optional] Check for given user. Default: current user.
:param parent_doctype: Required when checking permission for a child DocType (unless doc is specified).
"""
import frappe.permissions
if not doctype and doc:
doctype = doc.doctype
out = frappe.permissions.has_permission(
doctype,
ptype,
doc=doc,
user=user,
raise_exception=throw,
parent_doctype=parent_doctype,
)
if throw and not out:
# mimics frappe.throw
document_label = f"{_(doc.doctype)} {doc.name}" if doc else _(doctype)
msgprint(
_("No permission for {0}").format(document_label),
raise_exception=ValidationError,
title=None,
indicator="red",
is_minimizable=None,
wide=None,
as_list=False,
)
return out
def has_website_permission(doc=None, ptype="read", user=None, verbose=False, doctype=None):
"""Raises `frappe.PermissionError` if not permitted.
:param doctype: DocType for which permission is to be check.
:param ptype: Permission type (`read`, `write`, `create`, `submit`, `cancel`, `amend`). Default: `read`.
:param doc: Checks User permissions for given doc.
:param user: [optional] Check for given user. Default: current user."""
if not user:
user = session.user
if doc:
if isinstance(doc, str):
doc = get_doc(doctype, doc)
doctype = doc.doctype
if doc.flags.ignore_permissions:
return True
# check permission in controller
if hasattr(doc, "has_website_permission"):
return doc.has_website_permission(ptype, user, verbose=verbose)
hooks = (get_hooks("has_website_permission") or {}).get(doctype, [])
if hooks:
for method in hooks: