"""
Pattern 277: Release Management
Purpose: Release coordination, orchestration, and automated rollback
"""
from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
class ReleaseStage(Enum):
"""Stage of release lifecycle."""
PLANNING = "planning"
TESTING = "testing"
STAGING = "staging"
DEPLOYMENT = "deployment"
VERIFICATION = "verification"
COMPLETED = "completed"
ROLLED_BACK = "rolled_back"
FAILED = "failed"
class ReleaseType(Enum):
"""Type of release."""
MAJOR = "major"
MINOR = "minor"
PATCH = "patch"
HOTFIX = "hotfix"
EMERGENCY = "emergency"
class ComponentStatus(Enum):
"""Status of release component."""
PENDING = "pending"
DEPLOYING = "deploying"
DEPLOYED = "deployed"
FAILED = "failed"
ROLLED_BACK = "rolled_back"
@dataclass
class ReleaseComponent:
"""Component in a release."""
name: str
version: str
status: ComponentStatus
deployment_order: int
dependencies: List[str] = field(default_factory=list)
deployed_at: Optional[datetime] = None
error_message: Optional[str] = None
@dataclass
class ReleaseCheckpoint:
"""Checkpoint during release."""
stage: ReleaseStage
timestamp: datetime
status: str
details: str
can_rollback: bool = True
@dataclass
class Release:
"""Represents a release."""
release_id: str
version: str
release_type: ReleaseType
stage: ReleaseStage
components: Dict[str, ReleaseComponent]
created_at: datetime = field(default_factory=datetime.now)
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
checkpoints: List[ReleaseCheckpoint] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
class ReleaseManagement:
"""
Implements release management pattern.
Features:
- Multi-component coordination
- Staged deployment process
- Dependency management
- Automated verification
- Rollback capability
- Release history tracking
- Checkpoint management
"""
def __init__(self):
self.releases: Dict[str, Release] = {}
self.active_release: Optional[str] = None
self.release_history: List[str] = []
# Statistics
self.statistics = {
"total_releases": 0,
"successful_releases": 0,
"failed_releases": 0,
"rolled_back_releases": 0,
"total_components_deployed": 0
}
def create_release(
self,
release_id: str,
version: str,
release_type: ReleaseType,
components: List[Dict[str, Any]],
metadata: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Create a new release.
Args:
release_id: Unique release identifier
version: Release version
release_type: Type of release
components: List of components with name, version, order, dependencies
metadata: Additional metadata
Returns:
Status of release creation
"""
if release_id in self.releases:
return {
"success": False,
"message": f"Release {release_id} already exists"
}
# Create components
release_components = {}
for comp in components:
component = ReleaseComponent(
name=comp["name"],
version=comp["version"],
status=ComponentStatus.PENDING,
deployment_order=comp.get("order", 0),
dependencies=comp.get("dependencies", [])
)
release_components[comp["name"]] = component
# Validate dependencies
for comp_name, component in release_components.items():
for dep in component.dependencies:
if dep not in release_components:
return {
"success": False,
"message": f"Component {comp_name} depends on {dep} which is not in release"
}
release = Release(
release_id=release_id,
version=version,
release_type=release_type,
stage=ReleaseStage.PLANNING,
components=release_components,
metadata=metadata or {}
)
self.releases[release_id] = release
self.statistics["total_releases"] += 1
# Add checkpoint
checkpoint = ReleaseCheckpoint(
stage=ReleaseStage.PLANNING,
timestamp=datetime.now(),
status="created",
details=f"Release {release_id} created with {len(components)} components"
)
release.checkpoints.append(checkpoint)
return {
"success": True,
"release_id": release_id,
"version": version,
"components": len(components),
"message": f"Release {release_id} created successfully"
}
def start_release(self, release_id: str) -> Dict[str, Any]:
"""
Start a release deployment.
Args:
release_id: Release identifier
Returns:
Status of release start
"""
if release_id not in self.releases:
return {
"success": False,
"message": f"Release {release_id} not found"
}
if self.active_release:
return {
"success": False,
"message": f"Another release {self.active_release} is already active"
}
release = self.releases[release_id]
if release.stage != ReleaseStage.PLANNING:
return {
"success": False,
"message": f"Release is in {release.stage.value} stage, cannot start"
}
self.active_release = release_id
release.started_at = datetime.now()
release.stage = ReleaseStage.TESTING
# Add checkpoint
checkpoint = ReleaseCheckpoint(
stage=ReleaseStage.TESTING,
timestamp=datetime.now(),
status="started",
details=f"Release {release_id} started"
)
release.checkpoints.append(checkpoint)
return {
"success": True,
"release_id": release_id,
"stage": release.stage.value,
"message": f"Release {release_id} started"
}
def advance_stage(self, release_id: str) -> Dict[str, Any]:
"""
Advance release to next stage.
Args:
release_id: Release identifier
Returns:
Status of stage advancement
"""
if release_id not in self.releases:
return {
"success": False,
"message": f"Release {release_id} not found"
}
release = self.releases[release_id]
current_stage = release.stage
# Define stage progression
stage_progression = {
ReleaseStage.TESTING: ReleaseStage.STAGING,
ReleaseStage.STAGING: ReleaseStage.DEPLOYMENT,
ReleaseStage.DEPLOYMENT: ReleaseStage.VERIFICATION,
ReleaseStage.VERIFICATION: ReleaseStage.COMPLETED
}
if current_stage not in stage_progression:
return {
"success": False,
"message": f"Cannot advance from {current_stage.value} stage"
}
next_stage = stage_progression[current_stage]
release.stage = next_stage
# Add checkpoint
checkpoint = ReleaseCheckpoint(
stage=next_stage,
timestamp=datetime.now(),
status="advanced",
details=f"Advanced from {current_stage.value} to {next_stage.value}"
)
release.checkpoints.append(checkpoint)
# If deployment stage, deploy components
if next_stage == ReleaseStage.DEPLOYMENT:
self._deploy_components(release)
# If completed, finalize
if next_stage == ReleaseStage.COMPLETED:
release.completed_at = datetime.now()
self.active_release = None
self.release_history.append(release_id)
self.statistics["successful_releases"] += 1
return {
"success": True,
"release_id": release_id,
"previous_stage": current_stage.value,
"current_stage": next_stage.value,
"message": f"Advanced to {next_stage.value} stage"
}
def _deploy_components(self, release: Release) -> None:
"""Deploy components in order respecting dependencies."""
# Sort components by deployment order
components_by_order = sorted(
release.components.items(),
key=lambda x: x[1].deployment_order
)
deployed = set()
for comp_name, component in components_by_order:
# Check dependencies
can_deploy = all(dep in deployed for dep in component.dependencies)
if can_deploy:
component.status = ComponentStatus.DEPLOYING
# Simulate deployment (90% success rate)
import random
if random.random() Dict[str, Any]:
"""
Rollback a release.
Args:
release_id: Release identifier
Returns:
Status of rollback
"""
if release_id not in self.releases:
return {
"success": False,
"message": f"Release {release_id} not found"
}
release = self.releases[release_id]
# Check if can rollback
if not release.checkpoints:
return {
"success": False,
"message": "No checkpoints available for rollback"
}
last_checkpoint = release.checkpoints[-1]
if not last_checkpoint.can_rollback:
return {
"success": False,
"message": f"Cannot rollback from {last_checkpoint.stage.value} stage"
}
# Rollback deployed components
rollback_count = 0
for component in release.components.values():
if component.status == ComponentStatus.DEPLOYED:
component.status = ComponentStatus.ROLLED_BACK
rollback_count += 1
release.stage = ReleaseStage.ROLLED_BACK
release.completed_at = datetime.now()
if self.active_release == release_id:
self.active_release = None
self.statistics["rolled_back_releases"] += 1
# Add checkpoint
checkpoint = ReleaseCheckpoint(
stage=ReleaseStage.ROLLED_BACK,
timestamp=datetime.now(),
status="rolled_back",
details=f"Release rolled back, {rollback_count} components reverted",
can_rollback=False
)
release.checkpoints.append(checkpoint)
return {
"success": True,
"release_id": release_id,
"components_rolled_back": rollback_count,
"message": f"Release {release_id} rolled back successfully"
}
def get_release_status(self, release_id: str) -> Dict[str, Any]:
"""Get status of a release."""
if release_id not in self.releases:
return {
"success": False,
"message": f"Release {release_id} not found"
}
release = self.releases[release_id]
# Component status summary
component_summary = {
"total": len(release.components),
"pending": sum(1 for c in release.components.values() if c.status == ComponentStatus.PENDING),
"deploying": sum(1 for c in release.components.values() if c.status == ComponentStatus.DEPLOYING),
"deployed": sum(1 for c in release.components.values() if c.status == ComponentStatus.DEPLOYED),
"failed": sum(1 for c in release.components.values() if c.status == ComponentStatus.FAILED),
}
components_list = []
for name, comp in sorted(release.components.items(), key=lambda x: x[1].deployment_order):
components_list.append({
"name": name,
"version": comp.version,
"status": comp.status.value,
"order": comp.deployment_order,
"deployed_at": comp.deployed_at.isoformat() if comp.deployed_at else None,
"error": comp.error_message
})
return {
"success": True,
"release_id": release_id,
"version": release.version,
"type": release.release_type.value,
"stage": release.stage.value,
"is_active": self.active_release == release_id,
"created_at": release.created_at.isoformat(),
"started_at": release.started_at.isoformat() if release.started_at else None,
"completed_at": release.completed_at.isoformat() if release.completed_at else None,
"components": components_list,
"component_summary": component_summary,
"checkpoints": len(release.checkpoints)
}
def list_releases(self) -> Dict[str, Any]:
"""List all releases."""
releases_list = []
for release_id, release in self.releases.items():
releases_list.append({
"release_id": release_id,
"version": release.version,
"type": release.release_type.value,
"stage": release.stage.value,
"components": len(release.components),
"is_active": self.active_release == release_id
})
return {
"total": len(releases_list),
"active_release": self.active_release,
"releases": releases_list
}
def get_statistics(self) -> Dict[str, Any]:
"""Get release management statistics."""
return {
"total_releases": self.statistics["total_releases"],
"successful_releases": self.statistics["successful_releases"],
"failed_releases": self.statistics["failed_releases"],
"rolled_back_releases": self.statistics["rolled_back_releases"],
"total_components_deployed": self.statistics["total_components_deployed"],
"active_release": self.active_release
}
class ReleaseManagementPattern:
"""
Coordinator for release management.
Manages release lifecycle with multi-component coordination.
"""
def __init__(self):
self.management = ReleaseManagement()
def create_release(self, release_id: str, version: str, release_type: ReleaseType, components: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Create a new release."""
return self.management.create_release(release_id, version, release_type, components)
def start_release(self, release_id: str) -> Dict[str, Any]:
"""Start a release."""
return self.management.start_release(release_id)
def advance_stage(self, release_id: str) -> Dict[str, Any]:
"""Advance release stage."""
return self.management.advance_stage(release_id)
def rollback_release(self, release_id: str) -> Dict[str, Any]:
"""Rollback a release."""
return self.management.rollback_release(release_id)
def get_release_status(self, release_id: str) -> Dict[str, Any]:
"""Get release status."""
return self.management.get_release_status(release_id)
def list_releases(self) -> Dict[str, Any]:
"""List releases."""
return self.management.list_releases()
def get_statistics(self) -> Dict[str, Any]:
"""Get statistics."""
return self.management.get_statistics()
# Example usage
if __name__ == "__main__":
print("=== Release Management Demo ===\n")
# Create release management
pattern = ReleaseManagementPattern()
print("1. Creating release...")
components = [
{"name": "database", "version": "2.0.0", "order": 1, "dependencies": []},
{"name": "backend", "version": "2.0.0", "order": 2, "dependencies": ["database"]},
{"name": "frontend", "version": "2.0.0", "order": 3, "dependencies": ["backend"]},
{"name": "api-gateway", "version": "2.0.0", "order": 4, "dependencies": ["backend"]}
]
result = pattern.create_release(
"release-2.0.0",
"2.0.0",
ReleaseType.MAJOR,
components
)
print(f" {result['message']}")
print(f" Components: {result['components']}")
# Start release
print("\n2. Starting release...")
result = pattern.start_release("release-2.0.0")
print(f" {result['message']}")
print(f" Current stage: {result['stage']}")
# Advance through stages
print("\n3. Advancing through stages...")
stages = ["staging", "deployment", "verification", "completed"]
for stage in stages:
result = pattern.advance_stage("release-2.0.0")
if result['success']:
print(f" ✓ Advanced to {result['current_stage']}")
else:
print(f" ✗ Failed: {result['message']}")
break
# Get release status
print("\n4. Release status:")
status = pattern.get_release_status("release-2.0.0")
if status['success']:
print(f" Release: {status['release_id']}")
print(f" Version: {status['version']}")
print(f" Stage: {status['stage']}")
print(f" Components deployed: {status['component_summary']['deployed']}/{status['component_summary']['total']}")
print(f"\n Components:")
for comp in status['components']:
print(f" - {comp['name']} v{comp['version']}: {comp['status']}")
# Statistics
print("\n5. Release statistics:")
stats = pattern.get_statistics()
print(f" Total releases: {stats['total_releases']}")
print(f" Successful: {stats['successful_releases']}")
print(f" Failed: {stats['failed_releases']}")
print(f" Rolled back: {stats['rolled_back_releases']}")
print(f" Total components deployed: {stats['total_components_deployed']}")
print("\n✓ Comprehensive release management with coordination complete!")
Agentic Architecture Pattern: Release Management
Pattern 277: 277 Release Management – Line-by-Line Explanation
Overview
Pattern 277: Release Management
Purpose: Release coordination, orchestration, and automated rollback
Code Structure Analysis
File Information
- Pattern Number: 277
- Pattern Name: 277 Release Management
- File Path:
patterns\deployment\277_release_management.py - Total Lines: 579
Detailed Code Explanation
Module Docstring
"""
Pattern 277: Release Management
Purpose: Release coordination, orchestration, and automated rollback
"""
Explanation: This module docstring describes the purpose and key features of the 277 Release Management pattern.
Imports Section
from typing import Dict, Any, List, Optional, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
import random # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional, Set: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime: Date and time utilities
– from enum import Enum: Enumeration support for fixed sets of values
Enumerations
ReleaseStage (Line 14)
class ReleaseStage(Enum):
PLANNING
TESTING
STAGING
DEPLOYMENT
VERIFICATION
COMPLETED
ROLLED_BACK
FAILED
Purpose: Stage of release lifecycle.
Values:
– PLANNING: Represents planning
– TESTING: Represents testing
– STAGING: Represents staging
– DEPLOYMENT: Represents deployment
– VERIFICATION: Represents verification
– COMPLETED: Successfully finished
– ROLLED_BACK: Represents rolled back
– FAILED: Execution failed
ReleaseType (Line 26)
class ReleaseType(Enum):
MAJOR
MINOR
PATCH
HOTFIX
EMERGENCY
Purpose: Type of release.
Values:
– MAJOR: Represents major
– MINOR: Represents minor
– PATCH: Represents patch
– HOTFIX: Represents hotfix
– EMERGENCY: Represents emergency
ComponentStatus (Line 35)
class ComponentStatus(Enum):
PENDING
DEPLOYING
DEPLOYED
FAILED
ROLLED_BACK
Purpose: Status of release component.
Values:
– PENDING: Initial state, not yet processed
– DEPLOYING: Represents deploying
– DEPLOYED: Represents deployed
– FAILED: Execution failed
– ROLLED_BACK: Represents rolled back
Classes
ReleaseStage (Line 14)
Description: Stage of release lifecycle.
ReleaseType (Line 26)
Description: Type of release.
ComponentStatus (Line 35)
Description: Status of release component.
ReleaseComponent (Line 45)
Description: Component in a release.
ReleaseCheckpoint (Line 57)
Description: Checkpoint during release.
Release (Line 67)
Description: Represents a release.
ReleaseManagement (Line 81)
Description: Implements release management pattern.
Features: – Multi-component coordination – Staged deployment process – Dependency management – Automated verification – Rollback capability – Release history tracking – Checkpoint management
Methods:
__init__ (Line 95)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
create_release (Line 109)
def create_release(self, release_id, version, release_type, components, metadata):
"""Create a new release.
Args:
release_id: Unique release identifier
version: Release version
release_type: Type of release
components: List of components with name, version, order, dependencies
metadata: Additional metadata
Returns:
Status of release creation"""
Purpose: Create a new release.
Args: release_id: Unique release identifier version: Release version release_type: Type of release components: List of components with name, version, order, dependencies metadata: Additional metadata
Returns: Status of release creation
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
– version: Parameter for version
– release_type: Parameter for release_type
– components: Parameter for components
– metadata: Parameter for metadata
start_release (Line 186)
def start_release(self, release_id):
"""Start a release deployment.
Args:
release_id: Release identifier
Returns:
Status of release start"""
Purpose: Start a release deployment.
Args: release_id: Release identifier
Returns: Status of release start
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
advance_stage (Line 236)
def advance_stage(self, release_id):
"""Advance release to next stage.
Args:
release_id: Release identifier
Returns:
Status of stage advancement"""
Purpose: Advance release to next stage.
Args: release_id: Release identifier
Returns: Status of stage advancement
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
_deploy_components (Line 300)
def _deploy_components(self, release):
"""Deploy components in order respecting dependencies."""
Purpose: Deploy components in order respecting dependencies.
Parameters:
– self: Instance reference
– release: Parameter for release
rollback_release (Line 337)
def rollback_release(self, release_id):
"""Rollback a release.
Args:
release_id: Release identifier
Returns:
Status of rollback"""
Purpose: Rollback a release.
Args: release_id: Release identifier
Returns: Status of rollback
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
get_release_status (Line 401)
def get_release_status(self, release_id):
"""Get status of a release."""
Purpose: Get status of a release.
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
list_releases (Line 446)
def list_releases(self):
"""List all releases."""
Purpose: List all releases.
Parameters:
– self: Instance reference
get_statistics (Line 465)
def get_statistics(self):
"""Get release management statistics."""
Purpose: Get release management statistics.
Parameters:
– self: Instance reference
ReleaseManagementPattern (Line 477)
Description: Coordinator for release management.
Manages release lifecycle with multi-component coordination.
Methods:
__init__ (Line 484)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
create_release (Line 487)
def create_release(self, release_id, version, release_type, components):
"""Create a new release."""
Purpose: Create a new release.
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
– version: Parameter for version
– release_type: Parameter for release_type
– components: Parameter for components
start_release (Line 491)
def start_release(self, release_id):
"""Start a release."""
Purpose: Start a release.
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
advance_stage (Line 495)
def advance_stage(self, release_id):
"""Advance release stage."""
Purpose: Advance release stage.
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
rollback_release (Line 499)
def rollback_release(self, release_id):
"""Rollback a release."""
Purpose: Rollback a release.
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
get_release_status (Line 503)
def get_release_status(self, release_id):
"""Get release status."""
Purpose: Get release status.
Parameters:
– self: Instance reference
– release_id: Parameter for release_id
list_releases (Line 507)
def list_releases(self):
"""List releases."""
Purpose: List releases.
Parameters:
– self: Instance reference
get_statistics (Line 511)
def get_statistics(self):
"""Get statistics."""
Purpose: Get statistics.
Parameters:
– self: Instance reference
Design Patterns Used
- Strategy Pattern: Different execution strategies for various scenarios
- State Pattern: Managing different states in the workflow
- Chain of Responsibility: Sequential processing through workflow nodes
- Observer Pattern: Tracking and monitoring pattern execution
- Factory Pattern: Creating instances with proper configuration
Usage Example
# Initialize the pattern
pattern = ReleaseStage(name="example", model_name="gpt-4")
# Configure and execute
result = pattern.execute(data={"key": "value"})
# Check results
print(result)
Key Concepts
- LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
- Workflow Orchestration: LangGraph manages execution flow through nodes
- State Management: TypedDict defines workflow state structure
- Error Handling: Comprehensive error handling and fallback mechanisms
- Extensibility: Designed for easy customization and extension
Extension Points
This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities