zulip/zilencer/management/commands/profile_request.py

60 lines
2.0 KiB
Python
Raw Normal View History

2017-11-16 00:55:49 +01:00
import cProfile
import logging
import tempfile
from typing import Any, Dict
from django.core.management.base import CommandParser
2017-11-16 00:55:49 +01:00
from django.http import HttpRequest, HttpResponse
from zerver.lib.management import ZulipBaseCommand
2017-11-16 00:55:49 +01:00
from zerver.middleware import LogRequests
from zerver.models import UserMessage, UserProfile
from zerver.views.messages import get_messages_backend
request_logger = LogRequests()
class MockSession:
def __init__(self) -> None:
self.modified = False
class MockRequest(HttpRequest):
def __init__(self, user: UserProfile) -> None:
self.user = user
self.path = '/'
self.method = "POST"
self.META = {"REMOTE_ADDR": "127.0.0.1"}
anchor = UserMessage.objects.filter(user_profile=self.user).order_by("-message")[200].message_id
self.REQUEST = {
"anchor": anchor,
"num_before": 1200,
"num_after": 200
}
python: Convert assignment type annotations to Python 3.6 style. This commit was split by tabbott; this piece covers the vast majority of files in Zulip, but excludes scripts/, tools/, and puppet/ to help ensure we at least show the right error messages for Xenial systems. We can likely further refine the remaining pieces with some testing. Generated by com2ann, with whitespace fixes and various manual fixes for runtime issues: - invoiced_through: Optional[LicenseLedger] = models.ForeignKey( + invoiced_through: Optional["LicenseLedger"] = models.ForeignKey( -_apns_client: Optional[APNsClient] = None +_apns_client: Optional["APNsClient"] = None - notifications_stream: Optional[Stream] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE) - signup_notifications_stream: Optional[Stream] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE) + notifications_stream: Optional["Stream"] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE) + signup_notifications_stream: Optional["Stream"] = models.ForeignKey('Stream', related_name='+', null=True, blank=True, on_delete=CASCADE) - author: Optional[UserProfile] = models.ForeignKey('UserProfile', blank=True, null=True, on_delete=CASCADE) + author: Optional["UserProfile"] = models.ForeignKey('UserProfile', blank=True, null=True, on_delete=CASCADE) - bot_owner: Optional[UserProfile] = models.ForeignKey('self', null=True, on_delete=models.SET_NULL) + bot_owner: Optional["UserProfile"] = models.ForeignKey('self', null=True, on_delete=models.SET_NULL) - default_sending_stream: Optional[Stream] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE) - default_events_register_stream: Optional[Stream] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE) + default_sending_stream: Optional["Stream"] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE) + default_events_register_stream: Optional["Stream"] = models.ForeignKey('zerver.Stream', null=True, related_name='+', on_delete=CASCADE) -descriptors_by_handler_id: Dict[int, ClientDescriptor] = {} +descriptors_by_handler_id: Dict[int, "ClientDescriptor"] = {} -worker_classes: Dict[str, Type[QueueProcessingWorker]] = {} -queues: Dict[str, Dict[str, Type[QueueProcessingWorker]]] = {} +worker_classes: Dict[str, Type["QueueProcessingWorker"]] = {} +queues: Dict[str, Dict[str, Type["QueueProcessingWorker"]]] = {} -AUTH_LDAP_REVERSE_EMAIL_SEARCH: Optional[LDAPSearch] = None +AUTH_LDAP_REVERSE_EMAIL_SEARCH: Optional["LDAPSearch"] = None Signed-off-by: Anders Kaseorg <anders@zulipchat.com>
2020-04-22 01:09:50 +02:00
self.GET: Dict[Any, Any] = {}
self.session = MockSession()
def get_full_path(self) -> str:
return self.path
def profile_request(request: HttpRequest) -> HttpResponse:
request_logger.process_request(request)
prof = cProfile.Profile()
prof.enable()
ret = get_messages_backend(request, request.user,
apply_markdown=True)
prof.disable()
with tempfile.NamedTemporaryFile(prefix='profile.data.', delete=False) as stats_file:
prof.dump_stats(stats_file.name)
request_logger.process_response(request, ret)
logging.info("Profiling data written to {}".format(stats_file.name))
return ret
class Command(ZulipBaseCommand):
def add_arguments(self, parser: CommandParser) -> None:
parser.add_argument("email", metavar="<email>", type=str, help="Email address of the user")
self.add_realm_args(parser)
def handle(self, *args: Any, **options: Any) -> None:
realm = self.get_realm(options)
user = self.get_user(options["email"], realm)
profile_request(MockRequest(user))