Circular Dependency Traps in Python Microservices and Solutions-Purple Clay Logic

Opening Thoughts

Have you encountered this situation: while coding, you suddenly discover two services that depend on each other, and the program just won't run? When I first encountered this problem, I was also puzzled. After continuous exploration and practice, I finally found some effective solutions. Today I'd like to share my experiences in handling circular dependency issues in Python microservices.

Origin of the Problem

I remember when I first started working with microservice architecture, I habitually wrote service calls in the form of direct imports. For example, if an order service needs to call the user service to get user information, I would write:

from user_service import get_user_info

def process_order(order_id):
    user_id = get_order_user_id(order_id)
    user_info = get_user_info(user_id)  # Direct call to user service
    # Order processing logic

This looks intuitive, right? But as business logic becomes more complex, the relationships between services become increasingly intricate. Until one day, I needed to query order information in the user service:

from order_service import get_user_orders

def get_user_info(user_id):
    orders = get_user_orders(user_id)  # Call order service
    return {
        "user_id": user_id,
        "orders": orders
    }

Now we're in trouble - the order service depends on the user service, and the user service depends on the order service, creating a circular dependency. The program throws an ImportError when run, which is frustrating.

Have you ever wondered why this happens? This actually reflects a deeper issue: when designing services, we often focus too much on implementing business functionality while neglecting the dependency relationships between services. It's like building a house - if you only focus on stacking materials without considering structural stability, problems will eventually arise.

Breaking the Deadlock

After repeated practice and exploration, I've summarized several effective solutions. Let me explain them one by one.

Dependency Injection Method

The first solution is to use dependency injection. The core idea is: rather than letting services fetch dependencies themselves, inject them from the outside.

class OrderService:
    def __init__(self, user_service):
        self.user_service = user_service

    def process_order(self, order_id):
        user_id = self.get_order_user_id(order_id)
        user_info = self.user_service.get_user_info(user_id)
        # Order processing logic

class UserService:
    def __init__(self, order_service):
        self.order_service = order_service

    def get_user_info(self, user_id):
        orders = self.order_service.get_user_orders(user_id)
        return {"user_id": user_id, "orders": orders}


container = DependencyContainer()
user_service = UserService(None)
order_service = OrderService(None)


user_service.order_service = order_service
order_service.user_service = user_service

See, through dependency injection, we've decoupled the tightly coupled services. It's like building with blocks - we prepare all the modules first, then assemble them one by one. This not only solves the circular dependency problem but also improves code testability and maintainability.

Event-Driven Architecture

The second solution is to adopt an event-driven architecture. We can introduce a message queue to let services communicate through publishing/subscribing to messages rather than direct calls.

import asyncio
from collections import deque

class EventBus:
    def __init__(self):
        self.subscribers = {}

    def subscribe(self, event_type, callback):
        if event_type not in self.subscribers:
            self.subscribers[event_type] = []
        self.subscribers[event_type].append(callback)

    async def publish(self, event_type, data):
        if event_type in self.subscribers:
            for callback in self.subscribers[event_type]:
                await callback(data)

class OrderService:
    def __init__(self, event_bus):
        self.event_bus = event_bus
        self.event_bus.subscribe("user_info_updated", self.handle_user_update)

    async def process_order(self, order_id):
        # Publish event requesting user information
        await self.event_bus.publish("get_user_info", {"order_id": order_id})

    async def handle_user_update(self, data):
        # Handle user information update event
        print(f"Order service received user update: {data}")

class UserService:
    def __init__(self, event_bus):
        self.event_bus = event_bus
        self.event_bus.subscribe("get_user_info", self.handle_get_user_info)

    async def handle_get_user_info(self, data):
        # Handle get user information request
        user_info = {"user_id": "123", "name": "Zhang San"}
        await self.event_bus.publish("user_info_updated", user_info)

This approach completely changes the communication pattern between services. See, services no longer directly call each other's methods but interact through publishing and subscribing to events. It's like two departments communicating via email rather than direct phone calls. This way, the services are completely decoupled.

Interface Abstraction

The third solution is to use interface abstraction. We can define service interfaces and have concrete implementations depend on abstract interfaces rather than concrete implementation classes.

from abc import ABC, abstractmethod

class UserServiceInterface(ABC):
    @abstractmethod
    def get_user_info(self, user_id):
        pass

class OrderServiceInterface(ABC):
    @abstractmethod
    def get_user_orders(self, user_id):
        pass

class OrderService(OrderServiceInterface):
    def __init__(self, user_service: UserServiceInterface):
        self.user_service = user_service

    def process_order(self, order_id):
        user_id = self.get_order_user_id(order_id)
        user_info = self.user_service.get_user_info(user_id)
        # Order processing logic

class UserService(UserServiceInterface):
    def __init__(self, order_service: OrderServiceInterface):
        self.order_service = order_service

    def get_user_info(self, user_id):
        orders = self.order_service.get_user_orders(user_id)
        return {"user_id": user_id, "orders": orders}

By defining interfaces, we elevate dependency relationships to an abstract level. It's like defining a contract that specifies the rules both parties need to follow, without concerning how they're implemented. This approach not only solves the circular dependency problem but also improves code extensibility.

Practical Experience

After all this theory, let me share some practical lessons from real projects.

In an e-commerce project, we initially used direct calls. As the business scale expanded, dependencies between services became increasingly complex, and circular dependency issues frequently occurred. Later we decided to refactor the entire system, adopting an event-driven architecture.

We used RabbitMQ as the message queue and rewrote the inter-service communication logic:

import pika
import json

class MessageBroker:
    def __init__(self):
        self.connection = pika.BlockingConnection(
            pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()

    def publish(self, routing_key, message):
        self.channel.basic_publish(
            exchange='',
            routing_key=routing_key,
            body=json.dumps(message)
        )

    def subscribe(self, queue_name, callback):
        self.channel.queue_declare(queue=queue_name)
        self.channel.basic_consume(
            queue=queue_name,
            on_message_callback=callback,
            auto_ack=True
        )
        self.channel.start_consuming()

class OrderService:
    def __init__(self):
        self.broker = MessageBroker()

    def process_order(self, order_id):
        # Publish message to get user information
        self.broker.publish('get_user_info', {
            'order_id': order_id,
            'callback_queue': 'order_service_callback'
        })

    def handle_user_info(self, ch, method, properties, body):
        user_info = json.loads(body)
        # Process user information
        print(f"Received user info: {user_info}")


order_service = OrderService()
order_service.broker.subscribe(
    'order_service_callback',
    order_service.handle_user_info
)

This transformation wasn't easy, but the results were significant:

System maintainability greatly improved. Each service is independent, and modifying one service doesn't affect others.
System scalability also improved. We can easily add new services without worrying about dependency issues.
System stability improved as well. Even if one service is temporarily unavailable, other services can continue running normally.

However, this architecture also brought new challenges:

Message delivery reliability. We need to ensure messages aren't lost, requiring message confirmation mechanisms.
Message ordering. In some scenarios, message processing order is important, requiring additional mechanisms to ensure order.
Increased debugging difficulty. Because service calls became asynchronous, debugging and tracking issues became more complex.

To address these challenges, we took several measures:

Implemented message retry mechanism:

class MessageBroker:
    def publish_with_retry(self, routing_key, message, max_retries=3):
        retries = 0
        while retries < max_retries:
            try:
                self.publish(routing_key, message)
                return True
            except Exception as e:
                retries += 1
                print(f"Failed to publish message, retry {retries}/{max_retries}")
                if retries == max_retries:
                    raise e

Added message tracking functionality:

class MessageTracer:
    def __init__(self):
        self.traces = {}

    def trace_message(self, message_id, service, action):
        if message_id not in self.traces:
            self.traces[message_id] = []
        self.traces[message_id].append({
            'service': service,
            'action': action,
            'timestamp': datetime.now()
        })

    def get_message_trace(self, message_id):
        return self.traces.get(message_id, [])

Implemented message idempotency handling:

class MessageHandler:
    def __init__(self):
        self.processed_messages = set()

    def handle_message(self, message_id, message):
        if message_id in self.processed_messages:
            print(f"Message {message_id} already processed, skipping")
            return

        # Process message
        self.process_message(message)

        # Mark message as processed
        self.processed_messages.add(message_id)

Final Thoughts

Through these practices, I deeply realized that handling service dependencies in microservice architecture is not just a technical issue but an architectural design issue. We need to consider these issues at the beginning of system design, rather than waiting until problems arise.

What do you think about these solutions? How do you handle service dependencies in your projects? Feel free to share your experiences and thoughts in the comments.

Finally, here's a question to ponder: if you discover circular dependency issues in an already live system, how would you gradually refactor while ensuring normal system operation?