mirror of https://github.com/zulip/zulip.git
python: Mark dict parameters with defaults as read-only.
Found by semgrep 0.115 more accurately applying the rule added in
commit 0d6c771baf
(#15349).
Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
parent
3f010bbcc4
commit
47c5deeccd
|
@ -1979,8 +1979,9 @@ class StripeTest(StripeTestCase):
|
||||||
invoice: bool,
|
invoice: bool,
|
||||||
licenses: Optional[int],
|
licenses: Optional[int],
|
||||||
min_licenses_in_response: int,
|
min_licenses_in_response: int,
|
||||||
upgrade_params: Dict[str, Any] = {},
|
upgrade_params: Mapping[str, Any] = {},
|
||||||
) -> None:
|
) -> None:
|
||||||
|
upgrade_params = dict(upgrade_params)
|
||||||
if licenses is None:
|
if licenses is None:
|
||||||
del_args = ["licenses"]
|
del_args = ["licenses"]
|
||||||
else:
|
else:
|
||||||
|
@ -2006,8 +2007,9 @@ class StripeTest(StripeTestCase):
|
||||||
)
|
)
|
||||||
|
|
||||||
def check_success(
|
def check_success(
|
||||||
invoice: bool, licenses: Optional[int], upgrade_params: Dict[str, Any] = {}
|
invoice: bool, licenses: Optional[int], upgrade_params: Mapping[str, Any] = {}
|
||||||
) -> None:
|
) -> None:
|
||||||
|
upgrade_params = dict(upgrade_params)
|
||||||
if licenses is None:
|
if licenses is None:
|
||||||
del_args = ["licenses"]
|
del_args = ["licenses"]
|
||||||
else:
|
else:
|
||||||
|
|
|
@ -13,6 +13,7 @@ from typing import (
|
||||||
Iterable,
|
Iterable,
|
||||||
Iterator,
|
Iterator,
|
||||||
List,
|
List,
|
||||||
|
Mapping,
|
||||||
Optional,
|
Optional,
|
||||||
Protocol,
|
Protocol,
|
||||||
Set,
|
Set,
|
||||||
|
@ -173,7 +174,7 @@ def make_user_messages(
|
||||||
subscriber_map: Dict[int, Set[int]],
|
subscriber_map: Dict[int, Set[int]],
|
||||||
is_pm_data: bool,
|
is_pm_data: bool,
|
||||||
mention_map: Dict[int, Set[int]],
|
mention_map: Dict[int, Set[int]],
|
||||||
wildcard_mention_map: Dict[int, bool] = {},
|
wildcard_mention_map: Mapping[int, bool] = {},
|
||||||
) -> List[ZerverFieldsT]:
|
) -> List[ZerverFieldsT]:
|
||||||
|
|
||||||
zerver_usermessage = []
|
zerver_usermessage = []
|
||||||
|
|
|
@ -2,7 +2,7 @@
|
||||||
# high-level documentation on how this system works.
|
# high-level documentation on how this system works.
|
||||||
import copy
|
import copy
|
||||||
import time
|
import time
|
||||||
from typing import Any, Callable, Collection, Dict, Iterable, Optional, Sequence, Set
|
from typing import Any, Callable, Collection, Dict, Iterable, Mapping, Optional, Sequence, Set
|
||||||
|
|
||||||
from django.conf import settings
|
from django.conf import settings
|
||||||
from django.utils.translation import gettext as _
|
from django.utils.translation import gettext as _
|
||||||
|
@ -1365,7 +1365,7 @@ def do_events_register(
|
||||||
all_public_streams: bool = False,
|
all_public_streams: bool = False,
|
||||||
include_subscribers: bool = True,
|
include_subscribers: bool = True,
|
||||||
include_streams: bool = True,
|
include_streams: bool = True,
|
||||||
client_capabilities: Dict[str, bool] = {},
|
client_capabilities: Mapping[str, bool] = {},
|
||||||
narrow: Collection[Sequence[str]] = [],
|
narrow: Collection[Sequence[str]] = [],
|
||||||
fetch_event_types: Optional[Collection[str]] = None,
|
fetch_event_types: Optional[Collection[str]] = None,
|
||||||
spectator_requested_language: Optional[str] = None,
|
spectator_requested_language: Optional[str] = None,
|
||||||
|
|
|
@ -6,7 +6,7 @@ import logging
|
||||||
import re
|
import re
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from functools import lru_cache
|
from functools import lru_cache
|
||||||
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Tuple, Type, Union
|
from typing import TYPE_CHECKING, Any, Dict, List, Mapping, Optional, Sequence, Tuple, Type, Union
|
||||||
|
|
||||||
import gcm
|
import gcm
|
||||||
import lxml.html
|
import lxml.html
|
||||||
|
@ -150,7 +150,7 @@ def apns_enabled() -> bool:
|
||||||
return settings.APNS_CERT_FILE is not None
|
return settings.APNS_CERT_FILE is not None
|
||||||
|
|
||||||
|
|
||||||
def modernize_apns_payload(data: Dict[str, Any]) -> Dict[str, Any]:
|
def modernize_apns_payload(data: Mapping[str, Any]) -> Mapping[str, Any]:
|
||||||
"""Take a payload in an unknown Zulip version's format, and return in current format."""
|
"""Take a payload in an unknown Zulip version's format, and return in current format."""
|
||||||
# TODO this isn't super robust as is -- if a buggy remote server
|
# TODO this isn't super robust as is -- if a buggy remote server
|
||||||
# sends a malformed payload, we are likely to raise an exception.
|
# sends a malformed payload, we are likely to raise an exception.
|
||||||
|
@ -183,7 +183,7 @@ APNS_MAX_RETRIES = 3
|
||||||
def send_apple_push_notification(
|
def send_apple_push_notification(
|
||||||
user_identity: UserPushIndentityCompat,
|
user_identity: UserPushIndentityCompat,
|
||||||
devices: Sequence[DeviceToken],
|
devices: Sequence[DeviceToken],
|
||||||
payload_data: Dict[str, Any],
|
payload_data: Mapping[str, Any],
|
||||||
remote: Optional["RemoteZulipServer"] = None,
|
remote: Optional["RemoteZulipServer"] = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
if not devices:
|
if not devices:
|
||||||
|
@ -222,7 +222,7 @@ def send_apple_push_notification(
|
||||||
user_identity,
|
user_identity,
|
||||||
len(devices),
|
len(devices),
|
||||||
)
|
)
|
||||||
payload_data = modernize_apns_payload(payload_data).copy()
|
payload_data = dict(modernize_apns_payload(payload_data))
|
||||||
message = {**payload_data.pop("custom", {}), "aps": payload_data}
|
message = {**payload_data.pop("custom", {}), "aps": payload_data}
|
||||||
for device in devices:
|
for device in devices:
|
||||||
# TODO obviously this should be made to actually use the async
|
# TODO obviously this should be made to actually use the async
|
||||||
|
|
|
@ -234,7 +234,7 @@ def send_email(
|
||||||
from_address: Optional[str] = None,
|
from_address: Optional[str] = None,
|
||||||
reply_to_email: Optional[str] = None,
|
reply_to_email: Optional[str] = None,
|
||||||
language: Optional[str] = None,
|
language: Optional[str] = None,
|
||||||
context: Dict[str, Any] = {},
|
context: Mapping[str, Any] = {},
|
||||||
realm: Optional[Realm] = None,
|
realm: Optional[Realm] = None,
|
||||||
connection: Optional[BaseEmailBackend] = None,
|
connection: Optional[BaseEmailBackend] = None,
|
||||||
dry_run: bool = False,
|
dry_run: bool = False,
|
||||||
|
@ -338,7 +338,7 @@ def send_future_email(
|
||||||
from_name: Optional[str] = None,
|
from_name: Optional[str] = None,
|
||||||
from_address: Optional[str] = None,
|
from_address: Optional[str] = None,
|
||||||
language: Optional[str] = None,
|
language: Optional[str] = None,
|
||||||
context: Dict[str, Any] = {},
|
context: Mapping[str, Any] = {},
|
||||||
delay: datetime.timedelta = datetime.timedelta(0),
|
delay: datetime.timedelta = datetime.timedelta(0),
|
||||||
) -> None:
|
) -> None:
|
||||||
template_name = template_prefix.split("/")[-1]
|
template_name = template_prefix.split("/")[-1]
|
||||||
|
@ -393,7 +393,7 @@ def send_email_to_admins(
|
||||||
from_name: Optional[str] = None,
|
from_name: Optional[str] = None,
|
||||||
from_address: Optional[str] = None,
|
from_address: Optional[str] = None,
|
||||||
language: Optional[str] = None,
|
language: Optional[str] = None,
|
||||||
context: Dict[str, Any] = {},
|
context: Mapping[str, Any] = {},
|
||||||
) -> None:
|
) -> None:
|
||||||
admins = realm.get_human_admin_users()
|
admins = realm.get_human_admin_users()
|
||||||
admin_user_ids = [admin.id for admin in admins]
|
admin_user_ids = [admin.id for admin in admins]
|
||||||
|
@ -413,7 +413,7 @@ def send_email_to_billing_admins_and_realm_owners(
|
||||||
from_name: Optional[str] = None,
|
from_name: Optional[str] = None,
|
||||||
from_address: Optional[str] = None,
|
from_address: Optional[str] = None,
|
||||||
language: Optional[str] = None,
|
language: Optional[str] = None,
|
||||||
context: Dict[str, Any] = {},
|
context: Mapping[str, Any] = {},
|
||||||
) -> None:
|
) -> None:
|
||||||
send_email(
|
send_email(
|
||||||
template_prefix,
|
template_prefix,
|
||||||
|
|
|
@ -253,7 +253,7 @@ Output:
|
||||||
url: str,
|
url: str,
|
||||||
method: str,
|
method: str,
|
||||||
result: "TestHttpResponse",
|
result: "TestHttpResponse",
|
||||||
data: Union[str, bytes, Dict[str, Any]],
|
data: Union[str, bytes, Mapping[str, Any]],
|
||||||
extra: Dict[str, str],
|
extra: Dict[str, str],
|
||||||
intentionally_undocumented: bool = False,
|
intentionally_undocumented: bool = False,
|
||||||
) -> None:
|
) -> None:
|
||||||
|
@ -296,7 +296,7 @@ Output:
|
||||||
def client_patch(
|
def client_patch(
|
||||||
self,
|
self,
|
||||||
url: str,
|
url: str,
|
||||||
info: Dict[str, Any] = {},
|
info: Mapping[str, Any] = {},
|
||||||
skip_user_agent: bool = False,
|
skip_user_agent: bool = False,
|
||||||
follow: bool = False,
|
follow: bool = False,
|
||||||
secure: bool = False,
|
secure: bool = False,
|
||||||
|
@ -325,7 +325,7 @@ Output:
|
||||||
def client_patch_multipart(
|
def client_patch_multipart(
|
||||||
self,
|
self,
|
||||||
url: str,
|
url: str,
|
||||||
info: Dict[str, Any] = {},
|
info: Mapping[str, Any] = {},
|
||||||
skip_user_agent: bool = False,
|
skip_user_agent: bool = False,
|
||||||
follow: bool = False,
|
follow: bool = False,
|
||||||
secure: bool = False,
|
secure: bool = False,
|
||||||
|
@ -340,7 +340,7 @@ Output:
|
||||||
with the Django test client, it deals with MULTIPART_CONTENT
|
with the Django test client, it deals with MULTIPART_CONTENT
|
||||||
automatically, but not patch.)
|
automatically, but not patch.)
|
||||||
"""
|
"""
|
||||||
encoded = encode_multipart(BOUNDARY, info)
|
encoded = encode_multipart(BOUNDARY, dict(info))
|
||||||
django_client = self.client # see WRAPPER_COMMENT
|
django_client = self.client # see WRAPPER_COMMENT
|
||||||
self.set_http_headers(extra, skip_user_agent)
|
self.set_http_headers(extra, skip_user_agent)
|
||||||
result = django_client.patch(
|
result = django_client.patch(
|
||||||
|
@ -359,7 +359,7 @@ Output:
|
||||||
def json_patch(
|
def json_patch(
|
||||||
self,
|
self,
|
||||||
url: str,
|
url: str,
|
||||||
payload: Dict[str, Any] = {},
|
payload: Mapping[str, Any] = {},
|
||||||
skip_user_agent: bool = False,
|
skip_user_agent: bool = False,
|
||||||
follow: bool = False,
|
follow: bool = False,
|
||||||
secure: bool = False,
|
secure: bool = False,
|
||||||
|
@ -376,7 +376,7 @@ Output:
|
||||||
def client_put(
|
def client_put(
|
||||||
self,
|
self,
|
||||||
url: str,
|
url: str,
|
||||||
info: Dict[str, Any] = {},
|
info: Mapping[str, Any] = {},
|
||||||
skip_user_agent: bool = False,
|
skip_user_agent: bool = False,
|
||||||
follow: bool = False,
|
follow: bool = False,
|
||||||
secure: bool = False,
|
secure: bool = False,
|
||||||
|
@ -391,7 +391,7 @@ Output:
|
||||||
def json_put(
|
def json_put(
|
||||||
self,
|
self,
|
||||||
url: str,
|
url: str,
|
||||||
payload: Dict[str, Any] = {},
|
payload: Mapping[str, Any] = {},
|
||||||
skip_user_agent: bool = False,
|
skip_user_agent: bool = False,
|
||||||
follow: bool = False,
|
follow: bool = False,
|
||||||
secure: bool = False,
|
secure: bool = False,
|
||||||
|
@ -408,7 +408,7 @@ Output:
|
||||||
def client_delete(
|
def client_delete(
|
||||||
self,
|
self,
|
||||||
url: str,
|
url: str,
|
||||||
info: Dict[str, Any] = {},
|
info: Mapping[str, Any] = {},
|
||||||
skip_user_agent: bool = False,
|
skip_user_agent: bool = False,
|
||||||
follow: bool = False,
|
follow: bool = False,
|
||||||
secure: bool = False,
|
secure: bool = False,
|
||||||
|
@ -434,7 +434,7 @@ Output:
|
||||||
def client_options(
|
def client_options(
|
||||||
self,
|
self,
|
||||||
url: str,
|
url: str,
|
||||||
info: Dict[str, Any] = {},
|
info: Mapping[str, Any] = {},
|
||||||
skip_user_agent: bool = False,
|
skip_user_agent: bool = False,
|
||||||
follow: bool = False,
|
follow: bool = False,
|
||||||
secure: bool = False,
|
secure: bool = False,
|
||||||
|
@ -442,13 +442,13 @@ Output:
|
||||||
) -> "TestHttpResponse":
|
) -> "TestHttpResponse":
|
||||||
django_client = self.client # see WRAPPER_COMMENT
|
django_client = self.client # see WRAPPER_COMMENT
|
||||||
self.set_http_headers(extra, skip_user_agent)
|
self.set_http_headers(extra, skip_user_agent)
|
||||||
return django_client.options(url, info, follow=follow, secure=secure, **extra)
|
return django_client.options(url, dict(info), follow=follow, secure=secure, **extra)
|
||||||
|
|
||||||
@instrument_url
|
@instrument_url
|
||||||
def client_head(
|
def client_head(
|
||||||
self,
|
self,
|
||||||
url: str,
|
url: str,
|
||||||
info: Dict[str, Any] = {},
|
info: Mapping[str, Any] = {},
|
||||||
skip_user_agent: bool = False,
|
skip_user_agent: bool = False,
|
||||||
follow: bool = False,
|
follow: bool = False,
|
||||||
secure: bool = False,
|
secure: bool = False,
|
||||||
|
@ -500,7 +500,7 @@ Output:
|
||||||
def client_get(
|
def client_get(
|
||||||
self,
|
self,
|
||||||
url: str,
|
url: str,
|
||||||
info: Dict[str, Any] = {},
|
info: Mapping[str, Any] = {},
|
||||||
skip_user_agent: bool = False,
|
skip_user_agent: bool = False,
|
||||||
follow: bool = False,
|
follow: bool = False,
|
||||||
secure: bool = False,
|
secure: bool = False,
|
||||||
|
@ -858,7 +858,7 @@ Output:
|
||||||
return "Basic " + base64.b64encode(credentials.encode()).decode()
|
return "Basic " + base64.b64encode(credentials.encode()).decode()
|
||||||
|
|
||||||
def uuid_get(
|
def uuid_get(
|
||||||
self, identifier: str, url: str, info: Dict[str, Any] = {}, **extra: str
|
self, identifier: str, url: str, info: Mapping[str, Any] = {}, **extra: str
|
||||||
) -> "TestHttpResponse":
|
) -> "TestHttpResponse":
|
||||||
extra["HTTP_AUTHORIZATION"] = self.encode_uuid(identifier)
|
extra["HTTP_AUTHORIZATION"] = self.encode_uuid(identifier)
|
||||||
return self.client_get(
|
return self.client_get(
|
||||||
|
@ -890,7 +890,7 @@ Output:
|
||||||
)
|
)
|
||||||
|
|
||||||
def api_get(
|
def api_get(
|
||||||
self, user: UserProfile, url: str, info: Dict[str, Any] = {}, **extra: str
|
self, user: UserProfile, url: str, info: Mapping[str, Any] = {}, **extra: str
|
||||||
) -> "TestHttpResponse":
|
) -> "TestHttpResponse":
|
||||||
extra["HTTP_AUTHORIZATION"] = self.encode_user(user)
|
extra["HTTP_AUTHORIZATION"] = self.encode_user(user)
|
||||||
return self.client_get(
|
return self.client_get(
|
||||||
|
@ -923,7 +923,7 @@ Output:
|
||||||
)
|
)
|
||||||
|
|
||||||
def api_patch(
|
def api_patch(
|
||||||
self, user: UserProfile, url: str, info: Dict[str, Any] = {}, **extra: str
|
self, user: UserProfile, url: str, info: Mapping[str, Any] = {}, **extra: str
|
||||||
) -> "TestHttpResponse":
|
) -> "TestHttpResponse":
|
||||||
extra["HTTP_AUTHORIZATION"] = self.encode_user(user)
|
extra["HTTP_AUTHORIZATION"] = self.encode_user(user)
|
||||||
return self.client_patch(
|
return self.client_patch(
|
||||||
|
@ -937,7 +937,7 @@ Output:
|
||||||
)
|
)
|
||||||
|
|
||||||
def api_delete(
|
def api_delete(
|
||||||
self, user: UserProfile, url: str, info: Dict[str, Any] = {}, **extra: str
|
self, user: UserProfile, url: str, info: Mapping[str, Any] = {}, **extra: str
|
||||||
) -> "TestHttpResponse":
|
) -> "TestHttpResponse":
|
||||||
extra["HTTP_AUTHORIZATION"] = self.encode_user(user)
|
extra["HTTP_AUTHORIZATION"] = self.encode_user(user)
|
||||||
return self.client_delete(
|
return self.client_delete(
|
||||||
|
@ -1247,7 +1247,7 @@ Output:
|
||||||
self,
|
self,
|
||||||
user: UserProfile,
|
user: UserProfile,
|
||||||
streams: Iterable[str],
|
streams: Iterable[str],
|
||||||
extra_post_data: Dict[str, Any] = {},
|
extra_post_data: Mapping[str, Any] = {},
|
||||||
invite_only: bool = False,
|
invite_only: bool = False,
|
||||||
is_web_public: bool = False,
|
is_web_public: bool = False,
|
||||||
allow_fail: bool = False,
|
allow_fail: bool = False,
|
||||||
|
|
|
@ -14,6 +14,7 @@ from typing import (
|
||||||
Iterable,
|
Iterable,
|
||||||
Iterator,
|
Iterator,
|
||||||
List,
|
List,
|
||||||
|
Mapping,
|
||||||
Optional,
|
Optional,
|
||||||
Tuple,
|
Tuple,
|
||||||
TypeVar,
|
TypeVar,
|
||||||
|
@ -305,7 +306,7 @@ class HostRequestMock(HttpRequest):
|
||||||
|
|
||||||
def __init__(
|
def __init__(
|
||||||
self,
|
self,
|
||||||
post_data: Dict[str, Any] = {},
|
post_data: Mapping[str, Any] = {},
|
||||||
user_profile: Union[UserProfile, None] = None,
|
user_profile: Union[UserProfile, None] = None,
|
||||||
remote_server: Optional[RemoteZulipServer] = None,
|
remote_server: Optional[RemoteZulipServer] = None,
|
||||||
host: str = settings.EXTERNAL_HOST,
|
host: str = settings.EXTERNAL_HOST,
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
import re
|
import re
|
||||||
import unicodedata
|
import unicodedata
|
||||||
from collections import defaultdict
|
from collections import defaultdict
|
||||||
from typing import Any, Dict, Iterable, List, Optional, Sequence, TypedDict
|
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, TypedDict
|
||||||
|
|
||||||
import dateutil.parser as date_parser
|
import dateutil.parser as date_parser
|
||||||
from django.conf import settings
|
from django.conf import settings
|
||||||
|
@ -79,7 +79,9 @@ def check_short_name(short_name_raw: str) -> str:
|
||||||
return short_name
|
return short_name
|
||||||
|
|
||||||
|
|
||||||
def check_valid_bot_config(bot_type: int, service_name: str, config_data: Dict[str, str]) -> None:
|
def check_valid_bot_config(
|
||||||
|
bot_type: int, service_name: str, config_data: Mapping[str, str]
|
||||||
|
) -> None:
|
||||||
if bot_type == UserProfile.INCOMING_WEBHOOK_BOT:
|
if bot_type == UserProfile.INCOMING_WEBHOOK_BOT:
|
||||||
from zerver.lib.integrations import WEBHOOK_INTEGRATIONS
|
from zerver.lib.integrations import WEBHOOK_INTEGRATIONS
|
||||||
|
|
||||||
|
|
|
@ -539,7 +539,7 @@ SKIP_JSON = {
|
||||||
def validate_request(
|
def validate_request(
|
||||||
url: str,
|
url: str,
|
||||||
method: str,
|
method: str,
|
||||||
data: Union[str, bytes, Dict[str, Any]],
|
data: Union[str, bytes, Mapping[str, Any]],
|
||||||
http_headers: Dict[str, str],
|
http_headers: Dict[str, str],
|
||||||
json_url: bool,
|
json_url: bool,
|
||||||
status_code: str,
|
status_code: str,
|
||||||
|
|
|
@ -3122,7 +3122,7 @@ class AppleAuthBackendNativeFlowTest(AppleAuthMixin, SocialAuthBase):
|
||||||
multiuse_object_key: str = "",
|
multiuse_object_key: str = "",
|
||||||
alternative_start_url: Optional[str] = None,
|
alternative_start_url: Optional[str] = None,
|
||||||
id_token: Optional[str] = None,
|
id_token: Optional[str] = None,
|
||||||
account_data_dict: Dict[str, str] = {},
|
account_data_dict: Mapping[str, str] = {},
|
||||||
*,
|
*,
|
||||||
user_agent: Optional[str] = None,
|
user_agent: Optional[str] = None,
|
||||||
) -> Tuple[str, Dict[str, Any]]:
|
) -> Tuple[str, Dict[str, Any]]:
|
||||||
|
|
|
@ -4,7 +4,7 @@ import datetime
|
||||||
import re
|
import re
|
||||||
import uuid
|
import uuid
|
||||||
from contextlib import contextmanager
|
from contextlib import contextmanager
|
||||||
from typing import Any, Dict, Iterator, List, Optional, Tuple, Union
|
from typing import Any, Dict, Iterator, List, Mapping, Optional, Tuple, Union
|
||||||
from unittest import mock, skipUnless
|
from unittest import mock, skipUnless
|
||||||
from urllib import parse
|
from urllib import parse
|
||||||
|
|
||||||
|
@ -1540,7 +1540,7 @@ class TestAPNs(PushNotificationTest):
|
||||||
def send(
|
def send(
|
||||||
self,
|
self,
|
||||||
devices: Optional[List[Union[PushDeviceToken, RemotePushDeviceToken]]] = None,
|
devices: Optional[List[Union[PushDeviceToken, RemotePushDeviceToken]]] = None,
|
||||||
payload_data: Dict[str, Any] = {},
|
payload_data: Mapping[str, Any] = {},
|
||||||
) -> None:
|
) -> None:
|
||||||
send_apple_push_notification(
|
send_apple_push_notification(
|
||||||
UserPushIndentityCompat(user_id=self.user_profile.id),
|
UserPushIndentityCompat(user_id=self.user_profile.id),
|
||||||
|
|
|
@ -517,7 +517,7 @@ def oauth_redirect_to_root(
|
||||||
url: str,
|
url: str,
|
||||||
sso_type: str,
|
sso_type: str,
|
||||||
is_signup: bool = False,
|
is_signup: bool = False,
|
||||||
extra_url_params: Dict[str, str] = {},
|
extra_url_params: Mapping[str, str] = {},
|
||||||
next: Optional[str] = REQ(default=None),
|
next: Optional[str] = REQ(default=None),
|
||||||
multiuse_object_key: str = REQ(default=""),
|
multiuse_object_key: str = REQ(default=""),
|
||||||
mobile_flow_otp: Optional[str] = REQ(default=None),
|
mobile_flow_otp: Optional[str] = REQ(default=None),
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
from email.headerregistry import Address
|
from email.headerregistry import Address
|
||||||
from typing import Any, Dict, List, Optional, Union
|
from typing import Any, Dict, List, Mapping, Optional, Union
|
||||||
|
|
||||||
from django.conf import settings
|
from django.conf import settings
|
||||||
from django.contrib.auth.models import AnonymousUser
|
from django.contrib.auth.models import AnonymousUser
|
||||||
|
@ -467,7 +467,7 @@ def add_bot_backend(
|
||||||
bot_type: int = REQ(json_validator=check_int, default=UserProfile.DEFAULT_BOT),
|
bot_type: int = REQ(json_validator=check_int, default=UserProfile.DEFAULT_BOT),
|
||||||
payload_url: str = REQ(json_validator=check_url, default=""),
|
payload_url: str = REQ(json_validator=check_url, default=""),
|
||||||
service_name: Optional[str] = REQ(default=None),
|
service_name: Optional[str] = REQ(default=None),
|
||||||
config_data: Dict[str, str] = REQ(
|
config_data: Mapping[str, str] = REQ(
|
||||||
default={}, json_validator=check_dict(value_validator=check_string)
|
default={}, json_validator=check_dict(value_validator=check_string)
|
||||||
),
|
),
|
||||||
interface_type: int = REQ(json_validator=check_int, default=Service.GENERIC),
|
interface_type: int = REQ(json_validator=check_int, default=Service.GENERIC),
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from typing import Any, Callable, Dict, List, Tuple
|
from typing import Any, Callable, Dict, List, Mapping, Tuple
|
||||||
|
|
||||||
import orjson
|
import orjson
|
||||||
from django.http import HttpRequest, HttpResponse
|
from django.http import HttpRequest, HttpResponse
|
||||||
|
@ -21,7 +21,7 @@ SNAPSHOT = "image_url"
|
||||||
class LibratoWebhookParser:
|
class LibratoWebhookParser:
|
||||||
ALERT_URL_TEMPLATE = "https://metrics.librato.com/alerts#/{alert_id}"
|
ALERT_URL_TEMPLATE = "https://metrics.librato.com/alerts#/{alert_id}"
|
||||||
|
|
||||||
def __init__(self, payload: Dict[str, Any], attachments: List[Dict[str, Any]]) -> None:
|
def __init__(self, payload: Mapping[str, Any], attachments: List[Dict[str, Any]]) -> None:
|
||||||
self.payload = payload
|
self.payload = payload
|
||||||
self.attachments = attachments
|
self.attachments = attachments
|
||||||
|
|
||||||
|
@ -65,7 +65,7 @@ class LibratoWebhookParser:
|
||||||
|
|
||||||
|
|
||||||
class LibratoWebhookHandler(LibratoWebhookParser):
|
class LibratoWebhookHandler(LibratoWebhookParser):
|
||||||
def __init__(self, payload: Dict[str, Any], attachments: List[Dict[str, Any]]) -> None:
|
def __init__(self, payload: Mapping[str, Any], attachments: List[Dict[str, Any]]) -> None:
|
||||||
super().__init__(payload, attachments)
|
super().__init__(payload, attachments)
|
||||||
self.payload_available_types = {
|
self.payload_available_types = {
|
||||||
ALERT_CLEAR: self.handle_alert_clear_message,
|
ALERT_CLEAR: self.handle_alert_clear_message,
|
||||||
|
@ -163,7 +163,7 @@ class LibratoWebhookHandler(LibratoWebhookParser):
|
||||||
def api_librato_webhook(
|
def api_librato_webhook(
|
||||||
request: HttpRequest,
|
request: HttpRequest,
|
||||||
user_profile: UserProfile,
|
user_profile: UserProfile,
|
||||||
payload: Dict[str, Any] = REQ(json_validator=check_dict(), default={}),
|
payload: Mapping[str, Any] = REQ(json_validator=check_dict(), default={}),
|
||||||
) -> HttpResponse:
|
) -> HttpResponse:
|
||||||
try:
|
try:
|
||||||
attachments = orjson.loads(request.body).get("attachments", [])
|
attachments = orjson.loads(request.body).get("attachments", [])
|
||||||
|
|
Loading…
Reference in New Issue