CallerAPI: An Innovative Microservice Processing Solution

Disclaimer

CallerAPI was developed exclusively for our company’s internal needs and is intended to simplify the interaction between microservices within our document processing platform. The main goal of developing this package was to create a convenient and effective tool that would simplify the integration and interaction of a large number of microservices, allowing our developers to focus on solving business problems, rather than on the routine work of setting up and calling services.

I would like to share an experience and best practices, as they may be useful to other developers facing similar problems.

Introduction

In the modern world, microservice architecture has become one of the standards for developing complex software solutions. It allows you to split your application into many small, independent services, making them easier to develop, test, and deploy. However, with the increase in the number of microservices, problems arise between their interaction and coordination. In the company, I was faced with the need to integrate more than 40 microservices and the developed CallerAPI package greatly simplified this process.

Problem Overview

With a large number of microservices, it becomes inconvenient and difficult to manage their interactions. Each service has its own API, and calling one service from another requires writing a lot of code to configure HTTP requests, error handling, etc. This leads to code duplication, increased development and testing complexity.

Below I listed some related problems of microservices interaction:

• Difficulty of multiple microservices integration.
• Necessity of writing a big code volume for each HTTP-request.
• Increased complexity of error handling and session management.
• Difficulties in maintaining and updating the code.

Solution

To solve these problems, the CallerAPI package was developed, which allows you to easily and conveniently interact with microservices through a unified interface. The main idea of CallerAPI is to collect all services and their APIs in one package that can be installed in any microservice and used with minimal effort.

The main advantages of CallerAPI

• Simplification of calling microservices by using a single interface.
• Reduction of code volume by unifying interaction methods.
• Convenient error handling and session management.
• Easy code maintenance and update.
• Convenient configuration of environment variables (env).

Realization

The main components of CallerAPI

1.1. AIOHttpClient

The AIOHttpClient class is responsible for managing HTTP client sessions. It provides creation and configuration of aiohttp.ClientSession for each service.

class AIOHttpClient:
    """
    A class for managing AIOHttp client sessions.
    """
    def __init__(self, base_url: str) -> None:
        self.base_url = base_url

    def aiohttp_client(
        self, service_id: Union[ServiceID, str], timeout: Optional[Union[int, aiohttp.ClientTimeout]] = None
    ) -> aiohttp.ClientSession:
        """
        Return aiohttp.ClientSession based on the given service ID and timeout.
        Creates a new session if it does not exist or the previous one is closed.
        """
        import logging
        logging.basicConfig(level=logging.CRITICAL)

        if not timeout:
            timeout_settings = aiohttp.ClientTimeout(total=300, sock_read=180)
        elif isinstance(timeout, int):
            timeout_settings = aiohttp.ClientTimeout(total=timeout, sock_read=180)
        elif isinstance(timeout, aiohttp.ClientTimeout):
            timeout_settings = timeout

        headers = {
            "Accept": "application/json",
        }

        client = aiohttp.ClientSession(
            headers=headers,
            base_url=self.base_url.format(use_service_id=service_id.value.lower()),
            timeout=timeout_settings,
        )

        return client

It also allows to flexibly configure timeouts for HTTP sessions:

def aiohttp_client(self, service_id: Union[ServiceID, str], timeout: Optional[Union[int, aiohttp.ClientTimeout]] = None) -> aiohttp.ClientSession:
    ...

1.2. BaseApi

The BaseApi class is the base class for all API classes. It combines all the methods for executing HTTP requests (GET, POST, DELETE, etc.), which allows you to minimize code repetition and centralize request processing.

class BaseApi:
    """
    A base class for making API calls
    """
    http_client = None

    def __init__(
        self,
        app_id: str,
        base_url: str,
        timeout: Optional[Union[int, aiohttp.ClientTimeout]] = None,
    ) -> None:
        self.app_id = app_id
        self.http_client = AIOHttpClient(base_url=base_url)
        self.timeout = timeout

    async def _get_client(self) -> aiohttp.ClientSession:
        """
        Return ClientSession instance.
        """
        return self.http_client.aiohttp_client(self.app_id, self.timeout)

    async def _request(self, method: str, ep: str, data: Any = None, encode: bool = True) -> aiohttp.ClientResponse:
        """
        Make an HTTP request using the given method, endpoint, and data.
        """
        http_client = await self._get_client()
        json_compatible_data = jsonable_encoder(data) if encode else data

        if method == "post":
            response = await http_client.post(ep, json=json_compatible_data)
        elif method == "get":
            response = await http_client.get(ep, params=data)
        elif method == "delete":
            response = await http_client.delete(ep, params=json_compatible_data)
        elif method == "patch":
            response = await http_client.patch(ep, json=json_compatible_data)
        elif method == "put":
            response = await http_client.put(ep, json=json_compatible_data)
        elif method == "delete_body":
            response = await http_client.delete(ep, json=json_compatible_data)
        elif method == "post_data":
            response = await http_client.post(ep, data=json_compatible_data)
        else:
            raise ValueError(f"Unsupported method: {method}")

        return response

Please note that there are 2 additional methods post_data and delete_body used here.

Environment Configuration Example

from starlette.config import Config

config = Config(".env")

DEPLOY = config("DEPLOY", default=False)

BASE_URLS = {
    "SERVICE_NAME": "http://localhost:6001",
    ...
}

def get_config(key: str) -> str:
    if DEPLOY:
        DEFAULT_URL = config("DEFAULT_URL", default="http://localhost:6000")
        return config(key, default=DEFAULT_URL)
    else:
        return config(key, default=BASE_URLS[key])

for key in BASE_URLS:
    globals()[key] = get_config(key)

Usage Examples

ServiceID class using Enum:

class ServiceID(str, Enum):
    foo = "Foo"
    ...

6.1. Example class for calling a service

class Foo(CommonEndpoints):
    """
    A class for calling Foo service.
    """
    def __init__(self, timeout: Optional[Union[int, aiohttp.ClientTimeout]] = None) -> None:
        super().__init__(
            app_id=ServiceID.foo,
            base_url=S.FOO_URL,
            timeout=timeout,
        )

    async def v1_process(self, data: model.FooInput) -> aiohttp.ClientResponse:
        """
        Make a post request to the /v1/process endpoint.
        """
        return await self._request("post", "/v1/process", data)

6.2. File transfer and multipart/form-data usage

class Bar(CommonEndpoints):
    """
    A class for calling Bar service.
    """
    def __init__(self, timeout: Optional[Union[int, aiohttp.ClientTimeout]] = None) -> None:
        super().__init__(
            ServiceID.bar,
            S.Bar_URL,
            timeout,
        )

    async def v1_file(self, subdomain: str, client_id: str, file: UploadFile) -> aiohttp.ClientResponse:
        """
        Make a post request to the /v1/file endpoint.
        """
        ep = f"/v1/file?subdomain={subdomain}&client_id={client_id}"
        form = aiohttp.FormData()
        form.add_field("file", await file.read(), filename=file.filename, content_type="application/octet-stream")
        return await self._request("post_data", ep, form, False)

Conclusion

CallerAPI greatly simplifies communication between microservices by providing a unified interface for calling APIs. It reduces the amount of code required to integrate services, improves session management and error handling, which ultimately improves productivity and ease of development. Additionally, CallerAPI provides convenient mechanisms for configuring environment variables, as well as support for file transfer and multipart/form-data.