Knowledge Base

cancel
Showing results for 
Search instead for 
Did you mean: 

How do I create custom exit points in Python?

How do I use the Python Agent API to add custom Exit Points?

 

The AppDynamics Python Agent provides automatic exit point detection for many common libraries. If your specific library is not supported, you can use the Agent API to add custom exit points. A perfect place for these instrumentations is application middleware. 

 

In this article, we provide you with two examples of how you can quickly add instrumentation for all your database calls, in a:

  1. Django application
  2. Flask application when SQLAlchemy is used.

 

Table of Contents 

 

How do I instrument database calls in Django?

Django comes with a feature called database instrumentation that allows you to intercept database calls. You can instrument all database calls following our guide.  

 

  1. If you already have a place to store your middleware, just add the code below.

    If this is your first additional middleware, create a file middleware.py in your application folder (here, “app”) and add the code:
    from appdynamics.agent import api as appd
    from django.db import connection
    
    def appd_wrapper(request):
        bt_handle = appd.get_active_bt_handle(request)
        def real_wrapper(execute, sql, params, many, context):
    
            settings = context['connection'].settings_dict
    
            print(settings)
    
            name = "sql://{}:{}/{}".format(settings['HOST'], settings['PORT'], settings['NAME'])
            with appd.exit_call(bt_handle, appd.EXIT_DB, name, {
                    'ENGINE': settings['ENGINE'],
                    'NAME': settings['NAME'],
                    'HOST': settings['HOST'],
                    'PORT': settings['PORT']
                }, operation=sql):
                return execute(sql, params, many, context)
    
        return real_wrapper
    
    class AppDynamicsMiddleware:
        def __init__(self, get_response):
            self.get_response = get_response
    
        def __call__(self, request):
            with connection.execute_wrapper(appd_wrapper(request)):
                return self.get_response(request)

     

  2. Put this code into action by using it as middleware in settings.py.
    For example:
    MIDDLEWARE = [
    ...
    'appd.middleware.AppDynamicsMiddleware',
    ...
    ]

 

How do I instrument database calls with SQLAlchemy?

Flask (and every other wsgi application) allows you to add middleware to your code by wrapping your wsgi_app. For example, you might have the following code in your __init.py__:

app = Flask(__name__)

 

  1. Add the following code:
    App = Flask(__name__)
    App.wsgi_app = AppDynamicsMiddleware(app.wsgi_app)

     

  2. Now, add the middleware code into a file. For example “app/middleware/appdynamics.py”:
    import logging
    from werkzeug.wrappers import Request, Response
    from appdynamics.agent import api as appd
    from sqlalchemy.engine import Engine
    from sqlalchemy.event import listen
    from sqlalchemy.pool import Pool
    
    class AppDynamics:
    
        def __init__(self, app):
            logging.info('MIDDLEWARE INIT')
            listen(Engine, "before_cursor_execute", self.before_cursor_execute)
            listen(Engine, "after_cursor_execute", self.after_cursor_execute)
            listen(Engine, "handle_error", self.handle_error)
            appd.init()
            self.bt_handle = None
            self.req = None
            self.app = app
            self.exit_call = None
    
        def __call__(self, environ, start_response):
            self.req = Request(environ, shallow=True)
            return self.app(environ, start_response)
    
        def on_connect(self, conn, *args):
            logging.warn("New DB connection:", conn)
    
        def before_cursor_execute(self, conn, cursor, statement, parameters, context, executemany):
            
            id_props = {
                'Host': conn.engine.url.host, 
                'Port': conn.engine.url.port,
                'Vendor': conn.engine.url.database
            }
    
            name = f'sql://{conn.engine.url.host}:{conn.engine.url.port}:{conn.engine.url.database}'
    
            # self.bt_handle = appd.start_bt(self.req.environ['PATH_INFO'])
            
            self.bt_handle = appd.get_active_bt_handle(self.req)
            self.exit_call = appd.start_exit_call(self.bt_handle, appd.EXIT_DB, name, id_props, operation=statement)
    
    
        def after_cursor_execute(self, conn, cursor, statement, parameters, context, executemany):
            # logging.warn("middleware after cursor execution", self.exit_call)
            if self.exit_call:
                appd.end_exit_call(self.exit_call)
    
        def handle_error(self, context):
            # logging.warn("middleware handle error", self.exit_call)
            if self.exit_call:
                appd.end_exit_call(self.exit_call, context)

 

Additional Resources

Version history
Revision #:
6 of 6
Last update:
‎05-25-2021 06:36 PM
Updated by:
 
Labels (1)
Tags (1)


Found this article helpful? Click the Thumbs Up button.
Have an additional comment? Post it below.
0 Kudos