A function replacing silk middleware

silk-profiler.py

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

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)