Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

30 NODES
πŸ€–system promptβ€’6 months ago

Cross-CLI MCP Config Sync

Keep Claude, Cursor, Gemini, and related CLI integrations aligned with a repeatable dry-run and apply workflow.

productivity
⭐1
# Cross-CLI MCP Config Sync Imported from curated first-party documentation sources. ## What this covers Use this skill when multiple AI clients need the same MCP configuration without drifting out of sync. ## Use this when - Rolling out a shared MCP config to multiple clients - Previewing config changes before applying them - Standardizing developer setup across tools ## Expected outcomes - Cross-client MCP setup becomes easier to repeat - Dry-run and apply modes reduce accidental changes - Environment-specific details stay documented near the workflow ## Source synthesis - EVOKORE-MCP/docs/CLI_INTEGRATION.md (https://github.com/mattmre/EVOKORE-MCP/blob/main/docs/CLI_INTEGRATION.md) ## Dedupe notes Uses the dedicated CLI integration guide as the canonical source for config sync instead of duplicating setup notes elsewhere. ## Source excerpts ### EVOKORE-MCP/docs/CLI_INTEGRATION.md EVOKORE-MCP isn't just an MCP Server-Ò€—it also ships with natively integrated UI hooks designed to make your AI CLI experience (like Gemini CLI or Claude Code) significantly more powerful and transparent. ## 🍨 The Interactive Status Line When you connect EVOKORE-MCP to your AI Assistant, you can optionally enable the **EVOKORE Status Line**. Every time the AI finishes a thought or a tool execution, this hook intercepts the internal JSON payload and renders a beautiful, color-coded ASCII status bar at the bottom of your terminal showing: - **Location**: Your current working directory. - **Model Identity**: The exact LLM model currently loaded. - **Skill Count**: A live count of the MCP Agent Skills currently indexed in your library. - **Context Window Health**: A dynamic, color-coded progress bar showing exactly how many tokens you have consumed. --- ### 💜 Enabling in Gemini CLI Gemini CLI features a robust native hook engine. You can configure it to execute the EVOKORE Status Line immediately after every model response (`AfterModel`). **Step 1:** Locate your global settings file (`~/.gemini/settings.json`). **Step 2:** Ensure hooks are enabled, and add the `AfterModel` event array to the root of the JSON object: ```json { "enableHooks": true, "hooks": { "AfterModel": [ { "type": "command", "command": "node /absolute/path/to/EVOKORE-MCP/scripts/status.js" } ] } } ``` **Step 3:** Restart your Gemini CLI! --- ### 💜 Enabling in Claude Code Claude Code features an undocumented internal hook architecture that natively supports this status line. *(Note: Because this feature is currently undocumented by Anthropic, Claude Code's `doctor` command will display "Found 1 settings issue". This is perfectly normal and the status line will still execute successfully)*. **Step 1:** Locate your Claude settings file (`~/.claude/settings.json`). **Step 2:** Add the `statusLine` block to the root of the JSON object: ```json { "statusLine": { "type": "command", "command": "node /absolute/path/to/EVOKORE-MCP/scripts/status.js" } } ``` **Step 3:** Restart Claude Code. --- ### Òő ï¸ A Note on GitHub Copilot and Codex Microsoft's GitHub Copilot CLI and OpenAI's Codex CLI **do not natively support** these JSON hook configurations. If you want the EVOKORE Status Line to appear after commands in these tools, you must configure a native PowerShell/Bash alias wrapper around the CLI execution. **Example (PowerShell Profile):** ```powershell function copilot-evokore { gh copilot $args node "/absolute/path/to/EV ...
πŸ‘0
πŸ‘οΈ0
docs
πŸ“textβ€’6 months ago

Release Command Center Workflow

Coordinate release freezes, candidate promotion, validation, publication, and rollback handling through a defined state model.

devops
⭐1
# Release Command Center Workflow Imported from curated first-party documentation sources. ## What this covers Use this workflow to move a release from planning to publication with explicit validation checkpoints and rollback branches. ## Use this when - Managing release freezes and candidate promotion - Tracking validation state before publication - Documenting rollback-friendly release operations ## Expected outcomes - Release state changes are visible and repeatable - Rollback paths are treated as first-class outcomes - Validation discipline is preserved between phases ## Source synthesis - AGENT33/docs/functionality-and-workflows.md (https://github.com/mattmre/AGENT33/blob/main/docs/functionality-and-workflows.md) - AGENT33/docs/use-cases.md (https://github.com/mattmre/AGENT33/blob/main/docs/use-cases.md) ## Dedupe notes Merges AGENT33 release lifecycle details with release-oriented use-case guidance. ## Source excerpts ### AGENT33/docs/functionality-and-workflows.md ### 4.2 Release Lifecycle States: - `planned -> frozen -> rc -> validating -> released` - Failure/rollback branches: `failed`, `rolled_back` Main APIs: - `/v1/releases/{id}/freeze` - `/v1/releases/{id}/rc` - `/v1/releases/{id}/validate` - `/v1/releases/{id}/publish` - `/v1/releases/{id}/rollback` ### AGENT33/docs/use-cases.md ## 2. Release Command Center Goal: - Move releases through frozen -> RC -> validate -> publish with checklist controls. Use these modules: - `api/routes/releases.py` - `release/service.py` - `release/checklist.py` - `release/sync.py` - `release/rollback.py` Typical flow: 1. Create release (`POST /v1/releases`). 2. Freeze, cut RC, validate. 3. Update/evaluate checklist items. 4. Publish when checks pass. 5. Run sync dry-runs or real syncs. 6. Initiate rollback if needed. Best fit: - Teams with repeatable release compliance requirements.
πŸ‘0
πŸ‘οΈ0
docs
πŸ€–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
πŸ€–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

context-driven-development

Creates and maintains project context artifacts (product.md,

coding
⭐1
# Context-Driven Development Guide for implementing and maintaining context as a managed artifact alongside code, enabling consistent AI interactions and team alignment through structured project documentation. ## When to Use This Skill - Setting up new projects with Conductor - Understanding the relationship between context artifacts - Maintaining consistency across AI-assisted development sessions - Onboarding team members to an existing Conductor project - Deciding when to update context documents - Managing greenfield vs brownfield project contexts ## Core Philosophy Context-Driven Development treats project context as a first-class artifact managed alongside code. Instead of relying on ad-hoc prompts or scattered documentation, establish a persistent, structured foundation that informs all AI interactions. Key principles: 1. **Context precedes code**: Define what you're building and how before implementation 2. **Living documentation**: Context artifacts evolve with the project 3. **Single source of truth**: One canonical location for each type of information 4. **AI alignment**: Consistent context produces consistent AI behavior ## The Workflow Follow the **Context β†’ Spec & Plan β†’ Implement** workflow: 1. **Context Phase**: Establish or verify project context artifacts exist and are current 2. **Specification Phase**: Define requirements and acceptance criteria for work units 3. **Planning Phase**: Break specifications into phased, actionable tasks 4. **Implementation Phase**: Execute tasks following established workflow patterns ## Artifact Relationships ### product.md - Defines WHAT and WHY Purpose: Captures product vision, goals, target users, and business context. Contents: - Product name and one-line description - Problem statement and solution approach - Target user personas - Core features and capabilities - Success metrics and KPIs - Product roadmap (high-level) Update when: - Product vision or goals change - New major features are planned - Target audience shifts - Business priorities evolve ### product-guidelines.md - Defines HOW to Communicate Purpose: Establishes brand voice, messaging standards, and communication patterns. Contents: - Brand voice and tone guidelines - Terminology and glossary - Error message conventions - User-facing copy standards - Documentation style Update when: - Brand guidelines change - New terminology is introduced - Communication patterns need refinement ### tech-stack.md - Defines WITH WHAT Purpose: Documents technology choices, dependencies, and architectural decisions. Contents: - Primary languages and frameworks - Key dependencies with versions - Infrastructure and deployment targets - Development tools and environment - Testing frameworks - Code quality tools Update when: - Adding new dependencies - Upgrading major versions - Changing infrastructure - Adopting new tools or patterns ### workflow.md - Defines HOW to Work Purpose: Establishes development practices, quality gates, and team workflows. Contents: - Development methodology (TDD, etc.) - Git workflow and commit conventions - Code review requirements - Testing requirements and coverage targets - Quality assurance gates - Deployment procedures Update when: - Team practices evolve - Quality standards change - New workflow patterns are adopted ### tracks.md - Tracks WHAT'S HAPPENING Purpose: Registry of all work units with status and metadata. Contents: - Active tracks with current status - Completed tracks with completion dates - Track metadata (type, priority, assignee) - Links to individual track directories Update when: - New tracks are created - Track status changes - Tracks are completed or archived See [references/artifact-templates.md](references/artifact-templates.md) for copy-paste starter templates. ## Context Maintenance Principles ### Keep Artifacts Synchronized Ensure changes in one artifact reflect in related documents: - New feature in product.md β†’ Update tech-stack.md if new dependencies needed - Completed track β†’ Update product.md to reflect new capabilities - Workflow change β†’ Update all affected track plans ### Update tech-stack.md When Adding Dependencies Before adding any new dependency: 1. Check if existing dependencies solve the need 2. Document the rationale for new dependencies 3. Add version constraints 4. Note any configuration requirements ### Update product.md When Features Complete After completing a feature track: 1. Move feature from "planned" to "implemented" in product.md 2. Update any affected success metrics 3. Document any scope changes from original plan ### Verify Context Before Implementation Before starting any track: 1. Read all context artifacts 2. Flag any outdated information 3. Propose updates before proceeding 4. Confirm context accuracy with stakeholders ## Greenfield vs Brownfield Handling ### Greenfield Projects (New) For new projects: 1. Run `/conductor:setup` to create all artifacts interactively 2. Answer questions about product vision, tech preferences, and workflow 3. Generate initial style guides for chosen languages 4. Create empty tracks registry Characteristics: - Full control over context structure - Define standards before code exists - Establish patterns early ### Brownfield Projects (Existing) For existing codebases: 1. Run `/conductor:setup` with existing codebase detection 2. System analyzes existing code, configs, and documentation 3. Pre-populate artifacts based on discovered patterns 4. Review and refine generated context Characteristics: - Extract implicit context from existing code - Reconcile existing patterns with desired patterns - Document technical debt and modernization plans - Preserve working patterns while establishing standards ## Benefits ### Team Alignment - New team members onboard faster with explicit context - Consistent terminology and conventions across the team - Shared understanding of product goals and technical decisions ### AI Consistency - AI assistants produce aligned outputs across sessions - Reduced need to re-explain context in each interaction - Predictable behavior based on documented standards ### Institutional Memory - Decisions and rationale are preserved - Context survives team changes - Historical context informs future decisions ### Quality Assurance - Standards are explicit and verifiable - Deviations from context are detectable - Quality gates are documented and enforceable ## Directory Structure ``` conductor/ β”œβ”€β”€ index.md # Navigation hub linking all artifacts β”œβ”€β”€ product.md # Product vision and goals β”œβ”€β”€ product-guidelines.md # Communication standards β”œβ”€β”€ tech-stack.md # Technology preferences β”œβ”€β”€ workflow.md # Development practices β”œβ”€β”€ tracks.md # Work unit registry β”œβ”€β”€ setup_state.json # Resumable setup state β”œβ”€β”€ code_styleguides/ # Language-specific conventions β”‚ β”œβ”€β”€ python.md β”‚ β”œβ”€β”€ typescript.md β”‚ └── ... └── tracks/ └── <track-id>/ β”œβ”€β”€ spec.md β”œβ”€β”€ plan.md β”œβ”€β”€ metadata.json └── index.md ``` ## Context Lifecycle 1. **Creation**: Initial setup via `/conductor:setup` 2. **Validation**: Verify before each track 3. **Evolution**: Update as project grows 4. **Synchronization**: Keep artifacts aligned 5. **Archival**: Document historical decisions ## Context Validation Checklist Before starting implementation on any track, validate context: ### Product Context - [ ] product.md reflects current product vision - [ ] Target users are accurately described - [ ] Feature list is up to date - [ ] Success metrics are defined ### Technical Context - [ ] tech-stack.md lists all current dependencies - [ ] Version numbers are accurate - [ ] Infrastructure targets are correct - [ ] Development tools are documented ### Workflow Context - [ ] workflow.md describes current practices - [ ] Quality gates are defined - [ ] Coverage targets are specified - [ ] Commit conventions are documented ### Track Context - [ ] tracks.md shows all active work - [ ] No stale or abandoned tracks - [ ] Dependencies between tracks are noted ## Common Anti-Patterns Avoid these context management mistakes: ### Stale Context Problem: Context documents become outdated and misleading. Solution: Update context as part of each track's completion process. ### Context Sprawl Problem: Information scattered across multiple locations. Solution: Use the defined artifact structure; resist creating new document types. ### Implicit Context Problem: Relying on knowledge not captured in artifacts. Solution: If you reference something repeatedly, add it to the appropriate artifact. ### Context Hoarding Problem: One person maintains context without team input. Solution: Review context artifacts in pull requests; make updates collaborative. ### Over-Specification Problem: Context becomes so detailed it's impossible to maintain. Solution: Keep artifacts focused on decisions that affect AI behavior and team alignment. ## Integration with Development Tools ### IDE Integration Configure your IDE to display context files prominently: - Pin conductor/product.md for quick reference - Add tech-stack.md to project notes - Create snippets for common patterns from style guides ### Git Hooks Consider pre-commit hooks that: - Warn when dependencies change without tech-stack.md update - Remind to update product.md when feature branches merge - Validate context artifact syntax ### CI/CD Integration Include context validation in pipelines: - Check tech-stack.md matches actual dependencies - Verify links in context documents resolve - Ensure tracks.md status matches git branch state ## Session Continuity Conductor supports multi-session development through context persistence: ### Starting a New Session 1. Read index.md to orient yourself 2. Check tracks.md for active work 3. Review relevant track's plan.md for current task 4. Verify context artifacts are current ### Ending a Session 1. Update plan.md with current progress 2. Note any blockers or decisions made 3. Commit in-progress work with clear status 4. Update tracks.md if status changed ### Handling Interruptions If interrupted mid-task: 1. Mark task as `[~]` with note about stopping point 2. Commit work-in-progress to feature branch 3. Document any uncommitted decisions in plan.md ## Best Practices 1. **Read context first**: Always read relevant artifacts before starting work 2. **Small updates**: Make incremental context changes, not massive rewrites 3. **Link decisions**: Reference context when making implementation choices 4. **Version context**: Commit context changes alongside code changes 5. **Review context**: Include context artifact reviews in code reviews 6. **Validate regularly**: Run context validation checklist before major work 7. **Communicate changes**: Notify team when context artifacts change significantly 8. **Preserve history**: Use git to track context evolution over time 9. **Question staleness**: If context feels wrong, investigate and update 10. **Keep it actionable**: Every context item should inform a decision or behavior
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

track-management

Use this skill when creating, managing, or working with Conductor

coding
⭐1
# Track Management Guide for creating, managing, and completing Conductor tracks - the logical work units that organize features, bugs, and refactors through specification, planning, and implementation phases. ## When to Use This Skill - Creating new feature, bug, or refactor tracks - Writing or reviewing spec.md files - Creating or updating plan.md files - Managing track lifecycle from creation to completion - Understanding track status markers and conventions - Working with the tracks.md registry - Interpreting or updating track metadata ## Track Concept A track is a logical work unit that encapsulates a complete piece of work. Each track has: - A unique identifier - A specification defining requirements - A phased plan breaking work into tasks - Metadata tracking status and progress Tracks provide semantic organization for work, enabling: - Clear scope boundaries - Progress tracking - Git-aware operations (revert by track) - Team coordination ## Track Types ### feature New functionality or capabilities. Use for: - New user-facing features - New API endpoints - New integrations - Significant enhancements ### bug Defect fixes. Use for: - Incorrect behavior - Error conditions - Performance regressions - Security vulnerabilities ### chore Maintenance and housekeeping. Use for: - Dependency updates - Configuration changes - Documentation updates - Cleanup tasks ### refactor Code improvement without behavior change. Use for: - Code restructuring - Pattern adoption - Technical debt reduction - Performance optimization (same behavior, better performance) ## Track ID Format Track IDs follow the pattern: `{shortname}_{YYYYMMDD}` - **shortname**: 2-4 word kebab-case description (e.g., `user-auth`, `api-rate-limit`) - **YYYYMMDD**: Creation date in ISO format Examples: - `user-auth_20250115` - `fix-login-error_20250115` - `upgrade-deps_20250115` - `refactor-api-client_20250115` ## Track Lifecycle ### 1. Creation (newTrack) **Define Requirements** 1. Gather requirements through interactive Q&A 2. Identify acceptance criteria 3. Determine scope boundaries 4. Identify dependencies **Generate Specification** 1. Create `spec.md` with structured requirements 2. Document functional and non-functional requirements 3. Define acceptance criteria 4. List dependencies and constraints **Generate Plan** 1. Create `plan.md` with phased task breakdown 2. Organize tasks into logical phases 3. Add verification tasks after phases 4. Estimate effort and complexity **Register Track** 1. Add entry to `tracks.md` registry 2. Create track directory structure 3. Generate `metadata.json` 4. Create track `index.md` ### 2. Implementation **Execute Tasks** 1. Select next pending task from plan 2. Mark task as in-progress 3. Implement following workflow (TDD) 4. Mark task complete with commit SHA **Update Status** 1. Update task markers in plan.md 2. Record commit SHAs for traceability 3. Update phase progress 4. Update track status in tracks.md **Verify Progress** 1. Complete verification tasks 2. Wait for checkpoint approval 3. Record checkpoint commits ### 3. Completion **Sync Documentation** 1. Update product.md if features added 2. Update tech-stack.md if dependencies changed 3. Verify all acceptance criteria met **Archive or Delete** 1. Mark track as completed in tracks.md 2. Record completion date 3. Archive or retain track directory ## Specification (spec.md) Structure ```markdown # {Track Title} ## Overview Brief description of what this track accomplishes and why. ## Functional Requirements ### FR-1: {Requirement Name} Description of the functional requirement. - Acceptance: How to verify this requirement is met ### FR-2: {Requirement Name} ... ## Non-Functional Requirements ### NFR-1: {Requirement Name} Description of the non-functional requirement (performance, security, etc.) - Target: Specific measurable target - Verification: How to test ## Acceptance Criteria - [ ] Criterion 1: Specific, testable condition - [ ] Criterion 2: Specific, testable condition - [ ] Criterion 3: Specific, testable condition ## Scope ### In Scope - Explicitly included items - Features to implement - Components to modify ### Out of Scope - Explicitly excluded items - Future considerations - Related but separate work ## Dependencies ### Internal - Other tracks or components this depends on - Required context artifacts ### External - Third-party services or APIs - External dependencies ## Risks and Mitigations | Risk | Impact | Mitigation | | ---------------- | --------------- | ------------------- | | Risk description | High/Medium/Low | Mitigation strategy | ## Open Questions - [ ] Question that needs resolution - [x] Resolved question - Answer ``` ## Plan (plan.md) Structure ```markdown # Implementation Plan: {Track Title} Track ID: `{track-id}` Created: YYYY-MM-DD Status: pending | in-progress | completed ## Overview Brief description of implementation approach. ## Phase 1: {Phase Name} ### Tasks - [ ] **Task 1.1**: Task description - Sub-task or detail - Sub-task or detail - [ ] **Task 1.2**: Task description - [ ] **Task 1.3**: Task description ### Verification - [ ] **Verify 1.1**: Verification step for phase ## Phase 2: {Phase Name} ### Tasks - [ ] **Task 2.1**: Task description - [ ] **Task 2.2**: Task description ### Verification - [ ] **Verify 2.1**: Verification step for phase ## Phase 3: Finalization ### Tasks - [ ] **Task 3.1**: Update documentation - [ ] **Task 3.2**: Final integration test ### Verification - [ ] **Verify 3.1**: All acceptance criteria met ## Checkpoints | Phase | Checkpoint SHA | Date | Status | | ------- | -------------- | ---- | ------- | | Phase 1 | | | pending | | Phase 2 | | | pending | | Phase 3 | | | pending | ``` ## Status Marker Conventions Use consistent markers in plan.md: | Marker | Meaning | Usage | | ------ | ----------- | --------------------------- | | `[ ]` | Pending | Task not started | | `[~]` | In Progress | Currently being worked | | `[x]` | Complete | Task finished (include SHA) | | `[-]` | Skipped | Intentionally not done | | `[!]` | Blocked | Waiting on dependency | Example: ```markdown - [x] **Task 1.1**: Set up database schema `abc1234` - [~] **Task 1.2**: Implement user model - [ ] **Task 1.3**: Add validation logic - [!] **Task 1.4**: Integrate auth service (blocked: waiting for API key) - [-] **Task 1.5**: Legacy migration (skipped: not needed) ``` ## Track Registry (tracks.md) Format ```markdown # Track Registry ## Active Tracks | Track ID | Type | Status | Phase | Started | Assignee | | ------------------------------------------------ | ------- | ----------- | ----- | ---------- | ---------- | | [user-auth_20250115](tracks/user-auth_20250115/) | feature | in-progress | 2/3 | 2025-01-15 | @developer | | [fix-login_20250114](tracks/fix-login_20250114/) | bug | pending | 0/2 | 2025-01-14 | - | ## Completed Tracks | Track ID | Type | Completed | Duration | | ---------------------------------------------- | ----- | ---------- | -------- | | [setup-ci_20250110](tracks/setup-ci_20250110/) | chore | 2025-01-12 | 2 days | ## Archived Tracks | Track ID | Reason | Archived | | ---------------------------------------------------- | ---------- | ---------- | | [old-feature_20241201](tracks/old-feature_20241201/) | Superseded | 2025-01-05 | ``` ## Metadata (metadata.json) Fields ```json { "id": "user-auth_20250115", "title": "User Authentication System", "type": "feature", "status": "in-progress", "priority": "high", "created": "2025-01-15T10:30:00Z", "updated": "2025-01-15T14:45:00Z", "started": "2025-01-15T11:00:00Z", "completed": null, "assignee": "@developer", "phases": { "total": 3, "current": 2, "completed": 1 }, "tasks": { "total": 12, "completed": 5, "in_progress": 1, "pending": 6 }, "checkpoints": [ { "phase": 1, "sha": "abc1234", "date": "2025-01-15T13:00:00Z" } ], "dependencies": [], "tags": ["auth", "security"] } ``` ## Track Operations ### Creating a Track 1. Run `/conductor:new-track` 2. Answer interactive questions 3. Review generated spec.md 4. Review generated plan.md 5. Confirm track creation ### Starting Implementation 1. Read spec.md and plan.md 2. Verify context artifacts are current 3. Mark first task as `[~]` 4. Begin TDD workflow ### Completing a Phase 1. Ensure all phase tasks are `[x]` 2. Complete verification tasks 3. Wait for checkpoint approval 4. Record checkpoint SHA 5. Proceed to next phase ### Completing a Track 1. Verify all phases complete 2. Verify all acceptance criteria met 3. Update product.md if needed 4. Mark track completed in tracks.md 5. Update metadata.json ### Reverting a Track 1. Run `/conductor:revert` 2. Select track to revert 3. Choose granularity (track/phase/task) 4. Confirm revert operation 5. Update status markers ## Handling Track Dependencies ### Identifying Dependencies During track creation, identify: - **Hard dependencies**: Must complete before this track can start - **Soft dependencies**: Can proceed in parallel but may affect integration - **External dependencies**: Third-party services, APIs, or team decisions ### Documenting Dependencies In spec.md, list dependencies with: - Dependency type (hard/soft/external) - Current status (available/pending/blocked) - Resolution path (what needs to happen) ### Managing Blocked Tracks When a track is blocked: 1. Mark blocked tasks with `[!]` and reason 2. Update tracks.md status 3. Document blocker in metadata.json 4. Consider creating dependency track if needed ## Track Sizing Guidelines ### Right-Sized Tracks Aim for tracks that: - Complete in 1-5 days of work - Have 2-4 phases - Contain 8-20 tasks total - Deliver a coherent, testable unit ### Too Large Signs a track is too large: - More than 5 phases - More than 25 tasks - Multiple unrelated features - Estimated duration > 1 week Solution: Split into multiple tracks with clear boundaries. ### Too Small Signs a track is too small: - Single phase with 1-2 tasks - No meaningful verification needed - Could be a sub-task of another track - Less than a few hours of work Solution: Combine with related work or handle as part of existing track. ## Specification Quality Checklist Before finalizing spec.md, verify: ### Requirements Quality - [ ] Each requirement has clear acceptance criteria - [ ] Requirements are testable - [ ] Requirements are independent (can verify separately) - [ ] No ambiguous language ("should be fast" β†’ "response < 200ms") ### Scope Clarity - [ ] In-scope items are specific - [ ] Out-of-scope items prevent scope creep - [ ] Boundaries are clear to implementer ### Dependencies Identified - [ ] All internal dependencies listed - [ ] External dependencies have owners/contacts - [ ] Dependency status is current ### Risks Addressed - [ ] Major risks identified - [ ] Impact assessment realistic - [ ] Mitigations are actionable ## Plan Quality Checklist Before starting implementation, verify plan.md: ### Task Quality - [ ] Tasks are atomic (one logical action) - [ ] Tasks are independently verifiable - [ ] Task descriptions are clear - [ ] Sub-tasks provide helpful detail ### Phase Organization - [ ] Phases group related tasks - [ ] Each phase delivers something testable - [ ] Verification tasks after each phase - [ ] Phases build on each other logically ### Completeness - [ ] All spec requirements have corresponding tasks - [ ] Documentation tasks included - [ ] Testing tasks included - [ ] Integration tasks included ## Common Track Patterns ### Feature Track Pattern ``` Phase 1: Foundation - Data models - Database migrations - Basic API structure Phase 2: Core Logic - Business logic implementation - Input validation - Error handling Phase 3: Integration - UI integration - API documentation - End-to-end tests ``` ### Bug Fix Track Pattern ``` Phase 1: Reproduction - Write failing test capturing bug - Document reproduction steps Phase 2: Fix - Implement fix - Verify test passes - Check for regressions Phase 3: Verification - Manual verification - Update documentation if needed ``` ### Refactor Track Pattern ``` Phase 1: Preparation - Add characterization tests - Document current behavior Phase 2: Refactoring - Apply changes incrementally - Maintain green tests throughout Phase 3: Cleanup - Remove dead code - Update documentation ``` ## Best Practices 1. **One track, one concern**: Keep tracks focused on a single logical change 2. **Small phases**: Break work into phases of 3-5 tasks maximum 3. **Verification after phases**: Always include verification tasks 4. **Update markers immediately**: Mark task status as you work 5. **Record SHAs**: Always note commit SHAs for completed tasks 6. **Review specs before planning**: Ensure spec is complete before creating plan 7. **Link dependencies**: Explicitly note track dependencies 8. **Archive, don't delete**: Preserve completed tracks for reference 9. **Size appropriately**: Keep tracks between 1-5 days of work 10. **Clear acceptance criteria**: Every requirement must be testable
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

dbt-transformation-patterns

Master dbt (data build tool) for analytics engineering with model

data
⭐1
# dbt Transformation Patterns Production-ready patterns for dbt (data build tool) including model organization, testing strategies, documentation, and incremental processing. ## When to Use This Skill - Building data transformation pipelines with dbt - Organizing models into staging, intermediate, and marts layers - Implementing data quality tests - Creating incremental models for large datasets - Documenting data models and lineage - Setting up dbt project structure ## Core Concepts ### 1. Model Layers (Medallion Architecture) ``` sources/ Raw data definitions ↓ staging/ 1:1 with source, light cleaning ↓ intermediate/ Business logic, joins, aggregations ↓ marts/ Final analytics tables ``` ### 2. Naming Conventions | Layer | Prefix | Example | | ------------ | -------------- | ----------------------------- | | Staging | `stg_` | `stg_stripe__payments` | | Intermediate | `int_` | `int_payments_pivoted` | | Marts | `dim_`, `fct_` | `dim_customers`, `fct_orders` | ## Quick Start ```yaml # dbt_project.yml name: "analytics" version: "1.0.0" profile: "analytics" model-paths: ["models"] analysis-paths: ["analyses"] test-paths: ["tests"] seed-paths: ["seeds"] macro-paths: ["macros"] vars: start_date: "2020-01-01" models: analytics: staging: +materialized: view +schema: staging intermediate: +materialized: ephemeral marts: +materialized: table +schema: analytics ``` ``` # Project structure models/ β”œβ”€β”€ staging/ β”‚ β”œβ”€β”€ stripe/ β”‚ β”‚ β”œβ”€β”€ _stripe__sources.yml β”‚ β”‚ β”œβ”€β”€ _stripe__models.yml β”‚ β”‚ β”œβ”€β”€ stg_stripe__customers.sql β”‚ β”‚ └── stg_stripe__payments.sql β”‚ └── shopify/ β”‚ β”œβ”€β”€ _shopify__sources.yml β”‚ └── stg_shopify__orders.sql β”œβ”€β”€ intermediate/ β”‚ └── finance/ β”‚ └── int_payments_pivoted.sql └── marts/ β”œβ”€β”€ core/ β”‚ β”œβ”€β”€ _core__models.yml β”‚ β”œβ”€β”€ dim_customers.sql β”‚ └── fct_orders.sql └── finance/ └── fct_revenue.sql ``` ## Patterns ### Pattern 1: Source Definitions ```yaml # models/staging/stripe/_stripe__sources.yml version: 2 sources: - name: stripe description: Raw Stripe data loaded via Fivetran database: raw schema: stripe loader: fivetran loaded_at_field: _fivetran_synced freshness: warn_after: { count: 12, period: hour } error_after: { count: 24, period: hour } tables: - name: customers description: Stripe customer records columns: - name: id description: Primary key tests: - unique - not_null - name: email description: Customer email - name: created description: Account creation timestamp - name: payments description: Stripe payment transactions columns: - name: id tests: - unique - not_null - name: customer_id tests: - not_null - relationships: to: source('stripe', 'customers') field: id ``` ### Pattern 2: Staging Models ```sql -- models/staging/stripe/stg_stripe__customers.sql with source as ( select * from {{ source('stripe', 'customers') }} ), renamed as ( select -- ids id as customer_id, -- strings lower(email) as email, name as customer_name, -- timestamps created as created_at, -- metadata _fivetran_synced as _loaded_at from source ) select * from renamed ``` ```sql -- models/staging/stripe/stg_stripe__payments.sql {{ config( materialized='incremental', unique_key='payment_id', on_schema_change='append_new_columns' ) }} with source as ( select * from {{ source('stripe', 'payments') }} {% if is_incremental() %} where _fivetran_synced > (select max(_loaded_at) from {{ this }}) {% endif %} ), renamed as ( select -- ids id as payment_id, customer_id, invoice_id, -- amounts (convert cents to dollars) amount / 100.0 as amount, amount_refunded / 100.0 as amount_refunded, -- status status as payment_status, -- timestamps created as created_at, -- metadata _fivetran_synced as _loaded_at from source ) select * from renamed ``` ### Pattern 3: Intermediate Models ```sql -- models/intermediate/finance/int_payments_pivoted_to_customer.sql with payments as ( select * from {{ ref('stg_stripe__payments') }} ), customers as ( select * from {{ ref('stg_stripe__customers') }} ), payment_summary as ( select customer_id, count(*) as total_payments, count(case when payment_status = 'succeeded' then 1 end) as successful_payments, sum(case when payment_status = 'succeeded' then amount else 0 end) as total_amount_paid, min(created_at) as first_payment_at, max(created_at) as last_payment_at from payments group by customer_id ) select customers.customer_id, customers.email, customers.created_at as customer_created_at, coalesce(payment_summary.total_payments, 0) as total_payments, coalesce(payment_summary.successful_payments, 0) as successful_payments, coalesce(payment_summary.total_amount_paid, 0) as lifetime_value, payment_summary.first_payment_at, payment_summary.last_payment_at from customers left join payment_summary using (customer_id) ``` ### Pattern 4: Mart Models (Dimensions and Facts) ```sql -- models/marts/core/dim_customers.sql {{ config( materialized='table', unique_key='customer_id' ) }} with customers as ( select * from {{ ref('int_payments_pivoted_to_customer') }} ), orders as ( select * from {{ ref('stg_shopify__orders') }} ), order_summary as ( select customer_id, count(*) as total_orders, sum(total_price) as total_order_value, min(created_at) as first_order_at, max(created_at) as last_order_at from orders group by customer_id ), final as ( select -- surrogate key {{ dbt_utils.generate_surrogate_key(['customers.customer_id']) }} as customer_key, -- natural key customers.customer_id, -- attributes customers.email, customers.customer_created_at, -- payment metrics customers.total_payments, customers.successful_payments, customers.lifetime_value, customers.first_payment_at, customers.last_payment_at, -- order metrics coalesce(order_summary.total_orders, 0) as total_orders, coalesce(order_summary.total_order_value, 0) as total_order_value, order_summary.first_order_at, order_summary.last_order_at, -- calculated fields case when customers.lifetime_value >= 1000 then 'high' when customers.lifetime_value >= 100 then 'medium' else 'low' end as customer_tier, -- timestamps current_timestamp as _loaded_at from customers left join order_summary using (customer_id) ) select * from final ``` ```sql -- models/marts/core/fct_orders.sql {{ config( materialized='incremental', unique_key='order_id', incremental_strategy='merge' ) }} with orders as ( select * from {{ ref('stg_shopify__orders') }} {% if is_incremental() %} where updated_at > (select max(updated_at) from {{ this }}) {% endif %} ), customers as ( select * from {{ ref('dim_customers') }} ), final as ( select -- keys orders.order_id, customers.customer_key, orders.customer_id, -- dimensions orders.order_status, orders.fulfillment_status, orders.payment_status, -- measures orders.subtotal, orders.tax, orders.shipping, orders.total_price, orders.total_discount, orders.item_count, -- timestamps orders.created_at, orders.updated_at, orders.fulfilled_at, -- metadata current_timestamp as _loaded_at from orders left join customers on orders.customer_id = customers.customer_id ) select * from final ``` ### Pattern 5: Testing and Documentation ```yaml # models/marts/core/_core__models.yml version: 2 models: - name: dim_customers description: Customer dimension with payment and order metrics columns: - name: customer_key description: Surrogate key for the customer dimension tests: - unique - not_null - name: customer_id description: Natural key from source system tests: - unique - not_null - name: email description: Customer email address tests: - not_null - name: customer_tier description: Customer value tier based on lifetime value tests: - accepted_values: values: ["high", "medium", "low"] - name: lifetime_value description: Total amount paid by customer tests: - dbt_utils.expression_is_true: expression: ">= 0" - name: fct_orders description: Order fact table with all order transactions tests: - dbt_utils.recency: datepart: day field: created_at interval: 1 columns: - name: order_id tests: - unique - not_null - name: customer_key tests: - not_null - relationships: to: ref('dim_customers') field: customer_key ``` ### Pattern 6: Macros and DRY Code ```sql -- macros/cents_to_dollars.sql {% macro cents_to_dollars(column_name, precision=2) %} round({{ column_name }} / 100.0, {{ precision }}) {% endmacro %} -- macros/generate_schema_name.sql {% macro generate_schema_name(custom_schema_name, node) %} {%- set default_schema = target.schema -%} {%- if custom_schema_name is none -%} {{ default_schema }} {%- else -%} {{ default_schema }}_{{ custom_schema_name }} {%- endif -%} {% endmacro %} -- macros/limit_data_in_dev.sql {% macro limit_data_in_dev(column_name, days=3) %} {% if target.name == 'dev' %} where {{ column_name }} >= dateadd(day, -{{ days }}, current_date) {% endif %} {% endmacro %} -- Usage in model select * from {{ ref('stg_orders') }} {{ limit_data_in_dev('created_at') }} ``` ### Pattern 7: Incremental Strategies ```sql -- Delete+Insert (default for most warehouses) {{ config( materialized='incremental', unique_key='id', incremental_strategy='delete+insert' ) }} -- Merge (best for late-arriving data) {{ config( materialized='incremental', unique_key='id', incremental_strategy='merge', merge_update_columns=['status', 'amount', 'updated_at'] ) }} -- Insert Overwrite (partition-based) {{ config( materialized='incremental', incremental_strategy='insert_overwrite', partition_by={ "field": "created_date", "data_type": "date", "granularity": "day" } ) }} select *, date(created_at) as created_date from {{ ref('stg_events') }} {% if is_incremental() %} where created_date >= dateadd(day, -3, current_date) {% endif %} ``` ## dbt Commands ```bash # Development dbt run # Run all models dbt run --select staging # Run staging models only dbt run --select +fct_orders # Run fct_orders and its upstream dbt run --select fct_orders+ # Run fct_orders and its downstream dbt run --full-refresh # Rebuild incremental models # Testing dbt test # Run all tests dbt test --select stg_stripe # Test specific models dbt build # Run + test in DAG order # Documentation dbt docs generate # Generate docs dbt docs serve # Serve docs locally # Debugging dbt compile # Compile SQL without running dbt debug # Test connection dbt ls --select tag:critical # List models by tag ``` ## Best Practices ### Do's - **Use staging layer** - Clean data once, use everywhere - **Test aggressively** - Not null, unique, relationships - **Document everything** - Column descriptions, model descriptions - **Use incremental** - For tables > 1M rows - **Version control** - dbt project in Git ### Don'ts - **Don't skip staging** - Raw β†’ mart is tech debt - **Don't hardcode dates** - Use `{{ var('start_date') }}` - **Don't repeat logic** - Extract to macros - **Don't test in prod** - Use dev target - **Don't ignore freshness** - Monitor source data ## Resources - [dbt Documentation](https://docs.getdbt.com/) - [dbt Best Practices](https://docs.getdbt.com/guides/best-practices) - [dbt-utils Package](https://hub.getdbt.com/dbt-labs/dbt_utils/latest/) - [dbt Discourse](https://discourse.getdbt.com/)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

debugging-strategies

Master systematic debugging techniques, profiling tools, and root

coding
⭐1
# Debugging Strategies Transform debugging from frustrating guesswork into systematic problem-solving with proven strategies, powerful tools, and methodical approaches. ## When to Use This Skill - Tracking down elusive bugs - Investigating performance issues - Understanding unfamiliar codebases - Debugging production issues - Analyzing crash dumps and stack traces - Profiling application performance - Investigating memory leaks - Debugging distributed systems ## Core Principles ### 1. The Scientific Method **1. Observe**: What's the actual behavior? **2. Hypothesize**: What could be causing it? **3. Experiment**: Test your hypothesis **4. Analyze**: Did it prove/disprove your theory? **5. Repeat**: Until you find the root cause ### 2. Debugging Mindset **Don't Assume:** - "It can't be X" - Yes it can - "I didn't change Y" - Check anyway - "It works on my machine" - Find out why **Do:** - Reproduce consistently - Isolate the problem - Keep detailed notes - Question everything - Take breaks when stuck ### 3. Rubber Duck Debugging Explain your code and problem out loud (to a rubber duck, colleague, or yourself). Often reveals the issue. ## Systematic Debugging Process ### Phase 1: Reproduce ```markdown ## Reproduction Checklist 1. **Can you reproduce it?** - Always? Sometimes? Randomly? - Specific conditions needed? - Can others reproduce it? 2. **Create minimal reproduction** - Simplify to smallest example - Remove unrelated code - Isolate the problem 3. **Document steps** - Write down exact steps - Note environment details - Capture error messages ``` ### Phase 2: Gather Information ```markdown ## Information Collection 1. **Error Messages** - Full stack trace - Error codes - Console/log output 2. **Environment** - OS version - Language/runtime version - Dependencies versions - Environment variables 3. **Recent Changes** - Git history - Deployment timeline - Configuration changes 4. **Scope** - Affects all users or specific ones? - All browsers or specific ones? - Production only or also dev? ``` ### Phase 3: Form Hypothesis ```markdown ## Hypothesis Formation Based on gathered info, ask: 1. **What changed?** - Recent code changes - Dependency updates - Infrastructure changes 2. **What's different?** - Working vs broken environment - Working vs broken user - Before vs after 3. **Where could this fail?** - Input validation - Business logic - Data layer - External services ``` ### Phase 4: Test & Verify ```markdown ## Testing Strategies 1. **Binary Search** - Comment out half the code - Narrow down problematic section - Repeat until found 2. **Add Logging** - Strategic console.log/print - Track variable values - Trace execution flow 3. **Isolate Components** - Test each piece separately - Mock dependencies - Remove complexity 4. **Compare Working vs Broken** - Diff configurations - Diff environments - Diff data ``` ## Debugging Tools ### JavaScript/TypeScript Debugging ```typescript // Chrome DevTools Debugger function processOrder(order: Order) { debugger; // Execution pauses here const total = calculateTotal(order); console.log("Total:", total); // Conditional breakpoint if (order.items.length > 10) { debugger; // Only breaks if condition true } return total; } // Console debugging techniques console.log("Value:", value); // Basic console.table(arrayOfObjects); // Table format console.time("operation"); /* code */ console.timeEnd("operation"); // Timing console.trace(); // Stack trace console.assert(value > 0, "Value must be positive"); // Assertion // Performance profiling performance.mark("start-operation"); // ... operation code performance.mark("end-operation"); performance.measure("operation", "start-operation", "end-operation"); console.log(performance.getEntriesByType("measure")); ``` **VS Code Debugger Configuration:** ```json // .vscode/launch.json { "version": "0.2.0", "configurations": [ { "type": "node", "request": "launch", "name": "Debug Program", "program": "${workspaceFolder}/src/index.ts", "preLaunchTask": "tsc: build - tsconfig.json", "outFiles": ["${workspaceFolder}/dist/**/*.js"], "skipFiles": ["<node_internals>/**"] }, { "type": "node", "request": "launch", "name": "Debug Tests", "program": "${workspaceFolder}/node_modules/jest/bin/jest", "args": ["--runInBand", "--no-cache"], "console": "integratedTerminal" } ] } ``` ### Python Debugging ```python # Built-in debugger (pdb) import pdb def calculate_total(items): total = 0 pdb.set_trace() # Debugger starts here for item in items: total += item.price * item.quantity return total # Breakpoint (Python 3.7+) def process_order(order): breakpoint() # More convenient than pdb.set_trace() # ... code # Post-mortem debugging try: risky_operation() except Exception: import pdb pdb.post_mortem() # Debug at exception point # IPython debugging (ipdb) from ipdb import set_trace set_trace() # Better interface than pdb # Logging for debugging import logging logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) def fetch_user(user_id): logger.debug(f'Fetching user: {user_id}') user = db.query(User).get(user_id) logger.debug(f'Found user: {user}') return user # Profile performance import cProfile import pstats cProfile.run('slow_function()', 'profile_stats') stats = pstats.Stats('profile_stats') stats.sort_stats('cumulative') stats.print_stats(10) # Top 10 slowest ``` ### Go Debugging ```go // Delve debugger // Install: go install github.com/go-delve/delve/cmd/dlv@latest // Run: dlv debug main.go import ( "fmt" "runtime" "runtime/debug" ) // Print stack trace func debugStack() { debug.PrintStack() } // Panic recovery with debugging func processRequest() { defer func() { if r := recover(); r != nil { fmt.Println("Panic:", r) debug.PrintStack() } }() // ... code that might panic } // Memory profiling import _ "net/http/pprof" // Visit http://localhost:6060/debug/pprof/ // CPU profiling import ( "os" "runtime/pprof" ) f, _ := os.Create("cpu.prof") pprof.StartCPUProfile(f) defer pprof.StopCPUProfile() // ... code to profile ``` ## Advanced Debugging Techniques ### Technique 1: Binary Search Debugging ```bash # Git bisect for finding regression git bisect start git bisect bad # Current commit is bad git bisect good v1.0.0 # v1.0.0 was good # Git checks out middle commit # Test it, then: git bisect good # if it works git bisect bad # if it's broken # Continue until bug found git bisect reset # when done ``` ### Technique 2: Differential Debugging Compare working vs broken: ```markdown ## What's Different? | Aspect | Working | Broken | | ------------ | ----------- | -------------- | | Environment | Development | Production | | Node version | 18.16.0 | 18.15.0 | | Data | Empty DB | 1M records | | User | Admin | Regular user | | Browser | Chrome | Safari | | Time | During day | After midnight | Hypothesis: Time-based issue? Check timezone handling. ``` ### Technique 3: Trace Debugging ```typescript // Function call tracing function trace( target: any, propertyKey: string, descriptor: PropertyDescriptor, ) { const originalMethod = descriptor.value; descriptor.value = function (...args: any[]) { console.log(`Calling ${propertyKey} with args:`, args); const result = originalMethod.apply(this, args); console.log(`${propertyKey} returned:`, result); return result; }; return descriptor; } class OrderService { @trace calculateTotal(items: Item[]): number { return items.reduce((sum, item) => sum + item.price, 0); } } ``` ### Technique 4: Memory Leak Detection ```typescript // Chrome DevTools Memory Profiler // 1. Take heap snapshot // 2. Perform action // 3. Take another snapshot // 4. Compare snapshots // Node.js memory debugging if (process.memoryUsage().heapUsed > 500 * 1024 * 1024) { console.warn("High memory usage:", process.memoryUsage()); // Generate heap dump require("v8").writeHeapSnapshot(); } // Find memory leaks in tests let beforeMemory: number; beforeEach(() => { beforeMemory = process.memoryUsage().heapUsed; }); afterEach(() => { const afterMemory = process.memoryUsage().heapUsed; const diff = afterMemory - beforeMemory; if (diff > 10 * 1024 * 1024) { // 10MB threshold console.warn(`Possible memory leak: ${diff / 1024 / 1024}MB`); } }); ``` ## Debugging Patterns by Issue Type ### Pattern 1: Intermittent Bugs ```markdown ## Strategies for Flaky Bugs 1. **Add extensive logging** - Log timing information - Log all state transitions - Log external interactions 2. **Look for race conditions** - Concurrent access to shared state - Async operations completing out of order - Missing synchronization 3. **Check timing dependencies** - setTimeout/setInterval - Promise resolution order - Animation frame timing 4. **Stress test** - Run many times - Vary timing - Simulate load ``` ### Pattern 2: Performance Issues ```markdown ## Performance Debugging 1. **Profile first** - Don't optimize blindly - Measure before and after - Find bottlenecks 2. **Common culprits** - N+1 queries - Unnecessary re-renders - Large data processing - Synchronous I/O 3. **Tools** - Browser DevTools Performance tab - Lighthouse - Python: cProfile, line_profiler - Node: clinic.js, 0x ``` ### Pattern 3: Production Bugs ```markdown ## Production Debugging 1. **Gather evidence** - Error tracking (Sentry, Bugsnag) - Application logs - User reports - Metrics/monitoring 2. **Reproduce locally** - Use production data (anonymized) - Match environment - Follow exact steps 3. **Safe investigation** - Don't change production - Use feature flags - Add monitoring/logging - Test fixes in staging ``` ## Best Practices 1. **Reproduce First**: Can't fix what you can't reproduce 2. **Isolate the Problem**: Remove complexity until minimal case 3. **Read Error Messages**: They're usually helpful 4. **Check Recent Changes**: Most bugs are recent 5. **Use Version Control**: Git bisect, blame, history 6. **Take Breaks**: Fresh eyes see better 7. **Document Findings**: Help future you 8. **Fix Root Cause**: Not just symptoms ## Common Debugging Mistakes - **Making Multiple Changes**: Change one thing at a time - **Not Reading Error Messages**: Read the full stack trace - **Assuming It's Complex**: Often it's simple - **Debug Logging in Prod**: Remove before shipping - **Not Using Debugger**: console.log isn't always best - **Giving Up Too Soon**: Persistence pays off - **Not Testing the Fix**: Verify it actually works ## Quick Debugging Checklist ```markdown ## When Stuck, Check: - [ ] Spelling errors (typos in variable names) - [ ] Case sensitivity (fileName vs filename) - [ ] Null/undefined values - [ ] Array index off-by-one - [ ] Async timing (race conditions) - [ ] Scope issues (closure, hoisting) - [ ] Type mismatches - [ ] Missing dependencies - [ ] Environment variables - [ ] File paths (absolute vs relative) - [ ] Cache issues (clear cache) - [ ] Stale data (refresh database) ``` ## Resources - **references/debugging-tools-guide.md**: Comprehensive tool documentation - **references/performance-profiling.md**: Performance debugging guide - **references/production-debugging.md**: Debugging live systems - **assets/debugging-checklist.md**: Quick reference checklist - **assets/common-bugs.md**: Common bug patterns - **scripts/debug-helper.ts**: Debugging utility functions
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

git-advanced-workflows

Master advanced Git workflows including rebasing, cherry-picking,

coding
⭐1
# Git Advanced Workflows Master advanced Git techniques to maintain clean history, collaborate effectively, and recover from any situation with confidence. ## When to Use This Skill - Cleaning up commit history before merging - Applying specific commits across branches - Finding commits that introduced bugs - Working on multiple features simultaneously - Recovering from Git mistakes or lost commits - Managing complex branch workflows - Preparing clean PRs for review - Synchronizing diverged branches ## Core Concepts ### 1. Interactive Rebase Interactive rebase is the Swiss Army knife of Git history editing. **Common Operations:** - `pick`: Keep commit as-is - `reword`: Change commit message - `edit`: Amend commit content - `squash`: Combine with previous commit - `fixup`: Like squash but discard message - `drop`: Remove commit entirely **Basic Usage:** ```bash # Rebase last 5 commits git rebase -i HEAD~5 # Rebase all commits on current branch git rebase -i $(git merge-base HEAD main) # Rebase onto specific commit git rebase -i abc123 ``` ### 2. Cherry-Picking Apply specific commits from one branch to another without merging entire branches. ```bash # Cherry-pick single commit git cherry-pick abc123 # Cherry-pick range of commits (exclusive start) git cherry-pick abc123..def456 # Cherry-pick without committing (stage changes only) git cherry-pick -n abc123 # Cherry-pick and edit commit message git cherry-pick -e abc123 ``` ### 3. Git Bisect Binary search through commit history to find the commit that introduced a bug. ```bash # Start bisect git bisect start # Mark current commit as bad git bisect bad # Mark known good commit git bisect good v1.0.0 # Git will checkout middle commit - test it # Then mark as good or bad git bisect good # or: git bisect bad # Continue until bug found # When done git bisect reset ``` **Automated Bisect:** ```bash # Use script to test automatically git bisect start HEAD v1.0.0 git bisect run ./test.sh # test.sh should exit 0 for good, 1-127 (except 125) for bad ``` ### 4. Worktrees Work on multiple branches simultaneously without stashing or switching. ```bash # List existing worktrees git worktree list # Add new worktree for feature branch git worktree add ../project-feature feature/new-feature # Add worktree and create new branch git worktree add -b bugfix/urgent ../project-hotfix main # Remove worktree git worktree remove ../project-feature # Prune stale worktrees git worktree prune ``` ### 5. Reflog Your safety net - tracks all ref movements, even deleted commits. ```bash # View reflog git reflog # View reflog for specific branch git reflog show feature/branch # Restore deleted commit git reflog # Find commit hash git checkout abc123 git branch recovered-branch # Restore deleted branch git reflog git branch deleted-branch abc123 ``` ## Practical Workflows ### Workflow 1: Clean Up Feature Branch Before PR ```bash # Start with feature branch git checkout feature/user-auth # Interactive rebase to clean history git rebase -i main # Example rebase operations: # - Squash "fix typo" commits # - Reword commit messages for clarity # - Reorder commits logically # - Drop unnecessary commits # Force push cleaned branch (safe if no one else is using it) git push --force-with-lease origin feature/user-auth ``` ### Workflow 2: Apply Hotfix to Multiple Releases ```bash # Create fix on main git checkout main git commit -m "fix: critical security patch" # Apply to release branches git checkout release/2.0 git cherry-pick abc123 git checkout release/1.9 git cherry-pick abc123 # Handle conflicts if they arise git cherry-pick --continue # or git cherry-pick --abort ``` ### Workflow 3: Find Bug Introduction ```bash # Start bisect git bisect start git bisect bad HEAD git bisect good v2.1.0 # Git checks out middle commit - run tests npm test # If tests fail git bisect bad # If tests pass git bisect good # Git will automatically checkout next commit to test # Repeat until bug found # Automated version git bisect start HEAD v2.1.0 git bisect run npm test ``` ### Workflow 4: Multi-Branch Development ```bash # Main project directory cd ~/projects/myapp # Create worktree for urgent bugfix git worktree add ../myapp-hotfix hotfix/critical-bug # Work on hotfix in separate directory cd ../myapp-hotfix # Make changes, commit git commit -m "fix: resolve critical bug" git push origin hotfix/critical-bug # Return to main work without interruption cd ~/projects/myapp git fetch origin git cherry-pick hotfix/critical-bug # Clean up when done git worktree remove ../myapp-hotfix ``` ### Workflow 5: Recover from Mistakes ```bash # Accidentally reset to wrong commit git reset --hard HEAD~5 # Oh no! # Use reflog to find lost commits git reflog # Output shows: # abc123 HEAD@{0}: reset: moving to HEAD~5 # def456 HEAD@{1}: commit: my important changes # Recover lost commits git reset --hard def456 # Or create branch from lost commit git branch recovery def456 ``` ## Advanced Techniques ### Rebase vs Merge Strategy **When to Rebase:** - Cleaning up local commits before pushing - Keeping feature branch up-to-date with main - Creating linear history for easier review **When to Merge:** - Integrating completed features into main - Preserving exact history of collaboration - Public branches used by others ```bash # Update feature branch with main changes (rebase) git checkout feature/my-feature git fetch origin git rebase origin/main # Handle conflicts git status # Fix conflicts in files git add . git rebase --continue # Or merge instead git merge origin/main ``` ### Autosquash Workflow Automatically squash fixup commits during rebase. ```bash # Make initial commit git commit -m "feat: add user authentication" # Later, fix something in that commit # Stage changes git commit --fixup HEAD # or specify commit hash # Make more changes git commit --fixup abc123 # Rebase with autosquash git rebase -i --autosquash main # Git automatically marks fixup commits ``` ### Split Commit Break one commit into multiple logical commits. ```bash # Start interactive rebase git rebase -i HEAD~3 # Mark commit to split with 'edit' # Git will stop at that commit # Reset commit but keep changes git reset HEAD^ # Stage and commit in logical chunks git add file1.py git commit -m "feat: add validation" git add file2.py git commit -m "feat: add error handling" # Continue rebase git rebase --continue ``` ### Partial Cherry-Pick Cherry-pick only specific files from a commit. ```bash # Show files in commit git show --name-only abc123 # Checkout specific files from commit git checkout abc123 -- path/to/file1.py path/to/file2.py # Stage and commit git commit -m "cherry-pick: apply specific changes from abc123" ``` ## Best Practices 1. **Always Use --force-with-lease**: Safer than --force, prevents overwriting others' work 2. **Rebase Only Local Commits**: Don't rebase commits that have been pushed and shared 3. **Descriptive Commit Messages**: Future you will thank present you 4. **Atomic Commits**: Each commit should be a single logical change 5. **Test Before Force Push**: Ensure history rewrite didn't break anything 6. **Keep Reflog Aware**: Remember reflog is your safety net for 90 days 7. **Branch Before Risky Operations**: Create backup branch before complex rebases ```bash # Safe force push git push --force-with-lease origin feature/branch # Create backup before risky operation git branch backup-branch git rebase -i main # If something goes wrong git reset --hard backup-branch ``` ## Common Pitfalls - **Rebasing Public Branches**: Causes history conflicts for collaborators - **Force Pushing Without Lease**: Can overwrite teammate's work - **Losing Work in Rebase**: Resolve conflicts carefully, test after rebase - **Forgetting Worktree Cleanup**: Orphaned worktrees consume disk space - **Not Backing Up Before Experiment**: Always create safety branch - **Bisect on Dirty Working Directory**: Commit or stash before bisecting ## Recovery Commands ```bash # Abort operations in progress git rebase --abort git merge --abort git cherry-pick --abort git bisect reset # Restore file to version from specific commit git restore --source=abc123 path/to/file # Undo last commit but keep changes git reset --soft HEAD^ # Undo last commit and discard changes git reset --hard HEAD^ # Recover deleted branch (within 90 days) git reflog git branch recovered-branch abc123 ``` ## Resources - **references/git-rebase-guide.md**: Deep dive into interactive rebase - **references/git-conflict-resolution.md**: Advanced conflict resolution strategies - **references/git-history-rewriting.md**: Safely rewriting Git history - **assets/git-workflow-checklist.md**: Pre-PR cleanup checklist - **assets/git-aliases.md**: Useful Git aliases for advanced workflows - **scripts/git-clean-branches.sh**: Clean up merged and stale branches
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

changelog-automation

Automate changelog generation from commits, PRs, and releases

coding
⭐1
# Changelog Automation Patterns and tools for automating changelog generation, release notes, and version management following industry standards. ## When to Use This Skill - Setting up automated changelog generation - Implementing Conventional Commits - Creating release note workflows - Standardizing commit message formats - Generating GitHub/GitLab release notes - Managing semantic versioning ## Core Concepts ### 1. Keep a Changelog Format ```markdown # Changelog All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] ### Added - New feature X ## [1.2.0] - 2024-01-15 ### Added - User profile avatars - Dark mode support ### Changed - Improved loading performance by 40% ### Deprecated - Old authentication API (use v2) ### Removed - Legacy payment gateway ### Fixed - Login timeout issue (#123) ### Security - Updated dependencies for CVE-2024-1234 [Unreleased]: https://github.com/user/repo/compare/v1.2.0...HEAD [1.2.0]: https://github.com/user/repo/compare/v1.1.0...v1.2.0 ``` ### 2. Conventional Commits ``` <type>[optional scope]: <description> [optional body] [optional footer(s)] ``` | Type | Description | Changelog Section | | ---------- | ---------------- | ------------------ | | `feat` | New feature | Added | | `fix` | Bug fix | Fixed | | `docs` | Documentation | (usually excluded) | | `style` | Formatting | (usually excluded) | | `refactor` | Code restructure | Changed | | `perf` | Performance | Changed | | `test` | Tests | (usually excluded) | | `chore` | Maintenance | (usually excluded) | | `ci` | CI changes | (usually excluded) | | `build` | Build system | (usually excluded) | | `revert` | Revert commit | Removed | ### 3. Semantic Versioning ``` MAJOR.MINOR.PATCH MAJOR: Breaking changes (feat! or BREAKING CHANGE) MINOR: New features (feat) PATCH: Bug fixes (fix) ``` ## Implementation ### Method 1: Conventional Changelog (Node.js) ```bash # Install tools npm install -D @commitlint/cli @commitlint/config-conventional npm install -D husky npm install -D standard-version # or npm install -D semantic-release # Setup commitlint cat > commitlint.config.js << 'EOF' module.exports = { extends: ['@commitlint/config-conventional'], rules: { 'type-enum': [ 2, 'always', [ 'feat', 'fix', 'docs', 'style', 'refactor', 'perf', 'test', 'chore', 'ci', 'build', 'revert', ], ], 'subject-case': [2, 'never', ['start-case', 'pascal-case', 'upper-case']], 'subject-max-length': [2, 'always', 72], }, }; EOF # Setup husky npx husky init echo "npx --no -- commitlint --edit \$1" > .husky/commit-msg ``` ### Method 2: standard-version Configuration ```javascript // .versionrc.js module.exports = { types: [ { type: "feat", section: "Features" }, { type: "fix", section: "Bug Fixes" }, { type: "perf", section: "Performance Improvements" }, { type: "revert", section: "Reverts" }, { type: "docs", section: "Documentation", hidden: true }, { type: "style", section: "Styles", hidden: true }, { type: "chore", section: "Miscellaneous", hidden: true }, { type: "refactor", section: "Code Refactoring", hidden: true }, { type: "test", section: "Tests", hidden: true }, { type: "build", section: "Build System", hidden: true }, { type: "ci", section: "CI/CD", hidden: true }, ], commitUrlFormat: "{{host}}/{{owner}}/{{repository}}/commit/{{hash}}", compareUrlFormat: "{{host}}/{{owner}}/{{repository}}/compare/{{previousTag}}...{{currentTag}}", issueUrlFormat: "{{host}}/{{owner}}/{{repository}}/issues/{{id}}", userUrlFormat: "{{host}}/{{user}}", releaseCommitMessageFormat: "chore(release): {{currentTag}}", scripts: { prebump: 'echo "Running prebump"', postbump: 'echo "Running postbump"', prechangelog: 'echo "Running prechangelog"', postchangelog: 'echo "Running postchangelog"', }, }; ``` ```json // package.json scripts { "scripts": { "release": "standard-version", "release:minor": "standard-version --release-as minor", "release:major": "standard-version --release-as major", "release:patch": "standard-version --release-as patch", "release:dry": "standard-version --dry-run" } } ``` ### Method 3: semantic-release (Full Automation) ```javascript // release.config.js module.exports = { branches: [ "main", { name: "beta", prerelease: true }, { name: "alpha", prerelease: true }, ], plugins: [ "@semantic-release/commit-analyzer", "@semantic-release/release-notes-generator", [ "@semantic-release/changelog", { changelogFile: "CHANGELOG.md", }, ], [ "@semantic-release/npm", { npmPublish: true, }, ], [ "@semantic-release/github", { assets: ["dist/**/*.js", "dist/**/*.css"], }, ], [ "@semantic-release/git", { assets: ["CHANGELOG.md", "package.json"], message: "chore(release): ${nextRelease.version} [skip ci]\n\n${nextRelease.notes}", }, ], ], }; ``` ### Method 4: GitHub Actions Workflow ```yaml # .github/workflows/release.yml name: Release on: push: branches: [main] workflow_dispatch: inputs: release_type: description: "Release type" required: true default: "patch" type: choice options: - patch - minor - major permissions: contents: write pull-requests: write jobs: release: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 with: fetch-depth: 0 token: ${{ secrets.GITHUB_TOKEN }} - uses: actions/setup-node@v4 with: node-version: "20" cache: "npm" - run: npm ci - name: Configure Git run: | git config user.name "github-actions[bot]" git config user.email "github-actions[bot]@users.noreply.github.com" - name: Run semantic-release env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} NPM_TOKEN: ${{ secrets.NPM_TOKEN }} run: npx semantic-release # Alternative: manual release with standard-version manual-release: if: github.event_name == 'workflow_dispatch' runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 with: fetch-depth: 0 - uses: actions/setup-node@v4 with: node-version: "20" - run: npm ci - name: Configure Git run: | git config user.name "github-actions[bot]" git config user.email "github-actions[bot]@users.noreply.github.com" - name: Bump version and generate changelog run: npx standard-version --release-as ${{ inputs.release_type }} - name: Push changes run: git push --follow-tags origin main - name: Create GitHub Release uses: softprops/action-gh-release@v1 with: tag_name: ${{ steps.version.outputs.tag }} body_path: RELEASE_NOTES.md generate_release_notes: true ``` ### Method 5: git-cliff (Rust-based, Fast) ```toml # cliff.toml [changelog] header = """ # Changelog All notable changes to this project will be documented in this file. """ body = """ {% if version %}\ ## [{{ version | trim_start_matches(pat="v") }}] - {{ timestamp | date(format="%Y-%m-%d") }} {% else %}\ ## [Unreleased] {% endif %}\ {% for group, commits in commits | group_by(attribute="group") %} ### {{ group | upper_first }} {% for commit in commits %} - {% if commit.scope %}**{{ commit.scope }}:** {% endif %}\ {{ commit.message | upper_first }}\ {% if commit.github.pr_number %} ([#{{ commit.github.pr_number }}](https://github.com/owner/repo/pull/{{ commit.github.pr_number }})){% endif %}\ {% endfor %} {% endfor %} """ footer = """ {% for release in releases -%} {% if release.version -%} {% if release.previous.version -%} [{{ release.version | trim_start_matches(pat="v") }}]: \ https://github.com/owner/repo/compare/{{ release.previous.version }}...{{ release.version }} {% endif -%} {% else -%} [unreleased]: https://github.com/owner/repo/compare/{{ release.previous.version }}...HEAD {% endif -%} {% endfor %} """ trim = true [git] conventional_commits = true filter_unconventional = true split_commits = false commit_parsers = [ { message = "^feat", group = "Features" }, { message = "^fix", group = "Bug Fixes" }, { message = "^doc", group = "Documentation" }, { message = "^perf", group = "Performance" }, { message = "^refactor", group = "Refactoring" }, { message = "^style", group = "Styling" }, { message = "^test", group = "Testing" }, { message = "^chore\\(release\\)", skip = true }, { message = "^chore", group = "Miscellaneous" }, ] filter_commits = false tag_pattern = "v[0-9]*" skip_tags = "" ignore_tags = "" topo_order = false sort_commits = "oldest" [github] owner = "owner" repo = "repo" ``` ```bash # Generate changelog git cliff -o CHANGELOG.md # Generate for specific range git cliff v1.0.0..v2.0.0 -o RELEASE_NOTES.md # Preview without writing git cliff --unreleased --dry-run ``` ### Method 6: Python (commitizen) ```toml # pyproject.toml [tool.commitizen] name = "cz_conventional_commits" version = "1.0.0" version_files = [ "pyproject.toml:version", "src/__init__.py:__version__", ] tag_format = "v$version" update_changelog_on_bump = true changelog_incremental = true changelog_start_rev = "v0.1.0" [tool.commitizen.customize] message_template = "{{change_type}}{% if scope %}({{scope}}){% endif %}: {{message}}" schema = "<type>(<scope>): <subject>" schema_pattern = "^(feat|fix|docs|style|refactor|perf|test|chore)(\\(\\w+\\))?:\\s.*" bump_pattern = "^(feat|fix|perf|refactor)" bump_map = {"feat" = "MINOR", "fix" = "PATCH", "perf" = "PATCH", "refactor" = "PATCH"} ``` ```bash # Install pip install commitizen # Create commit interactively cz commit # Bump version and update changelog cz bump --changelog # Check commits cz check --rev-range HEAD~5..HEAD ``` ## Release Notes Templates ### GitHub Release Template ```markdown ## What's Changed ### πŸš€ Features {{ range .Features }} - {{ .Title }} by @{{ .Author }} in #{{ .PR }} {{ end }} ### πŸ› Bug Fixes {{ range .Fixes }} - {{ .Title }} by @{{ .Author }} in #{{ .PR }} {{ end }} ### πŸ“š Documentation {{ range .Docs }} - {{ .Title }} by @{{ .Author }} in #{{ .PR }} {{ end }} ### πŸ”§ Maintenance {{ range .Chores }} - {{ .Title }} by @{{ .Author }} in #{{ .PR }} {{ end }} ## New Contributors {{ range .NewContributors }} - @{{ .Username }} made their first contribution in #{{ .PR }} {{ end }} **Full Changelog**: https://github.com/owner/repo/compare/v{{ .Previous }}...v{{ .Current }} ``` ### Internal Release Notes ```markdown # Release v2.1.0 - January 15, 2024 ## Summary This release introduces dark mode support and improves checkout performance by 40%. It also includes important security updates. ## Highlights ### πŸŒ™ Dark Mode Users can now switch to dark mode from settings. The preference is automatically saved and synced across devices. ### ⚑ Performance - Checkout flow is 40% faster - Reduced bundle size by 15% ## Breaking Changes None in this release. ## Upgrade Guide No special steps required. Standard deployment process applies. ## Known Issues - Dark mode may flicker on initial load (fix scheduled for v2.1.1) ## Dependencies Updated | Package | From | To | Reason | | ------- | ------- | ------- | ------------------------ | | react | 18.2.0 | 18.3.0 | Performance improvements | | lodash | 4.17.20 | 4.17.21 | Security patch | ``` ## Commit Message Examples ```bash # Feature with scope feat(auth): add OAuth2 support for Google login # Bug fix with issue reference fix(checkout): resolve race condition in payment processing Closes #123 # Breaking change feat(api)!: change user endpoint response format BREAKING CHANGE: The user endpoint now returns `userId` instead of `id`. Migration guide: Update all API consumers to use the new field name. # Multiple paragraphs fix(database): handle connection timeouts gracefully Previously, connection timeouts would cause the entire request to fail without retry. This change implements exponential backoff with up to 3 retries before failing. The timeout threshold has been increased from 5s to 10s based on p99 latency analysis. Fixes #456 Reviewed-by: @alice ``` ## Best Practices ### Do's - **Follow Conventional Commits** - Enables automation - **Write clear messages** - Future you will thank you - **Reference issues** - Link commits to tickets - **Use scopes consistently** - Define team conventions - **Automate releases** - Reduce manual errors ### Don'ts - **Don't mix changes** - One logical change per commit - **Don't skip validation** - Use commitlint - **Don't manual edit** - Generated changelogs only - **Don't forget breaking changes** - Mark with `!` or footer - **Don't ignore CI** - Validate commits in pipeline ## Resources - [Keep a Changelog](https://keepachangelog.com/) - [Conventional Commits](https://www.conventionalcommits.org/) - [Semantic Versioning](https://semver.org/) - [semantic-release](https://semantic-release.gitbook.io/) - [git-cliff](https://git-cliff.org/)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

react-native-architecture

Build production React Native apps with Expo, navigation, native

coding
⭐1
# React Native Architecture Production-ready patterns for React Native development with Expo, including navigation, state management, native modules, and offline-first architecture. ## When to Use This Skill - Starting a new React Native or Expo project - Implementing complex navigation patterns - Integrating native modules and platform APIs - Building offline-first mobile applications - Optimizing React Native performance - Setting up CI/CD for mobile releases ## Core Concepts ### 1. Project Structure ``` src/ β”œβ”€β”€ app/ # Expo Router screens β”‚ β”œβ”€β”€ (auth)/ # Auth group β”‚ β”œβ”€β”€ (tabs)/ # Tab navigation β”‚ └── _layout.tsx # Root layout β”œβ”€β”€ components/ β”‚ β”œβ”€β”€ ui/ # Reusable UI components β”‚ └── features/ # Feature-specific components β”œβ”€β”€ hooks/ # Custom hooks β”œβ”€β”€ services/ # API and native services β”œβ”€β”€ stores/ # State management β”œβ”€β”€ utils/ # Utilities └── types/ # TypeScript types ``` ### 2. Expo vs Bare React Native | Feature | Expo | Bare RN | | ------------------ | -------------- | -------------- | | Setup complexity | Low | High | | Native modules | EAS Build | Manual linking | | OTA updates | Built-in | Manual setup | | Build service | EAS | Custom CI | | Custom native code | Config plugins | Direct access | ## Quick Start ```bash # Create new Expo project npx create-expo-app@latest my-app -t expo-template-blank-typescript # Install essential dependencies npx expo install expo-router expo-status-bar react-native-safe-area-context npx expo install @react-native-async-storage/async-storage npx expo install expo-secure-store expo-haptics ``` ```typescript // app/_layout.tsx import { Stack } from 'expo-router' import { ThemeProvider } from '@/providers/ThemeProvider' import { QueryProvider } from '@/providers/QueryProvider' export default function RootLayout() { return ( <QueryProvider> <ThemeProvider> <Stack screenOptions={{ headerShown: false }}> <Stack.Screen name="(tabs)" /> <Stack.Screen name="(auth)" /> <Stack.Screen name="modal" options={{ presentation: 'modal' }} /> </Stack> </ThemeProvider> </QueryProvider> ) } ``` ## Patterns ### Pattern 1: Expo Router Navigation ```typescript // app/(tabs)/_layout.tsx import { Tabs } from 'expo-router' import { Home, Search, User, Settings } from 'lucide-react-native' import { useTheme } from '@/hooks/useTheme' export default function TabLayout() { const { colors } = useTheme() return ( <Tabs screenOptions={{ tabBarActiveTintColor: colors.primary, tabBarInactiveTintColor: colors.textMuted, tabBarStyle: { backgroundColor: colors.background }, headerShown: false, }} > <Tabs.Screen name="index" options={{ title: 'Home', tabBarIcon: ({ color, size }) => <Home size={size} color={color} />, }} /> <Tabs.Screen name="search" options={{ title: 'Search', tabBarIcon: ({ color, size }) => <Search size={size} color={color} />, }} /> <Tabs.Screen name="profile" options={{ title: 'Profile', tabBarIcon: ({ color, size }) => <User size={size} color={color} />, }} /> <Tabs.Screen name="settings" options={{ title: 'Settings', tabBarIcon: ({ color, size }) => <Settings size={size} color={color} />, }} /> </Tabs> ) } // app/(tabs)/profile/[id].tsx - Dynamic route import { useLocalSearchParams } from 'expo-router' export default function ProfileScreen() { const { id } = useLocalSearchParams<{ id: string }>() return <UserProfile userId={id} /> } // Navigation from anywhere import { router } from 'expo-router' // Programmatic navigation router.push('/profile/123') router.replace('/login') router.back() // With params router.push({ pathname: '/product/[id]', params: { id: '123', referrer: 'home' }, }) ``` ### Pattern 2: Authentication Flow ```typescript // providers/AuthProvider.tsx import { createContext, useContext, useEffect, useState } from 'react' import { useRouter, useSegments } from 'expo-router' import * as SecureStore from 'expo-secure-store' interface AuthContextType { user: User | null isLoading: boolean signIn: (credentials: Credentials) => Promise<void> signOut: () => Promise<void> } const AuthContext = createContext<AuthContextType | null>(null) export function AuthProvider({ children }: { children: React.ReactNode }) { const [user, setUser] = useState<User | null>(null) const [isLoading, setIsLoading] = useState(true) const segments = useSegments() const router = useRouter() // Check authentication on mount useEffect(() => { checkAuth() }, []) // Protect routes useEffect(() => { if (isLoading) return const inAuthGroup = segments[0] === '(auth)' if (!user && !inAuthGroup) { router.replace('/login') } else if (user && inAuthGroup) { router.replace('/(tabs)') } }, [user, segments, isLoading]) async function checkAuth() { try { const token = await SecureStore.getItemAsync('authToken') if (token) { const userData = await api.getUser(token) setUser(userData) } } catch (error) { await SecureStore.deleteItemAsync('authToken') } finally { setIsLoading(false) } } async function signIn(credentials: Credentials) { const { token, user } = await api.login(credentials) await SecureStore.setItemAsync('authToken', token) setUser(user) } async function signOut() { await SecureStore.deleteItemAsync('authToken') setUser(null) } if (isLoading) { return <SplashScreen /> } return ( <AuthContext.Provider value={{ user, isLoading, signIn, signOut }}> {children} </AuthContext.Provider> ) } export const useAuth = () => { const context = useContext(AuthContext) if (!context) throw new Error('useAuth must be used within AuthProvider') return context } ``` ### Pattern 3: Offline-First with React Query ```typescript // providers/QueryProvider.tsx import { QueryClient, QueryClientProvider } from '@tanstack/react-query' import { createAsyncStoragePersister } from '@tanstack/query-async-storage-persister' import { PersistQueryClientProvider } from '@tanstack/react-query-persist-client' import AsyncStorage from '@react-native-async-storage/async-storage' import NetInfo from '@react-native-community/netinfo' import { onlineManager } from '@tanstack/react-query' // Sync online status onlineManager.setEventListener((setOnline) => { return NetInfo.addEventListener((state) => { setOnline(!!state.isConnected) }) }) const queryClient = new QueryClient({ defaultOptions: { queries: { gcTime: 1000 * 60 * 60 * 24, // 24 hours staleTime: 1000 * 60 * 5, // 5 minutes retry: 2, networkMode: 'offlineFirst', }, mutations: { networkMode: 'offlineFirst', }, }, }) const asyncStoragePersister = createAsyncStoragePersister({ storage: AsyncStorage, key: 'REACT_QUERY_OFFLINE_CACHE', }) export function QueryProvider({ children }: { children: React.ReactNode }) { return ( <PersistQueryClientProvider client={queryClient} persistOptions={{ persister: asyncStoragePersister }} > {children} </PersistQueryClientProvider> ) } // hooks/useProducts.ts import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query' export function useProducts() { return useQuery({ queryKey: ['products'], queryFn: api.getProducts, // Use stale data while revalidating placeholderData: (previousData) => previousData, }) } export function useCreateProduct() { const queryClient = useQueryClient() return useMutation({ mutationFn: api.createProduct, // Optimistic update onMutate: async (newProduct) => { await queryClient.cancelQueries({ queryKey: ['products'] }) const previous = queryClient.getQueryData(['products']) queryClient.setQueryData(['products'], (old: Product[]) => [ ...old, { ...newProduct, id: 'temp-' + Date.now() }, ]) return { previous } }, onError: (err, newProduct, context) => { queryClient.setQueryData(['products'], context?.previous) }, onSettled: () => { queryClient.invalidateQueries({ queryKey: ['products'] }) }, }) } ``` ### Pattern 4: Native Module Integration ```typescript // services/haptics.ts import * as Haptics from "expo-haptics"; import { Platform } from "react-native"; export const haptics = { light: () => { if (Platform.OS !== "web") { Haptics.impactAsync(Haptics.ImpactFeedbackStyle.Light); } }, medium: () => { if (Platform.OS !== "web") { Haptics.impactAsync(Haptics.ImpactFeedbackStyle.Medium); } }, heavy: () => { if (Platform.OS !== "web") { Haptics.impactAsync(Haptics.ImpactFeedbackStyle.Heavy); } }, success: () => { if (Platform.OS !== "web") { Haptics.notificationAsync(Haptics.NotificationFeedbackType.Success); } }, error: () => { if (Platform.OS !== "web") { Haptics.notificationAsync(Haptics.NotificationFeedbackType.Error); } }, }; // services/biometrics.ts import * as LocalAuthentication from "expo-local-authentication"; export async function authenticateWithBiometrics(): Promise<boolean> { const hasHardware = await LocalAuthentication.hasHardwareAsync(); if (!hasHardware) return false; const isEnrolled = await LocalAuthentication.isEnrolledAsync(); if (!isEnrolled) return false; const result = await LocalAuthentication.authenticateAsync({ promptMessage: "Authenticate to continue", fallbackLabel: "Use passcode", disableDeviceFallback: false, }); return result.success; } // services/notifications.ts import * as Notifications from "expo-notifications"; import { Platform } from "react-native"; import Constants from "expo-constants"; Notifications.setNotificationHandler({ handleNotification: async () => ({ shouldShowAlert: true, shouldPlaySound: true, shouldSetBadge: true, }), }); export async function registerForPushNotifications() { let token: string | undefined; if (Platform.OS === "android") { await Notifications.setNotificationChannelAsync("default", { name: "default", importance: Notifications.AndroidImportance.MAX, vibrationPattern: [0, 250, 250, 250], }); } const { status: existingStatus } = await Notifications.getPermissionsAsync(); let finalStatus = existingStatus; if (existingStatus !== "granted") { const { status } = await Notifications.requestPermissionsAsync(); finalStatus = status; } if (finalStatus !== "granted") { return null; } const projectId = Constants.expoConfig?.extra?.eas?.projectId; token = (await Notifications.getExpoPushTokenAsync({ projectId })).data; return token; } ``` ### Pattern 5: Platform-Specific Code ```typescript // components/ui/Button.tsx import { Platform, Pressable, StyleSheet, Text, ViewStyle } from 'react-native' import * as Haptics from 'expo-haptics' import Animated, { useAnimatedStyle, useSharedValue, withSpring, } from 'react-native-reanimated' const AnimatedPressable = Animated.createAnimatedComponent(Pressable) interface ButtonProps { title: string onPress: () => void variant?: 'primary' | 'secondary' | 'outline' disabled?: boolean } export function Button({ title, onPress, variant = 'primary', disabled = false, }: ButtonProps) { const scale = useSharedValue(1) const animatedStyle = useAnimatedStyle(() => ({ transform: [{ scale: scale.value }], })) const handlePressIn = () => { scale.value = withSpring(0.95) if (Platform.OS !== 'web') { Haptics.impactAsync(Haptics.ImpactFeedbackStyle.Light) } } const handlePressOut = () => { scale.value = withSpring(1) } return ( <AnimatedPressable onPress={onPress} onPressIn={handlePressIn} onPressOut={handlePressOut} disabled={disabled} style={[ styles.button, styles[variant], disabled && styles.disabled, animatedStyle, ]} > <Text style={[styles.text, styles[`${variant}Text`]]}>{title}</Text> </AnimatedPressable> ) } // Platform-specific files // Button.ios.tsx - iOS-specific implementation // Button.android.tsx - Android-specific implementation // Button.web.tsx - Web-specific implementation // Or use Platform.select const styles = StyleSheet.create({ button: { paddingVertical: 12, paddingHorizontal: 24, borderRadius: 8, alignItems: 'center', ...Platform.select({ ios: { shadowColor: '#000', shadowOffset: { width: 0, height: 2 }, shadowOpacity: 0.1, shadowRadius: 4, }, android: { elevation: 4, }, }), }, primary: { backgroundColor: '#007AFF', }, secondary: { backgroundColor: '#5856D6', }, outline: { backgroundColor: 'transparent', borderWidth: 1, borderColor: '#007AFF', }, disabled: { opacity: 0.5, }, text: { fontSize: 16, fontWeight: '600', }, primaryText: { color: '#FFFFFF', }, secondaryText: { color: '#FFFFFF', }, outlineText: { color: '#007AFF', }, }) ``` ### Pattern 6: Performance Optimization ```typescript // components/ProductList.tsx import { FlashList } from '@shopify/flash-list' import { memo, useCallback } from 'react' interface ProductListProps { products: Product[] onProductPress: (id: string) => void } // Memoize list item const ProductItem = memo(function ProductItem({ item, onPress, }: { item: Product onPress: (id: string) => void }) { const handlePress = useCallback(() => onPress(item.id), [item.id, onPress]) return ( <Pressable onPress={handlePress} style={styles.item}> <FastImage source={{ uri: item.image }} style={styles.image} resizeMode="cover" /> <Text style={styles.title}>{item.name}</Text> <Text style={styles.price}>${item.price}</Text> </Pressable> ) }) export function ProductList({ products, onProductPress }: ProductListProps) { const renderItem = useCallback( ({ item }: { item: Product }) => ( <ProductItem item={item} onPress={onProductPress} /> ), [onProductPress] ) const keyExtractor = useCallback((item: Product) => item.id, []) return ( <FlashList data={products} renderItem={renderItem} keyExtractor={keyExtractor} estimatedItemSize={100} // Performance optimizations removeClippedSubviews={true} maxToRenderPerBatch={10} windowSize={5} // Pull to refresh onRefresh={onRefresh} refreshing={isRefreshing} /> ) } ``` ## EAS Build & Submit ```json // eas.json { "cli": { "version": ">= 5.0.0" }, "build": { "development": { "developmentClient": true, "distribution": "internal", "ios": { "simulator": true } }, "preview": { "distribution": "internal", "android": { "buildType": "apk" } }, "production": { "autoIncrement": true } }, "submit": { "production": { "ios": { "appleId": "your@email.com", "ascAppId": "123456789" }, "android": { "serviceAccountKeyPath": "./google-services.json" } } } } ``` ```bash # Build commands eas build --platform ios --profile development eas build --platform android --profile preview eas build --platform all --profile production # Submit to stores eas submit --platform ios eas submit --platform android # OTA updates eas update --branch production --message "Bug fixes" ``` ## Best Practices ### Do's - **Use Expo** - Faster development, OTA updates, managed native code - **FlashList over FlatList** - Better performance for long lists - **Memoize components** - Prevent unnecessary re-renders - **Use Reanimated** - 60fps animations on native thread - **Test on real devices** - Simulators miss real-world issues ### Don'ts - **Don't inline styles** - Use StyleSheet.create for performance - **Don't fetch in render** - Use useEffect or React Query - **Don't ignore platform differences** - Test on both iOS and Android - **Don't store secrets in code** - Use environment variables - **Don't skip error boundaries** - Mobile crashes are unforgiving ## Resources - [Expo Documentation](https://docs.expo.dev/) - [Expo Router](https://docs.expo.dev/router/introduction/) - [React Native Performance](https://reactnative.dev/docs/performance) - [FlashList](https://shopify.github.io/flash-list/)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

react-state-management

Master modern React state management with Redux Toolkit, Zustand,

coding
⭐1
# React State Management Comprehensive guide to modern React state management patterns, from local component state to global stores and server state synchronization. ## When to Use This Skill - Setting up global state management in a React app - Choosing between Redux Toolkit, Zustand, or Jotai - Managing server state with React Query or SWR - Implementing optimistic updates - Debugging state-related issues - Migrating from legacy Redux to modern patterns ## Core Concepts ### 1. State Categories | Type | Description | Solutions | | ---------------- | ---------------------------- | ----------------------------- | | **Local State** | Component-specific, UI state | useState, useReducer | | **Global State** | Shared across components | Redux Toolkit, Zustand, Jotai | | **Server State** | Remote data, caching | React Query, SWR, RTK Query | | **URL State** | Route parameters, search | React Router, nuqs | | **Form State** | Input values, validation | React Hook Form, Formik | ### 2. Selection Criteria ``` Small app, simple state β†’ Zustand or Jotai Large app, complex state β†’ Redux Toolkit Heavy server interaction β†’ React Query + light client state Atomic/granular updates β†’ Jotai ``` ## Quick Start ### Zustand (Simplest) ```typescript // store/useStore.ts import { create } from 'zustand' import { devtools, persist } from 'zustand/middleware' interface AppState { user: User | null theme: 'light' | 'dark' setUser: (user: User | null) => void toggleTheme: () => void } export const useStore = create<AppState>()( devtools( persist( (set) => ({ user: null, theme: 'light', setUser: (user) => set({ user }), toggleTheme: () => set((state) => ({ theme: state.theme === 'light' ? 'dark' : 'light' })), }), { name: 'app-storage' } ) ) ) // Usage in component function Header() { const { user, theme, toggleTheme } = useStore() return ( <header className={theme}> {user?.name} <button onClick={toggleTheme}>Toggle Theme</button> </header> ) } ``` ## Patterns ### Pattern 1: Redux Toolkit with TypeScript ```typescript // store/index.ts import { configureStore } from "@reduxjs/toolkit"; import { TypedUseSelectorHook, useDispatch, useSelector } from "react-redux"; import userReducer from "./slices/userSlice"; import cartReducer from "./slices/cartSlice"; export const store = configureStore({ reducer: { user: userReducer, cart: cartReducer, }, middleware: (getDefaultMiddleware) => getDefaultMiddleware({ serializableCheck: { ignoredActions: ["persist/PERSIST"], }, }), }); export type RootState = ReturnType<typeof store.getState>; export type AppDispatch = typeof store.dispatch; // Typed hooks export const useAppDispatch: () => AppDispatch = useDispatch; export const useAppSelector: TypedUseSelectorHook<RootState> = useSelector; ``` ```typescript // store/slices/userSlice.ts import { createSlice, createAsyncThunk, PayloadAction } from "@reduxjs/toolkit"; interface User { id: string; email: string; name: string; } interface UserState { current: User | null; status: "idle" | "loading" | "succeeded" | "failed"; error: string | null; } const initialState: UserState = { current: null, status: "idle", error: null, }; export const fetchUser = createAsyncThunk( "user/fetchUser", async (userId: string, { rejectWithValue }) => { try { const response = await fetch(`/api/users/${userId}`); if (!response.ok) throw new Error("Failed to fetch user"); return await response.json(); } catch (error) { return rejectWithValue((error as Error).message); } }, ); const userSlice = createSlice({ name: "user", initialState, reducers: { setUser: (state, action: PayloadAction<User>) => { state.current = action.payload; state.status = "succeeded"; }, clearUser: (state) => { state.current = null; state.status = "idle"; }, }, extraReducers: (builder) => { builder .addCase(fetchUser.pending, (state) => { state.status = "loading"; state.error = null; }) .addCase(fetchUser.fulfilled, (state, action) => { state.status = "succeeded"; state.current = action.payload; }) .addCase(fetchUser.rejected, (state, action) => { state.status = "failed"; state.error = action.payload as string; }); }, }); export const { setUser, clearUser } = userSlice.actions; export default userSlice.reducer; ``` ### Pattern 2: Zustand with Slices (Scalable) ```typescript // store/slices/createUserSlice.ts import { StateCreator } from "zustand"; export interface UserSlice { user: User | null; isAuthenticated: boolean; login: (credentials: Credentials) => Promise<void>; logout: () => void; } export const createUserSlice: StateCreator< UserSlice & CartSlice, // Combined store type [], [], UserSlice > = (set, get) => ({ user: null, isAuthenticated: false, login: async (credentials) => { const user = await authApi.login(credentials); set({ user, isAuthenticated: true }); }, logout: () => { set({ user: null, isAuthenticated: false }); // Can access other slices // get().clearCart() }, }); // store/index.ts import { create } from "zustand"; import { createUserSlice, UserSlice } from "./slices/createUserSlice"; import { createCartSlice, CartSlice } from "./slices/createCartSlice"; type StoreState = UserSlice & CartSlice; export const useStore = create<StoreState>()((...args) => ({ ...createUserSlice(...args), ...createCartSlice(...args), })); // Selective subscriptions (prevents unnecessary re-renders) export const useUser = () => useStore((state) => state.user); export const useCart = () => useStore((state) => state.cart); ``` ### Pattern 3: Jotai for Atomic State ```typescript // atoms/userAtoms.ts import { atom } from 'jotai' import { atomWithStorage } from 'jotai/utils' // Basic atom export const userAtom = atom<User | null>(null) // Derived atom (computed) export const isAuthenticatedAtom = atom((get) => get(userAtom) !== null) // Atom with localStorage persistence export const themeAtom = atomWithStorage<'light' | 'dark'>('theme', 'light') // Async atom export const userProfileAtom = atom(async (get) => { const user = get(userAtom) if (!user) return null const response = await fetch(`/api/users/${user.id}/profile`) return response.json() }) // Write-only atom (action) export const logoutAtom = atom(null, (get, set) => { set(userAtom, null) set(cartAtom, []) localStorage.removeItem('token') }) // Usage function Profile() { const [user] = useAtom(userAtom) const [, logout] = useAtom(logoutAtom) const [profile] = useAtom(userProfileAtom) // Suspense-enabled return ( <Suspense fallback={<Skeleton />}> <ProfileContent profile={profile} onLogout={logout} /> </Suspense> ) } ``` ### Pattern 4: React Query for Server State ```typescript // hooks/useUsers.ts import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"; // Query keys factory export const userKeys = { all: ["users"] as const, lists: () => [...userKeys.all, "list"] as const, list: (filters: UserFilters) => [...userKeys.lists(), filters] as const, details: () => [...userKeys.all, "detail"] as const, detail: (id: string) => [...userKeys.details(), id] as const, }; // Fetch hook export function useUsers(filters: UserFilters) { return useQuery({ queryKey: userKeys.list(filters), queryFn: () => fetchUsers(filters), staleTime: 5 * 60 * 1000, // 5 minutes gcTime: 30 * 60 * 1000, // 30 minutes (formerly cacheTime) }); } // Single user hook export function useUser(id: string) { return useQuery({ queryKey: userKeys.detail(id), queryFn: () => fetchUser(id), enabled: !!id, // Don't fetch if no id }); } // Mutation with optimistic update export function useUpdateUser() { const queryClient = useQueryClient(); return useMutation({ mutationFn: updateUser, onMutate: async (newUser) => { // Cancel outgoing refetches await queryClient.cancelQueries({ queryKey: userKeys.detail(newUser.id), }); // Snapshot previous value const previousUser = queryClient.getQueryData( userKeys.detail(newUser.id), ); // Optimistically update queryClient.setQueryData(userKeys.detail(newUser.id), newUser); return { previousUser }; }, onError: (err, newUser, context) => { // Rollback on error queryClient.setQueryData( userKeys.detail(newUser.id), context?.previousUser, ); }, onSettled: (data, error, variables) => { // Refetch after mutation queryClient.invalidateQueries({ queryKey: userKeys.detail(variables.id), }); }, }); } ``` ### Pattern 5: Combining Client + Server State ```typescript // Zustand for client state const useUIStore = create<UIState>((set) => ({ sidebarOpen: true, modal: null, toggleSidebar: () => set((s) => ({ sidebarOpen: !s.sidebarOpen })), openModal: (modal) => set({ modal }), closeModal: () => set({ modal: null }), })) // React Query for server state function Dashboard() { const { sidebarOpen, toggleSidebar } = useUIStore() const { data: users, isLoading } = useUsers({ active: true }) const { data: stats } = useStats() if (isLoading) return <DashboardSkeleton /> return ( <div className={sidebarOpen ? 'with-sidebar' : ''}> <Sidebar open={sidebarOpen} onToggle={toggleSidebar} /> <main> <StatsCards stats={stats} /> <UserTable users={users} /> </main> </div> ) } ``` ## Best Practices ### Do's - **Colocate state** - Keep state as close to where it's used as possible - **Use selectors** - Prevent unnecessary re-renders with selective subscriptions - **Normalize data** - Flatten nested structures for easier updates - **Type everything** - Full TypeScript coverage prevents runtime errors - **Separate concerns** - Server state (React Query) vs client state (Zustand) ### Don'ts - **Don't over-globalize** - Not everything needs to be in global state - **Don't duplicate server state** - Let React Query manage it - **Don't mutate directly** - Always use immutable updates - **Don't store derived data** - Compute it instead - **Don't mix paradigms** - Pick one primary solution per category ## Migration Guides ### From Legacy Redux to RTK ```typescript // Before (legacy Redux) const ADD_TODO = "ADD_TODO"; const addTodo = (text) => ({ type: ADD_TODO, payload: text }); function todosReducer(state = [], action) { switch (action.type) { case ADD_TODO: return [...state, { text: action.payload, completed: false }]; default: return state; } } // After (Redux Toolkit) const todosSlice = createSlice({ name: "todos", initialState: [], reducers: { addTodo: (state, action: PayloadAction<string>) => { // Immer allows "mutations" state.push({ text: action.payload, completed: false }); }, }, }); ``` ## Resources - [Redux Toolkit Documentation](https://redux-toolkit.js.org/) - [Zustand GitHub](https://github.com/pmndrs/zustand) - [Jotai Documentation](https://jotai.org/) - [TanStack Query](https://tanstack.com/query)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

unity-ecs-patterns

Master Unity ECS (Entity Component System) with DOTS, Jobs, and

coding
⭐1
# Unity ECS Patterns Production patterns for Unity's Data-Oriented Technology Stack (DOTS) including Entity Component System, Job System, and Burst Compiler. ## When to Use This Skill - Building high-performance Unity games - Managing thousands of entities efficiently - Implementing data-oriented game systems - Optimizing CPU-bound game logic - Converting OOP game code to ECS - Using Jobs and Burst for parallelization ## Core Concepts ### 1. ECS vs OOP | Aspect | Traditional OOP | ECS/DOTS | | ----------- | ----------------- | --------------- | | Data layout | Object-oriented | Data-oriented | | Memory | Scattered | Contiguous | | Processing | Per-object | Batched | | Scaling | Poor with count | Linear scaling | | Best for | Complex behaviors | Mass simulation | ### 2. DOTS Components ``` Entity: Lightweight ID (no data) Component: Pure data (no behavior) System: Logic that processes components World: Container for entities Archetype: Unique combination of components Chunk: Memory block for same-archetype entities ``` ## Patterns ### Pattern 1: Basic ECS Setup ```csharp using Unity.Entities; using Unity.Mathematics; using Unity.Transforms; using Unity.Burst; using Unity.Collections; // Component: Pure data, no methods public struct Speed : IComponentData { public float Value; } public struct Health : IComponentData { public float Current; public float Max; } public struct Target : IComponentData { public Entity Value; } // Tag component (zero-size marker) public struct EnemyTag : IComponentData { } public struct PlayerTag : IComponentData { } // Buffer component (variable-size array) [InternalBufferCapacity(8)] public struct InventoryItem : IBufferElementData { public int ItemId; public int Quantity; } // Shared component (grouped entities) public struct TeamId : ISharedComponentData { public int Value; } ``` ### Pattern 2: Systems with ISystem (Recommended) ```csharp using Unity.Entities; using Unity.Transforms; using Unity.Mathematics; using Unity.Burst; // ISystem: Unmanaged, Burst-compatible, highest performance [BurstCompile] public partial struct MovementSystem : ISystem { [BurstCompile] public void OnCreate(ref SystemState state) { // Require components before system runs state.RequireForUpdate<Speed>(); } [BurstCompile] public void OnUpdate(ref SystemState state) { float deltaTime = SystemAPI.Time.DeltaTime; // Simple foreach - auto-generates job foreach (var (transform, speed) in SystemAPI.Query<RefRW<LocalTransform>, RefRO<Speed>>()) { transform.ValueRW.Position += new float3(0, 0, speed.ValueRO.Value * deltaTime); } } [BurstCompile] public void OnDestroy(ref SystemState state) { } } // With explicit job for more control [BurstCompile] public partial struct MovementJobSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { var job = new MoveJob { DeltaTime = SystemAPI.Time.DeltaTime }; state.Dependency = job.ScheduleParallel(state.Dependency); } } [BurstCompile] public partial struct MoveJob : IJobEntity { public float DeltaTime; void Execute(ref LocalTransform transform, in Speed speed) { transform.Position += new float3(0, 0, speed.Value * DeltaTime); } } ``` ### Pattern 3: Entity Queries ```csharp [BurstCompile] public partial struct QueryExamplesSystem : ISystem { private EntityQuery _enemyQuery; public void OnCreate(ref SystemState state) { // Build query manually for complex cases _enemyQuery = new EntityQueryBuilder(Allocator.Temp) .WithAll<EnemyTag, Health, LocalTransform>() .WithNone<Dead>() .WithOptions(EntityQueryOptions.FilterWriteGroup) .Build(ref state); } [BurstCompile] public void OnUpdate(ref SystemState state) { // SystemAPI.Query - simplest approach foreach (var (health, entity) in SystemAPI.Query<RefRW<Health>>() .WithAll<EnemyTag>() .WithEntityAccess()) { if (health.ValueRO.Current <= 0) { // Mark for destruction SystemAPI.GetSingleton<EndSimulationEntityCommandBufferSystem.Singleton>() .CreateCommandBuffer(state.WorldUnmanaged) .DestroyEntity(entity); } } // Get count int enemyCount = _enemyQuery.CalculateEntityCount(); // Get all entities var enemies = _enemyQuery.ToEntityArray(Allocator.Temp); // Get component arrays var healths = _enemyQuery.ToComponentDataArray<Health>(Allocator.Temp); } } ``` ### Pattern 4: Entity Command Buffers (Structural Changes) ```csharp // Structural changes (create/destroy/add/remove) require command buffers [BurstCompile] [UpdateInGroup(typeof(SimulationSystemGroup))] public partial struct SpawnSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { var ecbSingleton = SystemAPI.GetSingleton<BeginSimulationEntityCommandBufferSystem.Singleton>(); var ecb = ecbSingleton.CreateCommandBuffer(state.WorldUnmanaged); foreach (var (spawner, transform) in SystemAPI.Query<RefRW<Spawner>, RefRO<LocalTransform>>()) { spawner.ValueRW.Timer -= SystemAPI.Time.DeltaTime; if (spawner.ValueRO.Timer <= 0) { spawner.ValueRW.Timer = spawner.ValueRO.Interval; // Create entity (deferred until sync point) Entity newEntity = ecb.Instantiate(spawner.ValueRO.Prefab); // Set component values ecb.SetComponent(newEntity, new LocalTransform { Position = transform.ValueRO.Position, Rotation = quaternion.identity, Scale = 1f }); // Add component ecb.AddComponent(newEntity, new Speed { Value = 5f }); } } } } // Parallel ECB usage [BurstCompile] public partial struct ParallelSpawnJob : IJobEntity { public EntityCommandBuffer.ParallelWriter ECB; void Execute([EntityIndexInQuery] int index, in Spawner spawner) { Entity e = ECB.Instantiate(index, spawner.Prefab); ECB.AddComponent(index, e, new Speed { Value = 5f }); } } ``` ### Pattern 5: Aspect (Grouping Components) ```csharp using Unity.Entities; using Unity.Transforms; using Unity.Mathematics; // Aspect: Groups related components for cleaner code public readonly partial struct CharacterAspect : IAspect { public readonly Entity Entity; private readonly RefRW<LocalTransform> _transform; private readonly RefRO<Speed> _speed; private readonly RefRW<Health> _health; // Optional component [Optional] private readonly RefRO<Shield> _shield; // Buffer private readonly DynamicBuffer<InventoryItem> _inventory; public float3 Position { get => _transform.ValueRO.Position; set => _transform.ValueRW.Position = value; } public float CurrentHealth => _health.ValueRO.Current; public float MaxHealth => _health.ValueRO.Max; public float MoveSpeed => _speed.ValueRO.Value; public bool HasShield => _shield.IsValid; public float ShieldAmount => HasShield ? _shield.ValueRO.Amount : 0f; public void TakeDamage(float amount) { float remaining = amount; if (HasShield && _shield.ValueRO.Amount > 0) { // Shield absorbs damage first remaining = math.max(0, amount - _shield.ValueRO.Amount); } _health.ValueRW.Current = math.max(0, _health.ValueRO.Current - remaining); } public void Move(float3 direction, float deltaTime) { _transform.ValueRW.Position += direction * _speed.ValueRO.Value * deltaTime; } public void AddItem(int itemId, int quantity) { _inventory.Add(new InventoryItem { ItemId = itemId, Quantity = quantity }); } } // Using aspect in system [BurstCompile] public partial struct CharacterSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { float dt = SystemAPI.Time.DeltaTime; foreach (var character in SystemAPI.Query<CharacterAspect>()) { character.Move(new float3(1, 0, 0), dt); if (character.CurrentHealth < character.MaxHealth * 0.5f) { // Low health logic } } } } ``` ### Pattern 6: Singleton Components ```csharp // Singleton: Exactly one entity with this component public struct GameConfig : IComponentData { public float DifficultyMultiplier; public int MaxEnemies; public float SpawnRate; } public struct GameState : IComponentData { public int Score; public int Wave; public float TimeRemaining; } // Create singleton on world creation public partial struct GameInitSystem : ISystem { public void OnCreate(ref SystemState state) { var entity = state.EntityManager.CreateEntity(); state.EntityManager.AddComponentData(entity, new GameConfig { DifficultyMultiplier = 1.0f, MaxEnemies = 100, SpawnRate = 2.0f }); state.EntityManager.AddComponentData(entity, new GameState { Score = 0, Wave = 1, TimeRemaining = 120f }); } } // Access singleton in system [BurstCompile] public partial struct ScoreSystem : ISystem { [BurstCompile] public void OnUpdate(ref SystemState state) { // Read singleton var config = SystemAPI.GetSingleton<GameConfig>(); // Write singleton ref var gameState = ref SystemAPI.GetSingletonRW<GameState>().ValueRW; gameState.TimeRemaining -= SystemAPI.Time.DeltaTime; // Check exists if (SystemAPI.HasSingleton<GameConfig>()) { // ... } } } ``` ### Pattern 7: Baking (Converting GameObjects) ```csharp using Unity.Entities; using UnityEngine; // Authoring component (MonoBehaviour in Editor) public class EnemyAuthoring : MonoBehaviour { public float Speed = 5f; public float Health = 100f; public GameObject ProjectilePrefab; class Baker : Baker<EnemyAuthoring> { public override void Bake(EnemyAuthoring authoring) { var entity = GetEntity(TransformUsageFlags.Dynamic); AddComponent(entity, new Speed { Value = authoring.Speed }); AddComponent(entity, new Health { Current = authoring.Health, Max = authoring.Health }); AddComponent(entity, new EnemyTag()); if (authoring.ProjectilePrefab != null) { AddComponent(entity, new ProjectilePrefab { Value = GetEntity(authoring.ProjectilePrefab, TransformUsageFlags.Dynamic) }); } } } } // Complex baking with dependencies public class SpawnerAuthoring : MonoBehaviour { public GameObject[] Prefabs; public float Interval = 1f; class Baker : Baker<SpawnerAuthoring> { public override void Bake(SpawnerAuthoring authoring) { var entity = GetEntity(TransformUsageFlags.Dynamic); AddComponent(entity, new Spawner { Interval = authoring.Interval, Timer = 0f }); // Bake buffer of prefabs var buffer = AddBuffer<SpawnPrefabElement>(entity); foreach (var prefab in authoring.Prefabs) { buffer.Add(new SpawnPrefabElement { Prefab = GetEntity(prefab, TransformUsageFlags.Dynamic) }); } // Declare dependencies DependsOn(authoring.Prefabs); } } } ``` ### Pattern 8: Jobs with Native Collections ```csharp using Unity.Jobs; using Unity.Collections; using Unity.Burst; using Unity.Mathematics; [BurstCompile] public struct SpatialHashJob : IJobParallelFor { [ReadOnly] public NativeArray<float3> Positions; // Thread-safe write to hash map public NativeParallelMultiHashMap<int, int>.ParallelWriter HashMap; public float CellSize; public void Execute(int index) { float3 pos = Positions[index]; int hash = GetHash(pos); HashMap.Add(hash, index); } int GetHash(float3 pos) { int x = (int)math.floor(pos.x / CellSize); int y = (int)math.floor(pos.y / CellSize); int z = (int)math.floor(pos.z / CellSize); return x * 73856093 ^ y * 19349663 ^ z * 83492791; } } [BurstCompile] public partial struct SpatialHashSystem : ISystem { private NativeParallelMultiHashMap<int, int> _hashMap; public void OnCreate(ref SystemState state) { _hashMap = new NativeParallelMultiHashMap<int, int>(10000, Allocator.Persistent); } public void OnDestroy(ref SystemState state) { _hashMap.Dispose(); } [BurstCompile] public void OnUpdate(ref SystemState state) { var query = SystemAPI.QueryBuilder() .WithAll<LocalTransform>() .Build(); int count = query.CalculateEntityCount(); // Resize if needed if (_hashMap.Capacity < count) { _hashMap.Capacity = count * 2; } _hashMap.Clear(); // Get positions var positions = query.ToComponentDataArray<LocalTransform>(Allocator.TempJob); var posFloat3 = new NativeArray<float3>(count, Allocator.TempJob); for (int i = 0; i < count; i++) { posFloat3[i] = positions[i].Position; } // Build hash map var hashJob = new SpatialHashJob { Positions = posFloat3, HashMap = _hashMap.AsParallelWriter(), CellSize = 10f }; state.Dependency = hashJob.Schedule(count, 64, state.Dependency); // Cleanup positions.Dispose(state.Dependency); posFloat3.Dispose(state.Dependency); } } ``` ## Performance Tips ```csharp // 1. Use Burst everywhere [BurstCompile] public partial struct MySystem : ISystem { } // 2. Prefer IJobEntity over manual iteration [BurstCompile] partial struct OptimizedJob : IJobEntity { void Execute(ref LocalTransform transform) { } } // 3. Schedule parallel when possible state.Dependency = job.ScheduleParallel(state.Dependency); // 4. Use ScheduleParallel with chunk iteration [BurstCompile] partial struct ChunkJob : IJobChunk { public ComponentTypeHandle<Health> HealthHandle; public void Execute(in ArchetypeChunk chunk, int unfilteredChunkIndex, bool useEnabledMask, in v128 chunkEnabledMask) { var healths = chunk.GetNativeArray(ref HealthHandle); for (int i = 0; i < chunk.Count; i++) { // Process } } } // 5. Avoid structural changes in hot paths // Use enableable components instead of add/remove public struct Disabled : IComponentData, IEnableableComponent { } ``` ## Best Practices ### Do's - **Use ISystem over SystemBase** - Better performance - **Burst compile everything** - Massive speedup - **Batch structural changes** - Use ECB - **Profile with Profiler** - Identify bottlenecks - **Use Aspects** - Clean component grouping ### Don'ts - **Don't use managed types** - Breaks Burst - **Don't structural change in jobs** - Use ECB - **Don't over-architect** - Start simple - **Don't ignore chunk utilization** - Group similar entities - **Don't forget disposal** - Native collections leak ## Resources - [Unity DOTS Documentation](https://docs.unity3d.com/Packages/com.unity.entities@latest) - [Unity DOTS Samples](https://github.com/Unity-Technologies/EntityComponentSystemSamples) - [Burst User Guide](https://docs.unity3d.com/Packages/com.unity.burst@latest)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

on-call-handoff-patterns

Master on-call shift handoffs with context transfer, escalation

coding
⭐1
# On-Call Handoff Patterns Effective patterns for on-call shift transitions, ensuring continuity, context transfer, and reliable incident response across shifts. ## When to Use This Skill - Transitioning on-call responsibilities - Writing shift handoff summaries - Documenting ongoing investigations - Establishing on-call rotation procedures - Improving handoff quality - Onboarding new on-call engineers ## Core Concepts ### 1. Handoff Components | Component | Purpose | | -------------------------- | ----------------------- | | **Active Incidents** | What's currently broken | | **Ongoing Investigations** | Issues being debugged | | **Recent Changes** | Deployments, configs | | **Known Issues** | Workarounds in place | | **Upcoming Events** | Maintenance, releases | ### 2. Handoff Timing ``` Recommended: 30 min overlap between shifts Outgoing: β”œβ”€β”€ 15 min: Write handoff document └── 15 min: Sync call with incoming Incoming: β”œβ”€β”€ 15 min: Review handoff document β”œβ”€β”€ 15 min: Sync call with outgoing └── 5 min: Verify alerting setup ``` ## Templates ### Template 1: Shift Handoff Document ````markdown # On-Call Handoff: Platform Team **Outgoing**: @alice (2024-01-15 to 2024-01-22) **Incoming**: @bob (2024-01-22 to 2024-01-29) **Handoff Time**: 2024-01-22 09:00 UTC --- ## πŸ”΄ Active Incidents ### None currently active No active incidents at handoff time. --- ## 🟑 Ongoing Investigations ### 1. Intermittent API Timeouts (ENG-1234) **Status**: Investigating **Started**: 2024-01-20 **Impact**: ~0.1% of requests timing out **Context**: - Timeouts correlate with database backup window (02:00-03:00 UTC) - Suspect backup process causing lock contention - Added extra logging in PR #567 (deployed 01/21) **Next Steps**: - [ ] Review new logs after tonight's backup - [ ] Consider moving backup window if confirmed **Resources**: - Dashboard: [API Latency](https://grafana/d/api-latency) - Thread: #platform-eng (01/20, 14:32) --- ### 2. Memory Growth in Auth Service (ENG-1235) **Status**: Monitoring **Started**: 2024-01-18 **Impact**: None yet (proactive) **Context**: - Memory usage growing ~5% per day - No memory leak found in profiling - Suspect connection pool not releasing properly **Next Steps**: - [ ] Review heap dump from 01/21 - [ ] Consider restart if usage > 80% **Resources**: - Dashboard: [Auth Service Memory](https://grafana/d/auth-memory) - Analysis doc: [Memory Investigation](https://docs/eng-1235) --- ## 🟒 Resolved This Shift ### Payment Service Outage (2024-01-19) - **Duration**: 23 minutes - **Root Cause**: Database connection exhaustion - **Resolution**: Rolled back v2.3.4, increased pool size - **Postmortem**: [POSTMORTEM-89](https://docs/postmortem-89) - **Follow-up tickets**: ENG-1230, ENG-1231 --- ## πŸ“‹ Recent Changes ### Deployments | Service | Version | Time | Notes | | ------------ | ------- | ----------- | -------------------------- | | api-gateway | v3.2.1 | 01/21 14:00 | Bug fix for header parsing | | user-service | v2.8.0 | 01/20 10:00 | New profile features | | auth-service | v4.1.2 | 01/19 16:00 | Security patch | ### Configuration Changes - 01/21: Increased API rate limit from 1000 to 1500 RPS - 01/20: Updated database connection pool max from 50 to 75 ### Infrastructure - 01/20: Added 2 nodes to Kubernetes cluster - 01/19: Upgraded Redis from 6.2 to 7.0 --- ## ⚠️ Known Issues & Workarounds ### 1. Slow Dashboard Loading **Issue**: Grafana dashboards slow on Monday mornings **Workaround**: Wait 5 min after 08:00 UTC for cache warm-up **Ticket**: OPS-456 (P3) ### 2. Flaky Integration Test **Issue**: `test_payment_flow` fails intermittently in CI **Workaround**: Re-run failed job (usually passes on retry) **Ticket**: ENG-1200 (P2) --- ## πŸ“… Upcoming Events | Date | Event | Impact | Contact | | ----------- | -------------------- | ------------------- | ------------- | | 01/23 02:00 | Database maintenance | 5 min read-only | @dba-team | | 01/24 14:00 | Major release v5.0 | Monitor closely | @release-team | | 01/25 | Marketing campaign | 2x traffic expected | @platform | --- ## πŸ“ž Escalation Reminders | Issue Type | First Escalation | Second Escalation | | --------------- | -------------------- | ----------------- | | Payment issues | @payments-oncall | @payments-manager | | Auth issues | @auth-oncall | @security-team | | Database issues | @dba-team | @infra-manager | | Unknown/severe | @engineering-manager | @vp-engineering | --- ## πŸ”§ Quick Reference ### Common Commands ```bash # Check service health kubectl get pods -A | grep -v Running # Recent deployments kubectl get events --sort-by='.lastTimestamp' | tail -20 # Database connections psql -c "SELECT count(*) FROM pg_stat_activity;" # Clear cache (emergency only) redis-cli FLUSHDB ``` ```` ### Important Links - [Runbooks](https://wiki/runbooks) - [Service Catalog](https://wiki/services) - [Incident Slack](https://slack.com/incidents) - [PagerDuty](https://pagerduty.com/schedules) --- ## Handoff Checklist ### Outgoing Engineer - [x] Document active incidents - [x] Document ongoing investigations - [x] List recent changes - [x] Note known issues - [x] Add upcoming events - [x] Sync with incoming engineer ### Incoming Engineer - [ ] Read this document - [ ] Join sync call - [ ] Verify PagerDuty is routing to you - [ ] Verify Slack notifications working - [ ] Check VPN/access working - [ ] Review critical dashboards ```` ### Template 2: Quick Handoff (Async) ```markdown # Quick Handoff: @alice β†’ @bob ## TL;DR - No active incidents - 1 investigation ongoing (API timeouts, see ENG-1234) - Major release tomorrow (01/24) - be ready for issues ## Watch List 1. API latency around 02:00-03:00 UTC (backup window) 2. Auth service memory (restart if > 80%) ## Recent - Deployed api-gateway v3.2.1 yesterday (stable) - Increased rate limits to 1500 RPS ## Coming Up - 01/23 02:00 - DB maintenance (5 min read-only) - 01/24 14:00 - v5.0 release ## Questions? I'll be available on Slack until 17:00 today. ```` ### Template 3: Incident Handoff (Mid-Incident) ```markdown # INCIDENT HANDOFF: Payment Service Degradation **Incident Start**: 2024-01-22 08:15 UTC **Current Status**: Mitigating **Severity**: SEV2 --- ## Current State - Error rate: 15% (down from 40%) - Mitigation in progress: scaling up pods - ETA to resolution: ~30 min ## What We Know 1. Root cause: Memory pressure on payment-service pods 2. Triggered by: Unusual traffic spike (3x normal) 3. Contributing: Inefficient query in checkout flow ## What We've Done - Scaled payment-service from 5 β†’ 15 pods - Enabled rate limiting on checkout endpoint - Disabled non-critical features ## What Needs to Happen 1. Monitor error rate - should reach <1% in ~15 min 2. If not improving, escalate to @payments-manager 3. Once stable, begin root cause investigation ## Key People - Incident Commander: @alice (handing off) - Comms Lead: @charlie - Technical Lead: @bob (incoming) ## Communication - Status page: Updated at 08:45 - Customer support: Notified - Exec team: Aware ## Resources - Incident channel: #inc-20240122-payment - Dashboard: [Payment Service](https://grafana/d/payments) - Runbook: [Payment Degradation](https://wiki/runbooks/payments) --- **Incoming on-call (@bob) - Please confirm you have:** - [ ] Joined #inc-20240122-payment - [ ] Access to dashboards - [ ] Understand current state - [ ] Know escalation path ``` ## Handoff Sync Meeting ### Agenda (15 minutes) ```markdown ## Handoff Sync: @alice β†’ @bob 1. **Active Issues** (5 min) - Walk through any ongoing incidents - Discuss investigation status - Transfer context and theories 2. **Recent Changes** (3 min) - Deployments to watch - Config changes - Known regressions 3. **Upcoming Events** (3 min) - Maintenance windows - Expected traffic changes - Releases planned 4. **Questions** (4 min) - Clarify anything unclear - Confirm access and alerting - Exchange contact info ``` ## On-Call Best Practices ### Before Your Shift ```markdown ## Pre-Shift Checklist ### Access Verification - [ ] VPN working - [ ] kubectl access to all clusters - [ ] Database read access - [ ] Log aggregator access (Splunk/Datadog) - [ ] PagerDuty app installed and logged in ### Alerting Setup - [ ] PagerDuty schedule shows you as primary - [ ] Phone notifications enabled - [ ] Slack notifications for incident channels - [ ] Test alert received and acknowledged ### Knowledge Refresh - [ ] Review recent incidents (past 2 weeks) - [ ] Check service changelog - [ ] Skim critical runbooks - [ ] Know escalation contacts ### Environment Ready - [ ] Laptop charged and accessible - [ ] Phone charged - [ ] Quiet space available for calls - [ ] Secondary contact identified (if traveling) ``` ### During Your Shift ```markdown ## Daily On-Call Routine ### Morning (start of day) - [ ] Check overnight alerts - [ ] Review dashboards for anomalies - [ ] Check for any P0/P1 tickets created - [ ] Skim incident channels for context ### Throughout Day - [ ] Respond to alerts within SLA - [ ] Document investigation progress - [ ] Update team on significant issues - [ ] Triage incoming pages ### End of Day - [ ] Hand off any active issues - [ ] Update investigation docs - [ ] Note anything for next shift ``` ### After Your Shift ```markdown ## Post-Shift Checklist - [ ] Complete handoff document - [ ] Sync with incoming on-call - [ ] Verify PagerDuty routing changed - [ ] Close/update investigation tickets - [ ] File postmortems for any incidents - [ ] Take time off if shift was stressful ``` ## Escalation Guidelines ### When to Escalate ```markdown ## Escalation Triggers ### Immediate Escalation - SEV1 incident declared - Data breach suspected - Unable to diagnose within 30 min - Customer or legal escalation received ### Consider Escalation - Issue spans multiple teams - Requires expertise you don't have - Business impact exceeds threshold - You're uncertain about next steps ### How to Escalate 1. Page the appropriate escalation path 2. Provide brief context in Slack 3. Stay engaged until escalation acknowledges 4. Hand off cleanly, don't just disappear ``` ## Best Practices ### Do's - **Document everything** - Future you will thank you - **Escalate early** - Better safe than sorry - **Take breaks** - Alert fatigue is real - **Keep handoffs synchronous** - Async loses context - **Test your setup** - Before incidents, not during ### Don'ts - **Don't skip handoffs** - Context loss causes incidents - **Don't hero** - Escalate when needed - **Don't ignore alerts** - Even if they seem minor - **Don't work sick** - Swap shifts instead - **Don't disappear** - Stay reachable during shift ## Resources - [Google SRE - Being On-Call](https://sre.google/sre-book/being-on-call/) - [PagerDuty On-Call Guide](https://www.pagerduty.com/resources/learn/on-call-management/) - [Increment On-Call Issue](https://increment.com/on-call/)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

gitops-workflow

Implement GitOps workflows with ArgoCD and Flux for automated,

architecture
⭐1
# GitOps Workflow Complete guide to implementing GitOps workflows with ArgoCD and Flux for automated Kubernetes deployments. ## Purpose Implement declarative, Git-based continuous delivery for Kubernetes using ArgoCD or Flux CD, following OpenGitOps principles. ## When to Use This Skill - Set up GitOps for Kubernetes clusters - Automate application deployments from Git - Implement progressive delivery strategies - Manage multi-cluster deployments - Configure automated sync policies - Set up secret management in GitOps ## OpenGitOps Principles 1. **Declarative** - Entire system described declaratively 2. **Versioned and Immutable** - Desired state stored in Git 3. **Pulled Automatically** - Software agents pull desired state 4. **Continuously Reconciled** - Agents reconcile actual vs desired state ## ArgoCD Setup ### 1. Installation ```bash # Create namespace kubectl create namespace argocd # Install ArgoCD kubectl apply -n argocd -f https://raw.githubusercontent.com/argoproj/argo-cd/stable/manifests/install.yaml # Get admin password kubectl -n argocd get secret argocd-initial-admin-secret -o jsonpath="{.data.password}" | base64 -d ``` **Reference:** See `references/argocd-setup.md` for detailed setup ### 2. Repository Structure ``` gitops-repo/ β”œβ”€β”€ apps/ β”‚ β”œβ”€β”€ production/ β”‚ β”‚ β”œβ”€β”€ app1/ β”‚ β”‚ β”‚ β”œβ”€β”€ kustomization.yaml β”‚ β”‚ β”‚ └── deployment.yaml β”‚ β”‚ └── app2/ β”‚ └── staging/ β”œβ”€β”€ infrastructure/ β”‚ β”œβ”€β”€ ingress-nginx/ β”‚ β”œβ”€β”€ cert-manager/ β”‚ └── monitoring/ └── argocd/ β”œβ”€β”€ applications/ └── projects/ ``` ### 3. Create Application ```yaml # argocd/applications/my-app.yaml apiVersion: argoproj.io/v1alpha1 kind: Application metadata: name: my-app namespace: argocd spec: project: default source: repoURL: https://github.com/org/gitops-repo targetRevision: main path: apps/production/my-app destination: server: https://kubernetes.default.svc namespace: production syncPolicy: automated: prune: true selfHeal: true syncOptions: - CreateNamespace=true ``` ### 4. App of Apps Pattern ```yaml apiVersion: argoproj.io/v1alpha1 kind: Application metadata: name: applications namespace: argocd spec: project: default source: repoURL: https://github.com/org/gitops-repo targetRevision: main path: argocd/applications destination: server: https://kubernetes.default.svc namespace: argocd syncPolicy: automated: {} ``` ## Flux CD Setup ### 1. Installation ```bash # Install Flux CLI curl -s https://fluxcd.io/install.sh | sudo bash # Bootstrap Flux flux bootstrap github \ --owner=org \ --repository=gitops-repo \ --branch=main \ --path=clusters/production \ --personal ``` ### 2. Create GitRepository ```yaml apiVersion: source.toolkit.fluxcd.io/v1 kind: GitRepository metadata: name: my-app namespace: flux-system spec: interval: 1m url: https://github.com/org/my-app ref: branch: main ``` ### 3. Create Kustomization ```yaml apiVersion: kustomize.toolkit.fluxcd.io/v1 kind: Kustomization metadata: name: my-app namespace: flux-system spec: interval: 5m path: ./deploy prune: true sourceRef: kind: GitRepository name: my-app ``` ## Sync Policies ### Auto-Sync Configuration **ArgoCD:** ```yaml syncPolicy: automated: prune: true # Delete resources not in Git selfHeal: true # Reconcile manual changes allowEmpty: false retry: limit: 5 backoff: duration: 5s factor: 2 maxDuration: 3m ``` **Flux:** ```yaml spec: interval: 1m prune: true wait: true timeout: 5m ``` **Reference:** See `references/sync-policies.md` ## Progressive Delivery ### Canary Deployment with ArgoCD Rollouts ```yaml apiVersion: argoproj.io/v1alpha1 kind: Rollout metadata: name: my-app spec: replicas: 5 strategy: canary: steps: - setWeight: 20 - pause: { duration: 1m } - setWeight: 50 - pause: { duration: 2m } - setWeight: 100 ``` ### Blue-Green Deployment ```yaml strategy: blueGreen: activeService: my-app previewService: my-app-preview autoPromotionEnabled: false ``` ## Secret Management ### External Secrets Operator ```yaml apiVersion: external-secrets.io/v1beta1 kind: ExternalSecret metadata: name: db-credentials spec: refreshInterval: 1h secretStoreRef: name: aws-secrets-manager kind: SecretStore target: name: db-credentials data: - secretKey: password remoteRef: key: prod/db/password ``` ### Sealed Secrets ```bash # Encrypt secret kubeseal --format yaml < secret.yaml > sealed-secret.yaml # Commit sealed-secret.yaml to Git ``` ## Best Practices 1. **Use separate repos or branches** for different environments 2. **Implement RBAC** for Git repositories 3. **Enable notifications** for sync failures 4. **Use health checks** for custom resources 5. **Implement approval gates** for production 6. **Keep secrets out of Git** (use External Secrets) 7. **Use App of Apps pattern** for organization 8. **Tag releases** for easy rollback 9. **Monitor sync status** with alerts 10. **Test changes** in staging first ## Troubleshooting **Sync failures:** ```bash argocd app get my-app argocd app sync my-app --prune ``` **Out of sync status:** ```bash argocd app diff my-app argocd app sync my-app --force ``` ## Related Skills - `k8s-manifest-generator` - For creating manifests - `helm-chart-scaffolding` - For packaging applications
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

fastapi-templates

Create production-ready FastAPI projects with async patterns,

coding
⭐1
# FastAPI Project Templates Production-ready FastAPI project structures with async patterns, dependency injection, middleware, and best practices for building high-performance APIs. ## When to Use This Skill - Starting new FastAPI projects from scratch - Implementing async REST APIs with Python - Building high-performance web services and microservices - Creating async applications with PostgreSQL, MongoDB - Setting up API projects with proper structure and testing ## Core Concepts ### 1. Project Structure **Recommended Layout:** ``` app/ β”œβ”€β”€ api/ # API routes β”‚ β”œβ”€β”€ v1/ β”‚ β”‚ β”œβ”€β”€ endpoints/ β”‚ β”‚ β”‚ β”œβ”€β”€ users.py β”‚ β”‚ β”‚ β”œβ”€β”€ auth.py β”‚ β”‚ β”‚ └── items.py β”‚ β”‚ └── router.py β”‚ └── dependencies.py # Shared dependencies β”œβ”€β”€ core/ # Core configuration β”‚ β”œβ”€β”€ config.py β”‚ β”œβ”€β”€ security.py β”‚ └── database.py β”œβ”€β”€ models/ # Database models β”‚ β”œβ”€β”€ user.py β”‚ └── item.py β”œβ”€β”€ schemas/ # Pydantic schemas β”‚ β”œβ”€β”€ user.py β”‚ └── item.py β”œβ”€β”€ services/ # Business logic β”‚ β”œβ”€β”€ user_service.py β”‚ └── auth_service.py β”œβ”€β”€ repositories/ # Data access β”‚ β”œβ”€β”€ user_repository.py β”‚ └── item_repository.py └── main.py # Application entry ``` ### 2. Dependency Injection FastAPI's built-in DI system using `Depends`: - Database session management - Authentication/authorization - Shared business logic - Configuration injection ### 3. Async Patterns Proper async/await usage: - Async route handlers - Async database operations - Async background tasks - Async middleware ## Implementation Patterns ### Pattern 1: Complete FastAPI Application ```python # main.py from fastapi import FastAPI, Depends from fastapi.middleware.cors import CORSMiddleware from contextlib import asynccontextmanager @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan events.""" # Startup await database.connect() yield # Shutdown await database.disconnect() app = FastAPI( title="API Template", version="1.0.0", lifespan=lifespan ) # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Include routers from app.api.v1.router import api_router app.include_router(api_router, prefix="/api/v1") # core/config.py from pydantic_settings import BaseSettings from functools import lru_cache class Settings(BaseSettings): """Application settings.""" DATABASE_URL: str SECRET_KEY: str ACCESS_TOKEN_EXPIRE_MINUTES: int = 30 API_V1_STR: str = "/api/v1" class Config: env_file = ".env" @lru_cache() def get_settings() -> Settings: return Settings() # core/database.py from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from app.core.config import get_settings settings = get_settings() engine = create_async_engine( settings.DATABASE_URL, echo=True, future=True ) AsyncSessionLocal = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) Base = declarative_base() async def get_db() -> AsyncSession: """Dependency for database session.""" async with AsyncSessionLocal() as session: try: yield session await session.commit() except Exception: await session.rollback() raise finally: await session.close() ``` ### Pattern 2: CRUD Repository Pattern ```python # repositories/base_repository.py from typing import Generic, TypeVar, Type, Optional, List from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from pydantic import BaseModel ModelType = TypeVar("ModelType") CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel) UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel) class BaseRepository(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): """Base repository for CRUD operations.""" def __init__(self, model: Type[ModelType]): self.model = model async def get(self, db: AsyncSession, id: int) -> Optional[ModelType]: """Get by ID.""" result = await db.execute( select(self.model).where(self.model.id == id) ) return result.scalars().first() async def get_multi( self, db: AsyncSession, skip: int = 0, limit: int = 100 ) -> List[ModelType]: """Get multiple records.""" result = await db.execute( select(self.model).offset(skip).limit(limit) ) return result.scalars().all() async def create( self, db: AsyncSession, obj_in: CreateSchemaType ) -> ModelType: """Create new record.""" db_obj = self.model(**obj_in.dict()) db.add(db_obj) await db.flush() await db.refresh(db_obj) return db_obj async def update( self, db: AsyncSession, db_obj: ModelType, obj_in: UpdateSchemaType ) -> ModelType: """Update record.""" update_data = obj_in.dict(exclude_unset=True) for field, value in update_data.items(): setattr(db_obj, field, value) await db.flush() await db.refresh(db_obj) return db_obj async def delete(self, db: AsyncSession, id: int) -> bool: """Delete record.""" obj = await self.get(db, id) if obj: await db.delete(obj) return True return False # repositories/user_repository.py from app.repositories.base_repository import BaseRepository from app.models.user import User from app.schemas.user import UserCreate, UserUpdate class UserRepository(BaseRepository[User, UserCreate, UserUpdate]): """User-specific repository.""" async def get_by_email(self, db: AsyncSession, email: str) -> Optional[User]: """Get user by email.""" result = await db.execute( select(User).where(User.email == email) ) return result.scalars().first() async def is_active(self, db: AsyncSession, user_id: int) -> bool: """Check if user is active.""" user = await self.get(db, user_id) return user.is_active if user else False user_repository = UserRepository(User) ``` ### Pattern 3: Service Layer ```python # services/user_service.py from typing import Optional from sqlalchemy.ext.asyncio import AsyncSession from app.repositories.user_repository import user_repository from app.schemas.user import UserCreate, UserUpdate, User from app.core.security import get_password_hash, verify_password class UserService: """Business logic for users.""" def __init__(self): self.repository = user_repository async def create_user( self, db: AsyncSession, user_in: UserCreate ) -> User: """Create new user with hashed password.""" # Check if email exists existing = await self.repository.get_by_email(db, user_in.email) if existing: raise ValueError("Email already registered") # Hash password user_in_dict = user_in.dict() user_in_dict["hashed_password"] = get_password_hash(user_in_dict.pop("password")) # Create user user = await self.repository.create(db, UserCreate(**user_in_dict)) return user async def authenticate( self, db: AsyncSession, email: str, password: str ) -> Optional[User]: """Authenticate user.""" user = await self.repository.get_by_email(db, email) if not user: return None if not verify_password(password, user.hashed_password): return None return user async def update_user( self, db: AsyncSession, user_id: int, user_in: UserUpdate ) -> Optional[User]: """Update user.""" user = await self.repository.get(db, user_id) if not user: return None if user_in.password: user_in_dict = user_in.dict(exclude_unset=True) user_in_dict["hashed_password"] = get_password_hash( user_in_dict.pop("password") ) user_in = UserUpdate(**user_in_dict) return await self.repository.update(db, user, user_in) user_service = UserService() ``` ### Pattern 4: API Endpoints with Dependencies ```python # api/v1/endpoints/users.py from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.ext.asyncio import AsyncSession from typing import List from app.core.database import get_db from app.schemas.user import User, UserCreate, UserUpdate from app.services.user_service import user_service from app.api.dependencies import get_current_user router = APIRouter() @router.post("/", response_model=User, status_code=status.HTTP_201_CREATED) async def create_user( user_in: UserCreate, db: AsyncSession = Depends(get_db) ): """Create new user.""" try: user = await user_service.create_user(db, user_in) return user except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) @router.get("/me", response_model=User) async def read_current_user( current_user: User = Depends(get_current_user) ): """Get current user.""" return current_user @router.get("/{user_id}", response_model=User) async def read_user( user_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """Get user by ID.""" user = await user_service.repository.get(db, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user @router.patch("/{user_id}", response_model=User) async def update_user( user_id: int, user_in: UserUpdate, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """Update user.""" if current_user.id != user_id: raise HTTPException(status_code=403, detail="Not authorized") user = await user_service.update_user(db, user_id, user_in) if not user: raise HTTPException(status_code=404, detail="User not found") return user @router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_user( user_id: int, db: AsyncSession = Depends(get_db), current_user: User = Depends(get_current_user) ): """Delete user.""" if current_user.id != user_id: raise HTTPException(status_code=403, detail="Not authorized") deleted = await user_service.repository.delete(db, user_id) if not deleted: raise HTTPException(status_code=404, detail="User not found") ``` ### Pattern 5: Authentication & Authorization ```python # core/security.py from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from app.core.config import get_settings settings = get_settings() pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") ALGORITHM = "HS256" def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): """Create JWT access token.""" to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=15) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify password against hash.""" return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: """Hash password.""" return pwd_context.hash(password) # api/dependencies.py from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from jose import JWTError, jwt from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.security import ALGORITHM from app.core.config import get_settings from app.repositories.user_repository import user_repository oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login") async def get_current_user( db: AsyncSession = Depends(get_db), token: str = Depends(oauth2_scheme) ): """Get current authenticated user.""" credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM]) user_id: int = payload.get("sub") if user_id is None: raise credentials_exception except JWTError: raise credentials_exception user = await user_repository.get(db, user_id) if user is None: raise credentials_exception return user ``` ## Testing ```python # tests/conftest.py import pytest import asyncio from httpx import AsyncClient from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from app.main import app from app.core.database import get_db, Base TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:" @pytest.fixture(scope="session") def event_loop(): loop = asyncio.get_event_loop_policy().new_event_loop() yield loop loop.close() @pytest.fixture async def db_session(): engine = create_async_engine(TEST_DATABASE_URL, echo=True) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) AsyncSessionLocal = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) async with AsyncSessionLocal() as session: yield session @pytest.fixture async def client(db_session): async def override_get_db(): yield db_session app.dependency_overrides[get_db] = override_get_db async with AsyncClient(app=app, base_url="http://test") as client: yield client # tests/test_users.py import pytest @pytest.mark.asyncio async def test_create_user(client): response = await client.post( "/api/v1/users/", json={ "email": "test@example.com", "password": "testpass123", "name": "Test User" } ) assert response.status_code == 201 data = response.json() assert data["email"] == "test@example.com" assert "id" in data ``` ## Resources - **references/fastapi-architecture.md**: Detailed architecture guide - **references/async-best-practices.md**: Async/await patterns - **references/testing-strategies.md**: Comprehensive testing guide - **assets/project-template/**: Complete FastAPI project - **assets/docker-compose.yml**: Development environment setup ## Best Practices 1. **Async All The Way**: Use async for database, external APIs 2. **Dependency Injection**: Leverage FastAPI's DI system 3. **Repository Pattern**: Separate data access from business logic 4. **Service Layer**: Keep business logic out of routes 5. **Pydantic Schemas**: Strong typing for request/response 6. **Error Handling**: Consistent error responses 7. **Testing**: Test all layers independently ## Common Pitfalls - **Blocking Code in Async**: Using synchronous database drivers - **No Service Layer**: Business logic in route handlers - **Missing Type Hints**: Loses FastAPI's benefits - **Ignoring Sessions**: Not properly managing database sessions - **No Testing**: Skipping integration tests - **Tight Coupling**: Direct database access in routes
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

async-python-patterns

Master Python asyncio, concurrent programming, and async/await

coding
⭐1
# Async Python Patterns Comprehensive guidance for implementing asynchronous Python applications using asyncio, concurrent programming patterns, and async/await for building high-performance, non-blocking systems. ## When to Use This Skill - Building async web APIs (FastAPI, aiohttp, Sanic) - Implementing concurrent I/O operations (database, file, network) - Creating web scrapers with concurrent requests - Developing real-time applications (WebSocket servers, chat systems) - Processing multiple independent tasks simultaneously - Building microservices with async communication - Optimizing I/O-bound workloads - Implementing async background tasks and queues ## Sync vs Async Decision Guide Before adopting async, consider whether it's the right choice for your use case. | Use Case | Recommended Approach | |----------|---------------------| | Many concurrent network/DB calls | `asyncio` | | CPU-bound computation | `multiprocessing` or thread pool | | Mixed I/O + CPU | Offload CPU work with `asyncio.to_thread()` | | Simple scripts, few connections | Sync (simpler, easier to debug) | | Web APIs with high concurrency | Async frameworks (FastAPI, aiohttp) | **Key Rule:** Stay fully sync or fully async within a call path. Mixing creates hidden blocking and complexity. ## Core Concepts ### 1. Event Loop The event loop is the heart of asyncio, managing and scheduling asynchronous tasks. **Key characteristics:** - Single-threaded cooperative multitasking - Schedules coroutines for execution - Handles I/O operations without blocking - Manages callbacks and futures ### 2. Coroutines Functions defined with `async def` that can be paused and resumed. **Syntax:** ```python async def my_coroutine(): result = await some_async_operation() return result ``` ### 3. Tasks Scheduled coroutines that run concurrently on the event loop. ### 4. Futures Low-level objects representing eventual results of async operations. ### 5. Async Context Managers Resources that support `async with` for proper cleanup. ### 6. Async Iterators Objects that support `async for` for iterating over async data sources. ## Quick Start ```python import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") # Python 3.7+ asyncio.run(main()) ``` ## Fundamental Patterns ### Pattern 1: Basic Async/Await ```python import asyncio async def fetch_data(url: str) -> dict: """Fetch data from URL asynchronously.""" await asyncio.sleep(1) # Simulate I/O return {"url": url, "data": "result"} async def main(): result = await fetch_data("https://api.example.com") print(result) asyncio.run(main()) ``` ### Pattern 2: Concurrent Execution with gather() ```python import asyncio from typing import List async def fetch_user(user_id: int) -> dict: """Fetch user data.""" await asyncio.sleep(0.5) return {"id": user_id, "name": f"User {user_id}"} async def fetch_all_users(user_ids: List[int]) -> List[dict]: """Fetch multiple users concurrently.""" tasks = [fetch_user(uid) for uid in user_ids] results = await asyncio.gather(*tasks) return results async def main(): user_ids = [1, 2, 3, 4, 5] users = await fetch_all_users(user_ids) print(f"Fetched {len(users)} users") asyncio.run(main()) ``` ### Pattern 3: Task Creation and Management ```python import asyncio async def background_task(name: str, delay: int): """Long-running background task.""" print(f"{name} started") await asyncio.sleep(delay) print(f"{name} completed") return f"Result from {name}" async def main(): # Create tasks task1 = asyncio.create_task(background_task("Task 1", 2)) task2 = asyncio.create_task(background_task("Task 2", 1)) # Do other work print("Main: doing other work") await asyncio.sleep(0.5) # Wait for tasks result1 = await task1 result2 = await task2 print(f"Results: {result1}, {result2}") asyncio.run(main()) ``` ### Pattern 4: Error Handling in Async Code ```python import asyncio from typing import List, Optional async def risky_operation(item_id: int) -> dict: """Operation that might fail.""" await asyncio.sleep(0.1) if item_id % 3 == 0: raise ValueError(f"Item {item_id} failed") return {"id": item_id, "status": "success"} async def safe_operation(item_id: int) -> Optional[dict]: """Wrapper with error handling.""" try: return await risky_operation(item_id) except ValueError as e: print(f"Error: {e}") return None async def process_items(item_ids: List[int]): """Process multiple items with error handling.""" tasks = [safe_operation(iid) for iid in item_ids] results = await asyncio.gather(*tasks, return_exceptions=True) # Filter out failures successful = [r for r in results if r is not None and not isinstance(r, Exception)] failed = [r for r in results if isinstance(r, Exception)] print(f"Success: {len(successful)}, Failed: {len(failed)}") return successful asyncio.run(process_items([1, 2, 3, 4, 5, 6])) ``` ### Pattern 5: Timeout Handling ```python import asyncio async def slow_operation(delay: int) -> str: """Operation that takes time.""" await asyncio.sleep(delay) return f"Completed after {delay}s" async def with_timeout(): """Execute operation with timeout.""" try: result = await asyncio.wait_for(slow_operation(5), timeout=2.0) print(result) except asyncio.TimeoutError: print("Operation timed out") asyncio.run(with_timeout()) ``` ## Advanced Patterns ### Pattern 6: Async Context Managers ```python import asyncio from typing import Optional class AsyncDatabaseConnection: """Async database connection context manager.""" def __init__(self, dsn: str): self.dsn = dsn self.connection: Optional[object] = None async def __aenter__(self): print("Opening connection") await asyncio.sleep(0.1) # Simulate connection self.connection = {"dsn": self.dsn, "connected": True} return self.connection async def __aexit__(self, exc_type, exc_val, exc_tb): print("Closing connection") await asyncio.sleep(0.1) # Simulate cleanup self.connection = None async def query_database(): """Use async context manager.""" async with AsyncDatabaseConnection("postgresql://localhost") as conn: print(f"Using connection: {conn}") await asyncio.sleep(0.2) # Simulate query return {"rows": 10} asyncio.run(query_database()) ``` ### Pattern 7: Async Iterators and Generators ```python import asyncio from typing import AsyncIterator async def async_range(start: int, end: int, delay: float = 0.1) -> AsyncIterator[int]: """Async generator that yields numbers with delay.""" for i in range(start, end): await asyncio.sleep(delay) yield i async def fetch_pages(url: str, max_pages: int) -> AsyncIterator[dict]: """Fetch paginated data asynchronously.""" for page in range(1, max_pages + 1): await asyncio.sleep(0.2) # Simulate API call yield { "page": page, "url": f"{url}?page={page}", "data": [f"item_{page}_{i}" for i in range(5)] } async def consume_async_iterator(): """Consume async iterator.""" async for number in async_range(1, 5): print(f"Number: {number}") print("\nFetching pages:") async for page_data in fetch_pages("https://api.example.com/items", 3): print(f"Page {page_data['page']}: {len(page_data['data'])} items") asyncio.run(consume_async_iterator()) ``` ### Pattern 8: Producer-Consumer Pattern ```python import asyncio from asyncio import Queue from typing import Optional async def producer(queue: Queue, producer_id: int, num_items: int): """Produce items and put them in queue.""" for i in range(num_items): item = f"Item-{producer_id}-{i}" await queue.put(item) print(f"Producer {producer_id} produced: {item}") await asyncio.sleep(0.1) await queue.put(None) # Signal completion async def consumer(queue: Queue, consumer_id: int): """Consume items from queue.""" while True: item = await queue.get() if item is None: queue.task_done() break print(f"Consumer {consumer_id} processing: {item}") await asyncio.sleep(0.2) # Simulate work queue.task_done() async def producer_consumer_example(): """Run producer-consumer pattern.""" queue = Queue(maxsize=10) # Create tasks producers = [ asyncio.create_task(producer(queue, i, 5)) for i in range(2) ] consumers = [ asyncio.create_task(consumer(queue, i)) for i in range(3) ] # Wait for producers await asyncio.gather(*producers) # Wait for queue to be empty await queue.join() # Cancel consumers for c in consumers: c.cancel() asyncio.run(producer_consumer_example()) ``` ### Pattern 9: Semaphore for Rate Limiting ```python import asyncio from typing import List async def api_call(url: str, semaphore: asyncio.Semaphore) -> dict: """Make API call with rate limiting.""" async with semaphore: print(f"Calling {url}") await asyncio.sleep(0.5) # Simulate API call return {"url": url, "status": 200} async def rate_limited_requests(urls: List[str], max_concurrent: int = 5): """Make multiple requests with rate limiting.""" semaphore = asyncio.Semaphore(max_concurrent) tasks = [api_call(url, semaphore) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [f"https://api.example.com/item/{i}" for i in range(20)] results = await rate_limited_requests(urls, max_concurrent=3) print(f"Completed {len(results)} requests") asyncio.run(main()) ``` ### Pattern 10: Async Locks and Synchronization ```python import asyncio class AsyncCounter: """Thread-safe async counter.""" def __init__(self): self.value = 0 self.lock = asyncio.Lock() async def increment(self): """Safely increment counter.""" async with self.lock: current = self.value await asyncio.sleep(0.01) # Simulate work self.value = current + 1 async def get_value(self) -> int: """Get current value.""" async with self.lock: return self.value async def worker(counter: AsyncCounter, worker_id: int): """Worker that increments counter.""" for _ in range(10): await counter.increment() print(f"Worker {worker_id} incremented") async def test_counter(): """Test concurrent counter.""" counter = AsyncCounter() workers = [asyncio.create_task(worker(counter, i)) for i in range(5)] await asyncio.gather(*workers) final_value = await counter.get_value() print(f"Final counter value: {final_value}") asyncio.run(test_counter()) ``` ## Real-World Applications ### Web Scraping with aiohttp ```python import asyncio import aiohttp from typing import List, Dict async def fetch_url(session: aiohttp.ClientSession, url: str) -> Dict: """Fetch single URL.""" try: async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as response: text = await response.text() return { "url": url, "status": response.status, "length": len(text) } except Exception as e: return {"url": url, "error": str(e)} async def scrape_urls(urls: List[str]) -> List[Dict]: """Scrape multiple URLs concurrently.""" async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] results = await asyncio.gather(*tasks) return results async def main(): urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/status/404", ] results = await scrape_urls(urls) for result in results: print(result) asyncio.run(main()) ``` ### Async Database Operations ```python import asyncio from typing import List, Optional # Simulated async database client class AsyncDB: """Simulated async database.""" async def execute(self, query: str) -> List[dict]: """Execute query.""" await asyncio.sleep(0.1) return [{"id": 1, "name": "Example"}] async def fetch_one(self, query: str) -> Optional[dict]: """Fetch single row.""" await asyncio.sleep(0.1) return {"id": 1, "name": "Example"} async def get_user_data(db: AsyncDB, user_id: int) -> dict: """Fetch user and related data concurrently.""" user_task = db.fetch_one(f"SELECT * FROM users WHERE id = {user_id}") orders_task = db.execute(f"SELECT * FROM orders WHERE user_id = {user_id}") profile_task = db.fetch_one(f"SELECT * FROM profiles WHERE user_id = {user_id}") user, orders, profile = await asyncio.gather(user_task, orders_task, profile_task) return { "user": user, "orders": orders, "profile": profile } async def main(): db = AsyncDB() user_data = await get_user_data(db, 1) print(user_data) asyncio.run(main()) ``` ### WebSocket Server ```python import asyncio from typing import Set # Simulated WebSocket connection class WebSocket: """Simulated WebSocket.""" def __init__(self, client_id: str): self.client_id = client_id async def send(self, message: str): """Send message.""" print(f"Sending to {self.client_id}: {message}") await asyncio.sleep(0.01) async def recv(self) -> str: """Receive message.""" await asyncio.sleep(1) return f"Message from {self.client_id}" class WebSocketServer: """Simple WebSocket server.""" def __init__(self): self.clients: Set[WebSocket] = set() async def register(self, websocket: WebSocket): """Register new client.""" self.clients.add(websocket) print(f"Client {websocket.client_id} connected") async def unregister(self, websocket: WebSocket): """Unregister client.""" self.clients.remove(websocket) print(f"Client {websocket.client_id} disconnected") async def broadcast(self, message: str): """Broadcast message to all clients.""" if self.clients: tasks = [client.send(message) for client in self.clients] await asyncio.gather(*tasks) async def handle_client(self, websocket: WebSocket): """Handle individual client connection.""" await self.register(websocket) try: async for message in self.message_iterator(websocket): await self.broadcast(f"{websocket.client_id}: {message}") finally: await self.unregister(websocket) async def message_iterator(self, websocket: WebSocket): """Iterate over messages from client.""" for _ in range(3): # Simulate 3 messages yield await websocket.recv() ``` ## Performance Best Practices ### 1. Use Connection Pools ```python import asyncio import aiohttp async def with_connection_pool(): """Use connection pool for efficiency.""" connector = aiohttp.TCPConnector(limit=100, limit_per_host=10) async with aiohttp.ClientSession(connector=connector) as session: tasks = [session.get(f"https://api.example.com/item/{i}") for i in range(50)] responses = await asyncio.gather(*tasks) return responses ``` ### 2. Batch Operations ```python async def batch_process(items: List[str], batch_size: int = 10): """Process items in batches.""" for i in range(0, len(items), batch_size): batch = items[i:i + batch_size] tasks = [process_item(item) for item in batch] await asyncio.gather(*tasks) print(f"Processed batch {i // batch_size + 1}") async def process_item(item: str): """Process single item.""" await asyncio.sleep(0.1) return f"Processed: {item}" ``` ### 3. Avoid Blocking Operations Never block the event loop with synchronous operations. A single blocking call stalls all concurrent tasks. ```python # BAD - blocks the entire event loop async def fetch_data_bad(): import time import requests time.sleep(1) # Blocks! response = requests.get(url) # Also blocks! # GOOD - use async-native libraries (e.g., httpx for async HTTP) import httpx async def fetch_data_good(url: str): await asyncio.sleep(1) async with httpx.AsyncClient() as client: response = await client.get(url) ``` **Wrapping Blocking Code with `asyncio.to_thread()` (Python 3.9+):** When you must use synchronous libraries, offload to a thread pool: ```python import asyncio from pathlib import Path async def read_file_async(path: str) -> str: """Read file without blocking event loop.""" # asyncio.to_thread() runs sync code in a thread pool return await asyncio.to_thread(Path(path).read_text) async def call_sync_library(data: dict) -> dict: """Wrap a synchronous library call.""" # Useful for sync database drivers, file I/O, CPU work return await asyncio.to_thread(sync_library.process, data) ``` **Lower-level approach with `run_in_executor()`:** ```python import asyncio import concurrent.futures from typing import Any def blocking_operation(data: Any) -> Any: """CPU-intensive blocking operation.""" import time time.sleep(1) return data * 2 async def run_in_executor(data: Any) -> Any: """Run blocking operation in thread pool.""" loop = asyncio.get_running_loop() with concurrent.futures.ThreadPoolExecutor() as pool: result = await loop.run_in_executor(pool, blocking_operation, data) return result async def main(): results = await asyncio.gather(*[run_in_executor(i) for i in range(5)]) print(results) asyncio.run(main()) ``` ## Common Pitfalls ### 1. Forgetting await ```python # Wrong - returns coroutine object, doesn't execute result = async_function() # Correct result = await async_function() ``` ### 2. Blocking the Event Loop ```python # Wrong - blocks event loop import time async def bad(): time.sleep(1) # Blocks! # Correct async def good(): await asyncio.sleep(1) # Non-blocking ``` ### 3. Not Handling Cancellation ```python async def cancelable_task(): """Task that handles cancellation.""" try: while True: await asyncio.sleep(1) print("Working...") except asyncio.CancelledError: print("Task cancelled, cleaning up...") # Perform cleanup raise # Re-raise to propagate cancellation ``` ### 4. Mixing Sync and Async Code ```python # Wrong - can't call async from sync directly def sync_function(): result = await async_function() # SyntaxError! # Correct def sync_function(): result = asyncio.run(async_function()) ``` ## Testing Async Code ```python import asyncio import pytest # Using pytest-asyncio @pytest.mark.asyncio async def test_async_function(): """Test async function.""" result = await fetch_data("https
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-performance-optimization

Profile and optimize Python code using cProfile, memory profilers,

coding
⭐1
# Python Performance Optimization Comprehensive guide to profiling, analyzing, and optimizing Python code for better performance, including CPU profiling, memory optimization, and implementation best practices. ## When to Use This Skill - Identifying performance bottlenecks in Python applications - Reducing application latency and response times - Optimizing CPU-intensive operations - Reducing memory consumption and memory leaks - Improving database query performance - Optimizing I/O operations - Speeding up data processing pipelines - Implementing high-performance algorithms - Profiling production applications ## Core Concepts ### 1. Profiling Types - **CPU Profiling**: Identify time-consuming functions - **Memory Profiling**: Track memory allocation and leaks - **Line Profiling**: Profile at line-by-line granularity - **Call Graph**: Visualize function call relationships ### 2. Performance Metrics - **Execution Time**: How long operations take - **Memory Usage**: Peak and average memory consumption - **CPU Utilization**: Processor usage patterns - **I/O Wait**: Time spent on I/O operations ### 3. Optimization Strategies - **Algorithmic**: Better algorithms and data structures - **Implementation**: More efficient code patterns - **Parallelization**: Multi-threading/processing - **Caching**: Avoid redundant computation - **Native Extensions**: C/Rust for critical paths ## Quick Start ### Basic Timing ```python import time def measure_time(): """Simple timing measurement.""" start = time.time() # Your code here result = sum(range(1000000)) elapsed = time.time() - start print(f"Execution time: {elapsed:.4f} seconds") return result # Better: use timeit for accurate measurements import timeit execution_time = timeit.timeit( "sum(range(1000000))", number=100 ) print(f"Average time: {execution_time/100:.6f} seconds") ``` ## Profiling Tools ### Pattern 1: cProfile - CPU Profiling ```python import cProfile import pstats from pstats import SortKey def slow_function(): """Function to profile.""" total = 0 for i in range(1000000): total += i return total def another_function(): """Another function.""" return [i**2 for i in range(100000)] def main(): """Main function to profile.""" result1 = slow_function() result2 = another_function() return result1, result2 # Profile the code if __name__ == "__main__": profiler = cProfile.Profile() profiler.enable() main() profiler.disable() # Print stats stats = pstats.Stats(profiler) stats.sort_stats(SortKey.CUMULATIVE) stats.print_stats(10) # Top 10 functions # Save to file for later analysis stats.dump_stats("profile_output.prof") ``` **Command-line profiling:** ```bash # Profile a script python -m cProfile -o output.prof script.py # View results python -m pstats output.prof # In pstats: # sort cumtime # stats 10 ``` ### Pattern 2: line_profiler - Line-by-Line Profiling ```python # Install: pip install line-profiler # Add @profile decorator (line_profiler provides this) @profile def process_data(data): """Process data with line profiling.""" result = [] for item in data: processed = item * 2 result.append(processed) return result # Run with: # kernprof -l -v script.py ``` **Manual line profiling:** ```python from line_profiler import LineProfiler def process_data(data): """Function to profile.""" result = [] for item in data: processed = item * 2 result.append(processed) return result if __name__ == "__main__": lp = LineProfiler() lp.add_function(process_data) data = list(range(100000)) lp_wrapper = lp(process_data) lp_wrapper(data) lp.print_stats() ``` ### Pattern 3: memory_profiler - Memory Usage ```python # Install: pip install memory-profiler from memory_profiler import profile @profile def memory_intensive(): """Function that uses lots of memory.""" # Create large list big_list = [i for i in range(1000000)] # Create large dict big_dict = {i: i**2 for i in range(100000)} # Process data result = sum(big_list) return result if __name__ == "__main__": memory_intensive() # Run with: # python -m memory_profiler script.py ``` ### Pattern 4: py-spy - Production Profiling ```bash # Install: pip install py-spy # Profile a running Python process py-spy top --pid 12345 # Generate flamegraph py-spy record -o profile.svg --pid 12345 # Profile a script py-spy record -o profile.svg -- python script.py # Dump current call stack py-spy dump --pid 12345 ``` ## Optimization Patterns ### Pattern 5: List Comprehensions vs Loops ```python import timeit # Slow: Traditional loop def slow_squares(n): """Create list of squares using loop.""" result = [] for i in range(n): result.append(i**2) return result # Fast: List comprehension def fast_squares(n): """Create list of squares using comprehension.""" return [i**2 for i in range(n)] # Benchmark n = 100000 slow_time = timeit.timeit(lambda: slow_squares(n), number=100) fast_time = timeit.timeit(lambda: fast_squares(n), number=100) print(f"Loop: {slow_time:.4f}s") print(f"Comprehension: {fast_time:.4f}s") print(f"Speedup: {slow_time/fast_time:.2f}x") # Even faster for simple operations: map def faster_squares(n): """Use map for even better performance.""" return list(map(lambda x: x**2, range(n))) ``` ### Pattern 6: Generator Expressions for Memory ```python import sys def list_approach(): """Memory-intensive list.""" data = [i**2 for i in range(1000000)] return sum(data) def generator_approach(): """Memory-efficient generator.""" data = (i**2 for i in range(1000000)) return sum(data) # Memory comparison list_data = [i for i in range(1000000)] gen_data = (i for i in range(1000000)) print(f"List size: {sys.getsizeof(list_data)} bytes") print(f"Generator size: {sys.getsizeof(gen_data)} bytes") # Generators use constant memory regardless of size ``` ### Pattern 7: String Concatenation ```python import timeit def slow_concat(items): """Slow string concatenation.""" result = "" for item in items: result += str(item) return result def fast_concat(items): """Fast string concatenation with join.""" return "".join(str(item) for item in items) def faster_concat(items): """Even faster with list.""" parts = [str(item) for item in items] return "".join(parts) items = list(range(10000)) # Benchmark slow = timeit.timeit(lambda: slow_concat(items), number=100) fast = timeit.timeit(lambda: fast_concat(items), number=100) faster = timeit.timeit(lambda: faster_concat(items), number=100) print(f"Concatenation (+): {slow:.4f}s") print(f"Join (generator): {fast:.4f}s") print(f"Join (list): {faster:.4f}s") ``` ### Pattern 8: Dictionary Lookups vs List Searches ```python import timeit # Create test data size = 10000 items = list(range(size)) lookup_dict = {i: i for i in range(size)} def list_search(items, target): """O(n) search in list.""" return target in items def dict_search(lookup_dict, target): """O(1) search in dict.""" return target in lookup_dict target = size - 1 # Worst case for list # Benchmark list_time = timeit.timeit( lambda: list_search(items, target), number=1000 ) dict_time = timeit.timeit( lambda: dict_search(lookup_dict, target), number=1000 ) print(f"List search: {list_time:.6f}s") print(f"Dict search: {dict_time:.6f}s") print(f"Speedup: {list_time/dict_time:.0f}x") ``` ### Pattern 9: Local Variable Access ```python import timeit # Global variable (slow) GLOBAL_VALUE = 100 def use_global(): """Access global variable.""" total = 0 for i in range(10000): total += GLOBAL_VALUE return total def use_local(): """Use local variable.""" local_value = 100 total = 0 for i in range(10000): total += local_value return total # Local is faster global_time = timeit.timeit(use_global, number=1000) local_time = timeit.timeit(use_local, number=1000) print(f"Global access: {global_time:.4f}s") print(f"Local access: {local_time:.4f}s") print(f"Speedup: {global_time/local_time:.2f}x") ``` ### Pattern 10: Function Call Overhead ```python import timeit def calculate_inline(): """Inline calculation.""" total = 0 for i in range(10000): total += i * 2 + 1 return total def helper_function(x): """Helper function.""" return x * 2 + 1 def calculate_with_function(): """Calculation with function calls.""" total = 0 for i in range(10000): total += helper_function(i) return total # Inline is faster due to no call overhead inline_time = timeit.timeit(calculate_inline, number=1000) function_time = timeit.timeit(calculate_with_function, number=1000) print(f"Inline: {inline_time:.4f}s") print(f"Function calls: {function_time:.4f}s") ``` ## Advanced Optimization ### Pattern 11: NumPy for Numerical Operations ```python import timeit import numpy as np def python_sum(n): """Sum using pure Python.""" return sum(range(n)) def numpy_sum(n): """Sum using NumPy.""" return np.arange(n).sum() n = 1000000 python_time = timeit.timeit(lambda: python_sum(n), number=100) numpy_time = timeit.timeit(lambda: numpy_sum(n), number=100) print(f"Python: {python_time:.4f}s") print(f"NumPy: {numpy_time:.4f}s") print(f"Speedup: {python_time/numpy_time:.2f}x") # Vectorized operations def python_multiply(): """Element-wise multiplication in Python.""" a = list(range(100000)) b = list(range(100000)) return [x * y for x, y in zip(a, b)] def numpy_multiply(): """Vectorized multiplication in NumPy.""" a = np.arange(100000) b = np.arange(100000) return a * b py_time = timeit.timeit(python_multiply, number=100) np_time = timeit.timeit(numpy_multiply, number=100) print(f"\nPython multiply: {py_time:.4f}s") print(f"NumPy multiply: {np_time:.4f}s") print(f"Speedup: {py_time/np_time:.2f}x") ``` ### Pattern 12: Caching with functools.lru_cache ```python from functools import lru_cache import timeit def fibonacci_slow(n): """Recursive fibonacci without caching.""" if n < 2: return n return fibonacci_slow(n-1) + fibonacci_slow(n-2) @lru_cache(maxsize=None) def fibonacci_fast(n): """Recursive fibonacci with caching.""" if n < 2: return n return fibonacci_fast(n-1) + fibonacci_fast(n-2) # Massive speedup for recursive algorithms n = 30 slow_time = timeit.timeit(lambda: fibonacci_slow(n), number=1) fast_time = timeit.timeit(lambda: fibonacci_fast(n), number=1000) print(f"Without cache (1 run): {slow_time:.4f}s") print(f"With cache (1000 runs): {fast_time:.4f}s") # Cache info print(f"Cache info: {fibonacci_fast.cache_info()}") ``` ### Pattern 13: Using **slots** for Memory ```python import sys class RegularClass: """Regular class with __dict__.""" def __init__(self, x, y, z): self.x = x self.y = y self.z = z class SlottedClass: """Class with __slots__ for memory efficiency.""" __slots__ = ['x', 'y', 'z'] def __init__(self, x, y, z): self.x = x self.y = y self.z = z # Memory comparison regular = RegularClass(1, 2, 3) slotted = SlottedClass(1, 2, 3) print(f"Regular class size: {sys.getsizeof(regular)} bytes") print(f"Slotted class size: {sys.getsizeof(slotted)} bytes") # Significant savings with many instances regular_objects = [RegularClass(i, i+1, i+2) for i in range(10000)] slotted_objects = [SlottedClass(i, i+1, i+2) for i in range(10000)] print(f"\nMemory for 10000 regular objects: ~{sys.getsizeof(regular) * 10000} bytes") print(f"Memory for 10000 slotted objects: ~{sys.getsizeof(slotted) * 10000} bytes") ``` ### Pattern 14: Multiprocessing for CPU-Bound Tasks ```python import multiprocessing as mp import time def cpu_intensive_task(n): """CPU-intensive calculation.""" return sum(i**2 for i in range(n)) def sequential_processing(): """Process tasks sequentially.""" start = time.time() results = [cpu_intensive_task(1000000) for _ in range(4)] elapsed = time.time() - start return elapsed, results def parallel_processing(): """Process tasks in parallel.""" start = time.time() with mp.Pool(processes=4) as pool: results = pool.map(cpu_intensive_task, [1000000] * 4) elapsed = time.time() - start return elapsed, results if __name__ == "__main__": seq_time, seq_results = sequential_processing() par_time, par_results = parallel_processing() print(f"Sequential: {seq_time:.2f}s") print(f"Parallel: {par_time:.2f}s") print(f"Speedup: {seq_time/par_time:.2f}x") ``` ### Pattern 15: Async I/O for I/O-Bound Tasks ```python import asyncio import aiohttp import time import requests urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/1", "https://httpbin.org/delay/1", "https://httpbin.org/delay/1", ] def synchronous_requests(): """Synchronous HTTP requests.""" start = time.time() results = [] for url in urls: response = requests.get(url) results.append(response.status_code) elapsed = time.time() - start return elapsed, results async def async_fetch(session, url): """Async HTTP request.""" async with session.get(url) as response: return response.status async def asynchronous_requests(): """Asynchronous HTTP requests.""" start = time.time() async with aiohttp.ClientSession() as session: tasks = [async_fetch(session, url) for url in urls] results = await asyncio.gather(*tasks) elapsed = time.time() - start return elapsed, results # Async is much faster for I/O-bound work sync_time, sync_results = synchronous_requests() async_time, async_results = asyncio.run(asynchronous_requests()) print(f"Synchronous: {sync_time:.2f}s") print(f"Asynchronous: {async_time:.2f}s") print(f"Speedup: {sync_time/async_time:.2f}x") ``` ## Database Optimization ### Pattern 16: Batch Database Operations ```python import sqlite3 import time def create_db(): """Create test database.""" conn = sqlite3.connect(":memory:") conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)") return conn def slow_inserts(conn, count): """Insert records one at a time.""" start = time.time() cursor = conn.cursor() for i in range(count): cursor.execute("INSERT INTO users (name) VALUES (?)", (f"User {i}",)) conn.commit() # Commit each insert elapsed = time.time() - start return elapsed def fast_inserts(conn, count): """Batch insert with single commit.""" start = time.time() cursor = conn.cursor() data = [(f"User {i}",) for i in range(count)] cursor.executemany("INSERT INTO users (name) VALUES (?)", data) conn.commit() # Single commit elapsed = time.time() - start return elapsed # Benchmark conn1 = create_db() slow_time = slow_inserts(conn1, 1000) conn2 = create_db() fast_time = fast_inserts(conn2, 1000) print(f"Individual inserts: {slow_time:.4f}s") print(f"Batch insert: {fast_time:.4f}s") print(f"Speedup: {slow_time/fast_time:.2f}x") ``` ### Pattern 17: Query Optimization ```python # Use indexes for frequently queried columns """ -- Slow: No index SELECT * FROM users WHERE email = 'user@example.com'; -- Fast: With index CREATE INDEX idx_users_email ON users(email); SELECT * FROM users WHERE email = 'user@example.com'; """ # Use query planning import sqlite3 conn = sqlite3.connect("example.db") cursor = conn.cursor() # Analyze query performance cursor.execute("EXPLAIN QUERY PLAN SELECT * FROM users WHERE email = ?", ("test@example.com",)) print(cursor.fetchall()) # Use SELECT only needed columns # Slow: SELECT * # Fast: SELECT id, name ``` ## Memory Optimization ### Pattern 18: Detecting Memory Leaks ```python import tracemalloc import gc def memory_leak_example(): """Example that leaks memory.""" leaked_objects = [] for i in range(100000): # Objects added but never removed leaked_objects.append([i] * 100) # In real code, this would be an unintended reference def track_memory_usage(): """Track memory allocations.""" tracemalloc.start() # Take snapshot before snapshot1 = tracemalloc.take_snapshot() # Run code memory_leak_example() # Take snapshot after snapshot2 = tracemalloc.take_snapshot() # Compare top_stats = snapshot2.compare_to(snapshot1, 'lineno') print("Top 10 memory allocations:") for stat in top_stats[:10]: print(stat) tracemalloc.stop() # Monitor memory track_memory_usage() # Force garbage collection gc.collect() ``` ### Pattern 19: Iterators vs Lists ```python import sys def process_file_list(filename): """Load entire file into memory.""" with open(filename) as f: lines = f.readlines() # Loads all lines return sum(1 for line in lines if line.strip()) def process_file_iterator(filename): """Process file line by line.""" with open(filename) as f: return sum(1 for line in f if line.strip()) # Iterator uses constant memory # List loads entire file into memory ``` ### Pattern 20: Weakref for Caches ```python import weakref class CachedResource: """Resource that can be garbage collected.""" def __init__(self, data): self.data = data # Regular cache prevents garbage collection regular_cache = {} def get_resource_regular(key): """Get resource from regular cache.""" if key not in regular_cache: regular_cache[key] = CachedResource(f"Data for {key}") return regular_cache[key] # Weak reference cache allows garbage collection weak_cache = weakref.WeakValueDictionary() def get_resource_weak(key): """Get resource from weak cache.""" resource = weak_cache.get(key) if resource is None: resource = CachedResource(f"Data for {key}") weak_cache[key] = resource return resource # When no strong references exist, objects can be GC'd ``` ## Benchmarking Tools ### Custom Benchmark Decorator ```python import time from functools import wraps def benchmark(func): """Decorator to benchmark function execution.""" @wraps(func) def wrapper(*args, **kwargs): start = time.perf_counter() result = func(*args, **kwargs) elapsed = time.perf_counter() - start print(f"{func.__name__} took {elapsed:.6f} seconds") return result return wrapper @benchmark def slow_function(): """Function to benchmark.""" time.sleep(0.5) return sum(range(1000000)) result = slow_function() ``` ### Performance Testing with pytest-benchmark ```python # Install: pip install pytest-benchmark def test_list_comprehension(benchmark): """Benchmark list comprehension.""" result = benchmark(lambda: [i**2 for i in range(10000)]) assert len(result) == 10000 def test_map_function(benchmark): """Benchmark map function.""" result = benchmark(lambda: list(m
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-resilience

Python resilience patterns including automatic retries, exponential

coding
⭐1
# Python Resilience Patterns Build fault-tolerant Python applications that gracefully handle transient failures, network issues, and service outages. Resilience patterns keep systems running when dependencies are unreliable. ## When to Use This Skill - Adding retry logic to external service calls - Implementing timeouts for network operations - Building fault-tolerant microservices - Handling rate limiting and backpressure - Creating infrastructure decorators - Designing circuit breakers ## Core Concepts ### 1. Transient vs Permanent Failures Retry transient errors (network timeouts, temporary service issues). Don't retry permanent errors (invalid credentials, bad requests). ### 2. Exponential Backoff Increase wait time between retries to avoid overwhelming recovering services. ### 3. Jitter Add randomness to backoff to prevent thundering herd when many clients retry simultaneously. ### 4. Bounded Retries Cap both attempt count and total duration to prevent infinite retry loops. ## Quick Start ```python from tenacity import retry, stop_after_attempt, wait_exponential_jitter @retry( stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=1, max=10), ) def call_external_service(request: dict) -> dict: return httpx.post("https://api.example.com", json=request).json() ``` ## Fundamental Patterns ### Pattern 1: Basic Retry with Tenacity Use the `tenacity` library for production-grade retry logic. For simpler cases, consider built-in retry functionality or a lightweight custom implementation. ```python from tenacity import ( retry, stop_after_attempt, stop_after_delay, wait_exponential_jitter, retry_if_exception_type, ) TRANSIENT_ERRORS = (ConnectionError, TimeoutError, OSError) @retry( retry=retry_if_exception_type(TRANSIENT_ERRORS), stop=stop_after_attempt(5) | stop_after_delay(60), wait=wait_exponential_jitter(initial=1, max=30), ) def fetch_data(url: str) -> dict: """Fetch data with automatic retry on transient failures.""" response = httpx.get(url, timeout=30) response.raise_for_status() return response.json() ``` ### Pattern 2: Retry Only Appropriate Errors Whitelist specific transient exceptions. Never retry: - `ValueError`, `TypeError` - These are bugs, not transient issues - `AuthenticationError` - Invalid credentials won't become valid - HTTP 4xx errors (except 429) - Client errors are permanent ```python from tenacity import retry, retry_if_exception_type import httpx # Define what's retryable RETRYABLE_EXCEPTIONS = ( ConnectionError, TimeoutError, httpx.ConnectTimeout, httpx.ReadTimeout, ) @retry( retry=retry_if_exception_type(RETRYABLE_EXCEPTIONS), stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=1, max=10), ) def resilient_api_call(endpoint: str) -> dict: """Make API call with retry on network issues.""" return httpx.get(endpoint, timeout=10).json() ``` ### Pattern 3: HTTP Status Code Retries Retry specific HTTP status codes that indicate transient issues. ```python from tenacity import retry, retry_if_result, stop_after_attempt import httpx RETRY_STATUS_CODES = {429, 502, 503, 504} def should_retry_response(response: httpx.Response) -> bool: """Check if response indicates a retryable error.""" return response.status_code in RETRY_STATUS_CODES @retry( retry=retry_if_result(should_retry_response), stop=stop_after_attempt(3), wait=wait_exponential_jitter(initial=1, max=10), ) def http_request(method: str, url: str, **kwargs) -> httpx.Response: """Make HTTP request with retry on transient status codes.""" return httpx.request(method, url, timeout=30, **kwargs) ``` ### Pattern 4: Combined Exception and Status Retry Handle both network exceptions and HTTP status codes. ```python from tenacity import ( retry, retry_if_exception_type, retry_if_result, stop_after_attempt, wait_exponential_jitter, before_sleep_log, ) import logging import httpx logger = logging.getLogger(__name__) TRANSIENT_EXCEPTIONS = ( ConnectionError, TimeoutError, httpx.ConnectError, httpx.ReadTimeout, ) RETRY_STATUS_CODES = {429, 500, 502, 503, 504} def is_retryable_response(response: httpx.Response) -> bool: return response.status_code in RETRY_STATUS_CODES @retry( retry=( retry_if_exception_type(TRANSIENT_EXCEPTIONS) | retry_if_result(is_retryable_response) ), stop=stop_after_attempt(5), wait=wait_exponential_jitter(initial=1, max=30), before_sleep=before_sleep_log(logger, logging.WARNING), ) def robust_http_call( method: str, url: str, **kwargs, ) -> httpx.Response: """HTTP call with comprehensive retry handling.""" return httpx.request(method, url, timeout=30, **kwargs) ``` ## Advanced Patterns ### Pattern 5: Logging Retry Attempts Track retry behavior for debugging and alerting. ```python from tenacity import retry, stop_after_attempt, wait_exponential import structlog logger = structlog.get_logger() def log_retry_attempt(retry_state): """Log detailed retry information.""" exception = retry_state.outcome.exception() logger.warning( "Retrying operation", attempt=retry_state.attempt_number, exception_type=type(exception).__name__, exception_message=str(exception), next_wait_seconds=retry_state.next_action.sleep if retry_state.next_action else None, ) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10), before_sleep=log_retry_attempt, ) def call_with_logging(request: dict) -> dict: """External call with retry logging.""" ... ``` ### Pattern 6: Timeout Decorator Create reusable timeout decorators for consistent timeout handling. ```python import asyncio from functools import wraps from typing import TypeVar, Callable T = TypeVar("T") def with_timeout(seconds: float): """Decorator to add timeout to async functions.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) async def wrapper(*args, **kwargs) -> T: return await asyncio.wait_for( func(*args, **kwargs), timeout=seconds, ) return wrapper return decorator @with_timeout(30) async def fetch_with_timeout(url: str) -> dict: """Fetch URL with 30 second timeout.""" async with httpx.AsyncClient() as client: response = await client.get(url) return response.json() ``` ### Pattern 7: Cross-Cutting Concerns via Decorators Stack decorators to separate infrastructure from business logic. ```python from functools import wraps from typing import TypeVar, Callable import structlog logger = structlog.get_logger() T = TypeVar("T") def traced(name: str | None = None): """Add tracing to function calls.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: span_name = name or func.__name__ @wraps(func) async def wrapper(*args, **kwargs) -> T: logger.info("Operation started", operation=span_name) try: result = await func(*args, **kwargs) logger.info("Operation completed", operation=span_name) return result except Exception as e: logger.error("Operation failed", operation=span_name, error=str(e)) raise return wrapper return decorator # Stack multiple concerns @traced("fetch_user_data") @with_timeout(30) @retry(stop=stop_after_attempt(3), wait=wait_exponential_jitter()) async def fetch_user_data(user_id: str) -> dict: """Fetch user with tracing, timeout, and retry.""" ... ``` ### Pattern 8: Dependency Injection for Testability Pass infrastructure components through constructors for easy testing. ```python from dataclasses import dataclass from typing import Protocol class Logger(Protocol): def info(self, msg: str, **kwargs) -> None: ... def error(self, msg: str, **kwargs) -> None: ... class MetricsClient(Protocol): def increment(self, metric: str, tags: dict | None = None) -> None: ... def timing(self, metric: str, value: float) -> None: ... @dataclass class UserService: """Service with injected infrastructure.""" repository: UserRepository logger: Logger metrics: MetricsClient async def get_user(self, user_id: str) -> User: self.logger.info("Fetching user", user_id=user_id) start = time.perf_counter() try: user = await self.repository.get(user_id) self.metrics.increment("user.fetch.success") return user except Exception as e: self.metrics.increment("user.fetch.error") self.logger.error("Failed to fetch user", user_id=user_id, error=str(e)) raise finally: elapsed = time.perf_counter() - start self.metrics.timing("user.fetch.duration", elapsed) # Easy to test with fakes service = UserService( repository=FakeRepository(), logger=FakeLogger(), metrics=FakeMetrics(), ) ``` ### Pattern 9: Fail-Safe Defaults Degrade gracefully when non-critical operations fail. ```python from typing import TypeVar from collections.abc import Callable T = TypeVar("T") def fail_safe(default: T, log_failure: bool = True): """Return default value on failure instead of raising.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) async def wrapper(*args, **kwargs) -> T: try: return await func(*args, **kwargs) except Exception as e: if log_failure: logger.warning( "Operation failed, using default", function=func.__name__, error=str(e), ) return default return wrapper return decorator @fail_safe(default=[]) async def get_recommendations(user_id: str) -> list[str]: """Get recommendations, return empty list on failure.""" ... ``` ## Best Practices Summary 1. **Retry only transient errors** - Don't retry bugs or authentication failures 2. **Use exponential backoff** - Give services time to recover 3. **Add jitter** - Prevent thundering herd from synchronized retries 4. **Cap total duration** - `stop_after_attempt(5) | stop_after_delay(60)` 5. **Log every retry** - Silent retries hide systemic problems 6. **Use decorators** - Keep retry logic separate from business logic 7. **Inject dependencies** - Make infrastructure testable 8. **Set timeouts everywhere** - Every network call needs a timeout 9. **Fail gracefully** - Return cached/default values for non-critical paths 10. **Monitor retry rates** - High retry rates indicate underlying issues
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-resource-management

Python resource management with context managers, cleanup patterns,

coding
⭐1
# Python Resource Management Manage resources deterministically using context managers. Resources like database connections, file handles, and network sockets should be released reliably, even when exceptions occur. ## When to Use This Skill - Managing database connections and connection pools - Working with file handles and I/O - Implementing custom context managers - Building streaming responses with state - Handling nested resource cleanup - Creating async context managers ## Core Concepts ### 1. Context Managers The `with` statement ensures resources are released automatically, even on exceptions. ### 2. Protocol Methods `__enter__`/`__exit__` for sync, `__aenter__`/`__aexit__` for async resource management. ### 3. Unconditional Cleanup `__exit__` always runs, regardless of whether an exception occurred. ### 4. Exception Handling Return `True` from `__exit__` to suppress exceptions, `False` to propagate them. ## Quick Start ```python from contextlib import contextmanager @contextmanager def managed_resource(): resource = acquire_resource() try: yield resource finally: resource.cleanup() with managed_resource() as r: r.do_work() ``` ## Fundamental Patterns ### Pattern 1: Class-Based Context Manager Implement the context manager protocol for complex resources. ```python class DatabaseConnection: """Database connection with automatic cleanup.""" def __init__(self, dsn: str) -> None: self._dsn = dsn self._conn: Connection | None = None def connect(self) -> None: """Establish database connection.""" self._conn = psycopg.connect(self._dsn) def close(self) -> None: """Close connection if open.""" if self._conn is not None: self._conn.close() self._conn = None def __enter__(self) -> "DatabaseConnection": """Enter context: connect and return self.""" self.connect() return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> None: """Exit context: always close connection.""" self.close() # Usage with context manager (preferred) with DatabaseConnection(dsn) as db: result = db.execute(query) # Manual management when needed db = DatabaseConnection(dsn) db.connect() try: result = db.execute(query) finally: db.close() ``` ### Pattern 2: Async Context Manager For async resources, implement the async protocol. ```python class AsyncDatabasePool: """Async database connection pool.""" def __init__(self, dsn: str, min_size: int = 1, max_size: int = 10) -> None: self._dsn = dsn self._min_size = min_size self._max_size = max_size self._pool: asyncpg.Pool | None = None async def __aenter__(self) -> "AsyncDatabasePool": """Create connection pool.""" self._pool = await asyncpg.create_pool( self._dsn, min_size=self._min_size, max_size=self._max_size, ) return self async def __aexit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> None: """Close all connections in pool.""" if self._pool is not None: await self._pool.close() async def execute(self, query: str, *args) -> list[dict]: """Execute query using pooled connection.""" async with self._pool.acquire() as conn: return await conn.fetch(query, *args) # Usage async with AsyncDatabasePool(dsn) as pool: users = await pool.execute("SELECT * FROM users WHERE active = $1", True) ``` ### Pattern 3: Using @contextmanager Decorator Simplify context managers with the decorator for straightforward cases. ```python from contextlib import contextmanager, asynccontextmanager import time import structlog logger = structlog.get_logger() @contextmanager def timed_block(name: str): """Time a block of code.""" start = time.perf_counter() try: yield finally: elapsed = time.perf_counter() - start logger.info(f"{name} completed", duration_seconds=round(elapsed, 3)) # Usage with timed_block("data_processing"): process_large_dataset() @asynccontextmanager async def database_transaction(conn: AsyncConnection): """Manage database transaction.""" await conn.execute("BEGIN") try: yield conn await conn.execute("COMMIT") except Exception: await conn.execute("ROLLBACK") raise # Usage async with database_transaction(conn) as tx: await tx.execute("INSERT INTO users ...") await tx.execute("INSERT INTO audit_log ...") ``` ### Pattern 4: Unconditional Resource Release Always clean up resources in `__exit__`, regardless of exceptions. ```python class FileProcessor: """Process file with guaranteed cleanup.""" def __init__(self, path: str) -> None: self._path = path self._file: IO | None = None self._temp_files: list[Path] = [] def __enter__(self) -> "FileProcessor": self._file = open(self._path, "r") return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> None: """Clean up all resources unconditionally.""" # Close main file if self._file is not None: self._file.close() # Clean up any temporary files for temp_file in self._temp_files: try: temp_file.unlink() except OSError: pass # Best effort cleanup # Return None/False to propagate any exception ``` ## Advanced Patterns ### Pattern 5: Selective Exception Suppression Only suppress specific, documented exceptions. ```python class StreamWriter: """Writer that handles broken pipe gracefully.""" def __init__(self, stream) -> None: self._stream = stream def __enter__(self) -> "StreamWriter": return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None, ) -> bool: """Clean up, suppressing BrokenPipeError on shutdown.""" self._stream.close() # Suppress BrokenPipeError (client disconnected) # This is expected behavior, not an error if exc_type is BrokenPipeError: return True # Exception suppressed return False # Propagate all other exceptions ``` ### Pattern 6: Streaming with Accumulated State Maintain both incremental chunks and accumulated state during streaming. ```python from collections.abc import Generator from dataclasses import dataclass, field @dataclass class StreamingResult: """Accumulated streaming result.""" chunks: list[str] = field(default_factory=list) _finalized: bool = False @property def content(self) -> str: """Get accumulated content.""" return "".join(self.chunks) def add_chunk(self, chunk: str) -> None: """Add chunk to accumulator.""" if self._finalized: raise RuntimeError("Cannot add to finalized result") self.chunks.append(chunk) def finalize(self) -> str: """Mark stream complete and return content.""" self._finalized = True return self.content def stream_with_accumulation( response: StreamingResponse, ) -> Generator[tuple[str, str], None, str]: """Stream response while accumulating content. Yields: Tuple of (accumulated_content, new_chunk) for each chunk. Returns: Final accumulated content. """ result = StreamingResult() for chunk in response.iter_content(): result.add_chunk(chunk) yield result.content, chunk return result.finalize() ``` ### Pattern 7: Efficient String Accumulation Avoid O(nΒ²) string concatenation when accumulating. ```python def accumulate_stream(stream) -> str: """Efficiently accumulate stream content.""" # BAD: O(nΒ²) due to string immutability # content = "" # for chunk in stream: # content += chunk # Creates new string each time # GOOD: O(n) with list and join chunks: list[str] = [] for chunk in stream: chunks.append(chunk) return "".join(chunks) # Single allocation ``` ### Pattern 8: Tracking Stream Metrics Measure time-to-first-byte and total streaming time. ```python import time from collections.abc import Generator def stream_with_metrics( response: StreamingResponse, ) -> Generator[str, None, dict]: """Stream response while collecting metrics. Yields: Content chunks. Returns: Metrics dictionary. """ start = time.perf_counter() first_chunk_time: float | None = None chunk_count = 0 total_bytes = 0 for chunk in response.iter_content(): if first_chunk_time is None: first_chunk_time = time.perf_counter() - start chunk_count += 1 total_bytes += len(chunk.encode()) yield chunk total_time = time.perf_counter() - start return { "time_to_first_byte_ms": round((first_chunk_time or 0) * 1000, 2), "total_time_ms": round(total_time * 1000, 2), "chunk_count": chunk_count, "total_bytes": total_bytes, } ``` ### Pattern 9: Managing Multiple Resources with ExitStack Handle a dynamic number of resources cleanly. ```python from contextlib import ExitStack, AsyncExitStack from pathlib import Path def process_files(paths: list[Path]) -> list[str]: """Process multiple files with automatic cleanup.""" results = [] with ExitStack() as stack: # Open all files - they'll all be closed when block exits files = [stack.enter_context(open(p)) for p in paths] for f in files: results.append(f.read()) return results async def process_connections(hosts: list[str]) -> list[dict]: """Process multiple async connections.""" results = [] async with AsyncExitStack() as stack: connections = [ await stack.enter_async_context(connect_to_host(host)) for host in hosts ] for conn in connections: results.append(await conn.fetch_data()) return results ``` ## Best Practices Summary 1. **Always use context managers** - For any resource that needs cleanup 2. **Clean up unconditionally** - `__exit__` runs even on exception 3. **Don't suppress unexpectedly** - Return `False` unless suppression is intentional 4. **Use @contextmanager** - For simple resource patterns 5. **Implement both protocols** - Support `with` and manual management 6. **Use ExitStack** - For dynamic numbers of resources 7. **Accumulate efficiently** - List + join, not string concatenation 8. **Track metrics** - Time-to-first-byte matters for streaming 9. **Document behavior** - Especially exception suppression 10. **Test cleanup paths** - Verify resources are released on errors
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered