Analyzing APIs with PyInterceptor

RustyTunaDev — Fri, 17 Jan 2025 09:39:20 +0000

Status January, 16th 2025: Currently, PyInterceptor is in a very early development phase and is available on GitHub.

Imagine, that you have a Python Client class, calling some Python API. This can be for example a processor calling some math library, a GUI calling an http-client, or something more low-level like a machine class calling a hardware API. In this fictive scenario imagine that you want to get detailed information about the executed API functions, such as their names, the parameters passed, their return values, the exact execution time, etc. but without changing the underlying code.

This is exactly what PyInterceptor is trying to achieve: the interception and analysis of function calls, as well as their further processing but without changing the actual implementation.

This article is the first one I publish on dev.to and it introduces the basic ideas and principles behind PyInterceptor, its use cases, and its application.

Python Call Interception and Processing

The Basic Principles

Figure 1 — Call Interception Overview: Blocking vs. Non-Blocking Interception

When discussing intercepting function calls or methods we have to distinguish between blocking and non-blocking interception (see Figure 1).

The former one (blocking) — like the name already, states — just intercepts calls, extracts relevant information, and returns immediately without forwarding the call to its destination target. This is very useful especially for unit testing when creating Mocks or Stubs.

The latter approach (non-blocking) also extracts information from the call but then forwards it to the target. The target method’s return value is also intercepted and finally forwarded back to the caller.

PyInterceptor supports both call interception methods.

Use Cases

To be able to intercept Python API calls enables lots of interesting and useful possibilities. Here are some examples:

Create Mocks or Stubs for unit tests semi-automatically using the blocking mode. In addition, PyInterceptor accepts a custom interceptor callable. In blocking mode, this allows the return of arbitrary data such as simulated or dummy data to the caller.
Forward intercepted arguments and meta-information to a (structured) logging library.
Helps during debugging: knowing when a function was called with which arguments can make finding bugs a lot easier
Create call statistics
Etc.

Going into the Details

Figure 2 — Details: Including the handler() function and interceptor() callable

The working principle of PyInterceptor is very easy to explain. A client wants to execute an API method. Instead of calling the method directly, a handler function — installed by PyInterceptor — is executed. This handler captures several (meta-) infos about the actual target method (arguments, time of call, etc.), stores it in a CallInfo object, and takes care of forwarding the call.

In blocking mode, the handler directly passes the CallInfo to the interceptor callable — a user-defined callback. The interceptor function is responsible for the actual processing of the CallInfo objects, e.g. to log them, to create call statistics, to pass them to a message broker, etc. PyInterceptor therefore only intercepts and forwards the function calls — the user is responsible for implementing the actual processing logic. Finally, the handler delivers the result from the interceptor back to the calling function.

In non-blocking mode, the handler executes the actual target method, captures its return value, and stores it in the CallInfo before executing the interceptor callable. The rest of the sequence is then the same as in non-blocking mode. At this point, it should be noted that in non-blocking mode, the real value of the target function is returned instead of the interceptor return value (unlike blocking mode).

Explanation by a Code Example

In this section, I would like to explain how to use PyInterceptor using a small example. Given is an arithmetic API class with 4 methods: add, sub, mul, and div. There is also a Processor class with the methods calc_mean() and calc_variance(). The aim is to document all method calls of the API and Processor class in a JSON file.

import json
from pathlib import Path
from typing import List

from interceptor import intercept, get_methods, CallInfo


class API:

    def add(self, a, b):
        return a + b

    def sub(self, a, b):
        return a - b

    def mul(self, a, b):
        return a * b

    def div(self, a, b):
        return a / b


class Processor:

    def __init__(self, api):
        self._api = api

    def calc_mean(self, a, b):
        return self._api.div(a=self._api.add(a, b), b=2)

    def calc_variance(self, a, b):
        mean = self.calc_mean(a, b)
        x = self._api.sub(a, mean)
        y = self._api.sub(b, mean)
        return self._api.div(self._api.mul(x, x) + self._api.mul(y, y), 2)


class JSONLogger:

    def __init__(self):
        self._logs: List[CallInfo] = []

    def __call__(self, info: CallInfo):
        self._logs.append(info)

    def save_as_json(self, path: Path):
        data = {"logs": []}
        for log in self._logs:
            data["logs"].append({
                "name": log.name,
                "args": log.args,
                "kwargs": log.kwargs,
                "ret_value": log.ret_value,
                "target": log.target.__name__,
                "timestamp_ns": log.timestamp_ns
            })
        with open(path, 'w') as f:
            json.dump(data, f, indent=2)


if __name__ == '__main__':
    # Create Interceptor instance
    logger = JSONLogger()

    # Create API instance and mount it for interception
    api = API()
    intercept(get_methods(api), target=api, interceptor=logger, blocking=False)

    # Create Processor instance and mount it for interception
    processor = Processor(api)
    intercept(get_methods(processor), target=processor, interceptor=logger, blocking=False)

    # Execute processor methods and save logs
    print(processor.calc_variance(4, 2))
    logger.save_as_json(Path("logs.json"))

If we look at the main method, we first see that an instance of the JSONLogger class is created. This class contains the logic for saving the CallInfo objects and exporting them to a JSON file. JSONLogger implements the call method, which means that objects of this class can be called directly just like functions. This means we can use the logger object directly as an interceptor callable.

In the second step, we create the API and Processor objects and mount them for interception using the intercept function. This function accepts:

A list of strings representing the names of the target object’s methods to be intercepted
The target object (the API and Processor instance)
The interceptor callable (the JSONLogger instance)
A boolean denoting if the interception should be performed in blocking mode or not

In the last step, we call and print the result from the processor’s calc_variance() method and save the interception results in a JSON file which can be seen below. What we are getting is a list of all executed API and Processor methods together with their names, args, kwargs, return values, the target object’s class, and the timestamp in nanoseconds.

{
  "logs": [
    {
      "name": "add",
      "args": [4, 2],
      "kwargs": {},
      "ret_value": 6,
      "target": "API",
      "timestamp_ns": 1736970850230849900
    },
    {
      "name": "div",
      "args": [],
      "kwargs": {
        "a": 6,
        "b": 2
      },
      "ret_value": 3.0,
      "target": "API",
      "timestamp_ns": 1736970850230853600
    },
    {
      "name": "calc_mean",
      "args": [4, 2],
      "kwargs": {},
      "ret_value": 3.0,
      "target": "Processor",
      "timestamp_ns": 1736970850230847300
    },
    {
      "name": "sub",
      "args": [4, 3.0],
      "kwargs": {},
      "ret_value": 1.0,
      "target": "API",
      "timestamp_ns": 1736970850230875100
    },
    {
      "name": "sub",
      "args": [2, 3.0],
      "kwargs": {},
      "ret_value": -1.0,
      "target": "API",
      "timestamp_ns": 1736970850230881100
    },
    {
      "name": "mul",
      "args": [1.0, 1.0],
      "kwargs": {},
      "ret_value": 1.0,
      "target": "API",
      "timestamp_ns": 1736970850230883500
    },
    {
      "name": "mul",
      "args": [-1.0, -1.0],
      "kwargs": {},
      "ret_value": 1.0,
      "target": "API",
      "timestamp_ns": 1736970850230887300
    },
    {
      "name": "div",
      "args": [2.0, 2],
      "kwargs": {},
      "ret_value": 1.0,
      "target": "API",
      "timestamp_ns": 1736970850230890000
    },
    {
      "name": "calc_variance",
      "args": [4, 2],
      "kwargs": {},
      "ret_value": 1.0,
      "target": "Processor",
      "timestamp_ns": 1736970850230843000
    }
  ]
}

Future Plans and ToDos

There are currently some ideas that would make working with PyInterceptor even more efficient and flexible:

Create an API documentation and a CI pipeline.
Implement Python decorators which can be easily applied either to classes (to intercept all methods) or to specific methods.
Add more settings to PyInterceptor e.g., a flag that denotes if intercepted values (args, kwargs, return values) should be copied or stored as references in the CallInfo objects. This is important to prevent unexpectedly high memory consumption especially when dealing with huge size objects.
Provide some standard interceptors, (e.g., something similar to the JSONLogger class) that can be used out-of-the-box but can also be tailored for custom datatypes.

If you like this article and/or want me to write more articles about the development process and the ideas behind PyInterceptor, please leave a comment — Thank you 😃!