Building an Inventory Service with Event-Driven Stock Management

1. Conceptual Foundation: Event-Driven Inventory Management

In a real-world e-commerce system, managing inventory is fraught with challenges like overselling (selling more items than you have) and race conditions. A synchronous approach, where the Order Service calls the Inventory Service to deduct stock, can create tight coupling and bottlenecks.

A more robust pattern is an event-driven saga. Here, services communicate by publishing and subscribing to events, allowing them to operate independently.

Choosing a Message Broker

To facilitate this event-driven communication, we need a message broker. While powerful systems like Apache Kafka are common for high-throughput data pipelines, they can introduce significant operational complexity. For our system, we'll start with Redis Streams.

We're choosing Redis Streams because it's lightweight, extremely fast, and you'll likely be using Redis for caching later anyway, which keeps our technology stack lean.

2. Building the Inventory Service with FastAPI and Redis

Let's set up the project structure for the Inventory Service. It will be a FastAPI application responsible for basic CRUD operations on products and will house our event consumer logic.

In your microservices-platform/services/ directory, create a new folder inventory-service:

services/
└── inventory-service/
    ├── app/
    │   ├── __init__.py
    │   ├── consumer.py     # Listens for and processes events
    │   ├── main.py         # FastAPI app, API endpoints, data models
    └── requirements.txt

Step 1: Setting up the FastAPI Application and Data Model

We'll use the redis-om library to model our inventory data and store it in Redis. This gives us an object-oriented way to interact with Redis data, much like an ORM for a SQL database.

The following video provides a complete walkthrough of setting up a FastAPI service that uses redis-om to manage products. We will follow its approach to build the foundation of our Inventory Service.

After following the video, your main.py and requirements.txt should be set up. Your main.py should look something like this:

# services/inventory-service/app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from redis_om import get_redis_connection, HashModel




# In a real app, these would come from settings/environment variables
REDIS_HOST = "localhost"
REDIS_PORT = 6379



# Add your Redis password if you have one
# REDIS_PASSWORD = "your-password"

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"], # Adjust for your frontend if you have one
    allow_methods=["*"],
    allow_headers=["*"],
)

redis = get_redis_connection(
    host=REDIS_HOST,
    port=REDIS_PORT,



    # password=REDIS_PASSWORD,
    decode_responses=True
)

class Product(HashModel):
    name: str
    price: float
    quantity: int

    class Meta:
        database = redis

@app.get("/products")
def all():



    # redis-om provides the pk (primary key) automatically
    return [format(pk) for pk in Product.all_pks()]

def format(pk: str):
    product = Product.get(pk)
    return {
        'id': product.pk,
        'name': product.name,
        'price': product.price,
        'quantity': product.quantity
    }

@app.post("/products")
def create(product: Product):
    return product.save()

@app.get("/products/{pk}")
def get(pk: str):
    return Product.get(pk)

@app.delete("/products/{pk}")
def delete(pk: str):
    return Product.delete(pk)

Your requirements.txt should contain:

fastapi
uvicorn
redis-om

You can now run this service (uvicorn app.main:app --reload --port 8002) and use an API tool like Postman or Insomnia to add a few products. For example, POST to http://localhost:8002/products:

{
    "name": "Laptop",
    "price": 1200.00,
    "quantity": 50
}

3. Implementing the Event Consumer

Now for the core logic. We'll create a standalone consumer.py script. This script will run in an infinite loop, listening for events on a Redis Stream. When an OrderCreated event appears, it will update the stock and publish a corresponding StockReserved or StockRejected event.

This process is a cornerstone of event-driven architecture. The video you watched earlier demonstrates exactly how to build such a consumer and even how to handle complex failure scenarios.

Let's adapt the patterns from the video to create our consumer.py. We'll listen on a stream called order_events and publish results to stock_events.

# services/inventory-service/app/consumer.py

import time
import json
from main import redis, Product




# Configuration for streams and groups
ORDER_STREAM_KEY = 'order_events'
ORDER_GROUP_NAME = 'inventory-group'
STOCK_STREAM_KEY = 'stock_events'

try:



    # Create the consumer group
    redis.xgroup_create(ORDER_STREAM_KEY, ORDER_GROUP_NAME, id='0', mkstream=True)
    print("Consumer group 'inventory-group' created.")
except Exception as e:
    print(f"Group already exists: {e}")

while True:
    try:



        # Wait for new messages in the stream
        results = redis.xreadgroup(ORDER_GROUP_NAME, ORDER_GROUP_NAME, {ORDER_STREAM_KEY: '>'}, count=1)

        if results:
            for stream, messages in results:
                for message_id, data in messages:
                    try:
                        order = {k: v for k, v in data.items()}
                        print(f"Processing order: {order['order_id']}")

                        product = Product.get(order['product_id'])
                        
                        requested_quantity = int(order['quantity'])

                        if product.quantity >= requested_quantity:



                            # Sufficient stock, proceed with reservation
                            product.quantity -= requested_quantity
                            product.save()
                            



                            # Publish StockReserved event
                            event_data = {
                                'order_id': order['order_id'],
                                'product_id': product.pk,
                                'status': 'RESERVED'
                            }
                            redis.xadd(STOCK_STREAM_KEY, event_data)
                            print(f"Stock reserved for order {order['order_id']}. New quantity: {product.quantity}")

                        else:



                            # Insufficient stock, publish rejection
                            event_data = {
                                'order_id': order['order_id'],
                                'product_id': product.pk,
                                'status': 'REJECTED',
                                'reason': 'INSUFFICIENT_STOCK'
                            }
                            redis.xadd(STOCK_STREAM_KEY, event_data)
                            print(f"Stock rejected for order {order['order_id']}. Reason: Insufficient stock.")




                        # Acknowledge the message so it's not processed again
                        redis.xack(ORDER_STREAM_KEY, ORDER_GROUP_NAME, message_id)

                    except Exception as processing_error:
                        print(f"Error processing message {message_id}: {processing_error}")



                        # In production, you might move this to a dead-letter queue
    
    except Exception as e:
        print(f"Error reading from stream: {e}")
        time.sleep(1)

This consumer reliably processes events. By using a consumer group and acknowledging messages (xack), we ensure that even if the consumer crashes, another instance can pick up where it left off. This is a key feature for building resilient systems.

4. Testing the Event-Driven Flow

Now that you have the service and the consumer, let's test the entire flow.

1. Start the Inventory Service API:

   
   
   
   # In terminal 1
   cd services/inventory-service
   uvicorn app.main:app --reload --port 8002
   

   Use an API client to add a product if you haven't already. Note the product's ID (e.g., 01J3...).

2. Start the Consumer:

   
   
   
   # In terminal 2
   cd services/inventory-service
   python app/consumer.py
   

   You should see the "Consumer group created" or "Group already exists" message. It's now waiting for events.

3. Simulate an OrderCreated Event:
   Since we haven't modified the Order Service to publish events yet, we can simulate it by publishing an event directly to Redis. You can do this with a simple Python script or even the redis-cli.

   Create a file named test_publisher.py in the inventory-service directory:

   
   
   
   # services/inventory-service/test_publisher.py
   import redis
   import uuid
   
   r = redis.Redis(decode_responses=True)
   
   
   
   
   # Replace with a real product ID from your inventory
   product_id = "YOUR_PRODUCT_ID_HERE" 
   
   order = {
       'order_id': str(uuid.uuid4()),
       'product_id': product_id,
       'quantity': '2' # Redis stream values are strings
   }
   
   r.xadd('order_events', order)
   print(f"Published order event: {order}")
   

   Run this from a third terminal:

   
   
   
   # In terminal 3
   # Replace YOUR_PRODUCT_ID_HERE with an actual ID first!
   python test_publisher.py
   

   Immediately after running the publisher, look at the output in your consumer's terminal (Terminal 2). You should see it process the order, print "Stock reserved...", and update the quantity. If you run it again until stock runs out, you'll see the "Stock rejected..." message. You've just witnessed an event-driven workflow in action!

Conclusion

In this lesson, you successfully built the Inventory Service and implemented a complete event-driven loop. This is a significant step beyond the synchronous communication from the previous lesson and introduces a more scalable and resilient architectural style.

Key Takeaways:

• Event-Driven Architecture: You learned how services can be decoupled by communicating through a message broker (Redis Streams) instead of direct calls.
• Producer-Consumer Pattern: You implemented both a consumer (to process OrderCreated events) and a producer (to publish StockReserved/StockRejected events).
• Resilient Processing: You used Redis consumer groups and message acknowledgments (xack) to ensure reliable event processing, even in the face of failures.
• Saga Pattern in Practice: The logic of consuming an event, performing an action, and publishing a result event is a mini-saga, a foundational pattern for maintaining data consistency across microservices.

Preview of the Next Lesson

Our e-commerce system is taking shape. We have services for users, orders, and inventory. The next logical piece of the puzzle is payments. In the upcoming lesson, you will implement the Payment Service. It will listen for the StockReserved events you just created, process a mock payment, and publish its own events (PaymentSucceeded or PaymentFailed), continuing the event-driven saga and moving our orders closer to completion.