Versioning Microservices — Complete Guide
In this tutorial, you'll learn about Versioning Microservices. We cover key concepts, practical examples, and best practices to help you understand and apply this topic effectively.
Versioning microservices involves managing multiple versions of individual services that communicate via APIs, where each service can evolve independently while maintaining system-wide compatibility.
What You'll Learn
By the end of this lesson, you will implement consumer-driven contracts for microservices, route versioned requests through an API gateway, and manage independent service versioning.
Why It Matters
In a Microservices Architecture, different services may be at different versions simultaneously. Coordinating version upgrades without breaking consumers is a critical operational skill.
Real-World Use
Netflix uses API Versioning per service with consumer-driven contracts. Each team manages their service's version independently and provides a contract test suite for consumers.
Microservice Versioning Flow
flowchart LR
Client --> GW[API Gateway]
GW --> SV1[Service A v1]
GW --> SV2[Service A v2]
GW --> Other[Service B v1]
SV1 --> DB1[(DB v1)]
SV2 --> DB2[(DB v2)]
Consumer-Driven Contracts
# consumer_contract.py
from typing import Any, Callable, Dict, List, Optional
class ConsumerContract:
def __init__(self, name: str, version: str):
self.name = name
self.version = version
self.expectations: List[Dict] = []
def expect(self, method: str, path: str, status: int,
body_contains: Optional[List[str]] = None):
self.expectations.append({
"method": method,
"path": path,
"expected_status": status,
"body_contains": body_contains or [],
})
def validate(self, test_fn: Callable) -> List[str]:
failures = []
for exp in self.expectations:
try:
result = test_fn(exp["method"], exp["path"])
if result["status"] != exp["expected_status"]:
failures.append(
f"{exp['method']} {exp['path']}: expected {exp['expected_status']}, "
f"got {result['status']}")
body = result.get("body", {})
for field in exp["body_contains"]:
parts = field.split(".")
current = body
for p in parts:
if isinstance(current, dict):
current = current.get(p)
else:
current = None
break
if current is None:
failures.append(
f"{exp['method']} {exp['path']}: missing '{field}' in body")
except Exception as e:
failures.append(f"{exp['method']} {exp['path']}: error - {e}")
return failures
contract = ConsumerContract("user-service", "v2")
contract.expect("GET", "/users", 200, body_contains=["data", "data.0.id", "meta"])
def test_client(method: str, path: str) -> Dict:
responses = {
("GET", "/users"): {
"status": 200,
"body": {"data": [{"id": 1, "name": "Alice"}], "meta": {"count": 1}},
},
}
return responses.get((method, path), {"status": 404, "body": {}})
failures = contract.validate(test_client)
print(f"Contract validation failures: {len(failures)}")
for f in failures:
print(f" FAIL: {f}")
Expected output:
Contract validation failures: 0
API Gateway Version Routing
# gateway_version_routing.py
from typing import Any, Callable, Dict, Optional
class GatewayRouter:
def __init__(self):
self.services: Dict[str, Dict[int, Callable]] = {}
def register(self, service: str, version: int, handler: Callable):
if service not in self.services:
self.services[service] = {}
self.services[service][version] = handler
def route(self, service: str, version: int, **kwargs) -> Dict:
handlers = self.services.get(service, {})
handler = handlers.get(version)
if not handler:
fallback = handlers.get(max(h or 0 for h in handlers))
if fallback:
return {
"service": service,
"requested_version": version,
"served_version": max(handlers.keys()),
"data": fallback(**kwargs),
"warning": f"Version {version} not found, served latest",
}
return {"error": f"Service '{service}' not found"}
return {
"service": service,
"version": version,
"data": handler(**kwargs),
}
gw = GatewayRouter()
gw.register("users", 1, lambda: {"users": [{"id": 1, "name": "Alice"}]})
gw.register("users", 2, lambda: {"data": [{"id": 1, "name": "Alice"}], "meta": {"v": 2}})
print(gw.route("users", 2))
print(gw.route("users", 1))
print(gw.route("users", 3))
Expected output:
{'service': 'users', 'version': 2, 'data': {'data': [{'id': 1, 'name': 'Alice'}], 'meta': {'v': 2}}}
{'service': 'users', 'version': 1, 'data': {'users': [{'id': 1, 'name': 'Alice'}]}}
{'service': 'users', 'requested_version': 3, 'served_version': 2, ...}
Independent Service Version Management
# service_version_mgr.py
from typing import Dict, List, Optional
class ServiceVersionManager:
def __init__(self):
self.versions: Dict[str, List[int]] = {}
self.dependencies: Dict[str, List[str]] = {}
def register(self, service: str, versions: List[int],
depends_on: Optional[List[str]] = None):
self.versions[service] = versions
self.dependencies[service] = depends_on or []
def is_compatible(self, upgrades: Dict[str, int]) -> Dict[str, str]:
issues = {}
for svc, target_v in upgrades.items():
allowed = self.versions.get(svc, [])
if target_v not in allowed:
issues[svc] = f"Version {target_v} not available (options: {allowed})"
for dep in self.dependencies.get(svc, []):
dep_v = upgrades.get(dep)
if dep_v is None:
issues[svc] = f"Missing upgrade spec for dependency: {dep}"
return issues
def suggest_upgrade_order(self, services: List[str]) -> List[str]:
ordered = []
visited = set()
def visit(svc):
if svc in visited:
return
visited.add(svc)
for dep in self.dependencies.get(svc, []):
if dep in services:
visit(dep)
ordered.append(svc)
for svc in services:
visit(svc)
return ordered
mgr = ServiceVersionManager()
mgr.register("users", [1, 2, 3])
mgr.register("orders", [1, 2], depends_on=["users"])
mgr.register("notifications", [1], depends_on=["users", "orders"])
upgrades = {"users": 2, "orders": 2, "notifications": 1}
issues = mgr.is_compatible(upgrades)
print(f"Compatibility issues: {issues}")
order = mgr.suggest_upgrade_order(["notifications", "orders", "users"])
print(f"Upgrade order: {order}")
Expected output:
Compatibility issues: {}
Upgrade order: ['users', 'orders', 'notifications']
Common Mistakes
1. Synchronous Multi-Service Deployments
Deploying multiple services simultaneously increases risk. Use independent release cycles with backward compatibility.
2. Tight Coupling Between Service Versions
Service A v1 should not depend on Service B v3 specifically. Use API versioning and tolerate multiple backend versions.
3. No Consumer Contract Tests
Without contract tests, a service provider can break consumers without knowing. Automate contract validation in CI.
4. Ignoring Backward Compatibility in Internal APIs
Internal service APIs also need versioning. Not all services use public API versioning practices.
5. Monolithic Thinking in Versioning
Using a single API version across all services. Each service should independently track and communicate its version.
Practice Questions
1. What is consumer-driven Contract Testing?
A testing pattern where consumers define their expectations in contracts, and providers validate against these contracts to avoid breaking changes.
2. How does API gateway help with microservice versioning?
The gateway routes requests to the correct service version, enabling clients to use a single entry point while services evolve independently.
3. What is the recommended upgrade order for microservices?
Upgrade dependencies first. For example, if notification depends on users, upgrade users before notifications.
4. Why should microservices avoid shared database schemas?
Shared schemas create tight coupling. Each service should own its data and provide versioned APIs for access.
Challenge
Design a microservice versioning system for a three-service architecture (users, orders, inventory) with independent versioning, contract testing, and gateway routing.
FAQ
Mini Project: Multi-Service Version Orchestrator
# multi_orchestrator.py
from typing import Any, Callable, Dict, List
class ServiceNode:
def __init__(self, name: str):
self.name = name
self.services: Dict[int, Callable] = {}
def add(self, version: int, handler: Callable):
self.services[version] = handler
def call(self, version: int, **kwargs) -> Dict:
h = self.services.get(version)
if not h:
return {"error": f"{self.name} v{version} unavailable"}
return h(**kwargs)
class Orchestrator:
def __init__(self):
self.nodes: Dict[str, ServiceNode] = {}
def add_node(self, svc: ServiceNode): self.nodes[svc.name] = svc
def execute(self, workflow: List[Dict]) -> Dict:
context = {}
for step in workflow:
svc = self.nodes[step["service"]]
result = svc.call(step["version"], **{**context, **step.get("args", {})})
context.update(result)
return context
users = ServiceNode("users")
users.add(1, lambda: {"user_id": 1})
orders = ServiceNode("orders")
orders.add(1, lambda user_id: {"order_id": 100})
orch = Orchestrator()
orch.add_node(users)
orch.add_node(orders)
result = orch.execute([
{"service": "users", "version": 1},
{"service": "orders", "version": 1, "args": {"user_id": "$user_id"}},
])
print(f"Workflow result: {result}")
Expected output:
Workflow result: {'user_id': 1, 'order_id': 100}
What's Next
You understand microservice versioning. Next, learn about testing versioning strategies, then explore the versioning mini project.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro