Skip to content

Instantly share code, notes, and snippets.

@mh-keshavarz
Last active July 21, 2024 06:27
Show Gist options
  • Save mh-keshavarz/eb966905428ef9017ee1ff7d25518bba to your computer and use it in GitHub Desktop.
Save mh-keshavarz/eb966905428ef9017ee1ff7d25518bba to your computer and use it in GitHub Desktop.
A function replacing silk middleware
from django.test.client import RequestFactory
from django.http import HttpResponse
from django.db import transaction, DatabaseError
from django.db.models.sql.compiler import SQLCompiler
from django.utils import timezone
from silk.collector import DataCollector
from silk.config import SilkyConfig
from silk.model_factory import RequestModelFactory, ResponseModelFactory
from silk.profiling.profiler import silk_meta_profiler
from silk.sql import execute_sql
def profile(prefix):
def wrapepr(func):
def inner(*args, **kwargs):
DataCollector().clear()
rf = RequestFactory()
path = '/tasks/{}{}/'.format(prefix, func.__name__)
request = rf.get(path)
request.silk_is_intercepted = True
if not hasattr(SQLCompiler, '_execute_sql'):
SQLCompiler._execute_sql = SQLCompiler.execute_sql
SQLCompiler.execute_sql = execute_sql
silky_config = SilkyConfig()
silky_config.SILKY_PYTHON_PROFILER
request_model = RequestModelFactory(request).construct_request_model()
DataCollector().configure(request_model)
response = func(*args, **kwargs)
process_response(request, HttpResponse(str(response)))
return response
return inner
return wrapper
@transaction.atomic()
def _process_response(request, response):
with silk_meta_profiler():
collector = DataCollector()
collector.stop_python_profiler()
silk_request = collector.request
if silk_request:
response._headers = {}
silk_response = ResponseModelFactory(response).construct_response_model()
silk_response.save()
silk_request.end_time = timezone.now()
collector.finalise()
else:
pass
if silk_request:
silk_request.save()
def process_response(request, response):
if getattr(request, 'silk_is_intercepted', False):
while True:
try:
_process_response(request, response)
except (AttributeError, DatabaseError):
_process_response(request, response)
finally:
break
return response
@OscarVanL
Copy link

OscarVanL commented Mar 24, 2024

This was super helpful, thank you.

One minor thing, def wrapepr(func): has a typo.

I managed to figure it out without too much trouble, but it could be helpful to include an example too :)

    from silk-profiler import profile
    
    profile('my_profile_name')(my_function)(arg1, arg2)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment