Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

15 NODES
πŸ“textβ€’6 months ago

ARCH-AEP Tiered Remediation Cycle

Run a scope-locked remediation cycle that normalizes findings, revalidates them in parallel, and clears tiers with verification evidence.

architecture
⭐1
# ARCH-AEP Tiered Remediation Cycle Imported from curated first-party documentation sources. ## What this covers Use this workflow when you need an orchestrated engineering review cycle with strict scope lock, tiered execution, and audit-grade evidence. ## Use this when - Large remediation efforts spanning multiple PRs - Parallel specialist review with shared backlog authority - Tracking verification evidence through closure ## Expected outcomes - Scope changes are controlled instead of creeping mid-cycle - Severity tiers drive remediation order predictably - Verification evidence is treated as part of done ## Source synthesis - CHELATEDAI/docs/ARCH AGENTIC ENGINEERING AND PLANNING/workflow.md (https://github.com/mattmre/CHELATEDAI/blob/main/docs/ARCH%20AGENTIC%20ENGINEERING%20AND%20PLANNING/workflow.md) - CHELATEDAI/docs/ARCH AGENTIC ENGINEERING AND PLANNING/orchestrator-briefing.md (https://github.com/mattmre/CHELATEDAI/blob/main/docs/ARCH%20AGENTIC%20ENGINEERING%20AND%20PLANNING/orchestrator-briefing.md) ## Dedupe notes Condenses the core ARCH-AEP workflow and briefing docs into a single remediation-cycle entry instead of importing both separately. ## Source excerpts ### CHELATEDAI/docs/ARCH AGENTIC ENGINEERING AND PLANNING/workflow.md ## Workflow (ARCH phase) 1. Scope lock - Define PR range and time window. - Freeze inputs (refinement report + framework + PR list). 2. Discovery + normalization - Orchestrator ingests `agentic-review-framework.md` and the latest refinement report. - Normalize all findings into a single backlog with unique IDs. - De-duplicate, merge overlaps, and assign provisional severity. - Record backlog in `docs/ARCH AGENTIC ENGINEERING AND PLANNING/backlog-YYYY-MM-DD.md`. 3. Parallel re-validation - Spawn specialist agents to re-validate findings against current main. - Each agent must attach exact file paths and line ranges. - Propose the smallest safe fix, acceptance criteria, and effort sizing (S/M/L). 4. Architecture and planning synthesis - Orchestrator merges validated findings into a master backlog. - Re-score and re-rank using the sorting model. - Identify dependency chains and blockers. ### CHELATEDAI/docs/ARCH AGENTIC ENGINEERING AND PLANNING/orchestrator-briefing.md Purpose: Quick reference for the ARCH-AEP documentation set, plus the narrative to start a new session. Full file index: `docs/INDEX.md` ## What Exists In This Folder - `README.md`: narrative overview and intent for ARCH-AEP. - `workflow.md`: end-to-end workflow specification with phases, guardrails, and enhancements. - `templates.md`: ID and branch conventions, tracker table format, and status log. - `backlog-template.md`: template for the master backlog file. - `backlog-index.md`: index of backlog files. - `next-session.md`: short handoff checklist for resuming work. - `phase-planning.md`: long-running planning record for the current cycle. - `schedule-and-tracking.md`: cadence, gates, and milestone tracking. - `agent-learning.md`: cross-session learnings and reusable patterns. - `change-log.md`: scope/defer decisions with audit context. - `risk-memo-template.md`: Critical/High risk memo template. - `risk-memos/README.md`: storage location for risk memos. - `phase-summary-template.md`: template for per-PR or per-phase summaries. - `phase-summaries/README.md`: storage location for phase summaries. - `verification-log.md`: test/build evidence mirror for PRs. - `tracker-pointer.md`: pointer to the active tracker file. - `tracker-index.md`: index of tracker files. - `scope-lock-t ...
πŸ‘0
πŸ‘οΈ0
docs
πŸ“textβ€’6 months ago

PR Merge Boundary Validation Runbook

Validate chained pull requests at each merge boundary with explicit checks, stale-approval handling, and rollback awareness.

devops
⭐1
# PR Merge Boundary Validation Runbook Imported from curated first-party documentation sources. ## What this covers Use this runbook when a merge sequence spans dependent pull requests and each boundary needs fresh validation. ## Use this when - Managing dependency-chained pull requests - Refreshing stale approvals near merge time - Reducing merge surprises through checkpoint validation ## Expected outcomes - Each merge boundary has a concrete verification step - Review freshness stays visible as branches evolve - Rollback planning is captured before the merge happens ## Source synthesis - EVOKORE-MCP/docs/PR_MERGE_RUNBOOK.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/PR_MERGE_RUNBOOK.md) ## Dedupe notes Keeps the merge-boundary validation concept separate from more general workflow or orchestration docs. ## Source excerpts ### EVOKORE-MCP/docs/PR_MERGE_RUNBOOK.md Operator runbook for reliable merges and context-rot prevention. ## Pre-merge Checklist - [ ] PR scope matches approved plan - [ ] PR description is filled using `.github/PULL_REQUEST_TEMPLATE.md` - [ ] PR metadata automation check (`scripts/validate-pr-metadata.js`) is passing for pull_request CI runs - [ ] Required tests pass locally/CI - [ ] Docs updated for user-facing behavior changes - [ ] Release-impacting changes called out - [ ] Follow-up issues captured (if any) ## Required Checks by Change Type Use this as the minimum check set before approval and merge: | Change type | Required checks | | --- | --- | | Docs-only changes | `node test-docs-canonical-links.js` | | Ops/docs process changes (`docs/PR_MERGE_RUNBOOK.md`, `next-session.md`, orchestration docs) | `node test-ops-docs-validation.js` and `node test-docs-canonical-links.js` | | Source/tooling/config changes (`src/`, `scripts/`, workflow/config files) | Relevant targeted tests for touched area plus CI-required suite | | Release-flow changes | `node test-npm-release-flow-validation.js` plus docs/link checks | If a PR spans multiple change types, run the union of required checks. ## Reviewer Responsibilities - Confirm PR scope and dependency assumptions are explicit in description. - Confirm PR metadata fields from `.github/PULL_REQUEST_TEMPLATE.md` are complete. - Verify required checks for each change type are attached in PR evidence. - Block approval if dependency base PR is not merged or branch is stale. - Approve only the current chain head; do not pre-approve non-head dependent PRs. - Confirm all review conversations are resolved before final approval. - Ensure merge strategy and rollback notes are documented for risky changes. ## Merge Steps 1. Rebase or update branch with latest target branch. 2. Re-run required validations. 3. Confirm reviewer approvals and resolved conversations. 4. Merge via approved strategy. 5. Record merge commit/PR number in tracker. ## Merge-boundary Checkpoints At every dependency merge boundary (`base -> dependent`): 1. Rebase dependent PR branch on latest `main` immediately after parent merge. 2. Re-run required checks and attach updated evidence in PR metadata. 3. Revalidate approvals for the new head state (stale approvals must be refreshed). 4. Confirm merge-boundary revalidation notes are updated before merge. ## Merge-order Controls (Dependency Chain) 1. Define merge order explicitly in PR descriptions (`base -> dependent`). 2. Merge only the current chain head; hold dependents until parent merge is complete. 3. After each merge, rebase ...
πŸ‘0
πŸ‘οΈ0
docs
πŸ€–system promptβ€’6 months ago

Human-in-the-Loop Approval Token Workflow

Gate risky tool execution behind approval tokens so agents can retry safely after a human reviewer signs off.

security
⭐1
# Human-in-the-Loop Approval Token Workflow Imported from curated first-party documentation sources. ## What this covers Use this workflow when autonomous execution needs a durable handoff from human review back into the agent loop. ## Use this when - Retrying a blocked tool call after approval - Adding a human checkpoint to sensitive actions - Reducing insecure workarounds around approval flows ## Expected outcomes - Approval becomes a reusable tokenized workflow - Agents resume work without losing execution context - Security controls stay explicit instead of being implied ## Source synthesis - EVOKORE-MCP/docs/V2_MULTI_AGENT_WORKFLOWS.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/V2_MULTI_AGENT_WORKFLOWS.md) - EVOKORE-MCP/docs/AGENT33_IMPROVEMENT_INSTRUCTIONS.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/AGENT33_IMPROVEMENT_INSTRUCTIONS.md) ## Dedupe notes Synthesizes the approval-token concept from the main workflow doc and the Agent33 improvement transfer notes. ## Source excerpts ### EVOKORE-MCP/docs/V2_MULTI_AGENT_WORKFLOWS.md With EVOKORE-MCP v2.0 fully operational, we can now leverage 40+ proxied GitHub and Filesystem tools seamlessly within complex multi-agent workflows. The core architectural advancements include: ## 1. Dynamic Tool Prefixing & Indexing Instead of loading 40+ tools into an LLM's context window statically (which causes massive bloat), EVOKORE's `ProxyManager` boots child servers (like `@modelcontextprotocol/server-github` and `@modelcontextprotocol/server-filesystem`) and dynamically prefixes their tools (`github_create_issue`, `fs_write_file`). This prevents namespace collisions while keeping the tools accessible to native skills. ## 2. Human-in-the-Loop (HITL) Security Interceptor Automated multi-agent workflows involving sensitive endpoints (like GitHub write access or file deletion) are governed by EVOKORE's stateless `_evokore_approval_token` architecture. - When an agent attempts a restricted action, the tool call is intercepted and blocked. - The server returns an error explicitly commanding the agent to prompt the human for approval. - Upon approval, the agent retries the exact tool call with the injected token, securely fulfilling the workflow without severing the conversational context. ## 3. Active Skill Orchestration (Native Harnessing) Unlike v1.0 where skills merel ... ### EVOKORE-MCP/docs/AGENT33_IMPROVEMENT_INSTRUCTIONS.md > **Purpose**: Feed this file into Claude Code CLI when working on the Agent33 repo. It contains patterns, architectures, and capabilities proven in EVOKORE-MCP that Agent33 should adopt. --- ## 1. Multi-Server MCP Aggregation Pattern **What Agent33 lacks**: Agent33's MCP server (Phase 43) is a single-endpoint bridge. It doesn't aggregate multiple child MCP servers behind a unified namespace. **What to build**: A proxy layer that spawns and manages multiple child MCP servers from a single config file, presenting them as one unified tool surface. ### Implementation spec: ``` mcp.config.json { "servers": { "github": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-github"], "env": { "GITHUB_TOKEN": "${GITHUB_TOKEN}" } }, "fs": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "./"] }, "elevenlabs": { "command": "uvx", "args": ["elevenlabs-mcp"], "env": { "ELEVENLABS_API_KEY": "${ELEVENLABS_API_KEY}" } } } } ``` **Key patterns from EVOKORE**: - **Tool name prefixing**: Every proxied tool gets renamed `{serverId}_{originalName}` to prevent namespace collisions (e.g., `github_create_issue`, `fs_read_file`). First-registration-wins for duplicates. - **Environment interpolation**: `${VAR}` syntax in `env` blocks resolved ...
πŸ‘0
πŸ‘οΈ0
docs
πŸ€–system promptβ€’7 months ago

saga-orchestration

Implement saga patterns for distributed transactions and

coding
⭐1
# Saga Orchestration Patterns for managing distributed transactions and long-running business processes. ## When to Use This Skill - Coordinating multi-service transactions - Implementing compensating transactions - Managing long-running business workflows - Handling failures in distributed systems - Building order fulfillment processes - Implementing approval workflows ## Core Concepts ### 1. Saga Types ``` Choreography Orchestration β”Œβ”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚Svc A│─►│Svc B│─►│Svc Cβ”‚ β”‚ Orchestratorβ”‚ β””β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”‚ β”‚ β”‚ β–Ό β–Ό β–Ό β”Œβ”€β”€β”€β”€β”€β”Όβ”€β”€β”€β”€β”€β” Event Event Event β–Ό β–Ό β–Ό β”Œβ”€β”€β”€β”€β”β”Œβ”€β”€β”€β”€β”β”Œβ”€β”€β”€β”€β” β”‚Svc1β”‚β”‚Svc2β”‚β”‚Svc3β”‚ β””β”€β”€β”€β”€β”˜β””β”€β”€β”€β”€β”˜β””β”€β”€β”€β”€β”˜ ``` ### 2. Saga Execution States | State | Description | | ---------------- | ------------------------------ | | **Started** | Saga initiated | | **Pending** | Waiting for step completion | | **Compensating** | Rolling back due to failure | | **Completed** | All steps succeeded | | **Failed** | Saga failed after compensation | ## Templates ### Template 1: Saga Orchestrator Base ```python from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import Enum from typing import List, Dict, Any, Optional from datetime import datetime import uuid class SagaState(Enum): STARTED = "started" PENDING = "pending" COMPENSATING = "compensating" COMPLETED = "completed" FAILED = "failed" @dataclass class SagaStep: name: str action: str compensation: str status: str = "pending" result: Optional[Dict] = None error: Optional[str] = None executed_at: Optional[datetime] = None compensated_at: Optional[datetime] = None @dataclass class Saga: saga_id: str saga_type: str state: SagaState data: Dict[str, Any] steps: List[SagaStep] current_step: int = 0 created_at: datetime = field(default_factory=datetime.utcnow) updated_at: datetime = field(default_factory=datetime.utcnow) class SagaOrchestrator(ABC): """Base class for saga orchestrators.""" def __init__(self, saga_store, event_publisher): self.saga_store = saga_store self.event_publisher = event_publisher @abstractmethod def define_steps(self, data: Dict) -> List[SagaStep]: """Define the saga steps.""" pass @property @abstractmethod def saga_type(self) -> str: """Unique saga type identifier.""" pass async def start(self, data: Dict) -> Saga: """Start a new saga.""" saga = Saga( saga_id=str(uuid.uuid4()), saga_type=self.saga_type, state=SagaState.STARTED, data=data, steps=self.define_steps(data) ) await self.saga_store.save(saga) await self._execute_next_step(saga) return saga async def handle_step_completed(self, saga_id: str, step_name: str, result: Dict): """Handle successful step completion.""" saga = await self.saga_store.get(saga_id) # Update step for step in saga.steps: if step.name == step_name: step.status = "completed" step.result = result step.executed_at = datetime.utcnow() break saga.current_step += 1 saga.updated_at = datetime.utcnow() # Check if saga is complete if saga.current_step >= len(saga.steps): saga.state = SagaState.COMPLETED await self.saga_store.save(saga) await self._on_saga_completed(saga) else: saga.state = SagaState.PENDING await self.saga_store.save(saga) await self._execute_next_step(saga) async def handle_step_failed(self, saga_id: str, step_name: str, error: str): """Handle step failure - start compensation.""" saga = await self.saga_store.get(saga_id) # Mark step as failed for step in saga.steps: if step.name == step_name: step.status = "failed" step.error = error break saga.state = SagaState.COMPENSATING saga.updated_at = datetime.utcnow() await self.saga_store.save(saga) # Start compensation from current step backwards await self._compensate(saga) async def _execute_next_step(self, saga: Saga): """Execute the next step in the saga.""" if saga.current_step >= len(saga.steps): return step = saga.steps[saga.current_step] step.status = "executing" await self.saga_store.save(saga) # Publish command to execute step await self.event_publisher.publish( step.action, { "saga_id": saga.saga_id, "step_name": step.name, **saga.data } ) async def _compensate(self, saga: Saga): """Execute compensation for completed steps.""" # Compensate in reverse order for i in range(saga.current_step - 1, -1, -1): step = saga.steps[i] if step.status == "completed": step.status = "compensating" await self.saga_store.save(saga) await self.event_publisher.publish( step.compensation, { "saga_id": saga.saga_id, "step_name": step.name, "original_result": step.result, **saga.data } ) async def handle_compensation_completed(self, saga_id: str, step_name: str): """Handle compensation completion.""" saga = await self.saga_store.get(saga_id) for step in saga.steps: if step.name == step_name: step.status = "compensated" step.compensated_at = datetime.utcnow() break # Check if all compensations complete all_compensated = all( s.status in ("compensated", "pending", "failed") for s in saga.steps ) if all_compensated: saga.state = SagaState.FAILED await self._on_saga_failed(saga) await self.saga_store.save(saga) async def _on_saga_completed(self, saga: Saga): """Called when saga completes successfully.""" await self.event_publisher.publish( f"{self.saga_type}Completed", {"saga_id": saga.saga_id, **saga.data} ) async def _on_saga_failed(self, saga: Saga): """Called when saga fails after compensation.""" await self.event_publisher.publish( f"{self.saga_type}Failed", {"saga_id": saga.saga_id, "error": "Saga failed", **saga.data} ) ``` ### Template 2: Order Fulfillment Saga ```python class OrderFulfillmentSaga(SagaOrchestrator): """Orchestrates order fulfillment across services.""" @property def saga_type(self) -> str: return "OrderFulfillment" def define_steps(self, data: Dict) -> List[SagaStep]: return [ SagaStep( name="reserve_inventory", action="InventoryService.ReserveItems", compensation="InventoryService.ReleaseReservation" ), SagaStep( name="process_payment", action="PaymentService.ProcessPayment", compensation="PaymentService.RefundPayment" ), SagaStep( name="create_shipment", action="ShippingService.CreateShipment", compensation="ShippingService.CancelShipment" ), SagaStep( name="send_confirmation", action="NotificationService.SendOrderConfirmation", compensation="NotificationService.SendCancellationNotice" ) ] # Usage async def create_order(order_data: Dict): saga = OrderFulfillmentSaga(saga_store, event_publisher) return await saga.start({ "order_id": order_data["order_id"], "customer_id": order_data["customer_id"], "items": order_data["items"], "payment_method": order_data["payment_method"], "shipping_address": order_data["shipping_address"] }) # Event handlers in each service class InventoryService: async def handle_reserve_items(self, command: Dict): try: # Reserve inventory reservation = await self.reserve( command["items"], command["order_id"] ) # Report success await self.event_publisher.publish( "SagaStepCompleted", { "saga_id": command["saga_id"], "step_name": "reserve_inventory", "result": {"reservation_id": reservation.id} } ) except InsufficientInventoryError as e: await self.event_publisher.publish( "SagaStepFailed", { "saga_id": command["saga_id"], "step_name": "reserve_inventory", "error": str(e) } ) async def handle_release_reservation(self, command: Dict): # Compensating action await self.release_reservation( command["original_result"]["reservation_id"] ) await self.event_publisher.publish( "SagaCompensationCompleted", { "saga_id": command["saga_id"], "step_name": "reserve_inventory" } ) ``` ### Template 3: Choreography-Based Saga ```python from dataclasses import dataclass from typing import Dict, Any import asyncio @dataclass class SagaContext: """Passed through choreographed saga events.""" saga_id: str step: int data: Dict[str, Any] completed_steps: list class OrderChoreographySaga: """Choreography-based saga using events.""" def __init__(self, event_bus): self.event_bus = event_bus self._register_handlers() def _register_handlers(self): self.event_bus.subscribe("OrderCreated", self._on_order_created) self.event_bus.subscribe("InventoryReserved", self._on_inventory_reserved) self.event_bus.subscribe("PaymentProcessed", self._on_payment_processed) self.event_bus.subscribe("ShipmentCreated", self._on_shipment_created) # Compensation handlers self.event_bus.subscribe("PaymentFailed", self._on_payment_failed) self.event_bus.subscribe("ShipmentFailed", self._on_shipment_failed) async def _on_order_created(self, event: Dict): """Step 1: Order created, reserve inventory.""" await self.event_bus.publish("ReserveInventory", { "saga_id": event["order_id"], "order_id": event["order_id"], "items": event["items"] }) async def _on_inventory_reserved(self, event: Dict): """Step 2: Inventory reserved, process payment.""" await self.event_bus.publish("ProcessPayment", { "saga_id": event["saga_id"], "order_id": event["order_id"], "amount": event["total_amount"], "reservation_id": event["reservation_id"] }) async def _on_payment_processed(self, event: Dict): """Step 3: Payment done, create shipment.""" await self.event_bus.publish("CreateShipment", { "saga_id": event["saga_id"], "order_id": event["order_id"], "payment_id": event["payment_id"] }) async def _on_shipment_created(self, event: Dict): """Step 4: Complete - send confirmation.""" await self.event_bus.publish("OrderFulfilled", { "saga_id": event["saga_id"], "order_id": event["order_id"], "tracking_number": event["tracking_number"] }) # Compensation handlers async def _on_payment_failed(self, event: Dict): """Payment failed - release inventory.""" await self.event_bus.publish("ReleaseInventory", { "saga_id": event["saga_id"], "reservation_id": event["reservation_id"] }) await self.event_bus.publish("OrderFailed", { "order_id": event["order_id"], "reason": "Payment failed" }) async def _on_shipment_failed(self, event: Dict): """Shipment failed - refund payment and release inventory.""" await self.event_bus.publish("RefundPayment", { "saga_id": event["saga_id"], "payment_id": event["payment_id"] }) await self.event_bus.publish("ReleaseInventory", { "saga_id": event["saga_id"], "reservation_id": event["reservation_id"] }) ``` ### Template 4: Saga with Timeouts ```python class TimeoutSagaOrchestrator(SagaOrchestrator): """Saga orchestrator with step timeouts.""" def __init__(self, saga_store, event_publisher, scheduler): super().__init__(saga_store, event_publisher) self.scheduler = scheduler async def _execute_next_step(self, saga: Saga): if saga.current_step >= len(saga.steps): return step = saga.steps[saga.current_step] step.status = "executing" step.timeout_at = datetime.utcnow() + timedelta(minutes=5) await self.saga_store.save(saga) # Schedule timeout check await self.scheduler.schedule( f"saga_timeout_{saga.saga_id}_{step.name}", self._check_timeout, {"saga_id": saga.saga_id, "step_name": step.name}, run_at=step.timeout_at ) await self.event_publisher.publish( step.action, {"saga_id": saga.saga_id, "step_name": step.name, **saga.data} ) async def _check_timeout(self, data: Dict): """Check if step has timed out.""" saga = await self.saga_store.get(data["saga_id"]) step = next(s for s in saga.steps if s.name == data["step_name"]) if step.status == "executing": # Step timed out - fail it await self.handle_step_failed( data["saga_id"], data["step_name"], "Step timed out" ) ``` ## Best Practices ### Do's - **Make steps idempotent** - Safe to retry - **Design compensations carefully** - They must work - **Use correlation IDs** - For tracing across services - **Implement timeouts** - Don't wait forever - **Log everything** - For debugging failures ### Don'ts - **Don't assume instant completion** - Sagas take time - **Don't skip compensation testing** - Most critical part - **Don't couple services** - Use async messaging - **Don't ignore partial failures** - Handle gracefully ## Resources - [Saga Pattern](https://microservices.io/patterns/data/saga.html) - [Designing Data-Intensive Applications](https://dataintensive.net/)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

workflow-orchestration-patterns

Design durable workflows with Temporal for distributed systems.

coding
⭐1
# Workflow Orchestration Patterns Master workflow orchestration architecture with Temporal, covering fundamental design decisions, resilience patterns, and best practices for building reliable distributed systems. ## When to Use Workflow Orchestration ### Ideal Use Cases (Source: docs.temporal.io) - **Multi-step processes** spanning machines/services/databases - **Distributed transactions** requiring all-or-nothing semantics - **Long-running workflows** (hours to years) with automatic state persistence - **Failure recovery** that must resume from last successful step - **Business processes**: bookings, orders, campaigns, approvals - **Entity lifecycle management**: inventory tracking, account management, cart workflows - **Infrastructure automation**: CI/CD pipelines, provisioning, deployments - **Human-in-the-loop** systems requiring timeouts and escalations ### When NOT to Use - Simple CRUD operations (use direct API calls) - Pure data processing pipelines (use Airflow, batch processing) - Stateless request/response (use standard APIs) - Real-time streaming (use Kafka, event processors) ## Critical Design Decision: Workflows vs Activities **The Fundamental Rule** (Source: temporal.io/blog/workflow-engine-principles): - **Workflows** = Orchestration logic and decision-making - **Activities** = External interactions (APIs, databases, network calls) ### Workflows (Orchestration) **Characteristics:** - Contain business logic and coordination - **MUST be deterministic** (same inputs β†’ same outputs) - **Cannot** perform direct external calls - State automatically preserved across failures - Can run for years despite infrastructure failures **Example workflow tasks:** - Decide which steps to execute - Handle compensation logic - Manage timeouts and retries - Coordinate child workflows ### Activities (External Interactions) **Characteristics:** - Handle all external system interactions - Can be non-deterministic (API calls, DB writes) - Include built-in timeouts and retry logic - **Must be idempotent** (calling N times = calling once) - Short-lived (seconds to minutes typically) **Example activity tasks:** - Call payment gateway API - Write to database - Send emails or notifications - Query external services ### Design Decision Framework ``` Does it touch external systems? β†’ Activity Is it orchestration/decision logic? β†’ Workflow ``` ## Core Workflow Patterns ### 1. Saga Pattern with Compensation **Purpose**: Implement distributed transactions with rollback capability **Pattern** (Source: temporal.io/blog/compensating-actions-part-of-a-complete-breakfast-with-sagas): ``` For each step: 1. Register compensation BEFORE executing 2. Execute the step (via activity) 3. On failure, run all compensations in reverse order (LIFO) ``` **Example: Payment Workflow** 1. Reserve inventory (compensation: release inventory) 2. Charge payment (compensation: refund payment) 3. Fulfill order (compensation: cancel fulfillment) **Critical Requirements:** - Compensations must be idempotent - Register compensation BEFORE executing step - Run compensations in reverse order - Handle partial failures gracefully ### 2. Entity Workflows (Actor Model) **Purpose**: Long-lived workflow representing single entity instance **Pattern** (Source: docs.temporal.io/evaluate/use-cases-design-patterns): - One workflow execution = one entity (cart, account, inventory item) - Workflow persists for entity lifetime - Receives signals for state changes - Supports queries for current state **Example Use Cases:** - Shopping cart (add items, checkout, expiration) - Bank account (deposits, withdrawals, balance checks) - Product inventory (stock updates, reservations) **Benefits:** - Encapsulates entity behavior - Guarantees consistency per entity - Natural event sourcing ### 3. Fan-Out/Fan-In (Parallel Execution) **Purpose**: Execute multiple tasks in parallel, aggregate results **Pattern:** - Spawn child workflows or parallel activities - Wait for all to complete - Aggregate results - Handle partial failures **Scaling Rule** (Source: temporal.io/blog/workflow-engine-principles): - Don't scale individual workflows - For 1M tasks: spawn 1K child workflows Γ— 1K tasks each - Keep each workflow bounded ### 4. Async Callback Pattern **Purpose**: Wait for external event or human approval **Pattern:** - Workflow sends request and waits for signal - External system processes asynchronously - Sends signal to resume workflow - Workflow continues with response **Use Cases:** - Human approval workflows - Webhook callbacks - Long-running external processes ## State Management and Determinism ### Automatic State Preservation **How Temporal Works** (Source: docs.temporal.io/workflows): - Complete program state preserved automatically - Event History records every command and event - Seamless recovery from crashes - Applications restore pre-failure state ### Determinism Constraints **Workflows Execute as State Machines**: - Replay behavior must be consistent - Same inputs β†’ identical outputs every time **Prohibited in Workflows** (Source: docs.temporal.io/workflows): - ❌ Threading, locks, synchronization primitives - ❌ Random number generation (`random()`) - ❌ Global state or static variables - ❌ System time (`datetime.now()`) - ❌ Direct file I/O or network calls - ❌ Non-deterministic libraries **Allowed in Workflows**: - βœ… `workflow.now()` (deterministic time) - βœ… `workflow.random()` (deterministic random) - βœ… Pure functions and calculations - βœ… Calling activities (non-deterministic operations) ### Versioning Strategies **Challenge**: Changing workflow code while old executions still running **Solutions**: 1. **Versioning API**: Use `workflow.get_version()` for safe changes 2. **New Workflow Type**: Create new workflow, route new executions to it 3. **Backward Compatibility**: Ensure old events replay correctly ## Resilience and Error Handling ### Retry Policies **Default Behavior**: Temporal retries activities forever **Configure Retry**: - Initial retry interval - Backoff coefficient (exponential backoff) - Maximum interval (cap retry delay) - Maximum attempts (eventually fail) **Non-Retryable Errors**: - Invalid input (validation failures) - Business rule violations - Permanent failures (resource not found) ### Idempotency Requirements **Why Critical** (Source: docs.temporal.io/activities): - Activities may execute multiple times - Network failures trigger retries - Duplicate execution must be safe **Implementation Strategies**: - Idempotency keys (deduplication) - Check-then-act with unique constraints - Upsert operations instead of insert - Track processed request IDs ### Activity Heartbeats **Purpose**: Detect stalled long-running activities **Pattern**: - Activity sends periodic heartbeat - Includes progress information - Timeout if no heartbeat received - Enables progress-based retry ## Best Practices ### Workflow Design 1. **Keep workflows focused** - Single responsibility per workflow 2. **Small workflows** - Use child workflows for scalability 3. **Clear boundaries** - Workflow orchestrates, activities execute 4. **Test locally** - Use time-skipping test environment ### Activity Design 1. **Idempotent operations** - Safe to retry 2. **Short-lived** - Seconds to minutes, not hours 3. **Timeout configuration** - Always set timeouts 4. **Heartbeat for long tasks** - Report progress 5. **Error handling** - Distinguish retryable vs non-retryable ### Common Pitfalls **Workflow Violations**: - Using `datetime.now()` instead of `workflow.now()` - Threading or async operations in workflow code - Calling external APIs directly from workflow - Non-deterministic logic in workflows **Activity Mistakes**: - Non-idempotent operations (can't handle retries) - Missing timeouts (activities run forever) - No error classification (retry validation errors) - Ignoring payload limits (2MB per argument) ### Operational Considerations **Monitoring**: - Workflow execution duration - Activity failure rates - Retry attempts and backoff - Pending workflow counts **Scalability**: - Horizontal scaling with workers - Task queue partitioning - Child workflow decomposition - Activity batching when appropriate ## Additional Resources **Official Documentation**: - Temporal Core Concepts: docs.temporal.io/workflows - Workflow Patterns: docs.temporal.io/evaluate/use-cases-design-patterns - Best Practices: docs.temporal.io/develop/best-practices - Saga Pattern: temporal.io/blog/saga-pattern-made-easy **Key Principles**: 1. Workflows = orchestration, Activities = external calls 2. Determinism is non-negotiable for workflows 3. Idempotency is critical for activities 4. State preservation is automatic 5. Design for failure and recovery
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

deployment-pipeline-design

Design multi-stage CI/CD pipelines with approval gates, security

coding
⭐1
# Deployment Pipeline Design Architecture patterns for multi-stage CI/CD pipelines with approval gates and deployment strategies. ## Purpose Design robust, secure deployment pipelines that balance speed with safety through proper stage organization and approval workflows. ## When to Use - Design CI/CD architecture - Implement deployment gates - Configure multi-environment pipelines - Establish deployment best practices - Implement progressive delivery ## Pipeline Stages ### Standard Pipeline Flow ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Build β”‚ β†’ β”‚ Test β”‚ β†’ β”‚ Staging β”‚ β†’ β”‚ Approveβ”‚ β†’ β”‚Productionβ”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ### Detailed Stage Breakdown 1. **Source** - Code checkout 2. **Build** - Compile, package, containerize 3. **Test** - Unit, integration, security scans 4. **Staging Deploy** - Deploy to staging environment 5. **Integration Tests** - E2E, smoke tests 6. **Approval Gate** - Manual approval required 7. **Production Deploy** - Canary, blue-green, rolling 8. **Verification** - Health checks, monitoring 9. **Rollback** - Automated rollback on failure ## Approval Gate Patterns ### Pattern 1: Manual Approval ```yaml # GitHub Actions production-deploy: needs: staging-deploy environment: name: production url: https://app.example.com runs-on: ubuntu-latest steps: - name: Deploy to production run: | # Deployment commands ``` ### Pattern 2: Time-Based Approval ```yaml # GitLab CI deploy:production: stage: deploy script: - deploy.sh production environment: name: production when: delayed start_in: 30 minutes only: - main ``` ### Pattern 3: Multi-Approver ```yaml # Azure Pipelines stages: - stage: Production dependsOn: Staging jobs: - deployment: Deploy environment: name: production resourceType: Kubernetes strategy: runOnce: preDeploy: steps: - task: ManualValidation@0 inputs: notifyUsers: "team-leads@example.com" instructions: "Review staging metrics before approving" ``` **Reference:** See `assets/approval-gate-template.yml` ## Deployment Strategies ### 1. Rolling Deployment ```yaml apiVersion: apps/v1 kind: Deployment metadata: name: my-app spec: replicas: 10 strategy: type: RollingUpdate rollingUpdate: maxSurge: 2 maxUnavailable: 1 ``` **Characteristics:** - Gradual rollout - Zero downtime - Easy rollback - Best for most applications ### 2. Blue-Green Deployment ```yaml # Blue (current) kubectl apply -f blue-deployment.yaml kubectl label service my-app version=blue # Green (new) kubectl apply -f green-deployment.yaml # Test green environment kubectl label service my-app version=green # Rollback if needed kubectl label service my-app version=blue ``` **Characteristics:** - Instant switchover - Easy rollback - Doubles infrastructure cost temporarily - Good for high-risk deployments ### 3. Canary Deployment ```yaml apiVersion: argoproj.io/v1alpha1 kind: Rollout metadata: name: my-app spec: replicas: 10 strategy: canary: steps: - setWeight: 10 - pause: { duration: 5m } - setWeight: 25 - pause: { duration: 5m } - setWeight: 50 - pause: { duration: 5m } - setWeight: 100 ``` **Characteristics:** - Gradual traffic shift - Risk mitigation - Real user testing - Requires service mesh or similar ### 4. Feature Flags ```python from flagsmith import Flagsmith flagsmith = Flagsmith(environment_key="API_KEY") if flagsmith.has_feature("new_checkout_flow"): # New code path process_checkout_v2() else: # Existing code path process_checkout_v1() ``` **Characteristics:** - Deploy without releasing - A/B testing - Instant rollback - Granular control ## Pipeline Orchestration ### Multi-Stage Pipeline Example ```yaml name: Production Pipeline on: push: branches: [main] jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Build application run: make build - name: Build Docker image run: docker build -t myapp:${{ github.sha }} . - name: Push to registry run: docker push myapp:${{ github.sha }} test: needs: build runs-on: ubuntu-latest steps: - name: Unit tests run: make test - name: Security scan run: trivy image myapp:${{ github.sha }} deploy-staging: needs: test runs-on: ubuntu-latest environment: name: staging steps: - name: Deploy to staging run: kubectl apply -f k8s/staging/ integration-test: needs: deploy-staging runs-on: ubuntu-latest steps: - name: Run E2E tests run: npm run test:e2e deploy-production: needs: integration-test runs-on: ubuntu-latest environment: name: production steps: - name: Canary deployment run: | kubectl apply -f k8s/production/ kubectl argo rollouts promote my-app verify: needs: deploy-production runs-on: ubuntu-latest steps: - name: Health check run: curl -f https://app.example.com/health - name: Notify team run: | curl -X POST ${{ secrets.SLACK_WEBHOOK }} \ -d '{"text":"Production deployment successful!"}' ``` ## Pipeline Best Practices 1. **Fail fast** - Run quick tests first 2. **Parallel execution** - Run independent jobs concurrently 3. **Caching** - Cache dependencies between runs 4. **Artifact management** - Store build artifacts 5. **Environment parity** - Keep environments consistent 6. **Secrets management** - Use secret stores (Vault, etc.) 7. **Deployment windows** - Schedule deployments appropriately 8. **Monitoring integration** - Track deployment metrics 9. **Rollback automation** - Auto-rollback on failures 10. **Documentation** - Document pipeline stages ## Rollback Strategies ### Automated Rollback ```yaml deploy-and-verify: steps: - name: Deploy new version run: kubectl apply -f k8s/ - name: Wait for rollout run: kubectl rollout status deployment/my-app - name: Health check id: health run: | for i in {1..10}; do if curl -sf https://app.example.com/health; then exit 0 fi sleep 10 done exit 1 - name: Rollback on failure if: failure() run: kubectl rollout undo deployment/my-app ``` ### Manual Rollback ```bash # List revision history kubectl rollout history deployment/my-app # Rollback to previous version kubectl rollout undo deployment/my-app # Rollback to specific revision kubectl rollout undo deployment/my-app --to-revision=3 ``` ## Monitoring and Metrics ### Key Pipeline Metrics - **Deployment Frequency** - How often deployments occur - **Lead Time** - Time from commit to production - **Change Failure Rate** - Percentage of failed deployments - **Mean Time to Recovery (MTTR)** - Time to recover from failure - **Pipeline Success Rate** - Percentage of successful runs - **Average Pipeline Duration** - Time to complete pipeline ### Integration with Monitoring ```yaml - name: Post-deployment verification run: | # Wait for metrics stabilization sleep 60 # Check error rate ERROR_RATE=$(curl -s "$PROMETHEUS_URL/api/v1/query?query=rate(http_errors_total[5m])" | jq '.data.result[0].value[1]') if (( $(echo "$ERROR_RATE > 0.01" | bc -l) )); then echo "Error rate too high: $ERROR_RATE" exit 1 fi ``` ## Reference Files - `references/pipeline-orchestration.md` - Complex pipeline patterns - `assets/approval-gate-template.yml` - Approval workflow templates ## Related Skills - `github-actions-templates` - For GitHub Actions implementation - `gitlab-ci-patterns` - For GitLab CI implementation - `secrets-management` - For secrets handling
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

airflow-dag-patterns

Build production Apache Airflow DAGs with best practices for

data
⭐1
# Apache Airflow DAG Patterns Production-ready patterns for Apache Airflow including DAG design, operators, sensors, testing, and deployment strategies. ## When to Use This Skill - Creating data pipeline orchestration with Airflow - Designing DAG structures and dependencies - Implementing custom operators and sensors - Testing Airflow DAGs locally - Setting up Airflow in production - Debugging failed DAG runs ## Core Concepts ### 1. DAG Design Principles | Principle | Description | | --------------- | ----------------------------------- | | **Idempotent** | Running twice produces same result | | **Atomic** | Tasks succeed or fail completely | | **Incremental** | Process only new/changed data | | **Observable** | Logs, metrics, alerts at every step | ### 2. Task Dependencies ```python # Linear task1 >> task2 >> task3 # Fan-out task1 >> [task2, task3, task4] # Fan-in [task1, task2, task3] >> task4 # Complex task1 >> task2 >> task4 task1 >> task3 >> task4 ``` ## Quick Start ```python # dags/example_dag.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.operators.empty import EmptyOperator default_args = { 'owner': 'data-team', 'depends_on_past': False, 'email_on_failure': True, 'email_on_retry': False, 'retries': 3, 'retry_delay': timedelta(minutes=5), 'retry_exponential_backoff': True, 'max_retry_delay': timedelta(hours=1), } with DAG( dag_id='example_etl', default_args=default_args, description='Example ETL pipeline', schedule='0 6 * * *', # Daily at 6 AM start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'example'], max_active_runs=1, ) as dag: start = EmptyOperator(task_id='start') def extract_data(**context): execution_date = context['ds'] # Extract logic here return {'records': 1000} extract = PythonOperator( task_id='extract', python_callable=extract_data, ) end = EmptyOperator(task_id='end') start >> extract >> end ``` ## Patterns ### Pattern 1: TaskFlow API (Airflow 2.0+) ```python # dags/taskflow_example.py from datetime import datetime from airflow.decorators import dag, task from airflow.models import Variable @dag( dag_id='taskflow_etl', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'taskflow'], ) def taskflow_etl(): """ETL pipeline using TaskFlow API""" @task() def extract(source: str) -> dict: """Extract data from source""" import pandas as pd df = pd.read_csv(f's3://bucket/{source}/{{ ds }}.csv') return {'data': df.to_dict(), 'rows': len(df)} @task() def transform(extracted: dict) -> dict: """Transform extracted data""" import pandas as pd df = pd.DataFrame(extracted['data']) df['processed_at'] = datetime.now() df = df.dropna() return {'data': df.to_dict(), 'rows': len(df)} @task() def load(transformed: dict, target: str): """Load data to target""" import pandas as pd df = pd.DataFrame(transformed['data']) df.to_parquet(f's3://bucket/{target}/{{ ds }}.parquet') return transformed['rows'] @task() def notify(rows_loaded: int): """Send notification""" print(f'Loaded {rows_loaded} rows') # Define dependencies with XCom passing extracted = extract(source='raw_data') transformed = transform(extracted) loaded = load(transformed, target='processed_data') notify(loaded) # Instantiate the DAG taskflow_etl() ``` ### Pattern 2: Dynamic DAG Generation ```python # dags/dynamic_dag_factory.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.models import Variable import json # Configuration for multiple similar pipelines PIPELINE_CONFIGS = [ {'name': 'customers', 'schedule': '@daily', 'source': 's3://raw/customers'}, {'name': 'orders', 'schedule': '@hourly', 'source': 's3://raw/orders'}, {'name': 'products', 'schedule': '@weekly', 'source': 's3://raw/products'}, ] def create_dag(config: dict) -> DAG: """Factory function to create DAGs from config""" dag_id = f"etl_{config['name']}" default_args = { 'owner': 'data-team', 'retries': 3, 'retry_delay': timedelta(minutes=5), } dag = DAG( dag_id=dag_id, default_args=default_args, schedule=config['schedule'], start_date=datetime(2024, 1, 1), catchup=False, tags=['etl', 'dynamic', config['name']], ) with dag: def extract_fn(source, **context): print(f"Extracting from {source} for {context['ds']}") def transform_fn(**context): print(f"Transforming data for {context['ds']}") def load_fn(table_name, **context): print(f"Loading to {table_name} for {context['ds']}") extract = PythonOperator( task_id='extract', python_callable=extract_fn, op_kwargs={'source': config['source']}, ) transform = PythonOperator( task_id='transform', python_callable=transform_fn, ) load = PythonOperator( task_id='load', python_callable=load_fn, op_kwargs={'table_name': config['name']}, ) extract >> transform >> load return dag # Generate DAGs for config in PIPELINE_CONFIGS: globals()[f"dag_{config['name']}"] = create_dag(config) ``` ### Pattern 3: Branching and Conditional Logic ```python # dags/branching_example.py from airflow.decorators import dag, task from airflow.operators.python import BranchPythonOperator from airflow.operators.empty import EmptyOperator from airflow.utils.trigger_rule import TriggerRule @dag( dag_id='branching_pipeline', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, ) def branching_pipeline(): @task() def check_data_quality() -> dict: """Check data quality and return metrics""" quality_score = 0.95 # Simulated return {'score': quality_score, 'rows': 10000} def choose_branch(**context) -> str: """Determine which branch to execute""" ti = context['ti'] metrics = ti.xcom_pull(task_ids='check_data_quality') if metrics['score'] >= 0.9: return 'high_quality_path' elif metrics['score'] >= 0.7: return 'medium_quality_path' else: return 'low_quality_path' quality_check = check_data_quality() branch = BranchPythonOperator( task_id='branch', python_callable=choose_branch, ) high_quality = EmptyOperator(task_id='high_quality_path') medium_quality = EmptyOperator(task_id='medium_quality_path') low_quality = EmptyOperator(task_id='low_quality_path') # Join point - runs after any branch completes join = EmptyOperator( task_id='join', trigger_rule=TriggerRule.NONE_FAILED_MIN_ONE_SUCCESS, ) quality_check >> branch >> [high_quality, medium_quality, low_quality] >> join branching_pipeline() ``` ### Pattern 4: Sensors and External Dependencies ```python # dags/sensor_patterns.py from datetime import datetime, timedelta from airflow import DAG from airflow.sensors.filesystem import FileSensor from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor from airflow.sensors.external_task import ExternalTaskSensor from airflow.operators.python import PythonOperator with DAG( dag_id='sensor_example', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, ) as dag: # Wait for file on S3 wait_for_file = S3KeySensor( task_id='wait_for_s3_file', bucket_name='data-lake', bucket_key='raw/{{ ds }}/data.parquet', aws_conn_id='aws_default', timeout=60 * 60 * 2, # 2 hours poke_interval=60 * 5, # Check every 5 minutes mode='reschedule', # Free up worker slot while waiting ) # Wait for another DAG to complete wait_for_upstream = ExternalTaskSensor( task_id='wait_for_upstream_dag', external_dag_id='upstream_etl', external_task_id='final_task', execution_date_fn=lambda dt: dt, # Same execution date timeout=60 * 60 * 3, mode='reschedule', ) # Custom sensor using @task.sensor decorator @task.sensor(poke_interval=60, timeout=3600, mode='reschedule') def wait_for_api() -> PokeReturnValue: """Custom sensor for API availability""" import requests response = requests.get('https://api.example.com/health') is_done = response.status_code == 200 return PokeReturnValue(is_done=is_done, xcom_value=response.json()) api_ready = wait_for_api() def process_data(**context): api_result = context['ti'].xcom_pull(task_ids='wait_for_api') print(f"API returned: {api_result}") process = PythonOperator( task_id='process', python_callable=process_data, ) [wait_for_file, wait_for_upstream, api_ready] >> process ``` ### Pattern 5: Error Handling and Alerts ```python # dags/error_handling.py from datetime import datetime, timedelta from airflow import DAG from airflow.operators.python import PythonOperator from airflow.utils.trigger_rule import TriggerRule from airflow.models import Variable def task_failure_callback(context): """Callback on task failure""" task_instance = context['task_instance'] exception = context.get('exception') # Send to Slack/PagerDuty/etc message = f""" Task Failed! DAG: {task_instance.dag_id} Task: {task_instance.task_id} Execution Date: {context['ds']} Error: {exception} Log URL: {task_instance.log_url} """ # send_slack_alert(message) print(message) def dag_failure_callback(context): """Callback on DAG failure""" # Aggregate failures, send summary pass with DAG( dag_id='error_handling_example', schedule='@daily', start_date=datetime(2024, 1, 1), catchup=False, on_failure_callback=dag_failure_callback, default_args={ 'on_failure_callback': task_failure_callback, 'retries': 3, 'retry_delay': timedelta(minutes=5), }, ) as dag: def might_fail(**context): import random if random.random() < 0.3: raise ValueError("Random failure!") return "Success" risky_task = PythonOperator( task_id='risky_task', python_callable=might_fail, ) def cleanup(**context): """Cleanup runs regardless of upstream failures""" print("Cleaning up...") cleanup_task = PythonOperator( task_id='cleanup', python_callable=cleanup, trigger_rule=TriggerRule.ALL_DONE, # Run even if upstream fails ) def notify_success(**context): """Only runs if all upstream succeeded""" print("All tasks succeeded!") success_notification = PythonOperator( task_id='notify_success', python_callable=notify_success, trigger_rule=TriggerRule.ALL_SUCCESS, ) risky_task >> [cleanup_task, success_notification] ``` ### Pattern 6: Testing DAGs ```python # tests/test_dags.py import pytest from datetime import datetime from airflow.models import DagBag @pytest.fixture def dagbag(): return DagBag(dag_folder='dags/', include_examples=False) def test_dag_loaded(dagbag): """Test that all DAGs load without errors""" assert len(dagbag.import_errors) == 0, f"DAG import errors: {dagbag.import_errors}" def test_dag_structure(dagbag): """Test specific DAG structure""" dag = dagbag.get_dag('example_etl') assert dag is not None assert len(dag.tasks) == 3 assert dag.schedule_interval == '0 6 * * *' def test_task_dependencies(dagbag): """Test task dependencies are correct""" dag = dagbag.get_dag('example_etl') extract_task = dag.get_task('extract') assert 'start' in [t.task_id for t in extract_task.upstream_list] assert 'end' in [t.task_id for t in extract_task.downstream_list] def test_dag_integrity(dagbag): """Test DAG has no cycles and is valid""" for dag_id, dag in dagbag.dags.items(): assert dag.test_cycle() is None, f"Cycle detected in {dag_id}" # Test individual task logic def test_extract_function(): """Unit test for extract function""" from dags.example_dag import extract_data result = extract_data(ds='2024-01-01') assert 'records' in result assert isinstance(result['records'], int) ``` ## Project Structure ``` airflow/ β”œβ”€β”€ dags/ β”‚ β”œβ”€β”€ __init__.py β”‚ β”œβ”€β”€ common/ β”‚ β”‚ β”œβ”€β”€ __init__.py β”‚ β”‚ β”œβ”€β”€ operators.py # Custom operators β”‚ β”‚ β”œβ”€β”€ sensors.py # Custom sensors β”‚ β”‚ └── callbacks.py # Alert callbacks β”‚ β”œβ”€β”€ etl/ β”‚ β”‚ β”œβ”€β”€ customers.py β”‚ β”‚ └── orders.py β”‚ └── ml/ β”‚ └── training.py β”œβ”€β”€ plugins/ β”‚ └── custom_plugin.py β”œβ”€β”€ tests/ β”‚ β”œβ”€β”€ __init__.py β”‚ β”œβ”€β”€ test_dags.py β”‚ └── test_operators.py β”œβ”€β”€ docker-compose.yml └── requirements.txt ``` ## Best Practices ### Do's - **Use TaskFlow API** - Cleaner code, automatic XCom - **Set timeouts** - Prevent zombie tasks - **Use `mode='reschedule'`** - For sensors, free up workers - **Test DAGs** - Unit tests and integration tests - **Idempotent tasks** - Safe to retry ### Don'ts - **Don't use `depends_on_past=True`** - Creates bottlenecks - **Don't hardcode dates** - Use `{{ ds }}` macros - **Don't use global state** - Tasks should be stateless - **Don't skip catchup blindly** - Understand implications - **Don't put heavy logic in DAG file** - Import from modules ## Resources - [Airflow Documentation](https://airflow.apache.org/docs/) - [Astronomer Guides](https://docs.astronomer.io/learn) - [TaskFlow API](https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

team-composition-patterns

Design optimal agent team compositions with sizing heuristics,

coding
⭐1
# Team Composition Patterns Best practices for composing multi-agent teams, selecting team sizes, choosing agent types, and configuring display modes for Claude Code's Agent Teams feature. ## When to Use This Skill - Deciding how many teammates to spawn for a task - Choosing between preset team configurations - Selecting the right agent type (subagent_type) for each role - Configuring teammate display modes (tmux, iTerm2, in-process) - Building custom team compositions for non-standard workflows ## Team Sizing Heuristics | Complexity | Team Size | When to Use | | ------------ | --------- | ----------------------------------------------------------- | | Simple | 1-2 | Single-dimension review, isolated bug, small feature | | Moderate | 2-3 | Multi-file changes, 2-3 concerns, medium features | | Complex | 3-4 | Cross-cutting concerns, large features, deep debugging | | Very Complex | 4-5 | Full-stack features, comprehensive reviews, systemic issues | **Rule of thumb**: Start with the smallest team that covers all required dimensions. Adding teammates increases coordination overhead. ## Preset Team Compositions ### Review Team - **Size**: 3 reviewers - **Agents**: 3x `team-reviewer` - **Default dimensions**: security, performance, architecture - **Use when**: Code changes need multi-dimensional quality assessment ### Debug Team - **Size**: 3 investigators - **Agents**: 3x `team-debugger` - **Default hypotheses**: 3 competing hypotheses - **Use when**: Bug has multiple plausible root causes ### Feature Team - **Size**: 3 (1 lead + 2 implementers) - **Agents**: 1x `team-lead` + 2x `team-implementer` - **Use when**: Feature can be decomposed into parallel work streams ### Fullstack Team - **Size**: 4 (1 lead + 3 implementers) - **Agents**: 1x `team-lead` + 1x frontend `team-implementer` + 1x backend `team-implementer` + 1x test `team-implementer` - **Use when**: Feature spans frontend, backend, and test layers ### Research Team - **Size**: 3 researchers - **Agents**: 3x `general-purpose` - **Default areas**: Each assigned a different research question, module, or topic - **Capabilities**: Codebase search (Grep, Glob, Read), web search (WebSearch, WebFetch) - **Use when**: Need to understand a codebase, research libraries, compare approaches, or gather information from code and web sources in parallel ### Security Team - **Size**: 4 reviewers - **Agents**: 4x `team-reviewer` - **Default dimensions**: OWASP/vulnerabilities, auth/access control, dependencies/supply chain, secrets/configuration - **Use when**: Comprehensive security audit covering multiple attack surfaces ### Migration Team - **Size**: 4 (1 lead + 2 implementers + 1 reviewer) - **Agents**: 1x `team-lead` + 2x `team-implementer` + 1x `team-reviewer` - **Use when**: Large codebase migration (framework upgrade, language port, API version bump) requiring parallel work with correctness verification ## Agent Type Selection When spawning teammates with the Task tool, choose `subagent_type` based on what tools the teammate needs: | Agent Type | Tools Available | Use For | | ------------------------------ | ----------------------------------------- | ---------------------------------------------------------- | | `general-purpose` | All tools (Read, Write, Edit, Bash, etc.) | Implementation, debugging, any task requiring file changes | | `Explore` | Read-only tools (Read, Grep, Glob) | Research, code exploration, analysis | | `Plan` | Read-only tools | Architecture planning, task decomposition | | `agent-teams:team-reviewer` | All tools | Code review with structured findings | | `agent-teams:team-debugger` | All tools | Hypothesis-driven investigation | | `agent-teams:team-implementer` | All tools | Building features within file ownership boundaries | | `agent-teams:team-lead` | All tools | Team orchestration and coordination | **Key distinction**: Read-only agents (Explore, Plan) cannot modify files. Never assign implementation tasks to read-only agents. ## Display Mode Configuration Configure in `~/.claude/settings.json`: ```json { "teammateMode": "tmux" } ``` | Mode | Behavior | Best For | | -------------- | ------------------------------ | ------------------------------------------------- | | `"tmux"` | Each teammate in a tmux pane | Development workflows, monitoring multiple agents | | `"iterm2"` | Each teammate in an iTerm2 tab | macOS users who prefer iTerm2 | | `"in-process"` | All teammates in same process | Simple tasks, CI/CD environments | ## Custom Team Guidelines When building custom teams: 1. **Every team needs a coordinator** β€” Either designate a `team-lead` or have the user coordinate directly 2. **Match roles to agent types** β€” Use specialized agents (reviewer, debugger, implementer) when available 3. **Avoid duplicate roles** β€” Two agents doing the same thing wastes resources 4. **Define boundaries upfront** β€” Each teammate needs clear ownership of files or responsibilities 5. **Keep it small** β€” 2-4 teammates is the sweet spot; 5+ requires significant coordination overhead
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

langchain-architecture

Design LLM applications using LangChain 1.x and LangGraph for

coding
⭐1
# LangChain & LangGraph Architecture Master modern LangChain 1.x and LangGraph for building sophisticated LLM applications with agents, state management, memory, and tool integration. ## When to Use This Skill - Building autonomous AI agents with tool access - Implementing complex multi-step LLM workflows - Managing conversation memory and state - Integrating LLMs with external data sources and APIs - Creating modular, reusable LLM application components - Implementing document processing pipelines - Building production-grade LLM applications ## Package Structure (LangChain 1.x) ``` langchain (1.2.x) # High-level orchestration langchain-core (1.2.x) # Core abstractions (messages, prompts, tools) langchain-community # Third-party integrations langgraph # Agent orchestration and state management langchain-openai # OpenAI integrations langchain-anthropic # Anthropic/Claude integrations langchain-voyageai # Voyage AI embeddings langchain-pinecone # Pinecone vector store ``` ## Core Concepts ### 1. LangGraph Agents LangGraph is the standard for building agents in 2026. It provides: **Key Features:** - **StateGraph**: Explicit state management with typed state - **Durable Execution**: Agents persist through failures - **Human-in-the-Loop**: Inspect and modify state at any point - **Memory**: Short-term and long-term memory across sessions - **Checkpointing**: Save and resume agent state **Agent Patterns:** - **ReAct**: Reasoning + Acting with `create_react_agent` - **Plan-and-Execute**: Separate planning and execution nodes - **Multi-Agent**: Supervisor routing between specialized agents - **Tool-Calling**: Structured tool invocation with Pydantic schemas ### 2. State Management LangGraph uses TypedDict for explicit state: ```python from typing import Annotated, TypedDict from langgraph.graph import MessagesState # Simple message-based state class AgentState(MessagesState): """Extends MessagesState with custom fields.""" context: Annotated[list, "retrieved documents"] # Custom state for complex agents class CustomState(TypedDict): messages: Annotated[list, "conversation history"] context: Annotated[dict, "retrieved context"] current_step: str results: list ``` ### 3. Memory Systems Modern memory implementations: - **ConversationBufferMemory**: Stores all messages (short conversations) - **ConversationSummaryMemory**: Summarizes older messages (long conversations) - **ConversationTokenBufferMemory**: Token-based windowing - **VectorStoreRetrieverMemory**: Semantic similarity retrieval - **LangGraph Checkpointers**: Persistent state across sessions ### 4. Document Processing Loading, transforming, and storing documents: **Components:** - **Document Loaders**: Load from various sources - **Text Splitters**: Chunk documents intelligently - **Vector Stores**: Store and retrieve embeddings - **Retrievers**: Fetch relevant documents ### 5. Callbacks & Tracing LangSmith is the standard for observability: - Request/response logging - Token usage tracking - Latency monitoring - Error tracking - Trace visualization ## Quick Start ### Modern ReAct Agent with LangGraph ```python from langgraph.prebuilt import create_react_agent from langgraph.checkpoint.memory import MemorySaver from langchain_anthropic import ChatAnthropic from langchain_core.tools import tool import ast import operator # Initialize LLM (Claude Sonnet 4.6 recommended) llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0) # Define tools with Pydantic schemas @tool def search_database(query: str) -> str: """Search internal database for information.""" # Your database search logic return f"Results for: {query}" @tool def calculate(expression: str) -> str: """Safely evaluate a mathematical expression. Supports: +, -, *, /, **, %, parentheses Example: '(2 + 3) * 4' returns '20' """ # Safe math evaluation using ast allowed_operators = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.Mod: operator.mod, ast.USub: operator.neg, } def _eval(node): if isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.BinOp): left = _eval(node.left) right = _eval(node.right) return allowed_operators[type(node.op)](left, right) elif isinstance(node, ast.UnaryOp): operand = _eval(node.operand) return allowed_operators[type(node.op)](operand) else: raise ValueError(f"Unsupported operation: {type(node)}") try: tree = ast.parse(expression, mode='eval') return str(_eval(tree.body)) except Exception as e: return f"Error: {e}" tools = [search_database, calculate] # Create checkpointer for memory persistence checkpointer = MemorySaver() # Create ReAct agent agent = create_react_agent( llm, tools, checkpointer=checkpointer ) # Run agent with thread ID for memory config = {"configurable": {"thread_id": "user-123"}} result = await agent.ainvoke( {"messages": [("user", "Search for Python tutorials and calculate 25 * 4")]}, config=config ) ``` ## Architecture Patterns ### Pattern 1: RAG with LangGraph ```python from langgraph.graph import StateGraph, START, END from langchain_anthropic import ChatAnthropic from langchain_voyageai import VoyageAIEmbeddings from langchain_pinecone import PineconeVectorStore from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from typing import TypedDict, Annotated class RAGState(TypedDict): question: str context: Annotated[list[Document], "retrieved documents"] answer: str # Initialize components llm = ChatAnthropic(model="claude-sonnet-4-6") embeddings = VoyageAIEmbeddings(model="voyage-3-large") vectorstore = PineconeVectorStore(index_name="docs", embedding=embeddings) retriever = vectorstore.as_retriever(search_kwargs={"k": 4}) # Define nodes async def retrieve(state: RAGState) -> RAGState: """Retrieve relevant documents.""" docs = await retriever.ainvoke(state["question"]) return {"context": docs} async def generate(state: RAGState) -> RAGState: """Generate answer from context.""" prompt = ChatPromptTemplate.from_template( """Answer based on the context below. If you cannot answer, say so. Context: {context} Question: {question} Answer:""" ) context_text = "\n\n".join(doc.page_content for doc in state["context"]) response = await llm.ainvoke( prompt.format(context=context_text, question=state["question"]) ) return {"answer": response.content} # Build graph builder = StateGraph(RAGState) builder.add_node("retrieve", retrieve) builder.add_node("generate", generate) builder.add_edge(START, "retrieve") builder.add_edge("retrieve", "generate") builder.add_edge("generate", END) rag_chain = builder.compile() # Use the chain result = await rag_chain.ainvoke({"question": "What is the main topic?"}) ``` ### Pattern 2: Custom Agent with Structured Tools ```python from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class SearchInput(BaseModel): """Input for database search.""" query: str = Field(description="Search query") filters: dict = Field(default={}, description="Optional filters") class EmailInput(BaseModel): """Input for sending email.""" recipient: str = Field(description="Email recipient") subject: str = Field(description="Email subject") content: str = Field(description="Email body") async def search_database(query: str, filters: dict = {}) -> str: """Search internal database for information.""" # Your database search logic return f"Results for '{query}' with filters {filters}" async def send_email(recipient: str, subject: str, content: str) -> str: """Send an email to specified recipient.""" # Email sending logic return f"Email sent to {recipient}" tools = [ StructuredTool.from_function( coroutine=search_database, name="search_database", description="Search internal database", args_schema=SearchInput ), StructuredTool.from_function( coroutine=send_email, name="send_email", description="Send an email", args_schema=EmailInput ) ] agent = create_react_agent(llm, tools) ``` ### Pattern 3: Multi-Step Workflow with StateGraph ```python from langgraph.graph import StateGraph, START, END from typing import TypedDict, Literal class WorkflowState(TypedDict): text: str entities: list analysis: str summary: str current_step: str async def extract_entities(state: WorkflowState) -> WorkflowState: """Extract key entities from text.""" prompt = f"Extract key entities from: {state['text']}\n\nReturn as JSON list." response = await llm.ainvoke(prompt) return {"entities": response.content, "current_step": "analyze"} async def analyze_entities(state: WorkflowState) -> WorkflowState: """Analyze extracted entities.""" prompt = f"Analyze these entities: {state['entities']}\n\nProvide insights." response = await llm.ainvoke(prompt) return {"analysis": response.content, "current_step": "summarize"} async def generate_summary(state: WorkflowState) -> WorkflowState: """Generate final summary.""" prompt = f"""Summarize: Entities: {state['entities']} Analysis: {state['analysis']} Provide a concise summary.""" response = await llm.ainvoke(prompt) return {"summary": response.content, "current_step": "complete"} def route_step(state: WorkflowState) -> Literal["analyze", "summarize", "end"]: """Route to next step based on current state.""" step = state.get("current_step", "extract") if step == "analyze": return "analyze" elif step == "summarize": return "summarize" return "end" # Build workflow builder = StateGraph(WorkflowState) builder.add_node("extract", extract_entities) builder.add_node("analyze", analyze_entities) builder.add_node("summarize", generate_summary) builder.add_edge(START, "extract") builder.add_conditional_edges("extract", route_step, { "analyze": "analyze", "summarize": "summarize", "end": END }) builder.add_conditional_edges("analyze", route_step, { "summarize": "summarize", "end": END }) builder.add_edge("summarize", END) workflow = builder.compile() ``` ### Pattern 4: Multi-Agent Orchestration ```python from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import create_react_agent from langchain_core.messages import HumanMessage from typing import Literal class MultiAgentState(TypedDict): messages: list next_agent: str # Create specialized agents researcher = create_react_agent(llm, research_tools) writer = create_react_agent(llm, writing_tools) reviewer = create_react_agent(llm, review_tools) async def supervisor(state: MultiAgentState) -> MultiAgentState: """Route to appropriate agent based on task.""" prompt = f"""Based on the conversation, which agent should handle this? Options: - researcher: For finding information - writer: For creating content - reviewer: For reviewing and editing - FINISH: Task is complete Messages: {state['messages']} Respond with just the agent name.""" response = await llm.ainvoke(prompt) return {"next_agent": response.content.strip().lower()} def route_to_agent(state: MultiAgentState) -> Literal["researcher", "writer", "reviewer", "end"]: """Route based on supervisor decision.""" next_agent = state.get("next_agent", "").lower() if next_agent == "finish": return "end" return next_agent if next_agent in ["researcher", "writer", "reviewer"] else "end" # Build multi-agent graph builder = StateGraph(MultiAgentState) builder.add_node("supervisor", supervisor) builder.add_node("researcher", researcher) builder.add_node("writer", writer) builder.add_node("reviewer", reviewer) builder.add_edge(START, "supervisor") builder.add_conditional_edges("supervisor", route_to_agent, { "researcher": "researcher", "writer": "writer", "reviewer": "reviewer", "end": END }) # Each agent returns to supervisor for agent in ["researcher", "writer", "reviewer"]: builder.add_edge(agent, "supervisor") multi_agent = builder.compile() ``` ## Memory Management ### Token-Based Memory with LangGraph ```python from langgraph.checkpoint.memory import MemorySaver from langgraph.prebuilt import create_react_agent # In-memory checkpointer (development) checkpointer = MemorySaver() # Create agent with persistent memory agent = create_react_agent(llm, tools, checkpointer=checkpointer) # Each thread_id maintains separate conversation config = {"configurable": {"thread_id": "session-abc123"}} # Messages persist across invocations with same thread_id result1 = await agent.ainvoke({"messages": [("user", "My name is Alice")]}, config) result2 = await agent.ainvoke({"messages": [("user", "What's my name?")]}, config) # Agent remembers: "Your name is Alice" ``` ### Production Memory with PostgreSQL ```python from langgraph.checkpoint.postgres import PostgresSaver # Production checkpointer checkpointer = PostgresSaver.from_conn_string( "postgresql://user:pass@localhost/langgraph" ) agent = create_react_agent(llm, tools, checkpointer=checkpointer) ``` ### Vector Store Memory for Long-Term Context ```python from langchain_community.vectorstores import Chroma from langchain_voyageai import VoyageAIEmbeddings embeddings = VoyageAIEmbeddings(model="voyage-3-large") memory_store = Chroma( collection_name="conversation_memory", embedding_function=embeddings, persist_directory="./memory_db" ) async def retrieve_relevant_memory(query: str, k: int = 5) -> list: """Retrieve relevant past conversations.""" docs = await memory_store.asimilarity_search(query, k=k) return [doc.page_content for doc in docs] async def store_memory(content: str, metadata: dict = {}): """Store conversation in long-term memory.""" await memory_store.aadd_texts([content], metadatas=[metadata]) ``` ## Callback System & LangSmith ### LangSmith Tracing ```python import os from langchain_anthropic import ChatAnthropic # Enable LangSmith tracing os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_API_KEY"] = "your-api-key" os.environ["LANGCHAIN_PROJECT"] = "my-project" # All LangChain/LangGraph operations are automatically traced llm = ChatAnthropic(model="claude-sonnet-4-6") ``` ### Custom Callback Handler ```python from langchain_core.callbacks import BaseCallbackHandler from typing import Any, Dict, List class CustomCallbackHandler(BaseCallbackHandler): def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs ) -> None: print(f"LLM started with {len(prompts)} prompts") def on_llm_end(self, response, **kwargs) -> None: print(f"LLM completed: {len(response.generations)} generations") def on_llm_error(self, error: Exception, **kwargs) -> None: print(f"LLM error: {error}") def on_tool_start( self, serialized: Dict[str, Any], input_str: str, **kwargs ) -> None: print(f"Tool started: {serialized.get('name')}") def on_tool_end(self, output: str, **kwargs) -> None: print(f"Tool completed: {output[:100]}...") # Use callbacks result = await agent.ainvoke( {"messages": [("user", "query")]}, config={"callbacks": [CustomCallbackHandler()]} ) ``` ## Streaming Responses ```python from langchain_anthropic import ChatAnthropic llm = ChatAnthropic(model="claude-sonnet-4-6", streaming=True) # Stream tokens async for chunk in llm.astream("Tell me a story"): print(chunk.content, end="", flush=True) # Stream agent events async for event in agent.astream_events( {"messages": [("user", "Search and summarize")]}, version="v2" ): if event["event"] == "on_chat_model_stream": print(event["data"]["chunk"].content, end="") elif event["event"] == "on_tool_start": print(f"\n[Using tool: {event['name']}]") ``` ## Testing Strategies ```python import pytest from unittest.mock import AsyncMock, patch @pytest.mark.asyncio async def test_agent_tool_selection(): """Test agent selects correct tool.""" with patch.object(llm, 'ainvoke') as mock_llm: mock_llm.return_value = AsyncMock(content="Using search_database") result = await agent.ainvoke({ "messages": [("user", "search for documents")] }) # Verify tool was called assert "search_database" in str(result) @pytest.mark.asyncio async def test_memory_persistence(): """Test memory persists across invocations.""" config = {"configurable": {"thread_id": "test-thread"}} # First message await agent.ainvoke( {"messages": [("user", "Remember: the code is 12345")]}, config ) # Second message should remember result = await agent.ainvoke( {"messages": [("user", "What was the code?")]}, config ) assert "12345" in result["messages"][-1].content ``` ## Performance Optimization ### 1. Caching with Redis ```python from langchain_community.cache import RedisCache from langchain_core.globals import set_llm_cache import redis redis_client = redis.Redis.from_url("redis://localhost:6379") set_llm_cache(RedisCache(redis_client)) ``` ### 2. Async Batch Processing ```python import asyncio from langchain_core.documents import Document async def process_documents(documents: list[Document]) -> list: """Process documents in parallel.""" tasks = [process_single(doc) for doc in documents] return await asyncio.gather(*tasks) async def process_single(doc: Document) -> dict: """Process a single document.""" chunks = text_splitter.split_documents([doc]) embeddings = await embeddings_model.aembed_documents( [c.page_content for c in chunks] ) return {"doc_id": doc.metadata.get("id"), "embeddings": embeddings} ``` ### 3. Connection Pooling ```python from langchain_pinecone import PineconeVectorStore from pinecone import Pinecone # Reuse Pinecone client pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"]) index = pc.Index("my-index") # Create vector store with existing index vectorstore = PineconeVectorStore(index=index, embedding=embeddings) ``` ## Resources - [LangChain Documentation](https://python.langchain.com/docs/) - [LangGraph Documentation](https://langchain-ai.github.io/langgraph/) - [LangSmith Platform](https://smith.langchain.com/) - [LangChain GitHub](https://github.com/langchain-ai/langchain) - [LangGraph GitHub](https://github.com/langchain-ai/langgraph) ## Common Pitfalls 1. **Using Deprecated APIs**: Use LangGraph for agents, not `initialize_agent` 2. **Memory Overflow**: Use checkpointers with TTL for long-running agents 3. **Poor Tool Descriptions**: Clear descriptions help LLM select correct tools 4. **Context Window Exceeded**: Use summarization or sliding window memory 5. **No Error Handling**: Wrap too
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

ml-pipeline-workflow

Build end-to-end MLOps pipelines from data preparation through

coding
⭐1
# ML Pipeline Workflow Complete end-to-end MLOps pipeline orchestration from data preparation through model deployment. ## Overview This skill provides comprehensive guidance for building production ML pipelines that handle the full lifecycle: data ingestion β†’ preparation β†’ training β†’ validation β†’ deployment β†’ monitoring. ## When to Use This Skill - Building new ML pipelines from scratch - Designing workflow orchestration for ML systems - Implementing data β†’ model β†’ deployment automation - Setting up reproducible training workflows - Creating DAG-based ML orchestration - Integrating ML components into production systems ## What This Skill Provides ### Core Capabilities 1. **Pipeline Architecture** - End-to-end workflow design - DAG orchestration patterns (Airflow, Dagster, Kubeflow) - Component dependencies and data flow - Error handling and retry strategies 2. **Data Preparation** - Data validation and quality checks - Feature engineering pipelines - Data versioning and lineage - Train/validation/test splitting strategies 3. **Model Training** - Training job orchestration - Hyperparameter management - Experiment tracking integration - Distributed training patterns 4. **Model Validation** - Validation frameworks and metrics - A/B testing infrastructure - Performance regression detection - Model comparison workflows 5. **Deployment Automation** - Model serving patterns - Canary deployments - Blue-green deployment strategies - Rollback mechanisms ### Reference Documentation See the `references/` directory for detailed guides: - **data-preparation.md** - Data cleaning, validation, and feature engineering - **model-training.md** - Training workflows and best practices - **model-validation.md** - Validation strategies and metrics - **model-deployment.md** - Deployment patterns and serving architectures ### Assets and Templates The `assets/` directory contains: - **pipeline-dag.yaml.template** - DAG template for workflow orchestration - **training-config.yaml** - Training configuration template - **validation-checklist.md** - Pre-deployment validation checklist ## Usage Patterns ### Basic Pipeline Setup ```python # 1. Define pipeline stages stages = [ "data_ingestion", "data_validation", "feature_engineering", "model_training", "model_validation", "model_deployment" ] # 2. Configure dependencies # See assets/pipeline-dag.yaml.template for full example ``` ### Production Workflow 1. **Data Preparation Phase** - Ingest raw data from sources - Run data quality checks - Apply feature transformations - Version processed datasets 2. **Training Phase** - Load versioned training data - Execute training jobs - Track experiments and metrics - Save trained models 3. **Validation Phase** - Run validation test suite - Compare against baseline - Generate performance reports - Approve for deployment 4. **Deployment Phase** - Package model artifacts - Deploy to serving infrastructure - Configure monitoring - Validate production traffic ## Best Practices ### Pipeline Design - **Modularity**: Each stage should be independently testable - **Idempotency**: Re-running stages should be safe - **Observability**: Log metrics at every stage - **Versioning**: Track data, code, and model versions - **Failure Handling**: Implement retry logic and alerting ### Data Management - Use data validation libraries (Great Expectations, TFX) - Version datasets with DVC or similar tools - Document feature engineering transformations - Maintain data lineage tracking ### Model Operations - Separate training and serving infrastructure - Use model registries (MLflow, Weights & Biases) - Implement gradual rollouts for new models - Monitor model performance drift - Maintain rollback capabilities ### Deployment Strategies - Start with shadow deployments - Use canary releases for validation - Implement A/B testing infrastructure - Set up automated rollback triggers - Monitor latency and throughput ## Integration Points ### Orchestration Tools - **Apache Airflow**: DAG-based workflow orchestration - **Dagster**: Asset-based pipeline orchestration - **Kubeflow Pipelines**: Kubernetes-native ML workflows - **Prefect**: Modern dataflow automation ### Experiment Tracking - MLflow for experiment tracking and model registry - Weights & Biases for visualization and collaboration - TensorBoard for training metrics ### Deployment Platforms - AWS SageMaker for managed ML infrastructure - Google Vertex AI for GCP deployments - Azure ML for Azure cloud - Kubernetes + KServe for cloud-agnostic serving ## Progressive Disclosure Start with the basics and gradually add complexity: 1. **Level 1**: Simple linear pipeline (data β†’ train β†’ deploy) 2. **Level 2**: Add validation and monitoring stages 3. **Level 3**: Implement hyperparameter tuning 4. **Level 4**: Add A/B testing and gradual rollouts 5. **Level 5**: Multi-model pipelines with ensemble strategies ## Common Patterns ### Batch Training Pipeline ```yaml # See assets/pipeline-dag.yaml.template stages: - name: data_preparation dependencies: [] - name: model_training dependencies: [data_preparation] - name: model_evaluation dependencies: [model_training] - name: model_deployment dependencies: [model_evaluation] ``` ### Real-time Feature Pipeline ```python # Stream processing for real-time features # Combined with batch training # See references/data-preparation.md ``` ### Continuous Training ```python # Automated retraining on schedule # Triggered by data drift detection # See references/model-training.md ``` ## Troubleshooting ### Common Issues - **Pipeline failures**: Check dependencies and data availability - **Training instability**: Review hyperparameters and data quality - **Deployment issues**: Validate model artifacts and serving config - **Performance degradation**: Monitor data drift and model metrics ### Debugging Steps 1. Check pipeline logs for each stage 2. Validate input/output data at boundaries 3. Test components in isolation 4. Review experiment tracking metrics 5. Inspect model artifacts and metadata ## Next Steps After setting up your pipeline: 1. Explore **hyperparameter-tuning** skill for optimization 2. Learn **experiment-tracking-setup** for MLflow/W&B 3. Review **model-deployment-patterns** for serving strategies 4. Implement monitoring with observability tools ## Related Skills - **experiment-tracking-setup**: MLflow and Weights & Biases integration - **hyperparameter-tuning**: Automated hyperparameter optimization - **model-deployment-patterns**: Advanced deployment strategies
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-design-patterns

Python design patterns including KISS, Separation of Concerns,

coding
⭐1
# Python Design Patterns Write maintainable Python code using fundamental design principles. These patterns help you build systems that are easy to understand, test, and modify. ## When to Use This Skill - Designing new components or services - Refactoring complex or tangled code - Deciding whether to create an abstraction - Choosing between inheritance and composition - Evaluating code complexity and coupling - Planning modular architectures ## Core Concepts ### 1. KISS (Keep It Simple) Choose the simplest solution that works. Complexity must be justified by concrete requirements. ### 2. Single Responsibility (SRP) Each unit should have one reason to change. Separate concerns into focused components. ### 3. Composition Over Inheritance Build behavior by combining objects, not extending classes. ### 4. Rule of Three Wait until you have three instances before abstracting. Duplication is often better than premature abstraction. ## Quick Start ```python # Simple beats clever # Instead of a factory/registry pattern: FORMATTERS = {"json": JsonFormatter, "csv": CsvFormatter} def get_formatter(name: str) -> Formatter: return FORMATTERS[name]() ``` ## Fundamental Patterns ### Pattern 1: KISS - Keep It Simple Before adding complexity, ask: does a simpler solution work? ```python # Over-engineered: Factory with registration class OutputFormatterFactory: _formatters: dict[str, type[Formatter]] = {} @classmethod def register(cls, name: str): def decorator(formatter_cls): cls._formatters[name] = formatter_cls return formatter_cls return decorator @classmethod def create(cls, name: str) -> Formatter: return cls._formatters[name]() @OutputFormatterFactory.register("json") class JsonFormatter(Formatter): ... # Simple: Just use a dictionary FORMATTERS = { "json": JsonFormatter, "csv": CsvFormatter, "xml": XmlFormatter, } def get_formatter(name: str) -> Formatter: """Get formatter by name.""" if name not in FORMATTERS: raise ValueError(f"Unknown format: {name}") return FORMATTERS[name]() ``` The factory pattern adds code without adding value here. Save patterns for when they solve real problems. ### Pattern 2: Single Responsibility Principle Each class or function should have one reason to change. ```python # BAD: Handler does everything class UserHandler: async def create_user(self, request: Request) -> Response: # HTTP parsing data = await request.json() # Validation if not data.get("email"): return Response({"error": "email required"}, status=400) # Database access user = await db.execute( "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *", data["email"], data["name"] ) # Response formatting return Response({"id": user.id, "email": user.email}, status=201) # GOOD: Separated concerns class UserService: """Business logic only.""" def __init__(self, repo: UserRepository) -> None: self._repo = repo async def create_user(self, data: CreateUserInput) -> User: # Only business rules here user = User(email=data.email, name=data.name) return await self._repo.save(user) class UserHandler: """HTTP concerns only.""" def __init__(self, service: UserService) -> None: self._service = service async def create_user(self, request: Request) -> Response: data = CreateUserInput(**(await request.json())) user = await self._service.create_user(data) return Response(user.to_dict(), status=201) ``` Now HTTP changes don't affect business logic, and vice versa. ### Pattern 3: Separation of Concerns Organize code into distinct layers with clear responsibilities. ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ API Layer (handlers) β”‚ β”‚ - Parse requests β”‚ β”‚ - Call services β”‚ β”‚ - Format responses β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Service Layer (business logic) β”‚ β”‚ - Domain rules and validation β”‚ β”‚ - Orchestrate operations β”‚ β”‚ - Pure functions where possible β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β–Ό β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Repository Layer (data access) β”‚ β”‚ - SQL queries β”‚ β”‚ - External API calls β”‚ β”‚ - Cache operations β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` Each layer depends only on layers below it: ```python # Repository: Data access class UserRepository: async def get_by_id(self, user_id: str) -> User | None: row = await self._db.fetchrow( "SELECT * FROM users WHERE id = $1", user_id ) return User(**row) if row else None # Service: Business logic class UserService: def __init__(self, repo: UserRepository) -> None: self._repo = repo async def get_user(self, user_id: str) -> User: user = await self._repo.get_by_id(user_id) if user is None: raise UserNotFoundError(user_id) return user # Handler: HTTP concerns @app.get("/users/{user_id}") async def get_user(user_id: str) -> UserResponse: user = await user_service.get_user(user_id) return UserResponse.from_user(user) ``` ### Pattern 4: Composition Over Inheritance Build behavior by combining objects rather than inheriting. ```python # Inheritance: Rigid and hard to test class EmailNotificationService(NotificationService): def __init__(self): super().__init__() self._smtp = SmtpClient() # Hard to mock def notify(self, user: User, message: str) -> None: self._smtp.send(user.email, message) # Composition: Flexible and testable class NotificationService: """Send notifications via multiple channels.""" def __init__( self, email_sender: EmailSender, sms_sender: SmsSender | None = None, push_sender: PushSender | None = None, ) -> None: self._email = email_sender self._sms = sms_sender self._push = push_sender async def notify( self, user: User, message: str, channels: set[str] | None = None, ) -> None: channels = channels or {"email"} if "email" in channels: await self._email.send(user.email, message) if "sms" in channels and self._sms and user.phone: await self._sms.send(user.phone, message) if "push" in channels and self._push and user.device_token: await self._push.send(user.device_token, message) # Easy to test with fakes service = NotificationService( email_sender=FakeEmailSender(), sms_sender=FakeSmsSender(), ) ``` ## Advanced Patterns ### Pattern 5: Rule of Three Wait until you have three instances before abstracting. ```python # Two similar functions? Don't abstract yet def process_orders(orders: list[Order]) -> list[Result]: results = [] for order in orders: validated = validate_order(order) result = process_validated_order(validated) results.append(result) return results def process_returns(returns: list[Return]) -> list[Result]: results = [] for ret in returns: validated = validate_return(ret) result = process_validated_return(validated) results.append(result) return results # These look similar, but wait! Are they actually the same? # Different validation, different processing, different errors... # Duplication is often better than the wrong abstraction # Only after a third case, consider if there's a real pattern # But even then, sometimes explicit is better than abstract ``` ### Pattern 6: Function Size Guidelines Keep functions focused. Extract when a function: - Exceeds 20-50 lines (varies by complexity) - Serves multiple distinct purposes - Has deeply nested logic (3+ levels) ```python # Too long, multiple concerns mixed def process_order(order: Order) -> Result: # 50 lines of validation... # 30 lines of inventory check... # 40 lines of payment processing... # 20 lines of notification... pass # Better: Composed from focused functions def process_order(order: Order) -> Result: """Process a customer order through the complete workflow.""" validate_order(order) reserve_inventory(order) payment_result = charge_payment(order) send_confirmation(order, payment_result) return Result(success=True, order_id=order.id) ``` ### Pattern 7: Dependency Injection Pass dependencies through constructors for testability. ```python from typing import Protocol class Logger(Protocol): def info(self, msg: str, **kwargs) -> None: ... def error(self, msg: str, **kwargs) -> None: ... class Cache(Protocol): async def get(self, key: str) -> str | None: ... async def set(self, key: str, value: str, ttl: int) -> None: ... class UserService: """Service with injected dependencies.""" def __init__( self, repository: UserRepository, cache: Cache, logger: Logger, ) -> None: self._repo = repository self._cache = cache self._logger = logger async def get_user(self, user_id: str) -> User: # Check cache first cached = await self._cache.get(f"user:{user_id}") if cached: self._logger.info("Cache hit", user_id=user_id) return User.from_json(cached) # Fetch from database user = await self._repo.get_by_id(user_id) if user: await self._cache.set(f"user:{user_id}", user.to_json(), ttl=300) return user # Production service = UserService( repository=PostgresUserRepository(db), cache=RedisCache(redis), logger=StructlogLogger(), ) # Testing service = UserService( repository=InMemoryUserRepository(), cache=FakeCache(), logger=NullLogger(), ) ``` ### Pattern 8: Avoiding Common Anti-Patterns **Don't expose internal types:** ```python # BAD: Leaking ORM model to API @app.get("/users/{id}") def get_user(id: str) -> UserModel: # SQLAlchemy model return db.query(UserModel).get(id) # GOOD: Use response schemas @app.get("/users/{id}") def get_user(id: str) -> UserResponse: user = db.query(UserModel).get(id) return UserResponse.from_orm(user) ``` **Don't mix I/O with business logic:** ```python # BAD: SQL embedded in business logic def calculate_discount(user_id: str) -> float: user = db.query("SELECT * FROM users WHERE id = ?", user_id) orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id) # Business logic mixed with data access # GOOD: Repository pattern def calculate_discount(user: User, order_history: list[Order]) -> float: # Pure business logic, easily testable if len(order_history) > 10: return 0.15 return 0.0 ``` ## Best Practices Summary 1. **Keep it simple** - Choose the simplest solution that works 2. **Single responsibility** - Each unit has one reason to change 3. **Separate concerns** - Distinct layers with clear purposes 4. **Compose, don't inherit** - Combine objects for flexibility 5. **Rule of three** - Wait before abstracting 6. **Keep functions small** - 20-50 lines (varies by complexity), one purpose 7. **Inject dependencies** - Constructor injection for testability 8. **Delete before abstracting** - Remove dead code, then consider patterns 9. **Test each layer** - Isolated tests for each concern 10. **Explicit over clever** - Readable code beats elegant code
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

bash-defensive-patterns

Master defensive Bash programming techniques for production-grade

coding
⭐1
# Bash Defensive Patterns Comprehensive guidance for writing production-ready Bash scripts using defensive programming techniques, error handling, and safety best practices to prevent common pitfalls and ensure reliability. ## When to Use This Skill - Writing production automation scripts - Building CI/CD pipeline scripts - Creating system administration utilities - Developing error-resilient deployment automation - Writing scripts that must handle edge cases safely - Building maintainable shell script libraries - Implementing comprehensive logging and monitoring - Creating scripts that must work across different platforms ## Core Defensive Principles ### 1. Strict Mode Enable bash strict mode at the start of every script to catch errors early. ```bash #!/bin/bash set -Eeuo pipefail # Exit on error, unset variables, pipe failures ``` **Key flags:** - `set -E`: Inherit ERR trap in functions - `set -e`: Exit on any error (command returns non-zero) - `set -u`: Exit on undefined variable reference - `set -o pipefail`: Pipe fails if any command fails (not just last) ### 2. Error Trapping and Cleanup Implement proper cleanup on script exit or error. ```bash #!/bin/bash set -Eeuo pipefail trap 'echo "Error on line $LINENO"' ERR trap 'echo "Cleaning up..."; rm -rf "$TMPDIR"' EXIT TMPDIR=$(mktemp -d) # Script code here ``` ### 3. Variable Safety Always quote variables to prevent word splitting and globbing issues. ```bash # Wrong - unsafe cp $source $dest # Correct - safe cp "$source" "$dest" # Required variables - fail with message if unset : "${REQUIRED_VAR:?REQUIRED_VAR is not set}" ``` ### 4. Array Handling Use arrays safely for complex data handling. ```bash # Safe array iteration declare -a items=("item 1" "item 2" "item 3") for item in "${items[@]}"; do echo "Processing: $item" done # Reading output into array safely mapfile -t lines < <(some_command) readarray -t numbers < <(seq 1 10) ``` ### 5. Conditional Safety Use `[[ ]]` for Bash-specific features, `[ ]` for POSIX. ```bash # Bash - safer if [[ -f "$file" && -r "$file" ]]; then content=$(<"$file") fi # POSIX - portable if [ -f "$file" ] && [ -r "$file" ]; then content=$(cat "$file") fi # Test for existence before operations if [[ -z "${VAR:-}" ]]; then echo "VAR is not set or is empty" fi ``` ## Fundamental Patterns ### Pattern 1: Safe Script Directory Detection ```bash #!/bin/bash set -Eeuo pipefail # Correctly determine script directory SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)" SCRIPT_NAME="$(basename -- "${BASH_SOURCE[0]}")" echo "Script location: $SCRIPT_DIR/$SCRIPT_NAME" ``` ### Pattern 2: Comprehensive Function Templat ```bash #!/bin/bash set -Eeuo pipefail # Prefix for functions: handle_*, process_*, check_*, validate_* # Include documentation and error handling validate_file() { local -r file="$1" local -r message="${2:-File not found: $file}" if [[ ! -f "$file" ]]; then echo "ERROR: $message" >&2 return 1 fi return 0 } process_files() { local -r input_dir="$1" local -r output_dir="$2" # Validate inputs [[ -d "$input_dir" ]] || { echo "ERROR: input_dir not a directory" >&2; return 1; } # Create output directory if needed mkdir -p "$output_dir" || { echo "ERROR: Cannot create output_dir" >&2; return 1; } # Process files safely while IFS= read -r -d '' file; do echo "Processing: $file" # Do work done < <(find "$input_dir" -maxdepth 1 -type f -print0) return 0 } ``` ### Pattern 3: Safe Temporary File Handling ```bash #!/bin/bash set -Eeuo pipefail trap 'rm -rf -- "$TMPDIR"' EXIT # Create temporary directory TMPDIR=$(mktemp -d) || { echo "ERROR: Failed to create temp directory" >&2; exit 1; } # Create temporary files in directory TMPFILE1="$TMPDIR/temp1.txt" TMPFILE2="$TMPDIR/temp2.txt" # Use temporary files touch "$TMPFILE1" "$TMPFILE2" echo "Temp files created in: $TMPDIR" ``` ### Pattern 4: Robust Argument Parsing ```bash #!/bin/bash set -Eeuo pipefail # Default values VERBOSE=false DRY_RUN=false OUTPUT_FILE="" THREADS=4 usage() { cat <<EOF Usage: $0 [OPTIONS] Options: -v, --verbose Enable verbose output -d, --dry-run Run without making changes -o, --output FILE Output file path -j, --jobs NUM Number of parallel jobs -h, --help Show this help message EOF exit "${1:-0}" } # Parse arguments while [[ $# -gt 0 ]]; do case "$1" in -v|--verbose) VERBOSE=true shift ;; -d|--dry-run) DRY_RUN=true shift ;; -o|--output) OUTPUT_FILE="$2" shift 2 ;; -j|--jobs) THREADS="$2" shift 2 ;; -h|--help) usage 0 ;; --) shift break ;; *) echo "ERROR: Unknown option: $1" >&2 usage 1 ;; esac done # Validate required arguments [[ -n "$OUTPUT_FILE" ]] || { echo "ERROR: -o/--output is required" >&2; usage 1; } ``` ### Pattern 5: Structured Logging ```bash #!/bin/bash set -Eeuo pipefail # Logging functions log_info() { echo "[$(date +'%Y-%m-%d %H:%M:%S')] INFO: $*" >&2 } log_warn() { echo "[$(date +'%Y-%m-%d %H:%M:%S')] WARN: $*" >&2 } log_error() { echo "[$(date +'%Y-%m-%d %H:%M:%S')] ERROR: $*" >&2 } log_debug() { if [[ "${DEBUG:-0}" == "1" ]]; then echo "[$(date +'%Y-%m-%d %H:%M:%S')] DEBUG: $*" >&2 fi } # Usage log_info "Starting script" log_debug "Debug information" log_warn "Warning message" log_error "Error occurred" ``` ### Pattern 6: Process Orchestration with Signals ```bash #!/bin/bash set -Eeuo pipefail # Track background processes PIDS=() cleanup() { log_info "Shutting down..." # Terminate all background processes for pid in "${PIDS[@]}"; do if kill -0 "$pid" 2>/dev/null; then kill -TERM "$pid" 2>/dev/null || true fi done # Wait for graceful shutdown for pid in "${PIDS[@]}"; do wait "$pid" 2>/dev/null || true done } trap cleanup SIGTERM SIGINT # Start background tasks background_task & PIDS+=($!) another_task & PIDS+=($!) # Wait for all background processes wait ``` ### Pattern 7: Safe File Operations ```bash #!/bin/bash set -Eeuo pipefail # Use -i flag to move safely without overwriting safe_move() { local -r source="$1" local -r dest="$2" if [[ ! -e "$source" ]]; then echo "ERROR: Source does not exist: $source" >&2 return 1 fi if [[ -e "$dest" ]]; then echo "ERROR: Destination already exists: $dest" >&2 return 1 fi mv "$source" "$dest" } # Safe directory cleanup safe_rmdir() { local -r dir="$1" if [[ ! -d "$dir" ]]; then echo "ERROR: Not a directory: $dir" >&2 return 1 fi # Use -I flag to prompt before rm (BSD/GNU compatible) rm -rI -- "$dir" } # Atomic file writes atomic_write() { local -r target="$1" local -r tmpfile tmpfile=$(mktemp) || return 1 # Write to temp file first cat > "$tmpfile" # Atomic rename mv "$tmpfile" "$target" } ``` ### Pattern 8: Idempotent Script Design ```bash #!/bin/bash set -Eeuo pipefail # Check if resource already exists ensure_directory() { local -r dir="$1" if [[ -d "$dir" ]]; then log_info "Directory already exists: $dir" return 0 fi mkdir -p "$dir" || { log_error "Failed to create directory: $dir" return 1 } log_info "Created directory: $dir" } # Ensure configuration state ensure_config() { local -r config_file="$1" local -r default_value="$2" if [[ ! -f "$config_file" ]]; then echo "$default_value" > "$config_file" log_info "Created config: $config_file" fi } # Rerunning script multiple times should be safe ensure_directory "/var/cache/myapp" ensure_config "/etc/myapp/config" "DEBUG=false" ``` ### Pattern 9: Safe Command Substitution ```bash #!/bin/bash set -Eeuo pipefail # Use $() instead of backticks name=$(<"$file") # Modern, safe variable assignment from file output=$(command -v python3) # Get command location safely # Handle command substitution with error checking result=$(command -v node) || { log_error "node command not found" return 1 } # For multiple lines mapfile -t lines < <(grep "pattern" "$file") # NUL-safe iteration while IFS= read -r -d '' file; do echo "Processing: $file" done < <(find /path -type f -print0) ``` ### Pattern 10: Dry-Run Support ```bash #!/bin/bash set -Eeuo pipefail DRY_RUN="${DRY_RUN:-false}" run_cmd() { if [[ "$DRY_RUN" == "true" ]]; then echo "[DRY RUN] Would execute: $*" return 0 fi "$@" } # Usage run_cmd cp "$source" "$dest" run_cmd rm "$file" run_cmd chown "$owner" "$target" ``` ## Advanced Defensive Techniques ### Named Parameters Pattern ```bash #!/bin/bash set -Eeuo pipefail process_data() { local input_file="" local output_dir="" local format="json" # Parse named parameters while [[ $# -gt 0 ]]; do case "$1" in --input=*) input_file="${1#*=}" ;; --output=*) output_dir="${1#*=}" ;; --format=*) format="${1#*=}" ;; *) echo "ERROR: Unknown parameter: $1" >&2 return 1 ;; esac shift done # Validate required parameters [[ -n "$input_file" ]] || { echo "ERROR: --input is required" >&2; return 1; } [[ -n "$output_dir" ]] || { echo "ERROR: --output is required" >&2; return 1; } } ``` ### Dependency Checking ```bash #!/bin/bash set -Eeuo pipefail check_dependencies() { local -a missing_deps=() local -a required=("jq" "curl" "git") for cmd in "${required[@]}"; do if ! command -v "$cmd" &>/dev/null; then missing_deps+=("$cmd") fi done if [[ ${#missing_deps[@]} -gt 0 ]]; then echo "ERROR: Missing required commands: ${missing_deps[*]}" >&2 return 1 fi } check_dependencies ``` ## Best Practices Summary 1. **Always use strict mode** - `set -Eeuo pipefail` 2. **Quote all variables** - `"$variable"` prevents word splitting 3. **Use [[]] conditionals** - More robust than [ ] 4. **Implement error trapping** - Catch and handle errors gracefully 5. **Validate all inputs** - Check file existence, permissions, formats 6. **Use functions for reusability** - Prefix with meaningful names 7. **Implement structured logging** - Include timestamps and levels 8. **Support dry-run mode** - Allow users to preview changes 9. **Handle temporary files safely** - Use mktemp, cleanup with trap 10. **Design for idempotency** - Scripts should be safe to rerun 11. **Document requirements** - List dependencies and minimum versions 12. **Test error paths** - Ensure error handling works correctly 13. **Use `command -v`** - Safer than `which` for checking executables 14. **Prefer printf over echo** - More predictable across systems ## Resources - **Bash Strict Mode**: http://redsymbol.net/articles/unofficial-bash-strict-mode/ - **Google Shell Style Guide**: https://google.github.io/styleguide/shellguide.html - **Defensive BASH Programming**: https://www.lifepipe.net/
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

data-quality-frameworks

Implement data quality validation with Great Expectations, dbt

data
⭐1
# Data Quality Frameworks Production patterns for implementing data quality with Great Expectations, dbt tests, and data contracts to ensure reliable data pipelines. ## When to Use This Skill - Implementing data quality checks in pipelines - Setting up Great Expectations validation - Building comprehensive dbt test suites - Establishing data contracts between teams - Monitoring data quality metrics - Automating data validation in CI/CD ## Core Concepts ### 1. Data Quality Dimensions | Dimension | Description | Example Check | | ---------------- | ------------------------ | -------------------------------------------------- | | **Completeness** | No missing values | `expect_column_values_to_not_be_null` | | **Uniqueness** | No duplicates | `expect_column_values_to_be_unique` | | **Validity** | Values in expected range | `expect_column_values_to_be_in_set` | | **Accuracy** | Data matches reality | Cross-reference validation | | **Consistency** | No contradictions | `expect_column_pair_values_A_to_be_greater_than_B` | | **Timeliness** | Data is recent | `expect_column_max_to_be_between` | ### 2. Testing Pyramid for Data ``` /\ / \ Integration Tests (cross-table) /────\ / \ Unit Tests (single column) /────────\ / \ Schema Tests (structure) /────────────\ ``` ## Quick Start ### Great Expectations Setup ```bash # Install pip install great_expectations # Initialize project great_expectations init # Create datasource great_expectations datasource new ``` ```python # great_expectations/checkpoints/daily_validation.yml import great_expectations as gx # Create context context = gx.get_context() # Create expectation suite suite = context.add_expectation_suite("orders_suite") # Add expectations suite.add_expectation( gx.expectations.ExpectColumnValuesToNotBeNull(column="order_id") ) suite.add_expectation( gx.expectations.ExpectColumnValuesToBeUnique(column="order_id") ) # Validate results = context.run_checkpoint(checkpoint_name="daily_orders") ``` ## Patterns ### Pattern 1: Great Expectations Suite ```python # expectations/orders_suite.py import great_expectations as gx from great_expectations.core import ExpectationSuite from great_expectations.core.expectation_configuration import ExpectationConfiguration def build_orders_suite() -> ExpectationSuite: """Build comprehensive orders expectation suite""" suite = ExpectationSuite(expectation_suite_name="orders_suite") # Schema expectations suite.add_expectation(ExpectationConfiguration( expectation_type="expect_table_columns_to_match_set", kwargs={ "column_set": ["order_id", "customer_id", "amount", "status", "created_at"], "exact_match": False # Allow additional columns } )) # Primary key suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_not_be_null", kwargs={"column": "order_id"} )) suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_unique", kwargs={"column": "order_id"} )) # Foreign key suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_not_be_null", kwargs={"column": "customer_id"} )) # Categorical values suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_in_set", kwargs={ "column": "status", "value_set": ["pending", "processing", "shipped", "delivered", "cancelled"] } )) # Numeric ranges suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_between", kwargs={ "column": "amount", "min_value": 0, "max_value": 100000, "strict_min": True # amount > 0 } )) # Date validity suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_values_to_be_dateutil_parseable", kwargs={"column": "created_at"} )) # Freshness - data should be recent suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_max_to_be_between", kwargs={ "column": "created_at", "min_value": {"$PARAMETER": "now - timedelta(days=1)"}, "max_value": {"$PARAMETER": "now"} } )) # Row count sanity suite.add_expectation(ExpectationConfiguration( expectation_type="expect_table_row_count_to_be_between", kwargs={ "min_value": 1000, # Expect at least 1000 rows "max_value": 10000000 } )) # Statistical expectations suite.add_expectation(ExpectationConfiguration( expectation_type="expect_column_mean_to_be_between", kwargs={ "column": "amount", "min_value": 50, "max_value": 500 } )) return suite ``` ### Pattern 2: Great Expectations Checkpoint ```yaml # great_expectations/checkpoints/orders_checkpoint.yml name: orders_checkpoint config_version: 1.0 class_name: Checkpoint run_name_template: "%Y%m%d-%H%M%S-orders-validation" validations: - batch_request: datasource_name: warehouse data_connector_name: default_inferred_data_connector_name data_asset_name: orders data_connector_query: index: -1 # Latest batch expectation_suite_name: orders_suite action_list: - name: store_validation_result action: class_name: StoreValidationResultAction - name: store_evaluation_parameters action: class_name: StoreEvaluationParametersAction - name: update_data_docs action: class_name: UpdateDataDocsAction # Slack notification on failure - name: send_slack_notification action: class_name: SlackNotificationAction slack_webhook: ${SLACK_WEBHOOK} notify_on: failure renderer: module_name: great_expectations.render.renderer.slack_renderer class_name: SlackRenderer ``` ```python # Run checkpoint import great_expectations as gx context = gx.get_context() result = context.run_checkpoint(checkpoint_name="orders_checkpoint") if not result.success: failed_expectations = [ r for r in result.run_results.values() if not r.success ] raise ValueError(f"Data quality check failed: {failed_expectations}") ``` ### Pattern 3: dbt Data Tests ```yaml # models/marts/core/_core__models.yml version: 2 models: - name: fct_orders description: Order fact table tests: # Table-level tests - dbt_utils.recency: datepart: day field: created_at interval: 1 - dbt_utils.at_least_one - dbt_utils.expression_is_true: expression: "total_amount >= 0" columns: - name: order_id description: Primary key tests: - unique - not_null - name: customer_id description: Foreign key to dim_customers tests: - not_null - relationships: to: ref('dim_customers') field: customer_id - name: order_status tests: - accepted_values: values: ["pending", "processing", "shipped", "delivered", "cancelled"] - name: total_amount tests: - not_null - dbt_utils.expression_is_true: expression: ">= 0" - name: created_at tests: - not_null - dbt_utils.expression_is_true: expression: "<= current_timestamp" - name: dim_customers columns: - name: customer_id tests: - unique - not_null - name: email tests: - unique - not_null # Custom regex test - dbt_utils.expression_is_true: expression: "email ~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}$'" ``` ### Pattern 4: Custom dbt Tests ```sql -- tests/generic/test_row_count_in_range.sql {% test row_count_in_range(model, min_count, max_count) %} with row_count as ( select count(*) as cnt from {{ model }} ) select cnt from row_count where cnt < {{ min_count }} or cnt > {{ max_count }} {% endtest %} -- Usage in schema.yml: -- tests: -- - row_count_in_range: -- min_count: 1000 -- max_count: 10000000 ``` ```sql -- tests/generic/test_sequential_values.sql {% test sequential_values(model, column_name, interval=1) %} with lagged as ( select {{ column_name }}, lag({{ column_name }}) over (order by {{ column_name }}) as prev_value from {{ model }} ) select * from lagged where {{ column_name }} - prev_value != {{ interval }} and prev_value is not null {% endtest %} ``` ```sql -- tests/singular/assert_orders_customers_match.sql -- Singular test: specific business rule with orders_customers as ( select distinct customer_id from {{ ref('fct_orders') }} ), dim_customers as ( select customer_id from {{ ref('dim_customers') }} ), orphaned_orders as ( select o.customer_id from orders_customers o left join dim_customers c using (customer_id) where c.customer_id is null ) select * from orphaned_orders -- Test passes if this returns 0 rows ``` ### Pattern 5: Data Contracts ```yaml # contracts/orders_contract.yaml apiVersion: datacontract.com/v1.0.0 kind: DataContract metadata: name: orders version: 1.0.0 owner: data-platform-team contact: data-team@company.com info: title: Orders Data Contract description: Contract for order event data from the ecommerce platform purpose: Analytics, reporting, and ML features servers: production: type: snowflake account: company.us-east-1 database: ANALYTICS schema: CORE terms: usage: Internal analytics only limitations: PII must not be exposed in downstream marts billing: Charged per query TB scanned schema: type: object properties: order_id: type: string format: uuid description: Unique order identifier required: true unique: true pii: false customer_id: type: string format: uuid description: Customer identifier required: true pii: true piiClassification: indirect total_amount: type: number minimum: 0 maximum: 100000 description: Order total in USD created_at: type: string format: date-time description: Order creation timestamp required: true status: type: string enum: [pending, processing, shipped, delivered, cancelled] description: Current order status quality: type: SodaCL specification: checks for orders: - row_count > 0 - missing_count(order_id) = 0 - duplicate_count(order_id) = 0 - invalid_count(status) = 0: valid values: [pending, processing, shipped, delivered, cancelled] - freshness(created_at) < 24h sla: availability: 99.9% freshness: 1 hour latency: 5 minutes ``` ### Pattern 6: Automated Quality Pipeline ```python # quality_pipeline.py from dataclasses import dataclass from typing import List, Dict, Any import great_expectations as gx from datetime import datetime @dataclass class QualityResult: table: str passed: bool total_expectations: int failed_expectations: int details: List[Dict[str, Any]] timestamp: datetime class DataQualityPipeline: """Orchestrate data quality checks across tables""" def __init__(self, context: gx.DataContext): self.context = context self.results: List[QualityResult] = [] def validate_table(self, table: str, suite: str) -> QualityResult: """Validate a single table against expectation suite""" checkpoint_config = { "name": f"{table}_validation", "config_version": 1.0, "class_name": "Checkpoint", "validations": [{ "batch_request": { "datasource_name": "warehouse", "data_asset_name": table, }, "expectation_suite_name": suite, }], } result = self.context.run_checkpoint(**checkpoint_config) # Parse results validation_result = list(result.run_results.values())[0] results = validation_result.results failed = [r for r in results if not r.success] return QualityResult( table=table, passed=result.success, total_expectations=len(results), failed_expectations=len(failed), details=[{ "expectation": r.expectation_config.expectation_type, "success": r.success, "observed_value": r.result.get("observed_value"), } for r in results], timestamp=datetime.now() ) def run_all(self, tables: Dict[str, str]) -> Dict[str, QualityResult]: """Run validation for all tables""" results = {} for table, suite in tables.items(): print(f"Validating {table}...") results[table] = self.validate_table(table, suite) return results def generate_report(self, results: Dict[str, QualityResult]) -> str: """Generate quality report""" report = ["# Data Quality Report", f"Generated: {datetime.now()}", ""] total_passed = sum(1 for r in results.values() if r.passed) total_tables = len(results) report.append(f"## Summary: {total_passed}/{total_tables} tables passed") report.append("") for table, result in results.items(): status = "βœ…" if result.passed else "❌" report.append(f"### {status} {table}") report.append(f"- Expectations: {result.total_expectations}") report.append(f"- Failed: {result.failed_expectations}") if not result.passed: report.append("- Failed checks:") for detail in result.details: if not detail["success"]: report.append(f" - {detail['expectation']}: {detail['observed_value']}") report.append("") return "\n".join(report) # Usage context = gx.get_context() pipeline = DataQualityPipeline(context) tables_to_validate = { "orders": "orders_suite", "customers": "customers_suite", "products": "products_suite", } results = pipeline.run_all(tables_to_validate) report = pipeline.generate_report(results) # Fail pipeline if any table failed if not all(r.passed for r in results.values()): print(report) raise ValueError("Data quality checks failed!") ``` ## Best Practices ### Do's - **Test early** - Validate source data before transformations - **Test incrementally** - Add tests as you find issues - **Document expectations** - Clear descriptions for each test - **Alert on failures** - Integrate with monitoring - **Version contracts** - Track schema changes ### Don'ts - **Don't test everything** - Focus on critical columns - **Don't ignore warnings** - They often precede failures - **Don't skip freshness** - Stale data is bad data - **Don't hardcode thresholds** - Use dynamic baselines - **Don't test in isolation** - Test relationships too ## Resources - [Great Expectations Documentation](https://docs.greatexpectations.io/) - [dbt Testing Documentation](https://docs.getdbt.com/docs/build/tests) - [Data Contract Specification](https://datacontract.com/) - [Soda Core](https://docs.soda.io/soda-core/overview.html)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

architecture-patterns

Implement proven backend architecture patterns including Clean

coding
⭐1
# Architecture Patterns Master proven backend architecture patterns including Clean Architecture, Hexagonal Architecture, and Domain-Driven Design to build maintainable, testable, and scalable systems. ## When to Use This Skill - Designing new backend systems from scratch - Refactoring monolithic applications for better maintainability - Establishing architecture standards for your team - Migrating from tightly coupled to loosely coupled architectures - Implementing domain-driven design principles - Creating testable and mockable codebases - Planning microservices decomposition ## Core Concepts ### 1. Clean Architecture (Uncle Bob) **Layers (dependency flows inward):** - **Entities**: Core business models - **Use Cases**: Application business rules - **Interface Adapters**: Controllers, presenters, gateways - **Frameworks & Drivers**: UI, database, external services **Key Principles:** - Dependencies point inward - Inner layers know nothing about outer layers - Business logic independent of frameworks - Testable without UI, database, or external services ### 2. Hexagonal Architecture (Ports and Adapters) **Components:** - **Domain Core**: Business logic - **Ports**: Interfaces defining interactions - **Adapters**: Implementations of ports (database, REST, message queue) **Benefits:** - Swap implementations easily (mock for testing) - Technology-agnostic core - Clear separation of concerns ### 3. Domain-Driven Design (DDD) **Strategic Patterns:** - **Bounded Contexts**: Separate models for different domains - **Context Mapping**: How contexts relate - **Ubiquitous Language**: Shared terminology **Tactical Patterns:** - **Entities**: Objects with identity - **Value Objects**: Immutable objects defined by attributes - **Aggregates**: Consistency boundaries - **Repositories**: Data access abstraction - **Domain Events**: Things that happened ## Clean Architecture Pattern ### Directory Structure ``` app/ β”œβ”€β”€ domain/ # Entities & business rules β”‚ β”œβ”€β”€ entities/ β”‚ β”‚ β”œβ”€β”€ user.py β”‚ β”‚ └── order.py β”‚ β”œβ”€β”€ value_objects/ β”‚ β”‚ β”œβ”€β”€ email.py β”‚ β”‚ └── money.py β”‚ └── interfaces/ # Abstract interfaces β”‚ β”œβ”€β”€ user_repository.py β”‚ └── payment_gateway.py β”œβ”€β”€ use_cases/ # Application business rules β”‚ β”œβ”€β”€ create_user.py β”‚ β”œβ”€β”€ process_order.py β”‚ └── send_notification.py β”œβ”€β”€ adapters/ # Interface implementations β”‚ β”œβ”€β”€ repositories/ β”‚ β”‚ β”œβ”€β”€ postgres_user_repository.py β”‚ β”‚ └── redis_cache_repository.py β”‚ β”œβ”€β”€ controllers/ β”‚ β”‚ └── user_controller.py β”‚ └── gateways/ β”‚ β”œβ”€β”€ stripe_payment_gateway.py β”‚ └── sendgrid_email_gateway.py └── infrastructure/ # Framework & external concerns β”œβ”€β”€ database.py β”œβ”€β”€ config.py └── logging.py ``` ### Implementation Example ```python # domain/entities/user.py from dataclasses import dataclass from datetime import datetime from typing import Optional @dataclass class User: """Core user entity - no framework dependencies.""" id: str email: str name: str created_at: datetime is_active: bool = True def deactivate(self): """Business rule: deactivating user.""" self.is_active = False def can_place_order(self) -> bool: """Business rule: active users can order.""" return self.is_active # domain/interfaces/user_repository.py from abc import ABC, abstractmethod from typing import Optional, List from domain.entities.user import User class IUserRepository(ABC): """Port: defines contract, no implementation.""" @abstractmethod async def find_by_id(self, user_id: str) -> Optional[User]: pass @abstractmethod async def find_by_email(self, email: str) -> Optional[User]: pass @abstractmethod async def save(self, user: User) -> User: pass @abstractmethod async def delete(self, user_id: str) -> bool: pass # use_cases/create_user.py from domain.entities.user import User from domain.interfaces.user_repository import IUserRepository from dataclasses import dataclass from datetime import datetime import uuid @dataclass class CreateUserRequest: email: str name: str @dataclass class CreateUserResponse: user: User success: bool error: Optional[str] = None class CreateUserUseCase: """Use case: orchestrates business logic.""" def __init__(self, user_repository: IUserRepository): self.user_repository = user_repository async def execute(self, request: CreateUserRequest) -> CreateUserResponse: # Business validation existing = await self.user_repository.find_by_email(request.email) if existing: return CreateUserResponse( user=None, success=False, error="Email already exists" ) # Create entity user = User( id=str(uuid.uuid4()), email=request.email, name=request.name, created_at=datetime.now(), is_active=True ) # Persist saved_user = await self.user_repository.save(user) return CreateUserResponse( user=saved_user, success=True ) # adapters/repositories/postgres_user_repository.py from domain.interfaces.user_repository import IUserRepository from domain.entities.user import User from typing import Optional import asyncpg class PostgresUserRepository(IUserRepository): """Adapter: PostgreSQL implementation.""" def __init__(self, pool: asyncpg.Pool): self.pool = pool async def find_by_id(self, user_id: str) -> Optional[User]: async with self.pool.acquire() as conn: row = await conn.fetchrow( "SELECT * FROM users WHERE id = $1", user_id ) return self._to_entity(row) if row else None async def find_by_email(self, email: str) -> Optional[User]: async with self.pool.acquire() as conn: row = await conn.fetchrow( "SELECT * FROM users WHERE email = $1", email ) return self._to_entity(row) if row else None async def save(self, user: User) -> User: async with self.pool.acquire() as conn: await conn.execute( """ INSERT INTO users (id, email, name, created_at, is_active) VALUES ($1, $2, $3, $4, $5) ON CONFLICT (id) DO UPDATE SET email = $2, name = $3, is_active = $5 """, user.id, user.email, user.name, user.created_at, user.is_active ) return user async def delete(self, user_id: str) -> bool: async with self.pool.acquire() as conn: result = await conn.execute( "DELETE FROM users WHERE id = $1", user_id ) return result == "DELETE 1" def _to_entity(self, row) -> User: """Map database row to entity.""" return User( id=row["id"], email=row["email"], name=row["name"], created_at=row["created_at"], is_active=row["is_active"] ) # adapters/controllers/user_controller.py from fastapi import APIRouter, Depends, HTTPException from use_cases.create_user import CreateUserUseCase, CreateUserRequest from pydantic import BaseModel router = APIRouter() class CreateUserDTO(BaseModel): email: str name: str @router.post("/users") async def create_user( dto: CreateUserDTO, use_case: CreateUserUseCase = Depends(get_create_user_use_case) ): """Controller: handles HTTP concerns only.""" request = CreateUserRequest(email=dto.email, name=dto.name) response = await use_case.execute(request) if not response.success: raise HTTPException(status_code=400, detail=response.error) return {"user": response.user} ``` ## Hexagonal Architecture Pattern ```python # Core domain (hexagon center) class OrderService: """Domain service - no infrastructure dependencies.""" def __init__( self, order_repository: OrderRepositoryPort, payment_gateway: PaymentGatewayPort, notification_service: NotificationPort ): self.orders = order_repository self.payments = payment_gateway self.notifications = notification_service async def place_order(self, order: Order) -> OrderResult: # Business logic if not order.is_valid(): return OrderResult(success=False, error="Invalid order") # Use ports (interfaces) payment = await self.payments.charge( amount=order.total, customer=order.customer_id ) if not payment.success: return OrderResult(success=False, error="Payment failed") order.mark_as_paid() saved_order = await self.orders.save(order) await self.notifications.send( to=order.customer_email, subject="Order confirmed", body=f"Order {order.id} confirmed" ) return OrderResult(success=True, order=saved_order) # Ports (interfaces) class OrderRepositoryPort(ABC): @abstractmethod async def save(self, order: Order) -> Order: pass class PaymentGatewayPort(ABC): @abstractmethod async def charge(self, amount: Money, customer: str) -> PaymentResult: pass class NotificationPort(ABC): @abstractmethod async def send(self, to: str, subject: str, body: str): pass # Adapters (implementations) class StripePaymentAdapter(PaymentGatewayPort): """Primary adapter: connects to Stripe API.""" def __init__(self, api_key: str): self.stripe = stripe self.stripe.api_key = api_key async def charge(self, amount: Money, customer: str) -> PaymentResult: try: charge = self.stripe.Charge.create( amount=amount.cents, currency=amount.currency, customer=customer ) return PaymentResult(success=True, transaction_id=charge.id) except stripe.error.CardError as e: return PaymentResult(success=False, error=str(e)) class MockPaymentAdapter(PaymentGatewayPort): """Test adapter: no external dependencies.""" async def charge(self, amount: Money, customer: str) -> PaymentResult: return PaymentResult(success=True, transaction_id="mock-123") ``` ## Domain-Driven Design Pattern ```python # Value Objects (immutable) from dataclasses import dataclass from typing import Optional @dataclass(frozen=True) class Email: """Value object: validated email.""" value: str def __post_init__(self): if "@" not in self.value: raise ValueError("Invalid email") @dataclass(frozen=True) class Money: """Value object: amount with currency.""" amount: int # cents currency: str def add(self, other: "Money") -> "Money": if self.currency != other.currency: raise ValueError("Currency mismatch") return Money(self.amount + other.amount, self.currency) # Entities (with identity) class Order: """Entity: has identity, mutable state.""" def __init__(self, id: str, customer: Customer): self.id = id self.customer = customer self.items: List[OrderItem] = [] self.status = OrderStatus.PENDING self._events: List[DomainEvent] = [] def add_item(self, product: Product, quantity: int): """Business logic in entity.""" item = OrderItem(product, quantity) self.items.append(item) self._events.append(ItemAddedEvent(self.id, item)) def total(self) -> Money: """Calculated property.""" return sum(item.subtotal() for item in self.items) def submit(self): """State transition with business rules.""" if not self.items: raise ValueError("Cannot submit empty order") if self.status != OrderStatus.PENDING: raise ValueError("Order already submitted") self.status = OrderStatus.SUBMITTED self._events.append(OrderSubmittedEvent(self.id)) # Aggregates (consistency boundary) class Customer: """Aggregate root: controls access to entities.""" def __init__(self, id: str, email: Email): self.id = id self.email = email self._addresses: List[Address] = [] self._orders: List[str] = [] # Order IDs, not full objects def add_address(self, address: Address): """Aggregate enforces invariants.""" if len(self._addresses) >= 5: raise ValueError("Maximum 5 addresses allowed") self._addresses.append(address) @property def primary_address(self) -> Optional[Address]: return next((a for a in self._addresses if a.is_primary), None) # Domain Events @dataclass class OrderSubmittedEvent: order_id: str occurred_at: datetime = field(default_factory=datetime.now) # Repository (aggregate persistence) class OrderRepository: """Repository: persist/retrieve aggregates.""" async def find_by_id(self, order_id: str) -> Optional[Order]: """Reconstitute aggregate from storage.""" pass async def save(self, order: Order): """Persist aggregate and publish events.""" await self._persist(order) await self._publish_events(order._events) order._events.clear() ``` ## Resources - **references/clean-architecture-guide.md**: Detailed layer breakdown - **references/hexagonal-architecture-guide.md**: Ports and adapters patterns - **references/ddd-tactical-patterns.md**: Entities, value objects, aggregates - **assets/clean-architecture-template/**: Complete project structure - **assets/ddd-examples/**: Domain modeling examples ## Best Practices 1. **Dependency Rule**: Dependencies always point inward 2. **Interface Segregation**: Small, focused interfaces 3. **Business Logic in Domain**: Keep frameworks out of core 4. **Test Independence**: Core testable without infrastructure 5. **Bounded Contexts**: Clear domain boundaries 6. **Ubiquitous Language**: Consistent terminology 7. **Thin Controllers**: Delegate to use cases 8. **Rich Domain Models**: Behavior with data ## Common Pitfalls - **Anemic Domain**: Entities with only data, no behavior - **Framework Coupling**: Business logic depends on frameworks - **Fat Controllers**: Business logic in controllers - **Repository Leakage**: Exposing ORM objects - **Missing Abstractions**: Concrete dependencies in core - **Over-Engineering**: Clean architecture for simple CRUD
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

microservices-patterns

Design microservices architectures with service boundaries,

coding
⭐1
# Microservices Patterns Master microservices architecture patterns including service boundaries, inter-service communication, data management, and resilience patterns for building distributed systems. ## When to Use This Skill - Decomposing monoliths into microservices - Designing service boundaries and contracts - Implementing inter-service communication - Managing distributed data and transactions - Building resilient distributed systems - Implementing service discovery and load balancing - Designing event-driven architectures ## Core Concepts ### 1. Service Decomposition Strategies **By Business Capability** - Organize services around business functions - Each service owns its domain - Example: OrderService, PaymentService, InventoryService **By Subdomain (DDD)** - Core domain, supporting subdomains - Bounded contexts map to services - Clear ownership and responsibility **Strangler Fig Pattern** - Gradually extract from monolith - New functionality as microservices - Proxy routes to old/new systems ### 2. Communication Patterns **Synchronous (Request/Response)** - REST APIs - gRPC - GraphQL **Asynchronous (Events/Messages)** - Event streaming (Kafka) - Message queues (RabbitMQ, SQS) - Pub/Sub patterns ### 3. Data Management **Database Per Service** - Each service owns its data - No shared databases - Loose coupling **Saga Pattern** - Distributed transactions - Compensating actions - Eventual consistency ### 4. Resilience Patterns **Circuit Breaker** - Fail fast on repeated errors - Prevent cascade failures **Retry with Backoff** - Transient fault handling - Exponential backoff **Bulkhead** - Isolate resources - Limit impact of failures ## Service Decomposition Patterns ### Pattern 1: By Business Capability ```python # E-commerce example # Order Service class OrderService: """Handles order lifecycle.""" async def create_order(self, order_data: dict) -> Order: order = Order.create(order_data) # Publish event for other services await self.event_bus.publish( OrderCreatedEvent( order_id=order.id, customer_id=order.customer_id, items=order.items, total=order.total ) ) return order # Payment Service (separate service) class PaymentService: """Handles payment processing.""" async def process_payment(self, payment_request: PaymentRequest) -> PaymentResult: # Process payment result = await self.payment_gateway.charge( amount=payment_request.amount, customer=payment_request.customer_id ) if result.success: await self.event_bus.publish( PaymentCompletedEvent( order_id=payment_request.order_id, transaction_id=result.transaction_id ) ) return result # Inventory Service (separate service) class InventoryService: """Handles inventory management.""" async def reserve_items(self, order_id: str, items: List[OrderItem]) -> ReservationResult: # Check availability for item in items: available = await self.inventory_repo.get_available(item.product_id) if available < item.quantity: return ReservationResult( success=False, error=f"Insufficient inventory for {item.product_id}" ) # Reserve items reservation = await self.create_reservation(order_id, items) await self.event_bus.publish( InventoryReservedEvent( order_id=order_id, reservation_id=reservation.id ) ) return ReservationResult(success=True, reservation=reservation) ``` ### Pattern 2: API Gateway ```python from fastapi import FastAPI, HTTPException, Depends import httpx from circuitbreaker import circuit app = FastAPI() class APIGateway: """Central entry point for all client requests.""" def __init__(self): self.order_service_url = "http://order-service:8000" self.payment_service_url = "http://payment-service:8001" self.inventory_service_url = "http://inventory-service:8002" self.http_client = httpx.AsyncClient(timeout=5.0) @circuit(failure_threshold=5, recovery_timeout=30) async def call_order_service(self, path: str, method: str = "GET", **kwargs): """Call order service with circuit breaker.""" response = await self.http_client.request( method, f"{self.order_service_url}{path}", **kwargs ) response.raise_for_status() return response.json() async def create_order_aggregate(self, order_id: str) -> dict: """Aggregate data from multiple services.""" # Parallel requests order, payment, inventory = await asyncio.gather( self.call_order_service(f"/orders/{order_id}"), self.call_payment_service(f"/payments/order/{order_id}"), self.call_inventory_service(f"/reservations/order/{order_id}"), return_exceptions=True ) # Handle partial failures result = {"order": order} if not isinstance(payment, Exception): result["payment"] = payment if not isinstance(inventory, Exception): result["inventory"] = inventory return result @app.post("/api/orders") async def create_order( order_data: dict, gateway: APIGateway = Depends() ): """API Gateway endpoint.""" try: # Route to order service order = await gateway.call_order_service( "/orders", method="POST", json=order_data ) return {"order": order} except httpx.HTTPError as e: raise HTTPException(status_code=503, detail="Order service unavailable") ``` ## Communication Patterns ### Pattern 1: Synchronous REST Communication ```python # Service A calls Service B import httpx from tenacity import retry, stop_after_attempt, wait_exponential class ServiceClient: """HTTP client with retries and timeout.""" def __init__(self, base_url: str): self.base_url = base_url self.client = httpx.AsyncClient( timeout=httpx.Timeout(5.0, connect=2.0), limits=httpx.Limits(max_keepalive_connections=20) ) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10) ) async def get(self, path: str, **kwargs): """GET with automatic retries.""" response = await self.client.get(f"{self.base_url}{path}", **kwargs) response.raise_for_status() return response.json() async def post(self, path: str, **kwargs): """POST request.""" response = await self.client.post(f"{self.base_url}{path}", **kwargs) response.raise_for_status() return response.json() # Usage payment_client = ServiceClient("http://payment-service:8001") result = await payment_client.post("/payments", json=payment_data) ``` ### Pattern 2: Asynchronous Event-Driven ```python # Event-driven communication with Kafka from aiokafka import AIOKafkaProducer, AIOKafkaConsumer import json from dataclasses import dataclass, asdict from datetime import datetime @dataclass class DomainEvent: event_id: str event_type: str aggregate_id: str occurred_at: datetime data: dict class EventBus: """Event publishing and subscription.""" def __init__(self, bootstrap_servers: List[str]): self.bootstrap_servers = bootstrap_servers self.producer = None async def start(self): self.producer = AIOKafkaProducer( bootstrap_servers=self.bootstrap_servers, value_serializer=lambda v: json.dumps(v).encode() ) await self.producer.start() async def publish(self, event: DomainEvent): """Publish event to Kafka topic.""" topic = event.event_type await self.producer.send_and_wait( topic, value=asdict(event), key=event.aggregate_id.encode() ) async def subscribe(self, topic: str, handler: callable): """Subscribe to events.""" consumer = AIOKafkaConsumer( topic, bootstrap_servers=self.bootstrap_servers, value_deserializer=lambda v: json.loads(v.decode()), group_id="my-service" ) await consumer.start() try: async for message in consumer: event_data = message.value await handler(event_data) finally: await consumer.stop() # Order Service publishes event async def create_order(order_data: dict): order = await save_order(order_data) event = DomainEvent( event_id=str(uuid.uuid4()), event_type="OrderCreated", aggregate_id=order.id, occurred_at=datetime.now(), data={ "order_id": order.id, "customer_id": order.customer_id, "total": order.total } ) await event_bus.publish(event) # Inventory Service listens for OrderCreated async def handle_order_created(event_data: dict): """React to order creation.""" order_id = event_data["data"]["order_id"] items = event_data["data"]["items"] # Reserve inventory await reserve_inventory(order_id, items) ``` ### Pattern 3: Saga Pattern (Distributed Transactions) ```python # Saga orchestration for order fulfillment from enum import Enum from typing import List, Callable class SagaStep: """Single step in saga.""" def __init__( self, name: str, action: Callable, compensation: Callable ): self.name = name self.action = action self.compensation = compensation class SagaStatus(Enum): PENDING = "pending" COMPLETED = "completed" COMPENSATING = "compensating" FAILED = "failed" class OrderFulfillmentSaga: """Orchestrated saga for order fulfillment.""" def __init__(self): self.steps: List[SagaStep] = [ SagaStep( "create_order", action=self.create_order, compensation=self.cancel_order ), SagaStep( "reserve_inventory", action=self.reserve_inventory, compensation=self.release_inventory ), SagaStep( "process_payment", action=self.process_payment, compensation=self.refund_payment ), SagaStep( "confirm_order", action=self.confirm_order, compensation=self.cancel_order_confirmation ) ] async def execute(self, order_data: dict) -> SagaResult: """Execute saga steps.""" completed_steps = [] context = {"order_data": order_data} try: for step in self.steps: # Execute step result = await step.action(context) if not result.success: # Compensate await self.compensate(completed_steps, context) return SagaResult( status=SagaStatus.FAILED, error=result.error ) completed_steps.append(step) context.update(result.data) return SagaResult(status=SagaStatus.COMPLETED, data=context) except Exception as e: # Compensate on error await self.compensate(completed_steps, context) return SagaResult(status=SagaStatus.FAILED, error=str(e)) async def compensate(self, completed_steps: List[SagaStep], context: dict): """Execute compensating actions in reverse order.""" for step in reversed(completed_steps): try: await step.compensation(context) except Exception as e: # Log compensation failure print(f"Compensation failed for {step.name}: {e}") # Step implementations async def create_order(self, context: dict) -> StepResult: order = await order_service.create(context["order_data"]) return StepResult(success=True, data={"order_id": order.id}) async def cancel_order(self, context: dict): await order_service.cancel(context["order_id"]) async def reserve_inventory(self, context: dict) -> StepResult: result = await inventory_service.reserve( context["order_id"], context["order_data"]["items"] ) return StepResult( success=result.success, data={"reservation_id": result.reservation_id} ) async def release_inventory(self, context: dict): await inventory_service.release(context["reservation_id"]) async def process_payment(self, context: dict) -> StepResult: result = await payment_service.charge( context["order_id"], context["order_data"]["total"] ) return StepResult( success=result.success, data={"transaction_id": result.transaction_id}, error=result.error ) async def refund_payment(self, context: dict): await payment_service.refund(context["transaction_id"]) ``` ## Resilience Patterns ### Circuit Breaker Pattern ```python from enum import Enum from datetime import datetime, timedelta from typing import Callable, Any class CircuitState(Enum): CLOSED = "closed" # Normal operation OPEN = "open" # Failing, reject requests HALF_OPEN = "half_open" # Testing if recovered class CircuitBreaker: """Circuit breaker for service calls.""" def __init__( self, failure_threshold: int = 5, recovery_timeout: int = 30, success_threshold: int = 2 ): self.failure_threshold = failure_threshold self.recovery_timeout = recovery_timeout self.success_threshold = success_threshold self.failure_count = 0 self.success_count = 0 self.state = CircuitState.CLOSED self.opened_at = None async def call(self, func: Callable, *args, **kwargs) -> Any: """Execute function with circuit breaker.""" if self.state == CircuitState.OPEN: if self._should_attempt_reset(): self.state = CircuitState.HALF_OPEN else: raise CircuitBreakerOpenError("Circuit breaker is open") try: result = await func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() raise def _on_success(self): """Handle successful call.""" self.failure_count = 0 if self.state == CircuitState.HALF_OPEN: self.success_count += 1 if self.success_count >= self.success_threshold: self.state = CircuitState.CLOSED self.success_count = 0 def _on_failure(self): """Handle failed call.""" self.failure_count += 1 if self.failure_count >= self.failure_threshold: self.state = CircuitState.OPEN self.opened_at = datetime.now() if self.state == CircuitState.HALF_OPEN: self.state = CircuitState.OPEN self.opened_at = datetime.now() def _should_attempt_reset(self) -> bool: """Check if enough time passed to try again.""" return ( datetime.now() - self.opened_at > timedelta(seconds=self.recovery_timeout) ) # Usage breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30) async def call_payment_service(payment_data: dict): return await breaker.call( payment_client.process_payment, payment_data ) ``` ## Resources - **references/service-decomposition-guide.md**: Breaking down monoliths - **references/communication-patterns.md**: Sync vs async patterns - **references/saga-implementation.md**: Distributed transactions - **assets/circuit-breaker.py**: Production circuit breaker - **assets/event-bus-template.py**: Kafka event bus implementation - **assets/api-gateway-template.py**: Complete API gateway ## Best Practices 1. **Service Boundaries**: Align with business capabilities 2. **Database Per Service**: No shared databases 3. **API Contracts**: Versioned, backward compatible 4. **Async When Possible**: Events over direct calls 5. **Circuit Breakers**: Fail fast on service failures 6. **Distributed Tracing**: Track requests across services 7. **Service Registry**: Dynamic service discovery 8. **Health Checks**: Liveness and readiness probes ## Common Pitfalls - **Distributed Monolith**: Tightly coupled services - **Chatty Services**: Too many inter-service calls - **Shared Databases**: Tight coupling through data - **No Circuit Breakers**: Cascade failures - **Synchronous Everything**: Tight coupling, poor resilience - **Premature Microservices**: Starting with microservices - **Ignoring Network Failures**: Assuming reliable network - **No Compensation Logic**: Can't undo failed transactions
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered