Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

19 NODES
πŸ“textβ€’3 hours ago

Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering

Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering User centric prompting for analyzing inspiration pages or instructing on internal analysis cycles

architecture
⭐1
# Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering You are an **Expert Enterprise Architect, Product Director, and Lead Engineer**. Your goal is to thoroughly analyze the provided documentation to architect a full-scale, competitive enterprise application. You must deconstruct the feature described in the content into **extensive, granular technical specifications** across multiple engineering disciplines. **CRITICAL INSTRUCTIONS**: - **DO NOT SUMMARIZE**. Be exhaustive. - For every category below, aim to list **10+ specific items** if possible. - Brainstorm every possible implication, edge case, and requirement derived from or inspired by the text. - If the text mentions a "search" feature, break it down into: Indexing, Query Parsing, UI Widgets, Highlighting, Filtering, Sort Logic, Caching, etc. Please output the analysis in the following Markdown format: # 1. Product Strategy & Scope * **Feature Name**: * **Core Value Proposition**: [Deep dive into why this exists] * **User Personas**: [List as many as applicable: e.g. Admin, Power User, Viewer, Auditor, API Consumer...] * **User Stories**: [Extensive list of 10+ granular user stories e.g. "As a User, I want to..."] * **Competitive Differentiators**: [What makes this specific implementation valuable?] # 2. Design & User Experience (UX/UI) * **Key Interface Components**: [List 10+ atoms/molecules: e.g. Data Grid, Filter Chips, Modals, Tooltips, Empty States, Toasts, Dropdowns...] * **Interaction Patterns**: [List 10+ patterns: e.g. Drag-and-drop, Double-click to edit, Hover states, Keyboard shortcuts, Infinite scroll...] * **Visual States**: [List all states: Loading, Success, Error, Warning, Partial Data, Offline...] * **Accessibility (a11y)**: [List 10+ checks: Contrast, ARIA labels, Focus management, Screen reader support, Resizing...] # 3. Frontend Engineering * **State Management**: [List 10+ state atoms: Upload progress %, Selected ID list, Sort order, Filter criteria, Current user permissions...] * **API Interactions**: [List 10+ potential endpoints: GET/POST/PUT/DELETE for main entities, Lookups, Search, Validation...] * **Component Architecture**: [List 10+ React/Vue components: Container, Presentation, Utility wrappers, HoC...] * **Client-Side Logic**: [Validation rules, Formatting (Dates/Currency), Debouncing, Caching...] # 4. Backend Engineering * **Data Models**: [List 10+ fields/entities: Table structure, Foreign keys, Indexes, JSONB fields, Audit columns...] * **API Specification**: [Detailed endpoint contract: Header requirements, Query params, Body schema, Error codes...] * **Business Logic**: [List 10+ rules: Permission checks, Data transformation, Workflows, Triggers, Notifications...] * **Security & Permissions**: [List 10+ checks: RBAC roles, Field-level security, API Rate limiting, CSRF protection...] # 5. Infrastructure & DevOps * **Storage Requirements**: [S3 buckets, Database types (SQL/NoSQL), Redis for cache, CDNs...] * **Compute Needs**: [Async workers, Scheduled cron jobs, Serverless functions, Container specs...] * **Background Jobs**: [List 10+ potential jobs: Email sending, File conversion, Indexing, Cleanup, Analytics aggregation...] * **Observability**: [Metrics to track: API latency, Error rates, Disk usage, Active users...] # 6. Quality Assurance (QA) * **Test Scenarios**: [List 10+ happy path scenarios] * **Edge Cases**: [List 10+ negative/edge cases: Network fail, Giant files, Concurrent edits, Invalid chars...] * **Performance Metrics**: [Specific SLAs: <200ms API, <1s Page load, 99.9% Uptime...] * **Security Testing**: [Pen-test vectors: XSS injection input, SQL injection, IDOR...] # 7. Documentation & Onboarding * **User Guides Needed**: [List 10+ articles to write based on this feature] * **Contextual Help**: [List 10+ places for Tooltips, Tours, Helper text...] * **API Documentation**: [Swagger/OpenAPI requirements] # 8. Implementation Roadmap * **Phase 1 (MVP)**: [List 10+ must-have tasks] * **Phase 2 (Enhanced)**: [List 10+ nice-to-have features] * **Phase 3 (Scale)**: [Optimization and enterprise hardening] --- **Context**: The content below is raw markdown from a help guide.
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ“textβ€’3 hours ago

Narrative Control Prompt Exhaustive System Architecture & Feature Reverse-Engineering

Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering User concentric prompting for analyzing inspiration pages or instructing on internal analysis cycles

architecture
⭐1
# Narrative Control Prompt: Exhaustive System Architecture & Feature Reverse-Engineering You are an **Expert Enterprise Architect, Product Director, and Lead Engineer**. Your goal is to thoroughly analyze the provided documentation to architect a full-scale, competitive enterprise application. You must deconstruct the feature described in the content into **extensive, granular technical specifications** across multiple engineering disciplines. **CRITICAL INSTRUCTIONS**: - **DO NOT SUMMARIZE**. Be exhaustive. - For every category below, aim to list **10+ specific items** if possible. - Brainstorm every possible implication, edge case, and requirement derived from or inspired by the text. - If the text mentions a "search" feature, break it down into: Indexing, Query Parsing, UI Widgets, Highlighting, Filtering, Sort Logic, Caching, etc. Please output the analysis in the following Markdown format: # 1. Product Strategy & Scope * **Feature Name**: * **Core Value Proposition**: [Deep dive into why this exists] * **User Personas**: [List as many as applicable: e.g. Admin, Power User, Viewer, Auditor, API Consumer...] * **User Stories**: [Extensive list of 10+ granular user stories e.g. "As a User, I want to..."] * **Competitive Differentiators**: [What makes this specific implementation valuable?] # 2. Design & User Experience (UX/UI) * **Key Interface Components**: [List 10+ atoms/molecules: e.g. Data Grid, Filter Chips, Modals, Tooltips, Empty States, Toasts, Dropdowns...] * **Interaction Patterns**: [List 10+ patterns: e.g. Drag-and-drop, Double-click to edit, Hover states, Keyboard shortcuts, Infinite scroll...] * **Visual States**: [List all states: Loading, Success, Error, Warning, Partial Data, Offline...] * **Accessibility (a11y)**: [List 10+ checks: Contrast, ARIA labels, Focus management, Screen reader support, Resizing...] # 3. Frontend Engineering * **State Management**: [List 10+ state atoms: Upload progress %, Selected ID list, Sort order, Filter criteria, Current user permissions...] * **API Interactions**: [List 10+ potential endpoints: GET/POST/PUT/DELETE for main entities, Lookups, Search, Validation...] * **Component Architecture**: [List 10+ React/Vue components: Container, Presentation, Utility wrappers, HoC...] * **Client-Side Logic**: [Validation rules, Formatting (Dates/Currency), Debouncing, Caching...] # 4. Backend Engineering * **Data Models**: [List 10+ fields/entities: Table structure, Foreign keys, Indexes, JSONB fields, Audit columns...] * **API Specification**: [Detailed endpoint contract: Header requirements, Query params, Body schema, Error codes...] * **Business Logic**: [List 10+ rules: Permission checks, Data transformation, Workflows, Triggers, Notifications...] * **Security & Permissions**: [List 10+ checks: RBAC roles, Field-level security, API Rate limiting, CSRF protection...] # 5. Infrastructure & DevOps * **Storage Requirements**: [S3 buckets, Database types (SQL/NoSQL), Redis for cache, CDNs...] * **Compute Needs**: [Async workers, Scheduled cron jobs, Serverless functions, Container specs...] * **Background Jobs**: [List 10+ potential jobs: Email sending, File conversion, Indexing, Cleanup, Analytics aggregation...] * **Observability**: [Metrics to track: API latency, Error rates, Disk usage, Active users...] # 6. Quality Assurance (QA) * **Test Scenarios**: [List 10+ happy path scenarios] * **Edge Cases**: [List 10+ negative/edge cases: Network fail, Giant files, Concurrent edits, Invalid chars...] * **Performance Metrics**: [Specific SLAs: <200ms API, <1s Page load, 99.9% Uptime...] * **Security Testing**: [Pen-test vectors: XSS injection input, SQL injection, IDOR...] # 7. Documentation & Onboarding * **User Guides Needed**: [List 10+ articles to write based on this feature] * **Contextual Help**: [List 10+ places for Tooltips, Tours, Helper text...] * **API Documentation**: [Swagger/OpenAPI requirements] # 8. Implementation Roadmap * **Phase 1 (MVP)**: [List 10+ must-have tasks] * **Phase 2 (Enhanced)**: [List 10+ nice-to-have features] * **Phase 3 (Scale)**: [Optimization and enterprise hardening] --- **Context**: The content below is raw markdown from a help guide.
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

deployment-pipeline-design

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

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

postgresql-table-design

Design a PostgreSQL-specific schema. Covers best-practices, data

data
⭐1
# PostgreSQL Table Design ## Core Rules - Define a **PRIMARY KEY** for reference tables (users, orders, etc.). Not always needed for time-series/event/log data. When used, prefer `BIGINT GENERATED ALWAYS AS IDENTITY`; use `UUID` only when global uniqueness/opacity is needed. - **Normalize first (to 3NF)** to eliminate data redundancy and update anomalies; denormalize **only** for measured, high-ROI reads where join performance is proven problematic. Premature denormalization creates maintenance burden. - Add **NOT NULL** everywhere it’s semantically required; use **DEFAULT**s for common values. - Create **indexes for access paths you actually query**: PK/unique (auto), **FK columns (manual!)**, frequent filters/sorts, and join keys. - Prefer **TIMESTAMPTZ** for event time; **NUMERIC** for money; **TEXT** for strings; **BIGINT** for integer values, **DOUBLE PRECISION** for floats (or `NUMERIC` for exact decimal arithmetic). ## PostgreSQL β€œGotchas” - **Identifiers**: unquoted β†’ lowercased. Avoid quoted/mixed-case names. Convention: use `snake_case` for table/column names. - **Unique + NULLs**: UNIQUE allows multiple NULLs. Use `UNIQUE (...) NULLS NOT DISTINCT` (PG15+) to restrict to one NULL. - **FK indexes**: PostgreSQL **does not** auto-index FK columns. Add them. - **No silent coercions**: length/precision overflows error out (no truncation). Example: inserting 999 into `NUMERIC(2,0)` fails with error, unlike some databases that silently truncate or round. - **Sequences/identity have gaps** (normal; don't "fix"). Rollbacks, crashes, and concurrent transactions create gaps in ID sequences (1, 2, 5, 6...). This is expected behaviorβ€”don't try to make IDs consecutive. - **Heap storage**: no clustered PK by default (unlike SQL Server/MySQL InnoDB); `CLUSTER` is one-off reorganization, not maintained on subsequent inserts. Row order on disk is insertion order unless explicitly clustered. - **MVCC**: updates/deletes leave dead tuples; vacuum handles themβ€”design to avoid hot wide-row churn. ## Data Types - **IDs**: `BIGINT GENERATED ALWAYS AS IDENTITY` preferred (`GENERATED BY DEFAULT` also fine); `UUID` when merging/federating/used in a distributed system or for opaque IDs. Generate with `uuidv7()` (preferred if using PG18+) or `gen_random_uuid()` (if using an older PG version). - **Integers**: prefer `BIGINT` unless storage space is critical; `INTEGER` for smaller ranges; avoid `SMALLINT` unless constrained. - **Floats**: prefer `DOUBLE PRECISION` over `REAL` unless storage space is critical. Use `NUMERIC` for exact decimal arithmetic. - **Strings**: prefer `TEXT`; if length limits needed, use `CHECK (LENGTH(col) <= n)` instead of `VARCHAR(n)`; avoid `CHAR(n)`. Use `BYTEA` for binary data. Large strings/binary (>2KB default threshold) automatically stored in TOAST with compression. TOAST storage: `PLAIN` (no TOAST), `EXTENDED` (compress + out-of-line), `EXTERNAL` (out-of-line, no compress), `MAIN` (compress, keep in-line if possible). Default `EXTENDED` usually optimal. Control with `ALTER TABLE tbl ALTER COLUMN col SET STORAGE strategy` and `ALTER TABLE tbl SET (toast_tuple_target = 4096)` for threshold. Case-insensitive: for locale/accent handling use non-deterministic collations; for plain ASCII use expression indexes on `LOWER(col)` (preferred unless column needs case-insensitive PK/FK/UNIQUE) or `CITEXT`. - **Money**: `NUMERIC(p,s)` (never float). - **Time**: `TIMESTAMPTZ` for timestamps; `DATE` for date-only; `INTERVAL` for durations. Avoid `TIMESTAMP` (without timezone). Use `now()` for transaction start time, `clock_timestamp()` for current wall-clock time. - **Booleans**: `BOOLEAN` with `NOT NULL` constraint unless tri-state values are required. - **Enums**: `CREATE TYPE ... AS ENUM` for small, stable sets (e.g. US states, days of week). For business-logic-driven and evolving values (e.g. order statuses) β†’ use TEXT (or INT) + CHECK or lookup table. - **Arrays**: `TEXT[]`, `INTEGER[]`, etc. Use for ordered lists where you query elements. Index with **GIN** for containment (`@>`, `<@`) and overlap (`&&`) queries. Access: `arr[1]` (1-indexed), `arr[1:3]` (slicing). Good for tags, categories; avoid for relationsβ€”use junction tables instead. Literal syntax: `'{val1,val2}'` or `ARRAY[val1,val2]`. - **Range types**: `daterange`, `numrange`, `tstzrange` for intervals. Support overlap (`&&`), containment (`@>`), operators. Index with **GiST**. Good for scheduling, versioning, numeric ranges. Pick a bounds scheme and use it consistently; prefer `[)` (inclusive/exclusive) by default. - **Network types**: `INET` for IP addresses, `CIDR` for network ranges, `MACADDR` for MAC addresses. Support network operators (`<<`, `>>`, `&&`). - **Geometric types**: `POINT`, `LINE`, `POLYGON`, `CIRCLE` for 2D spatial data. Index with **GiST**. Consider **PostGIS** for advanced spatial features. - **Text search**: `TSVECTOR` for full-text search documents, `TSQUERY` for search queries. Index `tsvector` with **GIN**. Always specify language: `to_tsvector('english', col)` and `to_tsquery('english', 'query')`. Never use single-argument versions. This applies to both index expressions and queries. - **Domain types**: `CREATE DOMAIN email AS TEXT CHECK (VALUE ~ '^[^@]+@[^@]+$')` for reusable custom types with validation. Enforces constraints across tables. - **Composite types**: `CREATE TYPE address AS (street TEXT, city TEXT, zip TEXT)` for structured data within columns. Access with `(col).field` syntax. - **JSONB**: preferred over JSON; index with **GIN**. Use only for optional/semi-structured attrs. ONLY use JSON if the original ordering of the contents MUST be preserved. - **Vector types**: `vector` type by `pgvector` for vector similarity search for embeddings. ### Do not use the following data types - DO NOT use `timestamp` (without time zone); DO use `timestamptz` instead. - DO NOT use `char(n)` or `varchar(n)`; DO use `text` instead. - DO NOT use `money` type; DO use `numeric` instead. - DO NOT use `timetz` type; DO use `timestamptz` instead. - DO NOT use `timestamptz(0)` or any other precision specification; DO use `timestamptz` instead - DO NOT use `serial` type; DO use `generated always as identity` instead. ## Table Types - **Regular**: default; fully durable, logged. - **TEMPORARY**: session-scoped, auto-dropped, not logged. Faster for scratch work. - **UNLOGGED**: persistent but not crash-safe. Faster writes; good for caches/staging. ## Row-Level Security Enable with `ALTER TABLE tbl ENABLE ROW LEVEL SECURITY`. Create policies: `CREATE POLICY user_access ON orders FOR SELECT TO app_users USING (user_id = current_user_id())`. Built-in user-based access control at the row level. ## Constraints - **PK**: implicit UNIQUE + NOT NULL; creates a B-tree index. - **FK**: specify `ON DELETE/UPDATE` action (`CASCADE`, `RESTRICT`, `SET NULL`, `SET DEFAULT`). Add explicit index on referencing columnβ€”speeds up joins and prevents locking issues on parent deletes/updates. Use `DEFERRABLE INITIALLY DEFERRED` for circular FK dependencies checked at transaction end. - **UNIQUE**: creates a B-tree index; allows multiple NULLs unless `NULLS NOT DISTINCT` (PG15+). Standard behavior: `(1, NULL)` and `(1, NULL)` are allowed. With `NULLS NOT DISTINCT`: only one `(1, NULL)` allowed. Prefer `NULLS NOT DISTINCT` unless you specifically need duplicate NULLs. - **CHECK**: row-local constraints; NULL values pass the check (three-valued logic). Example: `CHECK (price > 0)` allows NULL prices. Combine with `NOT NULL` to enforce: `price NUMERIC NOT NULL CHECK (price > 0)`. - **EXCLUDE**: prevents overlapping values using operators. `EXCLUDE USING gist (room_id WITH =, booking_period WITH &&)` prevents double-booking rooms. Requires appropriate index type (often GiST). ## Indexing - **B-tree**: default for equality/range queries (`=`, `<`, `>`, `BETWEEN`, `ORDER BY`) - **Composite**: order mattersβ€”index used if equality on leftmost prefix (`WHERE a = ? AND b > ?` uses index on `(a,b)`, but `WHERE b = ?` does not). Put most selective/frequently filtered columns first. - **Covering**: `CREATE INDEX ON tbl (id) INCLUDE (name, email)` - includes non-key columns for index-only scans without visiting table. - **Partial**: for hot subsets (`WHERE status = 'active'` β†’ `CREATE INDEX ON tbl (user_id) WHERE status = 'active'`). Any query with `status = 'active'` can use this index. - **Expression**: for computed search keys (`CREATE INDEX ON tbl (LOWER(email))`). Expression must match exactly in WHERE clause: `WHERE LOWER(email) = 'user@example.com'`. - **GIN**: JSONB containment/existence, arrays (`@>`, `?`), full-text search (`@@`) - **GiST**: ranges, geometry, exclusion constraints - **BRIN**: very large, naturally ordered data (time-series)β€”minimal storage overhead. Effective when row order on disk correlates with indexed column (insertion order or after `CLUSTER`). ## Partitioning - Use for very large tables (>100M rows) where queries consistently filter on partition key (often time/date). - Alternate use: use for tables where data maintenance tasks dictates e.g. data pruned or bulk replaced periodically - **RANGE**: common for time-series (`PARTITION BY RANGE (created_at)`). Create partitions: `CREATE TABLE logs_2024_01 PARTITION OF logs FOR VALUES FROM ('2024-01-01') TO ('2024-02-01')`. **TimescaleDB** automates time-based or ID-based partitioning with retention policies and compression. - **LIST**: for discrete values (`PARTITION BY LIST (region)`). Example: `FOR VALUES IN ('us-east', 'us-west')`. - **HASH**: for even distribution when no natural key (`PARTITION BY HASH (user_id)`). Creates N partitions with modulus. - **Constraint exclusion**: requires `CHECK` constraints on partitions for query planner to prune. Auto-created for declarative partitioning (PG10+). - Prefer declarative partitioning or hypertables. Do NOT use table inheritance. - **Limitations**: no global UNIQUE constraintsβ€”include partition key in PK/UNIQUE. FKs from partitioned tables not supported; use triggers. ## Special Considerations ### Update-Heavy Tables - **Separate hot/cold columns**β€”put frequently updated columns in separate table to minimize bloat. - **Use `fillfactor=90`** to leave space for HOT updates that avoid index maintenance. - **Avoid updating indexed columns**β€”prevents beneficial HOT updates. - **Partition by update patterns**β€”separate frequently updated rows in a different partition from stable data. ### Insert-Heavy Workloads - **Minimize indexes**β€”only create what you query; every index slows inserts. - **Use `COPY` or multi-row `INSERT`** instead of single-row inserts. - **UNLOGGED tables** for rebuildable staging dataβ€”much faster writes. - **Defer index creation** for bulk loadsβ€”>drop index, load data, recreate indexes. - **Partition by time/hash** to distribute load. **TimescaleDB** automates partitioning and compression of insert-heavy data. - **Use a natural key for primary key** such as a (timestamp, device_id) if enforcing global uniqueness is important many insert-heavy tables don't need a primary key at all. - If you do need a surrogate key, **Prefer `BIGINT GENERATED ALWAYS AS IDENTITY` over `UUID`**. ### Upsert-Friendly Design - **Requires UNIQUE index** on conflict target columnsβ€”`ON CONFLICT (col1, col2)` needs exact matching unique index (partial indexes don't work). - **Use `EXCLUDED.column`** to reference would-be-inserted values; only update columns that actually changed to reduce write overhead. - **`DO NOTHING` faster** than `DO UPDATE` when no actual update needed. ### Safe Schema Evolution - **Transactional DDL**: most DDL operations can run in transactions and be rolled backβ€”`BEGIN; ALTER TABLE...; ROLLBACK;` for safe testing. - **Concurrent index creation**: `CREATE INDEX CONCURRENTLY` avoids blocking writes but can't run in transactions. - **Volatile defaults cause rewrites**: adding `NOT NULL` columns with volatile defaults (e.g., `now()`, `gen_random_uuid()`) rewrites entire table. Non-volatile defaults are fast. - **Drop constraints before columns**: `ALTER TABLE DROP CONSTRAINT` then `DROP COLUMN` to avoid dependency issues. - **Function signature changes**: `CREATE OR REPLACE` with different arguments creates overloads, not replacements. DROP old version if no overload desired. ## Generated Columns - `... GENERATED ALWAYS AS (<expr>) STORED` for computed, indexable fields. PG18+ adds `VIRTUAL` columns (computed on read, not stored). ## Extensions - **`pgcrypto`**: `crypt()` for password hashing. - **`uuid-ossp`**: alternative UUID functions; prefer `pgcrypto` for new projects. - **`pg_trgm`**: fuzzy text search with `%` operator, `similarity()` function. Index with GIN for `LIKE '%pattern%'` acceleration. - **`citext`**: case-insensitive text type. Prefer expression indexes on `LOWER(col)` unless you need case-insensitive constraints. - **`btree_gin`/`btree_gist`**: enable mixed-type indexes (e.g., GIN index on both JSONB and text columns). - **`hstore`**: key-value pairs; mostly superseded by JSONB but useful for simple string mappings. - **`timescaledb`**: essential for time-seriesβ€”automated partitioning, retention, compression, continuous aggregates. - **`postgis`**: comprehensive geospatial support beyond basic geometric typesβ€”essential for location-based applications. - **`pgvector`**: vector similarity search for embeddings. - **`pgaudit`**: audit logging for all database activity. ## JSONB Guidance - Prefer `JSONB` with **GIN** index. - Default: `CREATE INDEX ON tbl USING GIN (jsonb_col);` β†’ accelerates: - **Containment** `jsonb_col @> '{"k":"v"}'` - **Key existence** `jsonb_col ? 'k'`, **any/all keys** `?\|`, `?&` - **Path containment** on nested docs - **Disjunction** `jsonb_col @> ANY(ARRAY['{"status":"active"}', '{"status":"pending"}'])` - Heavy `@>` workloads: consider opclass `jsonb_path_ops` for smaller/faster containment-only indexes: - `CREATE INDEX ON tbl USING GIN (jsonb_col jsonb_path_ops);` - **Trade-off**: loses support for key existence (`?`, `?|`, `?&`) queriesβ€”only supports containment (`@>`) - Equality/range on a specific scalar field: extract and index with B-tree (generated column or expression): - `ALTER TABLE tbl ADD COLUMN price INT GENERATED ALWAYS AS ((jsonb_col->>'price')::INT) STORED;` - `CREATE INDEX ON tbl (price);` - Prefer queries like `WHERE price BETWEEN 100 AND 500` (uses B-tree) over `WHERE (jsonb_col->>'price')::INT BETWEEN 100 AND 500` without index. - Arrays inside JSONB: use GIN + `@>` for containment (e.g., tags). Consider `jsonb_path_ops` if only doing containment. - Keep core relations in tables; use JSONB for optional/variable attributes. - Use constraints to limit allowed JSONB values in a column e.g. `config JSONB NOT NULL CHECK(jsonb_typeof(config) = 'object')` ## Examples ### Users ```sql CREATE TABLE users ( user_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, email TEXT NOT NULL UNIQUE, name TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE UNIQUE INDEX ON users (LOWER(email)); CREATE INDEX ON users (created_at); ``` ### Orders ```sql CREATE TABLE orders ( order_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY, user_id BIGINT NOT NULL REFERENCES users(user_id), status TEXT NOT NULL DEFAULT 'PENDING' CHECK (status IN ('PENDING','PAID','CANCELED')), total NUMERIC(10,2) NOT NULL CHECK (total > 0), created_at TIMESTAMPTZ NOT NULL DEFAULT now() ); CREATE INDEX ON orders (user_id); CREATE INDEX ON orders (created_at); ``` ### JSONB ```sql CREATE TABLE profiles ( user_id BIGINT PRIMARY KEY REFERENCES users(user_id), attrs JSONB NOT NULL DEFAULT '{}', theme TEXT GENERATED ALWAYS AS (attrs->>'theme') STORED ); CREATE INDEX profiles_attrs_gin ON profiles USING GIN (attrs); ```
πŸ‘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

error-handling-patterns

Master error handling patterns across languages including

coding
⭐1
# Error Handling Patterns Build resilient applications with robust error handling strategies that gracefully handle failures and provide excellent debugging experiences. ## When to Use This Skill - Implementing error handling in new features - Designing error-resilient APIs - Debugging production issues - Improving application reliability - Creating better error messages for users and developers - Implementing retry and circuit breaker patterns - Handling async/concurrent errors - Building fault-tolerant distributed systems ## Core Concepts ### 1. Error Handling Philosophies **Exceptions vs Result Types:** - **Exceptions**: Traditional try-catch, disrupts control flow - **Result Types**: Explicit success/failure, functional approach - **Error Codes**: C-style, requires discipline - **Option/Maybe Types**: For nullable values **When to Use Each:** - Exceptions: Unexpected errors, exceptional conditions - Result Types: Expected errors, validation failures - Panics/Crashes: Unrecoverable errors, programming bugs ### 2. Error Categories **Recoverable Errors:** - Network timeouts - Missing files - Invalid user input - API rate limits **Unrecoverable Errors:** - Out of memory - Stack overflow - Programming bugs (null pointer, etc.) ## Language-Specific Patterns ### Python Error Handling **Custom Exception Hierarchy:** ```python class ApplicationError(Exception): """Base exception for all application errors.""" def __init__(self, message: str, code: str = None, details: dict = None): super().__init__(message) self.code = code self.details = details or {} self.timestamp = datetime.utcnow() class ValidationError(ApplicationError): """Raised when validation fails.""" pass class NotFoundError(ApplicationError): """Raised when resource not found.""" pass class ExternalServiceError(ApplicationError): """Raised when external service fails.""" def __init__(self, message: str, service: str, **kwargs): super().__init__(message, **kwargs) self.service = service # Usage def get_user(user_id: str) -> User: user = db.query(User).filter_by(id=user_id).first() if not user: raise NotFoundError( f"User not found", code="USER_NOT_FOUND", details={"user_id": user_id} ) return user ``` **Context Managers for Cleanup:** ```python from contextlib import contextmanager @contextmanager def database_transaction(session): """Ensure transaction is committed or rolled back.""" try: yield session session.commit() except Exception as e: session.rollback() raise finally: session.close() # Usage with database_transaction(db.session) as session: user = User(name="Alice") session.add(user) # Automatic commit or rollback ``` **Retry with Exponential Backoff:** ```python import time from functools import wraps from typing import TypeVar, Callable T = TypeVar('T') def retry( max_attempts: int = 3, backoff_factor: float = 2.0, exceptions: tuple = (Exception,) ): """Retry decorator with exponential backoff.""" def decorator(func: Callable[..., T]) -> Callable[..., T]: @wraps(func) def wrapper(*args, **kwargs) -> T: last_exception = None for attempt in range(max_attempts): try: return func(*args, **kwargs) except exceptions as e: last_exception = e if attempt < max_attempts - 1: sleep_time = backoff_factor ** attempt time.sleep(sleep_time) continue raise raise last_exception return wrapper return decorator # Usage @retry(max_attempts=3, exceptions=(NetworkError,)) def fetch_data(url: str) -> dict: response = requests.get(url, timeout=5) response.raise_for_status() return response.json() ``` ### TypeScript/JavaScript Error Handling **Custom Error Classes:** ```typescript // Custom error classes class ApplicationError extends Error { constructor( message: string, public code: string, public statusCode: number = 500, public details?: Record<string, any>, ) { super(message); this.name = this.constructor.name; Error.captureStackTrace(this, this.constructor); } } class ValidationError extends ApplicationError { constructor(message: string, details?: Record<string, any>) { super(message, "VALIDATION_ERROR", 400, details); } } class NotFoundError extends ApplicationError { constructor(resource: string, id: string) { super(`${resource} not found`, "NOT_FOUND", 404, { resource, id }); } } // Usage function getUser(id: string): User { const user = users.find((u) => u.id === id); if (!user) { throw new NotFoundError("User", id); } return user; } ``` **Result Type Pattern:** ```typescript // Result type for explicit error handling type Result<T, E = Error> = { ok: true; value: T } | { ok: false; error: E }; // Helper functions function Ok<T>(value: T): Result<T, never> { return { ok: true, value }; } function Err<E>(error: E): Result<never, E> { return { ok: false, error }; } // Usage function parseJSON<T>(json: string): Result<T, SyntaxError> { try { const value = JSON.parse(json) as T; return Ok(value); } catch (error) { return Err(error as SyntaxError); } } // Consuming Result const result = parseJSON<User>(userJson); if (result.ok) { console.log(result.value.name); } else { console.error("Parse failed:", result.error.message); } // Chaining Results function chain<T, U, E>( result: Result<T, E>, fn: (value: T) => Result<U, E>, ): Result<U, E> { return result.ok ? fn(result.value) : result; } ``` **Async Error Handling:** ```typescript // Async/await with proper error handling async function fetchUserOrders(userId: string): Promise<Order[]> { try { const user = await getUser(userId); const orders = await getOrders(user.id); return orders; } catch (error) { if (error instanceof NotFoundError) { return []; // Return empty array for not found } if (error instanceof NetworkError) { // Retry logic return retryFetchOrders(userId); } // Re-throw unexpected errors throw error; } } // Promise error handling function fetchData(url: string): Promise<Data> { return fetch(url) .then((response) => { if (!response.ok) { throw new NetworkError(`HTTP ${response.status}`); } return response.json(); }) .catch((error) => { console.error("Fetch failed:", error); throw error; }); } ``` ### Rust Error Handling **Result and Option Types:** ```rust use std::fs::File; use std::io::{self, Read}; // Result type for operations that can fail fn read_file(path: &str) -> Result<String, io::Error> { let mut file = File::open(path)?; // ? operator propagates errors let mut contents = String::new(); file.read_to_string(&mut contents)?; Ok(contents) } // Custom error types #[derive(Debug)] enum AppError { Io(io::Error), Parse(std::num::ParseIntError), NotFound(String), Validation(String), } impl From<io::Error> for AppError { fn from(error: io::Error) -> Self { AppError::Io(error) } } // Using custom error type fn read_number_from_file(path: &str) -> Result<i32, AppError> { let contents = read_file(path)?; // Auto-converts io::Error let number = contents.trim().parse() .map_err(AppError::Parse)?; // Explicitly convert ParseIntError Ok(number) } // Option for nullable values fn find_user(id: &str) -> Option<User> { users.iter().find(|u| u.id == id).cloned() } // Combining Option and Result fn get_user_age(id: &str) -> Result<u32, AppError> { find_user(id) .ok_or_else(|| AppError::NotFound(id.to_string())) .map(|user| user.age) } ``` ### Go Error Handling **Explicit Error Returns:** ```go // Basic error handling func getUser(id string) (*User, error) { user, err := db.QueryUser(id) if err != nil { return nil, fmt.Errorf("failed to query user: %w", err) } if user == nil { return nil, errors.New("user not found") } return user, nil } // Custom error types type ValidationError struct { Field string Message string } func (e *ValidationError) Error() string { return fmt.Sprintf("validation failed for %s: %s", e.Field, e.Message) } // Sentinel errors for comparison var ( ErrNotFound = errors.New("not found") ErrUnauthorized = errors.New("unauthorized") ErrInvalidInput = errors.New("invalid input") ) // Error checking user, err := getUser("123") if err != nil { if errors.Is(err, ErrNotFound) { // Handle not found } else { // Handle other errors } } // Error wrapping and unwrapping func processUser(id string) error { user, err := getUser(id) if err != nil { return fmt.Errorf("process user failed: %w", err) } // Process user return nil } // Unwrap errors err := processUser("123") if err != nil { var valErr *ValidationError if errors.As(err, &valErr) { fmt.Printf("Validation error: %s\n", valErr.Field) } } ``` ## Universal Patterns ### Pattern 1: Circuit Breaker Prevent cascading failures in distributed systems. ```python from enum import Enum from datetime import datetime, timedelta from typing import Callable, TypeVar T = TypeVar('T') class CircuitState(Enum): CLOSED = "closed" # Normal operation OPEN = "open" # Failing, reject requests HALF_OPEN = "half_open" # Testing if recovered class CircuitBreaker: def __init__( self, failure_threshold: int = 5, timeout: timedelta = timedelta(seconds=60), success_threshold: int = 2 ): self.failure_threshold = failure_threshold self.timeout = timeout self.success_threshold = success_threshold self.failure_count = 0 self.success_count = 0 self.state = CircuitState.CLOSED self.last_failure_time = None def call(self, func: Callable[[], T]) -> T: if self.state == CircuitState.OPEN: if datetime.now() - self.last_failure_time > self.timeout: self.state = CircuitState.HALF_OPEN self.success_count = 0 else: raise Exception("Circuit breaker is OPEN") try: result = func() self.on_success() return result except Exception as e: self.on_failure() raise def on_success(self): 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): self.failure_count += 1 self.last_failure_time = datetime.now() if self.failure_count >= self.failure_threshold: self.state = CircuitState.OPEN # Usage circuit_breaker = CircuitBreaker() def fetch_data(): return circuit_breaker.call(lambda: external_api.get_data()) ``` ### Pattern 2: Error Aggregation Collect multiple errors instead of failing on first error. ```typescript class ErrorCollector { private errors: Error[] = []; add(error: Error): void { this.errors.push(error); } hasErrors(): boolean { return this.errors.length > 0; } getErrors(): Error[] { return [...this.errors]; } throw(): never { if (this.errors.length === 1) { throw this.errors[0]; } throw new AggregateError( this.errors, `${this.errors.length} errors occurred`, ); } } // Usage: Validate multiple fields function validateUser(data: any): User { const errors = new ErrorCollector(); if (!data.email) { errors.add(new ValidationError("Email is required")); } else if (!isValidEmail(data.email)) { errors.add(new ValidationError("Email is invalid")); } if (!data.name || data.name.length < 2) { errors.add(new ValidationError("Name must be at least 2 characters")); } if (!data.age || data.age < 18) { errors.add(new ValidationError("Age must be 18 or older")); } if (errors.hasErrors()) { errors.throw(); } return data as User; } ``` ### Pattern 3: Graceful Degradation Provide fallback functionality when errors occur. ```python from typing import Optional, Callable, TypeVar T = TypeVar('T') def with_fallback( primary: Callable[[], T], fallback: Callable[[], T], log_error: bool = True ) -> T: """Try primary function, fall back to fallback on error.""" try: return primary() except Exception as e: if log_error: logger.error(f"Primary function failed: {e}") return fallback() # Usage def get_user_profile(user_id: str) -> UserProfile: return with_fallback( primary=lambda: fetch_from_cache(user_id), fallback=lambda: fetch_from_database(user_id) ) # Multiple fallbacks def get_exchange_rate(currency: str) -> float: return ( try_function(lambda: api_provider_1.get_rate(currency)) or try_function(lambda: api_provider_2.get_rate(currency)) or try_function(lambda: cache.get_rate(currency)) or DEFAULT_RATE ) def try_function(func: Callable[[], Optional[T]]) -> Optional[T]: try: return func() except Exception: return None ``` ## Best Practices 1. **Fail Fast**: Validate input early, fail quickly 2. **Preserve Context**: Include stack traces, metadata, timestamps 3. **Meaningful Messages**: Explain what happened and how to fix it 4. **Log Appropriately**: Error = log, expected failure = don't spam logs 5. **Handle at Right Level**: Catch where you can meaningfully handle 6. **Clean Up Resources**: Use try-finally, context managers, defer 7. **Don't Swallow Errors**: Log or re-throw, don't silently ignore 8. **Type-Safe Errors**: Use typed errors when possible ```python # Good error handling example def process_order(order_id: str) -> Order: """Process order with comprehensive error handling.""" try: # Validate input if not order_id: raise ValidationError("Order ID is required") # Fetch order order = db.get_order(order_id) if not order: raise NotFoundError("Order", order_id) # Process payment try: payment_result = payment_service.charge(order.total) except PaymentServiceError as e: # Log and wrap external service error logger.error(f"Payment failed for order {order_id}: {e}") raise ExternalServiceError( f"Payment processing failed", service="payment_service", details={"order_id": order_id, "amount": order.total} ) from e # Update order order.status = "completed" order.payment_id = payment_result.id db.save(order) return order except ApplicationError: # Re-raise known application errors raise except Exception as e: # Log unexpected errors logger.exception(f"Unexpected error processing order {order_id}") raise ApplicationError( "Order processing failed", code="INTERNAL_ERROR" ) from e ``` ## Common Pitfalls - **Catching Too Broadly**: `except Exception` hides bugs - **Empty Catch Blocks**: Silently swallowing errors - **Logging and Re-throwing**: Creates duplicate log entries - **Not Cleaning Up**: Forgetting to close files, connections - **Poor Error Messages**: "Error occurred" is not helpful - **Returning Error Codes**: Use exceptions or Result types - **Ignoring Async Errors**: Unhandled promise rejections ## Resources - **references/exception-hierarchy-design.md**: Designing error class hierarchies - **references/error-recovery-strategies.md**: Recovery patterns for different scenarios - **references/async-error-handling.md**: Handling errors in concurrent code - **assets/error-handling-checklist.md**: Review checklist for error handling - **assets/error-message-guide.md**: Writing helpful error messages - **scripts/error-analyzer.py**: Analyze error patterns in logs
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

sql-optimization-patterns

Master SQL query optimization, indexing strategies, and EXPLAIN

coding
⭐1
# SQL Optimization Patterns Transform slow database queries into lightning-fast operations through systematic optimization, proper indexing, and query plan analysis. ## When to Use This Skill - Debugging slow-running queries - Designing performant database schemas - Optimizing application response times - Reducing database load and costs - Improving scalability for growing datasets - Analyzing EXPLAIN query plans - Implementing efficient indexes - Resolving N+1 query problems ## Core Concepts ### 1. Query Execution Plans (EXPLAIN) Understanding EXPLAIN output is fundamental to optimization. **PostgreSQL EXPLAIN:** ```sql -- Basic explain EXPLAIN SELECT * FROM users WHERE email = 'user@example.com'; -- With actual execution stats EXPLAIN ANALYZE SELECT * FROM users WHERE email = 'user@example.com'; -- Verbose output with more details EXPLAIN (ANALYZE, BUFFERS, VERBOSE) SELECT u.*, o.order_total FROM users u JOIN orders o ON u.id = o.user_id WHERE u.created_at > NOW() - INTERVAL '30 days'; ``` **Key Metrics to Watch:** - **Seq Scan**: Full table scan (usually slow for large tables) - **Index Scan**: Using index (good) - **Index Only Scan**: Using index without touching table (best) - **Nested Loop**: Join method (okay for small datasets) - **Hash Join**: Join method (good for larger datasets) - **Merge Join**: Join method (good for sorted data) - **Cost**: Estimated query cost (lower is better) - **Rows**: Estimated rows returned - **Actual Time**: Real execution time ### 2. Index Strategies Indexes are the most powerful optimization tool. **Index Types:** - **B-Tree**: Default, good for equality and range queries - **Hash**: Only for equality (=) comparisons - **GIN**: Full-text search, array queries, JSONB - **GiST**: Geometric data, full-text search - **BRIN**: Block Range INdex for very large tables with correlation ```sql -- Standard B-Tree index CREATE INDEX idx_users_email ON users(email); -- Composite index (order matters!) CREATE INDEX idx_orders_user_status ON orders(user_id, status); -- Partial index (index subset of rows) CREATE INDEX idx_active_users ON users(email) WHERE status = 'active'; -- Expression index CREATE INDEX idx_users_lower_email ON users(LOWER(email)); -- Covering index (include additional columns) CREATE INDEX idx_users_email_covering ON users(email) INCLUDE (name, created_at); -- Full-text search index CREATE INDEX idx_posts_search ON posts USING GIN(to_tsvector('english', title || ' ' || body)); -- JSONB index CREATE INDEX idx_metadata ON events USING GIN(metadata); ``` ### 3. Query Optimization Patterns **Avoid SELECT \*:** ```sql -- Bad: Fetches unnecessary columns SELECT * FROM users WHERE id = 123; -- Good: Fetch only what you need SELECT id, email, name FROM users WHERE id = 123; ``` **Use WHERE Clause Efficiently:** ```sql -- Bad: Function prevents index usage SELECT * FROM users WHERE LOWER(email) = 'user@example.com'; -- Good: Create functional index or use exact match CREATE INDEX idx_users_email_lower ON users(LOWER(email)); -- Then: SELECT * FROM users WHERE LOWER(email) = 'user@example.com'; -- Or store normalized data SELECT * FROM users WHERE email = 'user@example.com'; ``` **Optimize JOINs:** ```sql -- Bad: Cartesian product then filter SELECT u.name, o.total FROM users u, orders o WHERE u.id = o.user_id AND u.created_at > '2024-01-01'; -- Good: Filter before join SELECT u.name, o.total FROM users u JOIN orders o ON u.id = o.user_id WHERE u.created_at > '2024-01-01'; -- Better: Filter both tables SELECT u.name, o.total FROM (SELECT * FROM users WHERE created_at > '2024-01-01') u JOIN orders o ON u.id = o.user_id; ``` ## Optimization Patterns ### Pattern 1: Eliminate N+1 Queries **Problem: N+1 Query Anti-Pattern** ```python # Bad: Executes N+1 queries users = db.query("SELECT * FROM users LIMIT 10") for user in users: orders = db.query("SELECT * FROM orders WHERE user_id = ?", user.id) # Process orders ``` **Solution: Use JOINs or Batch Loading** ```sql -- Solution 1: JOIN SELECT u.id, u.name, o.id as order_id, o.total FROM users u LEFT JOIN orders o ON u.id = o.user_id WHERE u.id IN (1, 2, 3, 4, 5); -- Solution 2: Batch query SELECT * FROM orders WHERE user_id IN (1, 2, 3, 4, 5); ``` ```python # Good: Single query with JOIN or batch load # Using JOIN results = db.query(""" SELECT u.id, u.name, o.id as order_id, o.total FROM users u LEFT JOIN orders o ON u.id = o.user_id WHERE u.id IN (1, 2, 3, 4, 5) """) # Or batch load users = db.query("SELECT * FROM users LIMIT 10") user_ids = [u.id for u in users] orders = db.query( "SELECT * FROM orders WHERE user_id IN (?)", user_ids ) # Group orders by user_id orders_by_user = {} for order in orders: orders_by_user.setdefault(order.user_id, []).append(order) ``` ### Pattern 2: Optimize Pagination **Bad: OFFSET on Large Tables** ```sql -- Slow for large offsets SELECT * FROM users ORDER BY created_at DESC LIMIT 20 OFFSET 100000; -- Very slow! ``` **Good: Cursor-Based Pagination** ```sql -- Much faster: Use cursor (last seen ID) SELECT * FROM users WHERE created_at < '2024-01-15 10:30:00' -- Last cursor ORDER BY created_at DESC LIMIT 20; -- With composite sorting SELECT * FROM users WHERE (created_at, id) < ('2024-01-15 10:30:00', 12345) ORDER BY created_at DESC, id DESC LIMIT 20; -- Requires index CREATE INDEX idx_users_cursor ON users(created_at DESC, id DESC); ``` ### Pattern 3: Aggregate Efficiently **Optimize COUNT Queries:** ```sql -- Bad: Counts all rows SELECT COUNT(*) FROM orders; -- Slow on large tables -- Good: Use estimates for approximate counts SELECT reltuples::bigint AS estimate FROM pg_class WHERE relname = 'orders'; -- Good: Filter before counting SELECT COUNT(*) FROM orders WHERE created_at > NOW() - INTERVAL '7 days'; -- Better: Use index-only scan CREATE INDEX idx_orders_created ON orders(created_at); SELECT COUNT(*) FROM orders WHERE created_at > NOW() - INTERVAL '7 days'; ``` **Optimize GROUP BY:** ```sql -- Bad: Group by then filter SELECT user_id, COUNT(*) as order_count FROM orders GROUP BY user_id HAVING COUNT(*) > 10; -- Better: Filter first, then group (if possible) SELECT user_id, COUNT(*) as order_count FROM orders WHERE status = 'completed' GROUP BY user_id HAVING COUNT(*) > 10; -- Best: Use covering index CREATE INDEX idx_orders_user_status ON orders(user_id, status); ``` ### Pattern 4: Subquery Optimization **Transform Correlated Subqueries:** ```sql -- Bad: Correlated subquery (runs for each row) SELECT u.name, u.email, (SELECT COUNT(*) FROM orders o WHERE o.user_id = u.id) as order_count FROM users u; -- Good: JOIN with aggregation SELECT u.name, u.email, COUNT(o.id) as order_count FROM users u LEFT JOIN orders o ON o.user_id = u.id GROUP BY u.id, u.name, u.email; -- Better: Use window functions SELECT DISTINCT ON (u.id) u.name, u.email, COUNT(o.id) OVER (PARTITION BY u.id) as order_count FROM users u LEFT JOIN orders o ON o.user_id = u.id; ``` **Use CTEs for Clarity:** ```sql -- Using Common Table Expressions WITH recent_users AS ( SELECT id, name, email FROM users WHERE created_at > NOW() - INTERVAL '30 days' ), user_order_counts AS ( SELECT user_id, COUNT(*) as order_count FROM orders WHERE created_at > NOW() - INTERVAL '30 days' GROUP BY user_id ) SELECT ru.name, ru.email, COALESCE(uoc.order_count, 0) as orders FROM recent_users ru LEFT JOIN user_order_counts uoc ON ru.id = uoc.user_id; ``` ### Pattern 5: Batch Operations **Batch INSERT:** ```sql -- Bad: Multiple individual inserts INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com'); INSERT INTO users (name, email) VALUES ('Bob', 'bob@example.com'); INSERT INTO users (name, email) VALUES ('Carol', 'carol@example.com'); -- Good: Batch insert INSERT INTO users (name, email) VALUES ('Alice', 'alice@example.com'), ('Bob', 'bob@example.com'), ('Carol', 'carol@example.com'); -- Better: Use COPY for bulk inserts (PostgreSQL) COPY users (name, email) FROM '/tmp/users.csv' CSV HEADER; ``` **Batch UPDATE:** ```sql -- Bad: Update in loop UPDATE users SET status = 'active' WHERE id = 1; UPDATE users SET status = 'active' WHERE id = 2; -- ... repeat for many IDs -- Good: Single UPDATE with IN clause UPDATE users SET status = 'active' WHERE id IN (1, 2, 3, 4, 5, ...); -- Better: Use temporary table for large batches CREATE TEMP TABLE temp_user_updates (id INT, new_status VARCHAR); INSERT INTO temp_user_updates VALUES (1, 'active'), (2, 'active'), ...; UPDATE users u SET status = t.new_status FROM temp_user_updates t WHERE u.id = t.id; ``` ## Advanced Techniques ### Materialized Views Pre-compute expensive queries. ```sql -- Create materialized view CREATE MATERIALIZED VIEW user_order_summary AS SELECT u.id, u.name, COUNT(o.id) as total_orders, SUM(o.total) as total_spent, MAX(o.created_at) as last_order_date FROM users u LEFT JOIN orders o ON u.id = o.user_id GROUP BY u.id, u.name; -- Add index to materialized view CREATE INDEX idx_user_summary_spent ON user_order_summary(total_spent DESC); -- Refresh materialized view REFRESH MATERIALIZED VIEW user_order_summary; -- Concurrent refresh (PostgreSQL) REFRESH MATERIALIZED VIEW CONCURRENTLY user_order_summary; -- Query materialized view (very fast) SELECT * FROM user_order_summary WHERE total_spent > 1000 ORDER BY total_spent DESC; ``` ### Partitioning Split large tables for better performance. ```sql -- Range partitioning by date (PostgreSQL) CREATE TABLE orders ( id SERIAL, user_id INT, total DECIMAL, created_at TIMESTAMP ) PARTITION BY RANGE (created_at); -- Create partitions CREATE TABLE orders_2024_q1 PARTITION OF orders FOR VALUES FROM ('2024-01-01') TO ('2024-04-01'); CREATE TABLE orders_2024_q2 PARTITION OF orders FOR VALUES FROM ('2024-04-01') TO ('2024-07-01'); -- Queries automatically use appropriate partition SELECT * FROM orders WHERE created_at BETWEEN '2024-02-01' AND '2024-02-28'; -- Only scans orders_2024_q1 partition ``` ### Query Hints and Optimization ```sql -- Force index usage (MySQL) SELECT * FROM users USE INDEX (idx_users_email) WHERE email = 'user@example.com'; -- Parallel query (PostgreSQL) SET max_parallel_workers_per_gather = 4; SELECT * FROM large_table WHERE condition; -- Join hints (PostgreSQL) SET enable_nestloop = OFF; -- Force hash or merge join ``` ## Best Practices 1. **Index Selectively**: Too many indexes slow down writes 2. **Monitor Query Performance**: Use slow query logs 3. **Keep Statistics Updated**: Run ANALYZE regularly 4. **Use Appropriate Data Types**: Smaller types = better performance 5. **Normalize Thoughtfully**: Balance normalization vs performance 6. **Cache Frequently Accessed Data**: Use application-level caching 7. **Connection Pooling**: Reuse database connections 8. **Regular Maintenance**: VACUUM, ANALYZE, rebuild indexes ```sql -- Update statistics ANALYZE users; ANALYZE VERBOSE orders; -- Vacuum (PostgreSQL) VACUUM ANALYZE users; VACUUM FULL users; -- Reclaim space (locks table) -- Reindex REINDEX INDEX idx_users_email; REINDEX TABLE users; ``` ## Common Pitfalls - **Over-Indexing**: Each index slows down INSERT/UPDATE/DELETE - **Unused Indexes**: Waste space and slow writes - **Missing Indexes**: Slow queries, full table scans - **Implicit Type Conversion**: Prevents index usage - **OR Conditions**: Can't use indexes efficiently - **LIKE with Leading Wildcard**: `LIKE '%abc'` can't use index - **Function in WHERE**: Prevents index usage unless functional index exists ## Monitoring Queries ```sql -- Find slow queries (PostgreSQL) SELECT query, calls, total_time, mean_time FROM pg_stat_statements ORDER BY mean_time DESC LIMIT 10; -- Find missing indexes (PostgreSQL) SELECT schemaname, tablename, seq_scan, seq_tup_read, idx_scan, seq_tup_read / seq_scan AS avg_seq_tup_read FROM pg_stat_user_tables WHERE seq_scan > 0 ORDER BY seq_tup_read DESC LIMIT 10; -- Find unused indexes (PostgreSQL) SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read, idx_tup_fetch FROM pg_stat_user_indexes WHERE idx_scan = 0 ORDER BY pg_relation_size(indexrelid) DESC; ``` ## Resources - **references/postgres-optimization-guide.md**: PostgreSQL-specific optimization - **references/mysql-optimization-guide.md**: MySQL/MariaDB optimization - **references/query-plan-analysis.md**: Deep dive into EXPLAIN plans - **assets/index-strategy-checklist.md**: When and how to create indexes - **assets/query-optimization-checklist.md**: Step-by-step optimization guide - **scripts/analyze-slow-queries.sql**: Identify slow queries in your database - **scripts/index-recommendations.sql**: Generate index recommendations
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

architecture-decision-records

Write and maintain Architecture Decision Records (ADRs) following

coding
⭐1
# Architecture Decision Records Comprehensive patterns for creating, maintaining, and managing Architecture Decision Records (ADRs) that capture the context and rationale behind significant technical decisions. ## When to Use This Skill - Making significant architectural decisions - Documenting technology choices - Recording design trade-offs - Onboarding new team members - Reviewing historical decisions - Establishing decision-making processes ## Core Concepts ### 1. What is an ADR? An Architecture Decision Record captures: - **Context**: Why we needed to make a decision - **Decision**: What we decided - **Consequences**: What happens as a result ### 2. When to Write an ADR | Write ADR | Skip ADR | | -------------------------- | ---------------------- | | New framework adoption | Minor version upgrades | | Database technology choice | Bug fixes | | API design patterns | Implementation details | | Security architecture | Routine maintenance | | Integration patterns | Configuration changes | ### 3. ADR Lifecycle ``` Proposed β†’ Accepted β†’ Deprecated β†’ Superseded ↓ Rejected ``` ## Templates ### Template 1: Standard ADR (MADR Format) ```markdown # ADR-0001: Use PostgreSQL as Primary Database ## Status Accepted ## Context We need to select a primary database for our new e-commerce platform. The system will handle: - ~10,000 concurrent users - Complex product catalog with hierarchical categories - Transaction processing for orders and payments - Full-text search for products - Geospatial queries for store locator The team has experience with MySQL, PostgreSQL, and MongoDB. We need ACID compliance for financial transactions. ## Decision Drivers - **Must have ACID compliance** for payment processing - **Must support complex queries** for reporting - **Should support full-text search** to reduce infrastructure complexity - **Should have good JSON support** for flexible product attributes - **Team familiarity** reduces onboarding time ## Considered Options ### Option 1: PostgreSQL - **Pros**: ACID compliant, excellent JSON support (JSONB), built-in full-text search, PostGIS for geospatial, team has experience - **Cons**: Slightly more complex replication setup than MySQL ### Option 2: MySQL - **Pros**: Very familiar to team, simple replication, large community - **Cons**: Weaker JSON support, no built-in full-text search (need Elasticsearch), no geospatial without extensions ### Option 3: MongoDB - **Pros**: Flexible schema, native JSON, horizontal scaling - **Cons**: No ACID for multi-document transactions (at decision time), team has limited experience, requires schema design discipline ## Decision We will use **PostgreSQL 15** as our primary database. ## Rationale PostgreSQL provides the best balance of: 1. **ACID compliance** essential for e-commerce transactions 2. **Built-in capabilities** (full-text search, JSONB, PostGIS) reduce infrastructure complexity 3. **Team familiarity** with SQL databases reduces learning curve 4. **Mature ecosystem** with excellent tooling and community support The slight complexity in replication is outweighed by the reduction in additional services (no separate Elasticsearch needed). ## Consequences ### Positive - Single database handles transactions, search, and geospatial queries - Reduced operational complexity (fewer services to manage) - Strong consistency guarantees for financial data - Team can leverage existing SQL expertise ### Negative - Need to learn PostgreSQL-specific features (JSONB, full-text search syntax) - Vertical scaling limits may require read replicas sooner - Some team members need PostgreSQL-specific training ### Risks - Full-text search may not scale as well as dedicated search engines - Mitigation: Design for potential Elasticsearch addition if needed ## Implementation Notes - Use JSONB for flexible product attributes - Implement connection pooling with PgBouncer - Set up streaming replication for read replicas - Use pg_trgm extension for fuzzy search ## Related Decisions - ADR-0002: Caching Strategy (Redis) - complements database choice - ADR-0005: Search Architecture - may supersede if Elasticsearch needed ## References - [PostgreSQL JSON Documentation](https://www.postgresql.org/docs/current/datatype-json.html) - [PostgreSQL Full Text Search](https://www.postgresql.org/docs/current/textsearch.html) - Internal: Performance benchmarks in `/docs/benchmarks/database-comparison.md` ``` ### Template 2: Lightweight ADR ```markdown # ADR-0012: Adopt TypeScript for Frontend Development **Status**: Accepted **Date**: 2024-01-15 **Deciders**: @alice, @bob, @charlie ## Context Our React codebase has grown to 50+ components with increasing bug reports related to prop type mismatches and undefined errors. PropTypes provide runtime-only checking. ## Decision Adopt TypeScript for all new frontend code. Migrate existing code incrementally. ## Consequences **Good**: Catch type errors at compile time, better IDE support, self-documenting code. **Bad**: Learning curve for team, initial slowdown, build complexity increase. **Mitigations**: TypeScript training sessions, allow gradual adoption with `allowJs: true`. ``` ### Template 3: Y-Statement Format ```markdown # ADR-0015: API Gateway Selection In the context of **building a microservices architecture**, facing **the need for centralized API management, authentication, and rate limiting**, we decided for **Kong Gateway** and against **AWS API Gateway and custom Nginx solution**, to achieve **vendor independence, plugin extensibility, and team familiarity with Lua**, accepting that **we need to manage Kong infrastructure ourselves**. ``` ### Template 4: ADR for Deprecation ```markdown # ADR-0020: Deprecate MongoDB in Favor of PostgreSQL ## Status Accepted (Supersedes ADR-0003) ## Context ADR-0003 (2021) chose MongoDB for user profile storage due to schema flexibility needs. Since then: - MongoDB's multi-document transactions remain problematic for our use case - Our schema has stabilized and rarely changes - We now have PostgreSQL expertise from other services - Maintaining two databases increases operational burden ## Decision Deprecate MongoDB and migrate user profiles to PostgreSQL. ## Migration Plan 1. **Phase 1** (Week 1-2): Create PostgreSQL schema, dual-write enabled 2. **Phase 2** (Week 3-4): Backfill historical data, validate consistency 3. **Phase 3** (Week 5): Switch reads to PostgreSQL, monitor 4. **Phase 4** (Week 6): Remove MongoDB writes, decommission ## Consequences ### Positive - Single database technology reduces operational complexity - ACID transactions for user data - Team can focus PostgreSQL expertise ### Negative - Migration effort (~4 weeks) - Risk of data issues during migration - Lose some schema flexibility ## Lessons Learned Document from ADR-0003 experience: - Schema flexibility benefits were overestimated - Operational cost of multiple databases was underestimated - Consider long-term maintenance in technology decisions ``` ### Template 5: Request for Comments (RFC) Style ```markdown # RFC-0025: Adopt Event Sourcing for Order Management ## Summary Propose adopting event sourcing pattern for the order management domain to improve auditability, enable temporal queries, and support business analytics. ## Motivation Current challenges: 1. Audit requirements need complete order history 2. "What was the order state at time X?" queries are impossible 3. Analytics team needs event stream for real-time dashboards 4. Order state reconstruction for customer support is manual ## Detailed Design ### Event Store ``` OrderCreated { orderId, customerId, items[], timestamp } OrderItemAdded { orderId, item, timestamp } OrderItemRemoved { orderId, itemId, timestamp } PaymentReceived { orderId, amount, paymentId, timestamp } OrderShipped { orderId, trackingNumber, timestamp } ``` ### Projections - **CurrentOrderState**: Materialized view for queries - **OrderHistory**: Complete timeline for audit - **DailyOrderMetrics**: Analytics aggregation ### Technology - Event Store: EventStoreDB (purpose-built, handles projections) - Alternative considered: Kafka + custom projection service ## Drawbacks - Learning curve for team - Increased complexity vs. CRUD - Need to design events carefully (immutable once stored) - Storage growth (events never deleted) ## Alternatives 1. **Audit tables**: Simpler but doesn't enable temporal queries 2. **CDC from existing DB**: Complex, doesn't change data model 3. **Hybrid**: Event source only for order state changes ## Unresolved Questions - [ ] Event schema versioning strategy - [ ] Retention policy for events - [ ] Snapshot frequency for performance ## Implementation Plan 1. Prototype with single order type (2 weeks) 2. Team training on event sourcing (1 week) 3. Full implementation and migration (4 weeks) 4. Monitoring and optimization (ongoing) ## References - [Event Sourcing by Martin Fowler](https://martinfowler.com/eaaDev/EventSourcing.html) - [EventStoreDB Documentation](https://www.eventstore.com/docs) ``` ## ADR Management ### Directory Structure ``` docs/ β”œβ”€β”€ adr/ β”‚ β”œβ”€β”€ README.md # Index and guidelines β”‚ β”œβ”€β”€ template.md # Team's ADR template β”‚ β”œβ”€β”€ 0001-use-postgresql.md β”‚ β”œβ”€β”€ 0002-caching-strategy.md β”‚ β”œβ”€β”€ 0003-mongodb-user-profiles.md # [DEPRECATED] β”‚ └── 0020-deprecate-mongodb.md # Supersedes 0003 ``` ### ADR Index (README.md) ```markdown # Architecture Decision Records This directory contains Architecture Decision Records (ADRs) for [Project Name]. ## Index | ADR | Title | Status | Date | | ------------------------------------- | ---------------------------------- | ---------- | ---------- | | [0001](0001-use-postgresql.md) | Use PostgreSQL as Primary Database | Accepted | 2024-01-10 | | [0002](0002-caching-strategy.md) | Caching Strategy with Redis | Accepted | 2024-01-12 | | [0003](0003-mongodb-user-profiles.md) | MongoDB for User Profiles | Deprecated | 2023-06-15 | | [0020](0020-deprecate-mongodb.md) | Deprecate MongoDB | Accepted | 2024-01-15 | ## Creating a New ADR 1. Copy `template.md` to `NNNN-title-with-dashes.md` 2. Fill in the template 3. Submit PR for review 4. Update this index after approval ## ADR Status - **Proposed**: Under discussion - **Accepted**: Decision made, implementing - **Deprecated**: No longer relevant - **Superseded**: Replaced by another ADR - **Rejected**: Considered but not adopted ``` ### Automation (adr-tools) ```bash # Install adr-tools brew install adr-tools # Initialize ADR directory adr init docs/adr # Create new ADR adr new "Use PostgreSQL as Primary Database" # Supersede an ADR adr new -s 3 "Deprecate MongoDB in Favor of PostgreSQL" # Generate table of contents adr generate toc > docs/adr/README.md # Link related ADRs adr link 2 "Complements" 1 "Is complemented by" ``` ## Review Process ```markdown ## ADR Review Checklist ### Before Submission - [ ] Context clearly explains the problem - [ ] All viable options considered - [ ] Pros/cons balanced and honest - [ ] Consequences (positive and negative) documented - [ ] Related ADRs linked ### During Review - [ ] At least 2 senior engineers reviewed - [ ] Affected teams consulted - [ ] Security implications considered - [ ] Cost implications documented - [ ] Reversibility assessed ### After Acceptance - [ ] ADR index updated - [ ] Team notified - [ ] Implementation tickets created - [ ] Related documentation updated ``` ## Best Practices ### Do's - **Write ADRs early** - Before implementation starts - **Keep them short** - 1-2 pages maximum - **Be honest about trade-offs** - Include real cons - **Link related decisions** - Build decision graph - **Update status** - Deprecate when superseded ### Don'ts - **Don't change accepted ADRs** - Write new ones to supersede - **Don't skip context** - Future readers need background - **Don't hide failures** - Rejected decisions are valuable - **Don't be vague** - Specific decisions, specific consequences - **Don't forget implementation** - ADR without action is waste ## Resources - [Documenting Architecture Decisions (Michael Nygard)](https://cognitect.com/blog/2011/11/15/documenting-architecture-decisions) - [MADR Template](https://adr.github.io/madr/) - [ADR GitHub Organization](https://adr.github.io/) - [adr-tools](https://github.com/npryce/adr-tools)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

react-modernization

Upgrade React applications to latest versions, migrate from class

coding
⭐1
# React Modernization Master React version upgrades, class to hooks migration, concurrent features adoption, and codemods for automated transformation. ## When to Use This Skill - Upgrading React applications to latest versions - Migrating class components to functional components with hooks - Adopting concurrent React features (Suspense, transitions) - Applying codemods for automated refactoring - Modernizing state management patterns - Updating to TypeScript - Improving performance with React 18+ features ## Version Upgrade Path ### React 16 β†’ 17 β†’ 18 **Breaking Changes by Version:** **React 17:** - Event delegation changes - No event pooling - Effect cleanup timing - JSX transform (no React import needed) **React 18:** - Automatic batching - Concurrent rendering - Strict Mode changes (double invocation) - New root API - Suspense on server ## Class to Hooks Migration ### State Management ```javascript // Before: Class component class Counter extends React.Component { constructor(props) { super(props); this.state = { count: 0, name: "", }; } increment = () => { this.setState({ count: this.state.count + 1 }); }; render() { return ( <div> <p>Count: {this.state.count}</p> <button onClick={this.increment}>Increment</button> </div> ); } } // After: Functional component with hooks function Counter() { const [count, setCount] = useState(0); const [name, setName] = useState(""); const increment = () => { setCount(count + 1); }; return ( <div> <p>Count: {count}</p> <button onClick={increment}>Increment</button> </div> ); } ``` ### Lifecycle Methods to Hooks ```javascript // Before: Lifecycle methods class DataFetcher extends React.Component { state = { data: null, loading: true }; componentDidMount() { this.fetchData(); } componentDidUpdate(prevProps) { if (prevProps.id !== this.props.id) { this.fetchData(); } } componentWillUnmount() { this.cancelRequest(); } fetchData = async () => { const data = await fetch(`/api/${this.props.id}`); this.setState({ data, loading: false }); }; cancelRequest = () => { // Cleanup }; render() { if (this.state.loading) return <div>Loading...</div>; return <div>{this.state.data}</div>; } } // After: useEffect hook function DataFetcher({ id }) { const [data, setData] = useState(null); const [loading, setLoading] = useState(true); useEffect(() => { let cancelled = false; const fetchData = async () => { try { const response = await fetch(`/api/${id}`); const result = await response.json(); if (!cancelled) { setData(result); setLoading(false); } } catch (error) { if (!cancelled) { console.error(error); } } }; fetchData(); // Cleanup function return () => { cancelled = true; }; }, [id]); // Re-run when id changes if (loading) return <div>Loading...</div>; return <div>{data}</div>; } ``` ### Context and HOCs to Hooks ```javascript // Before: Context consumer and HOC const ThemeContext = React.createContext(); class ThemedButton extends React.Component { static contextType = ThemeContext; render() { return ( <button style={{ background: this.context.theme }}> {this.props.children} </button> ); } } // After: useContext hook function ThemedButton({ children }) { const { theme } = useContext(ThemeContext); return <button style={{ background: theme }}>{children}</button>; } // Before: HOC for data fetching function withUser(Component) { return class extends React.Component { state = { user: null }; componentDidMount() { fetchUser().then((user) => this.setState({ user })); } render() { return <Component {...this.props} user={this.state.user} />; } }; } // After: Custom hook function useUser() { const [user, setUser] = useState(null); useEffect(() => { fetchUser().then(setUser); }, []); return user; } function UserProfile() { const user = useUser(); if (!user) return <div>Loading...</div>; return <div>{user.name}</div>; } ``` ## React 18 Concurrent Features ### New Root API ```javascript // Before: React 17 import ReactDOM from "react-dom"; ReactDOM.render(<App />, document.getElementById("root")); // After: React 18 import { createRoot } from "react-dom/client"; const root = createRoot(document.getElementById("root")); root.render(<App />); ``` ### Automatic Batching ```javascript // React 18: All updates are batched function handleClick() { setCount((c) => c + 1); setFlag((f) => !f); // Only one re-render (batched) } // Even in async: setTimeout(() => { setCount((c) => c + 1); setFlag((f) => !f); // Still batched in React 18! }, 1000); // Opt out if needed import { flushSync } from "react-dom"; flushSync(() => { setCount((c) => c + 1); }); // Re-render happens here setFlag((f) => !f); // Another re-render ``` ### Transitions ```javascript import { useState, useTransition } from "react"; function SearchResults() { const [query, setQuery] = useState(""); const [results, setResults] = useState([]); const [isPending, startTransition] = useTransition(); const handleChange = (e) => { // Urgent: Update input immediately setQuery(e.target.value); // Non-urgent: Update results (can be interrupted) startTransition(() => { setResults(searchResults(e.target.value)); }); }; return ( <> <input value={query} onChange={handleChange} /> {isPending && <Spinner />} <Results data={results} /> </> ); } ``` ### Suspense for Data Fetching ```javascript import { Suspense } from "react"; // Resource-based data fetching (with React 18) const resource = fetchProfileData(); function ProfilePage() { return ( <Suspense fallback={<Loading />}> <ProfileDetails /> <Suspense fallback={<Loading />}> <ProfileTimeline /> </Suspense> </Suspense> ); } function ProfileDetails() { // This will suspend if data not ready const user = resource.user.read(); return <h1>{user.name}</h1>; } function ProfileTimeline() { const posts = resource.posts.read(); return <Timeline posts={posts} />; } ``` ## Codemods for Automation ### Run React Codemods ```bash # Rename unsafe lifecycle methods npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/rename-unsafe-lifecycles.js src/ # Update React imports (React 17+) npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/update-react-imports.js src/ # Add error boundaries npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/error-boundaries.js src/ # For TypeScript files npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/rename-unsafe-lifecycles.js --parser=tsx src/ # Dry run to preview changes npx jscodeshift -t https://raw.githubusercontent.com/reactjs/react-codemod/master/transforms/rename-unsafe-lifecycles.js --dry --print src/ # Class to Hooks (third-party) npx codemod react/hooks/convert-class-to-function src/ ``` ### Custom Codemod Example ```javascript // custom-codemod.js module.exports = function (file, api) { const j = api.jscodeshift; const root = j(file.source); // Find setState calls root .find(j.CallExpression, { callee: { type: "MemberExpression", property: { name: "setState" }, }, }) .forEach((path) => { // Transform to useState // ... transformation logic }); return root.toSource(); }; // Run: jscodeshift -t custom-codemod.js src/ ``` ## Performance Optimization ### useMemo and useCallback ```javascript function ExpensiveComponent({ items, filter }) { // Memoize expensive calculation const filteredItems = useMemo(() => { return items.filter((item) => item.category === filter); }, [items, filter]); // Memoize callback to prevent child re-renders const handleClick = useCallback((id) => { console.log("Clicked:", id); }, []); // No dependencies, never changes return <List items={filteredItems} onClick={handleClick} />; } // Child component with memo const List = React.memo(({ items, onClick }) => { return items.map((item) => ( <Item key={item.id} item={item} onClick={onClick} /> )); }); ``` ### Code Splitting ```javascript import { lazy, Suspense } from "react"; // Lazy load components const Dashboard = lazy(() => import("./Dashboard")); const Settings = lazy(() => import("./Settings")); function App() { return ( <Suspense fallback={<Loading />}> <Routes> <Route path="/dashboard" element={<Dashboard />} /> <Route path="/settings" element={<Settings />} /> </Routes> </Suspense> ); } ``` ## TypeScript Migration ```typescript // Before: JavaScript function Button({ onClick, children }) { return <button onClick={onClick}>{children}</button>; } // After: TypeScript interface ButtonProps { onClick: () => void; children: React.ReactNode; } function Button({ onClick, children }: ButtonProps) { return <button onClick={onClick}>{children}</button>; } // Generic components interface ListProps<T> { items: T[]; renderItem: (item: T) => React.ReactNode; } function List<T>({ items, renderItem }: ListProps<T>) { return <>{items.map(renderItem)}</>; } ``` ## Migration Checklist ```markdown ### Pre-Migration - [ ] Update dependencies incrementally (not all at once) - [ ] Review breaking changes in release notes - [ ] Set up testing suite - [ ] Create feature branch ### Class β†’ Hooks Migration - [ ] Identify class components to migrate - [ ] Start with leaf components (no children) - [ ] Convert state to useState - [ ] Convert lifecycle to useEffect - [ ] Convert context to useContext - [ ] Extract custom hooks - [ ] Test thoroughly ### React 18 Upgrade - [ ] Update to React 17 first (if needed) - [ ] Update react and react-dom to 18 - [ ] Update @types/react if using TypeScript - [ ] Change to createRoot API - [ ] Test with StrictMode (double invocation) - [ ] Address concurrent rendering issues - [ ] Adopt Suspense/Transitions where beneficial ### Performance - [ ] Identify performance bottlenecks - [ ] Add React.memo where appropriate - [ ] Use useMemo/useCallback for expensive operations - [ ] Implement code splitting - [ ] Optimize re-renders ### Testing - [ ] Update test utilities (React Testing Library) - [ ] Test with React 18 features - [ ] Check for warnings in console - [ ] Performance testing ``` ## Resources - **references/breaking-changes.md**: Version-specific breaking changes - **references/codemods.md**: Codemod usage guide - **references/hooks-migration.md**: Comprehensive hooks patterns - **references/concurrent-features.md**: React 18 concurrent features - **assets/codemod-config.json**: Codemod configurations - **assets/migration-checklist.md**: Step-by-step checklist - **scripts/apply-codemods.sh**: Automated codemod script ## Best Practices 1. **Incremental Migration**: Don't migrate everything at once 2. **Test Thoroughly**: Comprehensive testing at each step 3. **Use Codemods**: Automate repetitive transformations 4. **Start Simple**: Begin with leaf components 5. **Leverage StrictMode**: Catch issues early 6. **Monitor Performance**: Measure before and after 7. **Document Changes**: Keep migration log ## Common Pitfalls - Forgetting useEffect dependencies - Over-using useMemo/useCallback - Not handling cleanup in useEffect - Mixing class and functional patterns - Ignoring StrictMode warnings - Breaking change assumptions
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

parallel-debugging

Debug complex issues using competing hypotheses with parallel

coding
⭐1
# Parallel Debugging Framework for debugging complex issues using the Analysis of Competing Hypotheses (ACH) methodology with parallel agent investigation. ## When to Use This Skill - Bug has multiple plausible root causes - Initial debugging attempts haven't identified the issue - Issue spans multiple modules or components - Need systematic root cause analysis with evidence - Want to avoid confirmation bias in debugging ## Hypothesis Generation Framework Generate hypotheses across 6 failure mode categories: ### 1. Logic Error - Incorrect conditional logic (wrong operator, missing case) - Off-by-one errors in loops or array access - Missing edge case handling - Incorrect algorithm implementation ### 2. Data Issue - Invalid or unexpected input data - Type mismatch or coercion error - Null/undefined/None where value expected - Encoding or serialization problem - Data truncation or overflow ### 3. State Problem - Race condition between concurrent operations - Stale cache returning outdated data - Incorrect initialization or default values - Unintended mutation of shared state - State machine transition error ### 4. Integration Failure - API contract violation (request/response mismatch) - Version incompatibility between components - Configuration mismatch between environments - Missing or incorrect environment variables - Network timeout or connection failure ### 5. Resource Issue - Memory leak causing gradual degradation - Connection pool exhaustion - File descriptor or handle leak - Disk space or quota exceeded - CPU saturation from inefficient processing ### 6. Environment - Missing runtime dependency - Wrong library or framework version - Platform-specific behavior difference - Permission or access control issue - Timezone or locale-related behavior ## Evidence Collection Standards ### What Constitutes Evidence | Evidence Type | Strength | Example | | ----------------- | -------- | --------------------------------------------------------------- | | **Direct** | Strong | Code at `file.ts:42` shows `if (x > 0)` should be `if (x >= 0)` | | **Correlational** | Medium | Error rate increased after commit `abc123` | | **Testimonial** | Weak | "It works on my machine" | | **Absence** | Variable | No null check found in the code path | ### Citation Format Always cite evidence with file:line references: ``` **Evidence**: The validation function at `src/validators/user.ts:87` does not check for empty strings, only null/undefined. This allows empty email addresses to pass validation. ``` ### Confidence Levels | Level | Criteria | | ------------------- | ----------------------------------------------------------------------------------- | | **High (>80%)** | Multiple direct evidence pieces, clear causal chain, no contradicting evidence | | **Medium (50-80%)** | Some direct evidence, plausible causal chain, minor ambiguities | | **Low (<50%)** | Mostly correlational evidence, incomplete causal chain, some contradicting evidence | ## Result Arbitration Protocol After all investigators report: ### Step 1: Categorize Results - **Confirmed**: High confidence, strong evidence, clear causal chain - **Plausible**: Medium confidence, some evidence, reasonable causal chain - **Falsified**: Evidence contradicts the hypothesis - **Inconclusive**: Insufficient evidence to confirm or falsify ### Step 2: Compare Confirmed Hypotheses If multiple hypotheses are confirmed, rank by: 1. Confidence level 2. Number of supporting evidence pieces 3. Strength of causal chain 4. Absence of contradicting evidence ### Step 3: Determine Root Cause - If one hypothesis clearly dominates: declare as root cause - If multiple hypotheses are equally likely: may be compound issue (multiple contributing causes) - If no hypotheses confirmed: generate new hypotheses based on evidence gathered ### Step 4: Validate Fix Before declaring the bug fixed: - [ ] Fix addresses the identified root cause - [ ] Fix doesn't introduce new issues - [ ] Original reproduction case no longer fails - [ ] Related edge cases are covered - [ ] Relevant tests are added or updated
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-code-style

Python code style, linting, formatting, naming conventions, and

coding
⭐1
# Python Code Style & Documentation Consistent code style and clear documentation make codebases maintainable and collaborative. This skill covers modern Python tooling, naming conventions, and documentation standards. ## When to Use This Skill - Setting up linting and formatting for a new project - Writing or reviewing docstrings - Establishing team coding standards - Configuring ruff, mypy, or pyright - Reviewing code for style consistency - Creating project documentation ## Core Concepts ### 1. Automated Formatting Let tools handle formatting debates. Configure once, enforce automatically. ### 2. Consistent Naming Follow PEP 8 conventions with meaningful, descriptive names. ### 3. Documentation as Code Docstrings should be maintained alongside the code they describe. ### 4. Type Annotations Modern Python code should include type hints for all public APIs. ## Quick Start ```bash # Install modern tooling pip install ruff mypy # Configure in pyproject.toml [tool.ruff] line-length = 120 target-version = "py312" # Adjust based on your project's minimum Python version [tool.mypy] strict = true ``` ## Fundamental Patterns ### Pattern 1: Modern Python Tooling Use `ruff` as an all-in-one linter and formatter. It replaces flake8, isort, and black with a single fast tool. ```toml # pyproject.toml [tool.ruff] line-length = 120 target-version = "py312" # Adjust based on your project's minimum Python version [tool.ruff.lint] select = [ "E", # pycodestyle errors "W", # pycodestyle warnings "F", # pyflakes "I", # isort "B", # flake8-bugbear "C4", # flake8-comprehensions "UP", # pyupgrade "SIM", # flake8-simplify ] ignore = ["E501"] # Line length handled by formatter [tool.ruff.format] quote-style = "double" indent-style = "space" ``` Run with: ```bash ruff check --fix . # Lint and auto-fix ruff format . # Format code ``` ### Pattern 2: Type Checking Configuration Configure strict type checking for production code. ```toml # pyproject.toml [tool.mypy] python_version = "3.12" strict = true warn_return_any = true warn_unused_ignores = true disallow_untyped_defs = true disallow_incomplete_defs = true [[tool.mypy.overrides]] module = "tests.*" disallow_untyped_defs = false ``` Alternative: Use `pyright` for faster checking. ```toml [tool.pyright] pythonVersion = "3.12" typeCheckingMode = "strict" ``` ### Pattern 3: Naming Conventions Follow PEP 8 with emphasis on clarity over brevity. **Files and Modules:** ```python # Good: Descriptive snake_case user_repository.py order_processing.py http_client.py # Avoid: Abbreviations usr_repo.py ord_proc.py http_cli.py ``` **Classes and Functions:** ```python # Classes: PascalCase class UserRepository: pass class HTTPClientFactory: # Acronyms stay uppercase pass # Functions and variables: snake_case def get_user_by_email(email: str) -> User | None: retry_count = 3 max_connections = 100 ``` **Constants:** ```python # Module-level constants: SCREAMING_SNAKE_CASE MAX_RETRY_ATTEMPTS = 3 DEFAULT_TIMEOUT_SECONDS = 30 API_BASE_URL = "https://api.example.com" ``` ### Pattern 4: Import Organization Group imports in a consistent order: standard library, third-party, local. ```python # Standard library import os from collections.abc import Callable from typing import Any # Third-party packages import httpx from pydantic import BaseModel from sqlalchemy import Column # Local imports from myproject.models import User from myproject.services import UserService ``` Use absolute imports exclusively: ```python # Preferred from myproject.utils import retry_decorator # Avoid relative imports from ..utils import retry_decorator ``` ## Advanced Patterns ### Pattern 5: Google-Style Docstrings Write docstrings for all public classes, methods, and functions. **Simple Function:** ```python def get_user(user_id: str) -> User: """Retrieve a user by their unique identifier.""" ... ``` **Complex Function:** ```python def process_batch( items: list[Item], max_workers: int = 4, on_progress: Callable[[int, int], None] | None = None, ) -> BatchResult: """Process items concurrently using a worker pool. Processes each item in the batch using the configured number of workers. Progress can be monitored via the optional callback. Args: items: The items to process. Must not be empty. max_workers: Maximum concurrent workers. Defaults to 4. on_progress: Optional callback receiving (completed, total) counts. Returns: BatchResult containing succeeded items and any failures with their associated exceptions. Raises: ValueError: If items is empty. ProcessingError: If the batch cannot be processed. Example: >>> result = process_batch(items, max_workers=8) >>> print(f"Processed {len(result.succeeded)} items") """ ... ``` **Class Docstring:** ```python class UserService: """Service for managing user operations. Provides methods for creating, retrieving, updating, and deleting users with proper validation and error handling. Attributes: repository: The data access layer for user persistence. logger: Logger instance for operation tracking. Example: >>> service = UserService(repository, logger) >>> user = service.create_user(CreateUserInput(...)) """ def __init__(self, repository: UserRepository, logger: Logger) -> None: """Initialize the user service. Args: repository: Data access layer for users. logger: Logger for tracking operations. """ self.repository = repository self.logger = logger ``` ### Pattern 6: Line Length and Formatting Set line length to 120 characters for modern displays while maintaining readability. ```python # Good: Readable line breaks def create_user( email: str, name: str, role: UserRole = UserRole.MEMBER, notify: bool = True, ) -> User: ... # Good: Chain method calls clearly result = ( db.query(User) .filter(User.active == True) .order_by(User.created_at.desc()) .limit(10) .all() ) # Good: Format long strings error_message = ( f"Failed to process user {user_id}: " f"received status {response.status_code} " f"with body {response.text[:100]}" ) ``` ### Pattern 7: Project Documentation **README Structure:** ```markdown # Project Name Brief description of what the project does. ## Installation \`\`\`bash pip install myproject \`\`\` ## Quick Start \`\`\`python from myproject import Client client = Client(api_key="...") result = client.process(data) \`\`\` ## Configuration Document environment variables and configuration options. ## Development \`\`\`bash pip install -e ".[dev]" pytest \`\`\` ``` **CHANGELOG Format (Keep a Changelog):** ```markdown # Changelog ## [Unreleased] ### Added - New feature X ### Changed - Modified behavior of Y ### Fixed - Bug in Z ``` ## Best Practices Summary 1. **Use ruff** - Single tool for linting and formatting 2. **Enable strict mypy** - Catch type errors before runtime 3. **120 character lines** - Modern standard for readability 4. **Descriptive names** - Clarity over brevity 5. **Absolute imports** - More maintainable than relative 6. **Google-style docstrings** - Consistent, readable documentation 7. **Document public APIs** - Every public function needs a docstring 8. **Keep docs updated** - Treat documentation as code 9. **Automate in CI** - Run linters on every commit 10. **Target Python 3.10+** - For new projects, Python 3.12+ is recommended for modern language features
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-testing-patterns

Implement comprehensive testing strategies with pytest, fixtures,

coding
⭐1
# Python Testing Patterns Comprehensive guide to implementing robust testing strategies in Python using pytest, fixtures, mocking, parameterization, and test-driven development practices. ## When to Use This Skill - Writing unit tests for Python code - Setting up test suites and test infrastructure - Implementing test-driven development (TDD) - Creating integration tests for APIs and services - Mocking external dependencies and services - Testing async code and concurrent operations - Setting up continuous testing in CI/CD - Implementing property-based testing - Testing database operations - Debugging failing tests ## Core Concepts ### 1. Test Types - **Unit Tests**: Test individual functions/classes in isolation - **Integration Tests**: Test interaction between components - **Functional Tests**: Test complete features end-to-end - **Performance Tests**: Measure speed and resource usage ### 2. Test Structure (AAA Pattern) - **Arrange**: Set up test data and preconditions - **Act**: Execute the code under test - **Assert**: Verify the results ### 3. Test Coverage - Measure what code is exercised by tests - Identify untested code paths - Aim for meaningful coverage, not just high percentages ### 4. Test Isolation - Tests should be independent - No shared state between tests - Each test should clean up after itself ## Quick Start ```python # test_example.py def add(a, b): return a + b def test_add(): """Basic test example.""" result = add(2, 3) assert result == 5 def test_add_negative(): """Test with negative numbers.""" assert add(-1, 1) == 0 # Run with: pytest test_example.py ``` ## Fundamental Patterns ### Pattern 1: Basic pytest Tests ```python # test_calculator.py import pytest class Calculator: """Simple calculator for testing.""" def add(self, a: float, b: float) -> float: return a + b def subtract(self, a: float, b: float) -> float: return a - b def multiply(self, a: float, b: float) -> float: return a * b def divide(self, a: float, b: float) -> float: if b == 0: raise ValueError("Cannot divide by zero") return a / b def test_addition(): """Test addition.""" calc = Calculator() assert calc.add(2, 3) == 5 assert calc.add(-1, 1) == 0 assert calc.add(0, 0) == 0 def test_subtraction(): """Test subtraction.""" calc = Calculator() assert calc.subtract(5, 3) == 2 assert calc.subtract(0, 5) == -5 def test_multiplication(): """Test multiplication.""" calc = Calculator() assert calc.multiply(3, 4) == 12 assert calc.multiply(0, 5) == 0 def test_division(): """Test division.""" calc = Calculator() assert calc.divide(6, 3) == 2 assert calc.divide(5, 2) == 2.5 def test_division_by_zero(): """Test division by zero raises error.""" calc = Calculator() with pytest.raises(ValueError, match="Cannot divide by zero"): calc.divide(5, 0) ``` ### Pattern 2: Fixtures for Setup and Teardown ```python # test_database.py import pytest from typing import Generator class Database: """Simple database class.""" def __init__(self, connection_string: str): self.connection_string = connection_string self.connected = False def connect(self): """Connect to database.""" self.connected = True def disconnect(self): """Disconnect from database.""" self.connected = False def query(self, sql: str) -> list: """Execute query.""" if not self.connected: raise RuntimeError("Not connected") return [{"id": 1, "name": "Test"}] @pytest.fixture def db() -> Generator[Database, None, None]: """Fixture that provides connected database.""" # Setup database = Database("sqlite:///:memory:") database.connect() # Provide to test yield database # Teardown database.disconnect() def test_database_query(db): """Test database query with fixture.""" results = db.query("SELECT * FROM users") assert len(results) == 1 assert results[0]["name"] == "Test" @pytest.fixture(scope="session") def app_config(): """Session-scoped fixture - created once per test session.""" return { "database_url": "postgresql://localhost/test", "api_key": "test-key", "debug": True } @pytest.fixture(scope="module") def api_client(app_config): """Module-scoped fixture - created once per test module.""" # Setup expensive resource client = {"config": app_config, "session": "active"} yield client # Cleanup client["session"] = "closed" def test_api_client(api_client): """Test using api client fixture.""" assert api_client["session"] == "active" assert api_client["config"]["debug"] is True ``` ### Pattern 3: Parameterized Tests ```python # test_validation.py import pytest def is_valid_email(email: str) -> bool: """Check if email is valid.""" return "@" in email and "." in email.split("@")[1] @pytest.mark.parametrize("email,expected", [ ("user@example.com", True), ("test.user@domain.co.uk", True), ("invalid.email", False), ("@example.com", False), ("user@domain", False), ("", False), ]) def test_email_validation(email, expected): """Test email validation with various inputs.""" assert is_valid_email(email) == expected @pytest.mark.parametrize("a,b,expected", [ (2, 3, 5), (0, 0, 0), (-1, 1, 0), (100, 200, 300), (-5, -5, -10), ]) def test_addition_parameterized(a, b, expected): """Test addition with multiple parameter sets.""" from test_calculator import Calculator calc = Calculator() assert calc.add(a, b) == expected # Using pytest.param for special cases @pytest.mark.parametrize("value,expected", [ pytest.param(1, True, id="positive"), pytest.param(0, False, id="zero"), pytest.param(-1, False, id="negative"), ]) def test_is_positive(value, expected): """Test with custom test IDs.""" assert (value > 0) == expected ``` ### Pattern 4: Mocking with unittest.mock ```python # test_api_client.py import pytest from unittest.mock import Mock, patch, MagicMock import requests class APIClient: """Simple API client.""" def __init__(self, base_url: str): self.base_url = base_url def get_user(self, user_id: int) -> dict: """Fetch user from API.""" response = requests.get(f"{self.base_url}/users/{user_id}") response.raise_for_status() return response.json() def create_user(self, data: dict) -> dict: """Create new user.""" response = requests.post(f"{self.base_url}/users", json=data) response.raise_for_status() return response.json() def test_get_user_success(): """Test successful API call with mock.""" client = APIClient("https://api.example.com") mock_response = Mock() mock_response.json.return_value = {"id": 1, "name": "John Doe"} mock_response.raise_for_status.return_value = None with patch("requests.get", return_value=mock_response) as mock_get: user = client.get_user(1) assert user["id"] == 1 assert user["name"] == "John Doe" mock_get.assert_called_once_with("https://api.example.com/users/1") def test_get_user_not_found(): """Test API call with 404 error.""" client = APIClient("https://api.example.com") mock_response = Mock() mock_response.raise_for_status.side_effect = requests.HTTPError("404 Not Found") with patch("requests.get", return_value=mock_response): with pytest.raises(requests.HTTPError): client.get_user(999) @patch("requests.post") def test_create_user(mock_post): """Test user creation with decorator syntax.""" client = APIClient("https://api.example.com") mock_post.return_value.json.return_value = {"id": 2, "name": "Jane Doe"} mock_post.return_value.raise_for_status.return_value = None user_data = {"name": "Jane Doe", "email": "jane@example.com"} result = client.create_user(user_data) assert result["id"] == 2 mock_post.assert_called_once() call_args = mock_post.call_args assert call_args.kwargs["json"] == user_data ``` ### Pattern 5: Testing Exceptions ```python # test_exceptions.py import pytest def divide(a: float, b: float) -> float: """Divide a by b.""" if b == 0: raise ZeroDivisionError("Division by zero") if not isinstance(a, (int, float)) or not isinstance(b, (int, float)): raise TypeError("Arguments must be numbers") return a / b def test_zero_division(): """Test exception is raised for division by zero.""" with pytest.raises(ZeroDivisionError): divide(10, 0) def test_zero_division_with_message(): """Test exception message.""" with pytest.raises(ZeroDivisionError, match="Division by zero"): divide(5, 0) def test_type_error(): """Test type error exception.""" with pytest.raises(TypeError, match="must be numbers"): divide("10", 5) def test_exception_info(): """Test accessing exception info.""" with pytest.raises(ValueError) as exc_info: int("not a number") assert "invalid literal" in str(exc_info.value) ``` ## Advanced Patterns ### Pattern 6: Testing Async Code ```python # test_async.py import pytest import asyncio async def fetch_data(url: str) -> dict: """Fetch data asynchronously.""" await asyncio.sleep(0.1) return {"url": url, "data": "result"} @pytest.mark.asyncio async def test_fetch_data(): """Test async function.""" result = await fetch_data("https://api.example.com") assert result["url"] == "https://api.example.com" assert "data" in result @pytest.mark.asyncio async def test_concurrent_fetches(): """Test concurrent async operations.""" urls = ["url1", "url2", "url3"] tasks = [fetch_data(url) for url in urls] results = await asyncio.gather(*tasks) assert len(results) == 3 assert all("data" in r for r in results) @pytest.fixture async def async_client(): """Async fixture.""" client = {"connected": True} yield client client["connected"] = False @pytest.mark.asyncio async def test_with_async_fixture(async_client): """Test using async fixture.""" assert async_client["connected"] is True ``` ### Pattern 7: Monkeypatch for Testing ```python # test_environment.py import os import pytest def get_database_url() -> str: """Get database URL from environment.""" return os.environ.get("DATABASE_URL", "sqlite:///:memory:") def test_database_url_default(): """Test default database URL.""" # Will use actual environment variable if set url = get_database_url() assert url def test_database_url_custom(monkeypatch): """Test custom database URL with monkeypatch.""" monkeypatch.setenv("DATABASE_URL", "postgresql://localhost/test") assert get_database_url() == "postgresql://localhost/test" def test_database_url_not_set(monkeypatch): """Test when env var is not set.""" monkeypatch.delenv("DATABASE_URL", raising=False) assert get_database_url() == "sqlite:///:memory:" class Config: """Configuration class.""" def __init__(self): self.api_key = "production-key" def get_api_key(self): return self.api_key def test_monkeypatch_attribute(monkeypatch): """Test monkeypatching object attributes.""" config = Config() monkeypatch.setattr(config, "api_key", "test-key") assert config.get_api_key() == "test-key" ``` ### Pattern 8: Temporary Files and Directories ```python # test_file_operations.py import pytest from pathlib import Path def save_data(filepath: Path, data: str): """Save data to file.""" filepath.write_text(data) def load_data(filepath: Path) -> str: """Load data from file.""" return filepath.read_text() def test_file_operations(tmp_path): """Test file operations with temporary directory.""" # tmp_path is a pathlib.Path object test_file = tmp_path / "test_data.txt" # Save data save_data(test_file, "Hello, World!") # Verify file exists assert test_file.exists() # Load and verify data data = load_data(test_file) assert data == "Hello, World!" def test_multiple_files(tmp_path): """Test with multiple temporary files.""" files = { "file1.txt": "Content 1", "file2.txt": "Content 2", "file3.txt": "Content 3" } for filename, content in files.items(): filepath = tmp_path / filename save_data(filepath, content) # Verify all files created assert len(list(tmp_path.iterdir())) == 3 # Verify contents for filename, expected_content in files.items(): filepath = tmp_path / filename assert load_data(filepath) == expected_content ``` ### Pattern 9: Custom Fixtures and Conftest ```python # conftest.py """Shared fixtures for all tests.""" import pytest @pytest.fixture(scope="session") def database_url(): """Provide database URL for all tests.""" return "postgresql://localhost/test_db" @pytest.fixture(autouse=True) def reset_database(database_url): """Auto-use fixture that runs before each test.""" # Setup: Clear database print(f"Clearing database: {database_url}") yield # Teardown: Clean up print("Test completed") @pytest.fixture def sample_user(): """Provide sample user data.""" return { "id": 1, "name": "Test User", "email": "test@example.com" } @pytest.fixture def sample_users(): """Provide list of sample users.""" return [ {"id": 1, "name": "User 1"}, {"id": 2, "name": "User 2"}, {"id": 3, "name": "User 3"}, ] # Parametrized fixture @pytest.fixture(params=["sqlite", "postgresql", "mysql"]) def db_backend(request): """Fixture that runs tests with different database backends.""" return request.param def test_with_db_backend(db_backend): """This test will run 3 times with different backends.""" print(f"Testing with {db_backend}") assert db_backend in ["sqlite", "postgresql", "mysql"] ``` ### Pattern 10: Property-Based Testing ```python # test_properties.py from hypothesis import given, strategies as st import pytest def reverse_string(s: str) -> str: """Reverse a string.""" return s[::-1] @given(st.text()) def test_reverse_twice_is_original(s): """Property: reversing twice returns original.""" assert reverse_string(reverse_string(s)) == s @given(st.text()) def test_reverse_length(s): """Property: reversed string has same length.""" assert len(reverse_string(s)) == len(s) @given(st.integers(), st.integers()) def test_addition_commutative(a, b): """Property: addition is commutative.""" assert a + b == b + a @given(st.lists(st.integers())) def test_sorted_list_properties(lst): """Property: sorted list is ordered.""" sorted_lst = sorted(lst) # Same length assert len(sorted_lst) == len(lst) # All elements present assert set(sorted_lst) == set(lst) # Is ordered for i in range(len(sorted_lst) - 1): assert sorted_lst[i] <= sorted_lst[i + 1] ``` ## Test Design Principles ### One Behavior Per Test Each test should verify exactly one behavior. This makes failures easy to diagnose and tests easy to maintain. ```python # BAD - testing multiple behaviors def test_user_service(): user = service.create_user(data) assert user.id is not None assert user.email == data["email"] updated = service.update_user(user.id, {"name": "New"}) assert updated.name == "New" # GOOD - focused tests def test_create_user_assigns_id(): user = service.create_user(data) assert user.id is not None def test_create_user_stores_email(): user = service.create_user(data) assert user.email == data["email"] def test_update_user_changes_name(): user = service.create_user(data) updated = service.update_user(user.id, {"name": "New"}) assert updated.name == "New" ``` ### Test Error Paths Always test failure cases, not just happy paths. ```python def test_get_user_raises_not_found(): with pytest.raises(UserNotFoundError) as exc_info: service.get_user("nonexistent-id") assert "nonexistent-id" in str(exc_info.value) def test_create_user_rejects_invalid_email(): with pytest.raises(ValueError, match="Invalid email format"): service.create_user({"email": "not-an-email"}) ``` ## Testing Best Practices ### Test Organization ```python # tests/ # __init__.py # conftest.py # Shared fixtures # test_unit/ # Unit tests # test_models.py # test_utils.py # test_integration/ # Integration tests # test_api.py # test_database.py # test_e2e/ # End-to-end tests # test_workflows.py ``` ### Test Naming Convention A common pattern: `test_<unit>_<scenario>_<expected_outcome>`. Adapt to your team's preferences. ```python # Pattern: test_<unit>_<scenario>_<expected> def test_create_user_with_valid_data_returns_user(): ... def test_create_user_with_duplicate_email_raises_conflict(): ... def test_get_user_with_unknown_id_returns_none(): ... # Good test names - clear and descriptive def test_user_creation_with_valid_data(): """Clear name describes what is being tested.""" pass def test_login_fails_with_invalid_password(): """Name describes expected behavior.""" pass def test_api_returns_404_for_missing_resource(): """Specific about inputs and expected outcomes.""" pass # Bad test names - avoid these def test_1(): # Not descriptive pass def test_user(): # Too vague pass def test_function(): # Doesn't explain what's tested pass ``` ### Testing Retry Behavior Verify that retry logic works correctly using mock side effects. ```python from unittest.mock import Mock def test_retries_on_transient_error(): """Test that service retries on transient failures.""" client = Mock() # Fail twice, then succeed client.request.side_effect = [ ConnectionError("Failed"), ConnectionError("Failed"), {"status": "ok"}, ] service = ServiceWithRetry(client, max_retries=3) result = service.fetch() assert result == {"status": "ok"} assert client.request.call_count == 3 def test_gives_up_after_max_retries(): """Test that service stops retrying after max attempts.""" client = Mock() client.request.side_effect = ConnectionError("Failed") service = ServiceWithRetry(client, max_retries=3) with pytest.raises(ConnectionError): service.fetch() assert client.request.call_count == 3 def test_does_not_retry_on_permanent_error(): """Test that permanent errors are not retried.""" client = Mock() client.request.side_effect = ValueError("Invalid input") service = ServiceWithRetry(client, max_retries=3) with pytest.raises(ValueError): service.fetch() # Only called once - no retry for ValueError assert client.request.call_count == 1 ``` ### Mocking Time with Freezegun Use freezegun
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

python-type-safety

Python type safety with type hints, generics, protocols, and strict

coding
⭐1
# Python Type Safety Leverage Python's type system to catch errors at static analysis time. Type annotations serve as enforced documentation that tooling validates automatically. ## When to Use This Skill - Adding type hints to existing code - Creating generic, reusable classes - Defining structural interfaces with protocols - Configuring mypy or pyright for strict checking - Understanding type narrowing and guards - Building type-safe APIs and libraries ## Core Concepts ### 1. Type Annotations Declare expected types for function parameters, return values, and variables. ### 2. Generics Write reusable code that preserves type information across different types. ### 3. Protocols Define structural interfaces without inheritance (duck typing with type safety). ### 4. Type Narrowing Use guards and conditionals to narrow types within code blocks. ## Quick Start ```python def get_user(user_id: str) -> User | None: """Return type makes 'might not exist' explicit.""" ... # Type checker enforces handling None case user = get_user("123") if user is None: raise UserNotFoundError("123") print(user.name) # Type checker knows user is User here ``` ## Fundamental Patterns ### Pattern 1: Annotate All Public Signatures Every public function, method, and class should have type annotations. ```python def get_user(user_id: str) -> User: """Retrieve user by ID.""" ... def process_batch( items: list[Item], max_workers: int = 4, ) -> BatchResult[ProcessedItem]: """Process items concurrently.""" ... class UserRepository: def __init__(self, db: Database) -> None: self._db = db async def find_by_id(self, user_id: str) -> User | None: """Return User if found, None otherwise.""" ... async def find_by_email(self, email: str) -> User | None: ... async def save(self, user: User) -> User: """Save and return user with generated ID.""" ... ``` Use `mypy --strict` or `pyright` in CI to catch type errors early. For existing projects, enable strict mode incrementally using per-module overrides. ### Pattern 2: Use Modern Union Syntax Python 3.10+ provides cleaner union syntax. ```python # Preferred (3.10+) def find_user(user_id: str) -> User | None: ... def parse_value(v: str) -> int | float | str: ... # Older style (still valid, needed for 3.9) from typing import Optional, Union def find_user(user_id: str) -> Optional[User]: ... ``` ### Pattern 3: Type Narrowing with Guards Use conditionals to narrow types for the type checker. ```python def process_user(user_id: str) -> UserData: user = find_user(user_id) if user is None: raise UserNotFoundError(f"User {user_id} not found") # Type checker knows user is User here, not User | None return UserData( name=user.name, email=user.email, ) def process_items(items: list[Item | None]) -> list[ProcessedItem]: # Filter and narrow types valid_items = [item for item in items if item is not None] # valid_items is now list[Item] return [process(item) for item in valid_items] ``` ### Pattern 4: Generic Classes Create type-safe reusable containers. ```python from typing import TypeVar, Generic T = TypeVar("T") E = TypeVar("E", bound=Exception) class Result(Generic[T, E]): """Represents either a success value or an error.""" def __init__( self, value: T | None = None, error: E | None = None, ) -> None: if (value is None) == (error is None): raise ValueError("Exactly one of value or error must be set") self._value = value self._error = error @property def is_success(self) -> bool: return self._error is None @property def is_failure(self) -> bool: return self._error is not None def unwrap(self) -> T: """Get value or raise the error.""" if self._error is not None: raise self._error return self._value # type: ignore[return-value] def unwrap_or(self, default: T) -> T: """Get value or return default.""" if self._error is not None: return default return self._value # type: ignore[return-value] # Usage preserves types def parse_config(path: str) -> Result[Config, ConfigError]: try: return Result(value=Config.from_file(path)) except ConfigError as e: return Result(error=e) result = parse_config("config.yaml") if result.is_success: config = result.unwrap() # Type: Config ``` ## Advanced Patterns ### Pattern 5: Generic Repository Create type-safe data access patterns. ```python from typing import TypeVar, Generic from abc import ABC, abstractmethod T = TypeVar("T") ID = TypeVar("ID") class Repository(ABC, Generic[T, ID]): """Generic repository interface.""" @abstractmethod async def get(self, id: ID) -> T | None: """Get entity by ID.""" ... @abstractmethod async def save(self, entity: T) -> T: """Save and return entity.""" ... @abstractmethod async def delete(self, id: ID) -> bool: """Delete entity, return True if existed.""" ... class UserRepository(Repository[User, str]): """Concrete repository for Users with string IDs.""" async def get(self, id: str) -> User | None: row = await self._db.fetchrow( "SELECT * FROM users WHERE id = $1", id ) return User(**row) if row else None async def save(self, entity: User) -> User: ... async def delete(self, id: str) -> bool: ... ``` ### Pattern 6: TypeVar with Bounds Restrict generic parameters to specific types. ```python from typing import TypeVar from pydantic import BaseModel ModelT = TypeVar("ModelT", bound=BaseModel) def validate_and_create(model_cls: type[ModelT], data: dict) -> ModelT: """Create a validated Pydantic model from dict.""" return model_cls.model_validate(data) # Works with any BaseModel subclass class User(BaseModel): name: str email: str user = validate_and_create(User, {"name": "Alice", "email": "a@b.com"}) # user is typed as User # Type error: str is not a BaseModel subclass result = validate_and_create(str, {"name": "Alice"}) # Error! ``` ### Pattern 7: Protocols for Structural Typing Define interfaces without requiring inheritance. ```python from typing import Protocol, runtime_checkable @runtime_checkable class Serializable(Protocol): """Any class that can be serialized to/from dict.""" def to_dict(self) -> dict: ... @classmethod def from_dict(cls, data: dict) -> "Serializable": ... # User satisfies Serializable without inheriting from it class User: def __init__(self, id: str, name: str) -> None: self.id = id self.name = name def to_dict(self) -> dict: return {"id": self.id, "name": self.name} @classmethod def from_dict(cls, data: dict) -> "User": return cls(id=data["id"], name=data["name"]) def serialize(obj: Serializable) -> str: """Works with any Serializable object.""" return json.dumps(obj.to_dict()) # Works - User matches the protocol serialize(User("1", "Alice")) # Runtime checking with @runtime_checkable isinstance(User("1", "Alice"), Serializable) # True ``` ### Pattern 8: Common Protocol Patterns Define reusable structural interfaces. ```python from typing import Protocol class Closeable(Protocol): """Resource that can be closed.""" def close(self) -> None: ... class AsyncCloseable(Protocol): """Async resource that can be closed.""" async def close(self) -> None: ... class Readable(Protocol): """Object that can be read from.""" def read(self, n: int = -1) -> bytes: ... class HasId(Protocol): """Object with an ID property.""" @property def id(self) -> str: ... class Comparable(Protocol): """Object that supports comparison.""" def __lt__(self, other: "Comparable") -> bool: ... def __le__(self, other: "Comparable") -> bool: ... ``` ### Pattern 9: Type Aliases Create meaningful type names. **Note:** The `type` statement was introduced in Python 3.10 for simple aliases. Generic type statements require Python 3.12+. ```python # Python 3.10+ type statement for simple aliases type UserId = str type UserDict = dict[str, Any] # Python 3.12+ type statement with generics type Handler[T] = Callable[[Request], T] type AsyncHandler[T] = Callable[[Request], Awaitable[T]] # Python 3.9-3.11 style (needed for broader compatibility) from typing import TypeAlias from collections.abc import Callable, Awaitable UserId: TypeAlias = str Handler: TypeAlias = Callable[[Request], Response] # Usage def register_handler(path: str, handler: Handler[Response]) -> None: ... ``` ### Pattern 10: Callable Types Type function parameters and callbacks. ```python from collections.abc import Callable, Awaitable # Sync callback ProgressCallback = Callable[[int, int], None] # (current, total) # Async callback AsyncHandler = Callable[[Request], Awaitable[Response]] # With named parameters (using Protocol) class OnProgress(Protocol): def __call__( self, current: int, total: int, *, message: str = "", ) -> None: ... def process_items( items: list[Item], on_progress: ProgressCallback | None = None, ) -> list[Result]: for i, item in enumerate(items): if on_progress: on_progress(i, len(items)) ... ``` ## Configuration ### Strict Mode Checklist For `mypy --strict` compliance: ```toml # pyproject.toml [tool.mypy] python_version = "3.12" strict = true warn_return_any = true warn_unused_ignores = true disallow_untyped_defs = true disallow_incomplete_defs = true no_implicit_optional = true ``` Incremental adoption goals: - All function parameters annotated - All return types annotated - Class attributes annotated - Minimize `Any` usage (acceptable for truly dynamic data) - Generic collections use type parameters (`list[str]` not `list`) For existing codebases, enable strict mode per-module using `# mypy: strict` or configure per-module overrides in `pyproject.toml`. ## Best Practices Summary 1. **Annotate all public APIs** - Functions, methods, class attributes 2. **Use `T | None`** - Modern union syntax over `Optional[T]` 3. **Run strict type checking** - `mypy --strict` in CI 4. **Use generics** - Preserve type info in reusable code 5. **Define protocols** - Structural typing for interfaces 6. **Narrow types** - Use guards to help the type checker 7. **Bound type vars** - Restrict generics to meaningful types 8. **Create type aliases** - Meaningful names for complex types 9. **Minimize `Any`** - Use specific types or generics. `Any` is acceptable for truly dynamic data or when interfacing with untyped third-party code 10. **Document with types** - Types are enforceable documentation
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

bats-testing-patterns

Master Bash Automated Testing System (Bats) for comprehensive shell

coding
⭐1
# Bats Testing Patterns Comprehensive guidance for writing comprehensive unit tests for shell scripts using Bats (Bash Automated Testing System), including test patterns, fixtures, and best practices for production-grade shell testing. ## When to Use This Skill - Writing unit tests for shell scripts - Implementing test-driven development (TDD) for scripts - Setting up automated testing in CI/CD pipelines - Testing edge cases and error conditions - Validating behavior across different shell environments - Building maintainable test suites for scripts - Creating fixtures for complex test scenarios - Testing multiple shell dialects (bash, sh, dash) ## Bats Fundamentals ### What is Bats? Bats (Bash Automated Testing System) is a TAP (Test Anything Protocol) compliant testing framework for shell scripts that provides: - Simple, natural test syntax - TAP output format compatible with CI systems - Fixtures and setup/teardown support - Assertion helpers - Parallel test execution ### Installation ```bash # macOS with Homebrew brew install bats-core # Ubuntu/Debian git clone https://github.com/bats-core/bats-core.git cd bats-core ./install.sh /usr/local # From npm (Node.js) npm install --global bats # Verify installation bats --version ``` ### File Structure ``` project/ β”œβ”€β”€ bin/ β”‚ β”œβ”€β”€ script.sh β”‚ └── helper.sh β”œβ”€β”€ tests/ β”‚ β”œβ”€β”€ test_script.bats β”‚ β”œβ”€β”€ test_helper.sh β”‚ β”œβ”€β”€ fixtures/ β”‚ β”‚ β”œβ”€β”€ input.txt β”‚ β”‚ └── expected_output.txt β”‚ └── helpers/ β”‚ └── mocks.bash └── README.md ``` ## Basic Test Structure ### Simple Test File ```bash #!/usr/bin/env bats # Load test helper if present load test_helper # Setup runs before each test setup() { export TMPDIR=$(mktemp -d) } # Teardown runs after each test teardown() { rm -rf "$TMPDIR" } # Test: simple assertion @test "Function returns 0 on success" { run my_function "input" [ "$status" -eq 0 ] } # Test: output verification @test "Function outputs correct result" { run my_function "test" [ "$output" = "expected output" ] } # Test: error handling @test "Function returns 1 on missing argument" { run my_function [ "$status" -eq 1 ] } ``` ## Assertion Patterns ### Exit Code Assertions ```bash #!/usr/bin/env bats @test "Command succeeds" { run true [ "$status" -eq 0 ] } @test "Command fails as expected" { run false [ "$status" -ne 0 ] } @test "Command returns specific exit code" { run my_function --invalid [ "$status" -eq 127 ] } @test "Can capture command result" { run echo "hello" [ $status -eq 0 ] [ "$output" = "hello" ] } ``` ### Output Assertions ```bash #!/usr/bin/env bats @test "Output matches string" { result=$(echo "hello world") [ "$result" = "hello world" ] } @test "Output contains substring" { result=$(echo "hello world") [[ "$result" == *"world"* ]] } @test "Output matches pattern" { result=$(date +%Y) [[ "$result" =~ ^[0-9]{4}$ ]] } @test "Multi-line output" { run printf "line1\nline2\nline3" [ "$output" = "line1 line2 line3" ] } @test "Lines variable contains output" { run printf "line1\nline2\nline3" [ "${lines[0]}" = "line1" ] [ "${lines[1]}" = "line2" ] [ "${lines[2]}" = "line3" ] } ``` ### File Assertions ```bash #!/usr/bin/env bats @test "File is created" { [ ! -f "$TMPDIR/output.txt" ] my_function > "$TMPDIR/output.txt" [ -f "$TMPDIR/output.txt" ] } @test "File contents match expected" { my_function > "$TMPDIR/output.txt" [ "$(cat "$TMPDIR/output.txt")" = "expected content" ] } @test "File is readable" { touch "$TMPDIR/test.txt" [ -r "$TMPDIR/test.txt" ] } @test "File has correct permissions" { touch "$TMPDIR/test.txt" chmod 644 "$TMPDIR/test.txt" [ "$(stat -f %OLp "$TMPDIR/test.txt")" = "644" ] } @test "File size is correct" { echo -n "12345" > "$TMPDIR/test.txt" [ "$(wc -c < "$TMPDIR/test.txt")" -eq 5 ] } ``` ## Setup and Teardown Patterns ### Basic Setup and Teardown ```bash #!/usr/bin/env bats setup() { # Create test directory TEST_DIR=$(mktemp -d) export TEST_DIR # Source script under test source "${BATS_TEST_DIRNAME}/../bin/script.sh" } teardown() { # Clean up temporary directory rm -rf "$TEST_DIR" } @test "Test using TEST_DIR" { touch "$TEST_DIR/file.txt" [ -f "$TEST_DIR/file.txt" ] } ``` ### Setup with Resources ```bash #!/usr/bin/env bats setup() { # Create directory structure mkdir -p "$TMPDIR/data/input" mkdir -p "$TMPDIR/data/output" # Create test fixtures echo "line1" > "$TMPDIR/data/input/file1.txt" echo "line2" > "$TMPDIR/data/input/file2.txt" # Initialize environment export DATA_DIR="$TMPDIR/data" export INPUT_DIR="$DATA_DIR/input" export OUTPUT_DIR="$DATA_DIR/output" } teardown() { rm -rf "$TMPDIR/data" } @test "Processes input files" { run my_process_script "$INPUT_DIR" "$OUTPUT_DIR" [ "$status" -eq 0 ] [ -f "$OUTPUT_DIR/file1.txt" ] } ``` ### Global Setup/Teardown ```bash #!/usr/bin/env bats # Load shared setup from test_helper.sh load test_helper # setup_file runs once before all tests setup_file() { export SHARED_RESOURCE=$(mktemp -d) echo "Expensive setup" > "$SHARED_RESOURCE/data.txt" } # teardown_file runs once after all tests teardown_file() { rm -rf "$SHARED_RESOURCE" } @test "First test uses shared resource" { [ -f "$SHARED_RESOURCE/data.txt" ] } @test "Second test uses shared resource" { [ -d "$SHARED_RESOURCE" ] } ``` ## Mocking and Stubbing Patterns ### Function Mocking ```bash #!/usr/bin/env bats # Mock external command my_external_tool() { echo "mocked output" return 0 } @test "Function uses mocked tool" { export -f my_external_tool run my_function [[ "$output" == *"mocked output"* ]] } ``` ### Command Stubbing ```bash #!/usr/bin/env bats setup() { # Create stub directory STUBS_DIR="$TMPDIR/stubs" mkdir -p "$STUBS_DIR" # Add to PATH export PATH="$STUBS_DIR:$PATH" } create_stub() { local cmd="$1" local output="$2" local code="${3:-0}" cat > "$STUBS_DIR/$cmd" <<EOF #!/bin/bash echo "$output" exit $code EOF chmod +x "$STUBS_DIR/$cmd" } @test "Function works with stubbed curl" { create_stub curl "{ \"status\": \"ok\" }" 0 run my_api_function [ "$status" -eq 0 ] } ``` ### Variable Stubbing ```bash #!/usr/bin/env bats @test "Function handles environment override" { export MY_SETTING="override_value" run my_function [ "$status" -eq 0 ] [[ "$output" == *"override_value"* ]] } @test "Function uses default when var unset" { unset MY_SETTING run my_function [ "$status" -eq 0 ] [[ "$output" == *"default"* ]] } ``` ## Fixture Management ### Using Fixture Files ```bash #!/usr/bin/env bats # Fixture directory: tests/fixtures/ setup() { FIXTURES_DIR="${BATS_TEST_DIRNAME}/fixtures" WORK_DIR=$(mktemp -d) export WORK_DIR } teardown() { rm -rf "$WORK_DIR" } @test "Process fixture file" { # Copy fixture to work directory cp "$FIXTURES_DIR/input.txt" "$WORK_DIR/input.txt" # Run function run my_process_function "$WORK_DIR/input.txt" # Compare output diff "$WORK_DIR/output.txt" "$FIXTURES_DIR/expected_output.txt" } ``` ### Dynamic Fixture Generation ```bash #!/usr/bin/env bats generate_fixture() { local lines="$1" local file="$2" for i in $(seq 1 "$lines"); do echo "Line $i content" >> "$file" done } @test "Handle large input file" { generate_fixture 1000 "$TMPDIR/large.txt" run my_function "$TMPDIR/large.txt" [ "$status" -eq 0 ] [ "$(wc -l < "$TMPDIR/large.txt")" -eq 1000 ] } ``` ## Advanced Patterns ### Testing Error Conditions ```bash #!/usr/bin/env bats @test "Function fails with missing file" { run my_function "/nonexistent/file.txt" [ "$status" -ne 0 ] [[ "$output" == *"not found"* ]] } @test "Function fails with invalid input" { run my_function "" [ "$status" -ne 0 ] } @test "Function fails with permission denied" { touch "$TMPDIR/readonly.txt" chmod 000 "$TMPDIR/readonly.txt" run my_function "$TMPDIR/readonly.txt" [ "$status" -ne 0 ] chmod 644 "$TMPDIR/readonly.txt" # Cleanup } @test "Function provides helpful error message" { run my_function --invalid-option [ "$status" -ne 0 ] [[ "$output" == *"Usage:"* ]] } ``` ### Testing with Dependencies ```bash #!/usr/bin/env bats setup() { # Check for required tools if ! command -v jq &>/dev/null; then skip "jq is not installed" fi export SCRIPT="${BATS_TEST_DIRNAME}/../bin/script.sh" } @test "JSON parsing works" { skip_if ! command -v jq &>/dev/null run my_json_parser '{"key": "value"}' [ "$status" -eq 0 ] } ``` ### Testing Shell Compatibility ```bash #!/usr/bin/env bats @test "Script works in bash" { bash "${BATS_TEST_DIRNAME}/../bin/script.sh" arg1 } @test "Script works in sh (POSIX)" { sh "${BATS_TEST_DIRNAME}/../bin/script.sh" arg1 } @test "Script works in dash" { if command -v dash &>/dev/null; then dash "${BATS_TEST_DIRNAME}/../bin/script.sh" arg1 else skip "dash not installed" fi } ``` ### Parallel Execution ```bash #!/usr/bin/env bats @test "Multiple independent operations" { run bash -c 'for i in {1..10}; do my_operation "$i" & done wait' [ "$status" -eq 0 ] } @test "Concurrent file operations" { for i in {1..5}; do my_function "$TMPDIR/file$i" & done wait [ -f "$TMPDIR/file1" ] [ -f "$TMPDIR/file5" ] } ``` ## Test Helper Pattern ### test_helper.sh ```bash #!/usr/bin/env bash # Source script under test export SCRIPT_DIR="${BATS_TEST_DIRNAME%/*}/bin" # Common test utilities assert_file_exists() { if [ ! -f "$1" ]; then echo "Expected file to exist: $1" return 1 fi } assert_file_equals() { local file="$1" local expected="$2" if [ ! -f "$file" ]; then echo "File does not exist: $file" return 1 fi local actual=$(cat "$file") if [ "$actual" != "$expected" ]; then echo "File contents do not match" echo "Expected: $expected" echo "Actual: $actual" return 1 fi } # Create temporary test directory setup_test_dir() { export TEST_DIR=$(mktemp -d) } cleanup_test_dir() { rm -rf "$TEST_DIR" } ``` ## Integration with CI/CD ### GitHub Actions Workflow ```yaml name: Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Install Bats run: | npm install --global bats - name: Run Tests run: | bats tests/*.bats - name: Run Tests with Tap Reporter run: | bats tests/*.bats --tap | tee test_output.tap ``` ### Makefile Integration ```makefile .PHONY: test test-verbose test-tap test: bats tests/*.bats test-verbose: bats tests/*.bats --verbose test-tap: bats tests/*.bats --tap test-parallel: bats tests/*.bats --parallel 4 coverage: test # Optional: Generate coverage reports ``` ## Best Practices 1. **Test one thing per test** - Single responsibility principle 2. **Use descriptive test names** - Clearly states what is being tested 3. **Clean up after tests** - Always remove temporary files in teardown 4. **Test both success and failure paths** - Don't just test happy path 5. **Mock external dependencies** - Isolate unit under test 6. **Use fixtures for complex data** - Makes tests more readable 7. **Run tests in CI/CD** - Catch regressions early 8. **Test across shell dialects** - Ensure portability 9. **Keep tests fast** - Run in parallel when possible 10. **Document complex test setup** - Explain unusual patterns ## Resources - **Bats GitHub**: https://github.com/bats-core/bats-core - **Bats Documentation**: https://bats-core.readthedocs.io/ - **TAP Protocol**: https://testanything.org/ - **Test-Driven Development**: https://en.wikipedia.org/wiki/Test-driven_development
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

go-concurrency-patterns

Master Go concurrency with goroutines, channels, sync primitives,

coding
⭐1
# Go Concurrency Patterns Production patterns for Go concurrency including goroutines, channels, synchronization primitives, and context management. ## When to Use This Skill - Building concurrent Go applications - Implementing worker pools and pipelines - Managing goroutine lifecycles - Using channels for communication - Debugging race conditions - Implementing graceful shutdown ## Core Concepts ### 1. Go Concurrency Primitives | Primitive | Purpose | | ----------------- | -------------------------------- | | `goroutine` | Lightweight concurrent execution | | `channel` | Communication between goroutines | | `select` | Multiplex channel operations | | `sync.Mutex` | Mutual exclusion | | `sync.WaitGroup` | Wait for goroutines to complete | | `context.Context` | Cancellation and deadlines | ### 2. Go Concurrency Mantra ``` Don't communicate by sharing memory; share memory by communicating. ``` ## Quick Start ```go package main import ( "context" "fmt" "sync" "time" ) func main() { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() results := make(chan string, 10) var wg sync.WaitGroup // Spawn workers for i := 0; i < 3; i++ { wg.Add(1) go worker(ctx, i, results, &wg) } // Close results when done go func() { wg.Wait() close(results) }() // Collect results for result := range results { fmt.Println(result) } } func worker(ctx context.Context, id int, results chan<- string, wg *sync.WaitGroup) { defer wg.Done() select { case <-ctx.Done(): return case results <- fmt.Sprintf("Worker %d done", id): } } ``` ## Patterns ### Pattern 1: Worker Pool ```go package main import ( "context" "fmt" "sync" ) type Job struct { ID int Data string } type Result struct { JobID int Output string Err error } func WorkerPool(ctx context.Context, numWorkers int, jobs <-chan Job) <-chan Result { results := make(chan Result, len(jobs)) var wg sync.WaitGroup for i := 0; i < numWorkers; i++ { wg.Add(1) go func(workerID int) { defer wg.Done() for job := range jobs { select { case <-ctx.Done(): return default: result := processJob(job) results <- result } } }(i) } go func() { wg.Wait() close(results) }() return results } func processJob(job Job) Result { // Simulate work return Result{ JobID: job.ID, Output: fmt.Sprintf("Processed: %s", job.Data), } } // Usage func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() jobs := make(chan Job, 100) // Send jobs go func() { for i := 0; i < 50; i++ { jobs <- Job{ID: i, Data: fmt.Sprintf("job-%d", i)} } close(jobs) }() // Process with 5 workers results := WorkerPool(ctx, 5, jobs) for result := range results { fmt.Printf("Result: %+v\n", result) } } ``` ### Pattern 2: Fan-Out/Fan-In Pipeline ```go package main import ( "context" "sync" ) // Stage 1: Generate numbers func generate(ctx context.Context, nums ...int) <-chan int { out := make(chan int) go func() { defer close(out) for _, n := range nums { select { case <-ctx.Done(): return case out <- n: } } }() return out } // Stage 2: Square numbers (can run multiple instances) func square(ctx context.Context, in <-chan int) <-chan int { out := make(chan int) go func() { defer close(out) for n := range in { select { case <-ctx.Done(): return case out <- n * n: } } }() return out } // Fan-in: Merge multiple channels into one func merge(ctx context.Context, cs ...<-chan int) <-chan int { var wg sync.WaitGroup out := make(chan int) // Start output goroutine for each input channel output := func(c <-chan int) { defer wg.Done() for n := range c { select { case <-ctx.Done(): return case out <- n: } } } wg.Add(len(cs)) for _, c := range cs { go output(c) } // Close out after all inputs are done go func() { wg.Wait() close(out) }() return out } func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() // Generate input in := generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10) // Fan out to multiple squarers c1 := square(ctx, in) c2 := square(ctx, in) c3 := square(ctx, in) // Fan in results for result := range merge(ctx, c1, c2, c3) { fmt.Println(result) } } ``` ### Pattern 3: Bounded Concurrency with Semaphore ```go package main import ( "context" "fmt" "golang.org/x/sync/semaphore" "sync" ) type RateLimitedWorker struct { sem *semaphore.Weighted } func NewRateLimitedWorker(maxConcurrent int64) *RateLimitedWorker { return &RateLimitedWorker{ sem: semaphore.NewWeighted(maxConcurrent), } } func (w *RateLimitedWorker) Do(ctx context.Context, tasks []func() error) []error { var ( wg sync.WaitGroup mu sync.Mutex errors []error ) for _, task := range tasks { // Acquire semaphore (blocks if at limit) if err := w.sem.Acquire(ctx, 1); err != nil { return []error{err} } wg.Add(1) go func(t func() error) { defer wg.Done() defer w.sem.Release(1) if err := t(); err != nil { mu.Lock() errors = append(errors, err) mu.Unlock() } }(task) } wg.Wait() return errors } // Alternative: Channel-based semaphore type Semaphore chan struct{} func NewSemaphore(n int) Semaphore { return make(chan struct{}, n) } func (s Semaphore) Acquire() { s <- struct{}{} } func (s Semaphore) Release() { <-s } ``` ### Pattern 4: Graceful Shutdown ```go package main import ( "context" "fmt" "os" "os/signal" "sync" "syscall" "time" ) type Server struct { shutdown chan struct{} wg sync.WaitGroup } func NewServer() *Server { return &Server{ shutdown: make(chan struct{}), } } func (s *Server) Start(ctx context.Context) { // Start workers for i := 0; i < 5; i++ { s.wg.Add(1) go s.worker(ctx, i) } } func (s *Server) worker(ctx context.Context, id int) { defer s.wg.Done() defer fmt.Printf("Worker %d stopped\n", id) ticker := time.NewTicker(time.Second) defer ticker.Stop() for { select { case <-ctx.Done(): // Cleanup fmt.Printf("Worker %d cleaning up...\n", id) time.Sleep(500 * time.Millisecond) // Simulated cleanup return case <-ticker.C: fmt.Printf("Worker %d working...\n", id) } } } func (s *Server) Shutdown(timeout time.Duration) { // Signal shutdown close(s.shutdown) // Wait with timeout done := make(chan struct{}) go func() { s.wg.Wait() close(done) }() select { case <-done: fmt.Println("Clean shutdown completed") case <-time.After(timeout): fmt.Println("Shutdown timed out, forcing exit") } } func main() { // Setup signal handling ctx, cancel := context.WithCancel(context.Background()) sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) server := NewServer() server.Start(ctx) // Wait for signal sig := <-sigCh fmt.Printf("\nReceived signal: %v\n", sig) // Cancel context to stop workers cancel() // Wait for graceful shutdown server.Shutdown(5 * time.Second) } ``` ### Pattern 5: Error Group with Cancellation ```go package main import ( "context" "fmt" "golang.org/x/sync/errgroup" "net/http" ) func fetchAllURLs(ctx context.Context, urls []string) ([]string, error) { g, ctx := errgroup.WithContext(ctx) results := make([]string, len(urls)) for i, url := range urls { i, url := i, url // Capture loop variables g.Go(func() error { req, err := http.NewRequestWithContext(ctx, "GET", url, nil) if err != nil { return fmt.Errorf("creating request for %s: %w", url, err) } resp, err := http.DefaultClient.Do(req) if err != nil { return fmt.Errorf("fetching %s: %w", url, err) } defer resp.Body.Close() results[i] = fmt.Sprintf("%s: %d", url, resp.StatusCode) return nil }) } // Wait for all goroutines to complete or one to fail if err := g.Wait(); err != nil { return nil, err // First error cancels all others } return results, nil } // With concurrency limit func fetchWithLimit(ctx context.Context, urls []string, limit int) ([]string, error) { g, ctx := errgroup.WithContext(ctx) g.SetLimit(limit) // Max concurrent goroutines results := make([]string, len(urls)) var mu sync.Mutex for i, url := range urls { i, url := i, url g.Go(func() error { result, err := fetchURL(ctx, url) if err != nil { return err } mu.Lock() results[i] = result mu.Unlock() return nil }) } if err := g.Wait(); err != nil { return nil, err } return results, nil } ``` ### Pattern 6: Concurrent Map with sync.Map ```go package main import ( "sync" ) // For frequent reads, infrequent writes type Cache struct { m sync.Map } func (c *Cache) Get(key string) (interface{}, bool) { return c.m.Load(key) } func (c *Cache) Set(key string, value interface{}) { c.m.Store(key, value) } func (c *Cache) GetOrSet(key string, value interface{}) (interface{}, bool) { return c.m.LoadOrStore(key, value) } func (c *Cache) Delete(key string) { c.m.Delete(key) } // For write-heavy workloads, use sharded map type ShardedMap struct { shards []*shard numShards int } type shard struct { sync.RWMutex data map[string]interface{} } func NewShardedMap(numShards int) *ShardedMap { m := &ShardedMap{ shards: make([]*shard, numShards), numShards: numShards, } for i := range m.shards { m.shards[i] = &shard{data: make(map[string]interface{})} } return m } func (m *ShardedMap) getShard(key string) *shard { // Simple hash h := 0 for _, c := range key { h = 31*h + int(c) } return m.shards[h%m.numShards] } func (m *ShardedMap) Get(key string) (interface{}, bool) { shard := m.getShard(key) shard.RLock() defer shard.RUnlock() v, ok := shard.data[key] return v, ok } func (m *ShardedMap) Set(key string, value interface{}) { shard := m.getShard(key) shard.Lock() defer shard.Unlock() shard.data[key] = value } ``` ### Pattern 7: Select with Timeout and Default ```go func selectPatterns() { ch := make(chan int) // Timeout pattern select { case v := <-ch: fmt.Println("Received:", v) case <-time.After(time.Second): fmt.Println("Timeout!") } // Non-blocking send/receive select { case ch <- 42: fmt.Println("Sent") default: fmt.Println("Channel full, skipping") } // Priority select (check high priority first) highPriority := make(chan int) lowPriority := make(chan int) for { select { case msg := <-highPriority: fmt.Println("High priority:", msg) default: select { case msg := <-highPriority: fmt.Println("High priority:", msg) case msg := <-lowPriority: fmt.Println("Low priority:", msg) } } } } ``` ## Race Detection ```bash # Run tests with race detector go test -race ./... # Build with race detector go build -race . # Run with race detector go run -race main.go ``` ## Best Practices ### Do's - **Use context** - For cancellation and deadlines - **Close channels** - From sender side only - **Use errgroup** - For concurrent operations with errors - **Buffer channels** - When you know the count - **Prefer channels** - Over mutexes when possible ### Don'ts - **Don't leak goroutines** - Always have exit path - **Don't close from receiver** - Causes panic - **Don't use shared memory** - Unless necessary - **Don't ignore context cancellation** - Check ctx.Done() - **Don't use time.Sleep for sync** - Use proper primitives ## Resources - [Go Concurrency Patterns](https://go.dev/blog/pipelines) - [Effective Go - Concurrency](https://go.dev/doc/effective_go#concurrency) - [Go by Example - Goroutines](https://gobyexample.com/goroutines)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

memory-safety-patterns

Implement memory-safe programming with RAII, ownership, smart

coding
⭐1
# Memory Safety Patterns Cross-language patterns for memory-safe programming including RAII, ownership, smart pointers, and resource management. ## When to Use This Skill - Writing memory-safe systems code - Managing resources (files, sockets, memory) - Preventing use-after-free and leaks - Implementing RAII patterns - Choosing between languages for safety - Debugging memory issues ## Core Concepts ### 1. Memory Bug Categories | Bug Type | Description | Prevention | | -------------------- | -------------------------------- | ----------------- | | **Use-after-free** | Access freed memory | Ownership, RAII | | **Double-free** | Free same memory twice | Smart pointers | | **Memory leak** | Never free memory | RAII, GC | | **Buffer overflow** | Write past buffer end | Bounds checking | | **Dangling pointer** | Pointer to freed memory | Lifetime tracking | | **Data race** | Concurrent unsynchronized access | Ownership, Sync | ### 2. Safety Spectrum ``` Manual (C) β†’ Smart Pointers (C++) β†’ Ownership (Rust) β†’ GC (Go, Java) Less safe More safe More control Less control ``` ## Patterns by Language ### Pattern 1: RAII in C++ ```cpp // RAII: Resource Acquisition Is Initialization // Resource lifetime tied to object lifetime #include <memory> #include <fstream> #include <mutex> // File handle with RAII class FileHandle { public: explicit FileHandle(const std::string& path) : file_(path) { if (!file_.is_open()) { throw std::runtime_error("Failed to open file"); } } // Destructor automatically closes file ~FileHandle() = default; // fstream closes in its destructor // Delete copy (prevent double-close) FileHandle(const FileHandle&) = delete; FileHandle& operator=(const FileHandle&) = delete; // Allow move FileHandle(FileHandle&&) = default; FileHandle& operator=(FileHandle&&) = default; void write(const std::string& data) { file_ << data; } private: std::fstream file_; }; // Lock guard (RAII for mutexes) class Database { public: void update(const std::string& key, const std::string& value) { std::lock_guard<std::mutex> lock(mutex_); // Released on scope exit data_[key] = value; } std::string get(const std::string& key) { std::shared_lock<std::shared_mutex> lock(shared_mutex_); return data_[key]; } private: std::mutex mutex_; std::shared_mutex shared_mutex_; std::map<std::string, std::string> data_; }; // Transaction with rollback (RAII) template<typename T> class Transaction { public: explicit Transaction(T& target) : target_(target), backup_(target), committed_(false) {} ~Transaction() { if (!committed_) { target_ = backup_; // Rollback } } void commit() { committed_ = true; } T& get() { return target_; } private: T& target_; T backup_; bool committed_; }; ``` ### Pattern 2: Smart Pointers in C++ ```cpp #include <memory> // unique_ptr: Single ownership class Engine { public: void start() { /* ... */ } }; class Car { public: Car() : engine_(std::make_unique<Engine>()) {} void start() { engine_->start(); } // Transfer ownership std::unique_ptr<Engine> extractEngine() { return std::move(engine_); } private: std::unique_ptr<Engine> engine_; }; // shared_ptr: Shared ownership class Node { public: std::string data; std::shared_ptr<Node> next; // Use weak_ptr to break cycles std::weak_ptr<Node> parent; }; void sharedPtrExample() { auto node1 = std::make_shared<Node>(); auto node2 = std::make_shared<Node>(); node1->next = node2; node2->parent = node1; // Weak reference prevents cycle // Access weak_ptr if (auto parent = node2->parent.lock()) { // parent is valid shared_ptr } } // Custom deleter for resources class Socket { public: static void close(int* fd) { if (fd && *fd >= 0) { ::close(*fd); delete fd; } } }; auto createSocket() { int fd = socket(AF_INET, SOCK_STREAM, 0); return std::unique_ptr<int, decltype(&Socket::close)>( new int(fd), &Socket::close ); } // make_unique/make_shared best practices void bestPractices() { // Good: Exception safe, single allocation auto ptr = std::make_shared<Widget>(); // Bad: Two allocations, not exception safe std::shared_ptr<Widget> ptr2(new Widget()); // For arrays auto arr = std::make_unique<int[]>(10); } ``` ### Pattern 3: Ownership in Rust ```rust // Move semantics (default) fn move_example() { let s1 = String::from("hello"); let s2 = s1; // s1 is MOVED, no longer valid // println!("{}", s1); // Compile error! println!("{}", s2); } // Borrowing (references) fn borrow_example() { let s = String::from("hello"); // Immutable borrow (multiple allowed) let len = calculate_length(&s); println!("{} has length {}", s, len); // Mutable borrow (only one allowed) let mut s = String::from("hello"); change(&mut s); } fn calculate_length(s: &String) -> usize { s.len() } // s goes out of scope, but doesn't drop since borrowed fn change(s: &mut String) { s.push_str(", world"); } // Lifetimes: Compiler tracks reference validity fn longest<'a>(x: &'a str, y: &'a str) -> &'a str { if x.len() > y.len() { x } else { y } } // Struct with references needs lifetime annotation struct ImportantExcerpt<'a> { part: &'a str, } impl<'a> ImportantExcerpt<'a> { fn level(&self) -> i32 { 3 } // Lifetime elision: compiler infers 'a for &self fn announce_and_return_part(&self, announcement: &str) -> &str { println!("Attention: {}", announcement); self.part } } // Interior mutability use std::cell::{Cell, RefCell}; use std::rc::Rc; struct Stats { count: Cell<i32>, // Copy types data: RefCell<Vec<String>>, // Non-Copy types } impl Stats { fn increment(&self) { self.count.set(self.count.get() + 1); } fn add_data(&self, item: String) { self.data.borrow_mut().push(item); } } // Rc for shared ownership (single-threaded) fn rc_example() { let data = Rc::new(vec![1, 2, 3]); let data2 = Rc::clone(&data); // Increment reference count println!("Count: {}", Rc::strong_count(&data)); // 2 } // Arc for shared ownership (thread-safe) use std::sync::Arc; use std::thread; fn arc_example() { let data = Arc::new(vec![1, 2, 3]); let handles: Vec<_> = (0..3) .map(|_| { let data = Arc::clone(&data); thread::spawn(move || { println!("{:?}", data); }) }) .collect(); for handle in handles { handle.join().unwrap(); } } ``` ### Pattern 4: Safe Resource Management in C ```c // C doesn't have RAII, but we can use patterns #include <stdlib.h> #include <stdio.h> // Pattern: goto cleanup int process_file(const char* path) { FILE* file = NULL; char* buffer = NULL; int result = -1; file = fopen(path, "r"); if (!file) { goto cleanup; } buffer = malloc(1024); if (!buffer) { goto cleanup; } // Process file... result = 0; cleanup: if (buffer) free(buffer); if (file) fclose(file); return result; } // Pattern: Opaque pointer with create/destroy typedef struct Context Context; Context* context_create(void); void context_destroy(Context* ctx); int context_process(Context* ctx, const char* data); // Implementation struct Context { int* data; size_t size; FILE* log; }; Context* context_create(void) { Context* ctx = calloc(1, sizeof(Context)); if (!ctx) return NULL; ctx->data = malloc(100 * sizeof(int)); if (!ctx->data) { free(ctx); return NULL; } ctx->log = fopen("log.txt", "w"); if (!ctx->log) { free(ctx->data); free(ctx); return NULL; } return ctx; } void context_destroy(Context* ctx) { if (ctx) { if (ctx->log) fclose(ctx->log); if (ctx->data) free(ctx->data); free(ctx); } } // Pattern: Cleanup attribute (GCC/Clang extension) #define AUTO_FREE __attribute__((cleanup(auto_free_func))) void auto_free_func(void** ptr) { free(*ptr); } void auto_free_example(void) { AUTO_FREE char* buffer = malloc(1024); // buffer automatically freed at end of scope } ``` ### Pattern 5: Bounds Checking ```cpp // C++: Use containers instead of raw arrays #include <vector> #include <array> #include <span> void safe_array_access() { std::vector<int> vec = {1, 2, 3, 4, 5}; // Safe: throws std::out_of_range try { int val = vec.at(10); } catch (const std::out_of_range& e) { // Handle error } // Unsafe but faster (no bounds check) int val = vec[2]; // Modern C++20: std::span for array views std::span<int> view(vec); // Iterators are bounds-safe for (int& x : view) { x *= 2; } } // Fixed-size arrays void fixed_array() { std::array<int, 5> arr = {1, 2, 3, 4, 5}; // Compile-time size known static_assert(arr.size() == 5); // Safe access int val = arr.at(2); } ``` ```rust // Rust: Bounds checking by default fn rust_bounds_checking() { let vec = vec![1, 2, 3, 4, 5]; // Runtime bounds check (panics if out of bounds) let val = vec[2]; // Explicit option (no panic) match vec.get(10) { Some(val) => println!("Got {}", val), None => println!("Index out of bounds"), } // Iterators (no bounds checking needed) for val in &vec { println!("{}", val); } // Slices are bounds-checked let slice = &vec[1..3]; // [2, 3] } ``` ### Pattern 6: Preventing Data Races ```cpp // C++: Thread-safe shared state #include <mutex> #include <shared_mutex> #include <atomic> class ThreadSafeCounter { public: void increment() { // Atomic operations count_.fetch_add(1, std::memory_order_relaxed); } int get() const { return count_.load(std::memory_order_relaxed); } private: std::atomic<int> count_{0}; }; class ThreadSafeMap { public: void write(const std::string& key, int value) { std::unique_lock lock(mutex_); data_[key] = value; } std::optional<int> read(const std::string& key) { std::shared_lock lock(mutex_); auto it = data_.find(key); if (it != data_.end()) { return it->second; } return std::nullopt; } private: mutable std::shared_mutex mutex_; std::map<std::string, int> data_; }; ``` ```rust // Rust: Data race prevention at compile time use std::sync::{Arc, Mutex, RwLock}; use std::sync::atomic::{AtomicI32, Ordering}; use std::thread; // Atomic for simple types fn atomic_example() { let counter = Arc::new(AtomicI32::new(0)); let handles: Vec<_> = (0..10) .map(|_| { let counter = Arc::clone(&counter); thread::spawn(move || { counter.fetch_add(1, Ordering::SeqCst); }) }) .collect(); for handle in handles { handle.join().unwrap(); } println!("Counter: {}", counter.load(Ordering::SeqCst)); } // Mutex for complex types fn mutex_example() { let data = Arc::new(Mutex::new(vec![])); let handles: Vec<_> = (0..10) .map(|i| { let data = Arc::clone(&data); thread::spawn(move || { let mut vec = data.lock().unwrap(); vec.push(i); }) }) .collect(); for handle in handles { handle.join().unwrap(); } } // RwLock for read-heavy workloads fn rwlock_example() { let data = Arc::new(RwLock::new(HashMap::new())); // Multiple readers OK let read_guard = data.read().unwrap(); // Writer blocks readers let write_guard = data.write().unwrap(); } ``` ## Best Practices ### Do's - **Prefer RAII** - Tie resource lifetime to scope - **Use smart pointers** - Avoid raw pointers in C++ - **Understand ownership** - Know who owns what - **Check bounds** - Use safe access methods - **Use tools** - AddressSanitizer, Valgrind, Miri ### Don'ts - **Don't use raw pointers** - Unless interfacing with C - **Don't return local references** - Dangling pointer - **Don't ignore compiler warnings** - They catch bugs - **Don't use `unsafe` carelessly** - In Rust, minimize it - **Don't assume thread safety** - Be explicit ## Debugging Tools ```bash # AddressSanitizer (Clang/GCC) clang++ -fsanitize=address -g source.cpp # Valgrind valgrind --leak-check=full ./program # Rust Miri (undefined behavior detector) cargo +nightly miri run # ThreadSanitizer clang++ -fsanitize=thread -g source.cpp ``` ## Resources - [C++ Core Guidelines](https://isocpp.github.io/CppCoreGuidelines/) - [Rust Ownership](https://doc.rust-lang.org/book/ch04-00-understanding-ownership.html) - [AddressSanitizer](https://clang.llvm.org/docs/AddressSanitizer.html)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered
πŸ€–system promptβ€’7 months ago

rust-async-patterns

Master Rust async programming with Tokio, async traits, error

coding
⭐1
# Rust Async Patterns Production patterns for async Rust programming with Tokio runtime, including tasks, channels, streams, and error handling. ## When to Use This Skill - Building async Rust applications - Implementing concurrent network services - Using Tokio for async I/O - Handling async errors properly - Debugging async code issues - Optimizing async performance ## Core Concepts ### 1. Async Execution Model ``` Future (lazy) β†’ poll() β†’ Ready(value) | Pending ↑ ↓ Waker ← Runtime schedules ``` ### 2. Key Abstractions | Concept | Purpose | | ---------- | ---------------------------------------- | | `Future` | Lazy computation that may complete later | | `async fn` | Function returning impl Future | | `await` | Suspend until future completes | | `Task` | Spawned future running concurrently | | `Runtime` | Executor that polls futures | ## Quick Start ```toml # Cargo.toml [dependencies] tokio = { version = "1", features = ["full"] } futures = "0.3" async-trait = "0.1" anyhow = "1.0" tracing = "0.1" tracing-subscriber = "0.3" ``` ```rust use tokio::time::{sleep, Duration}; use anyhow::Result; #[tokio::main] async fn main() -> Result<()> { // Initialize tracing tracing_subscriber::fmt::init(); // Async operations let result = fetch_data("https://api.example.com").await?; println!("Got: {}", result); Ok(()) } async fn fetch_data(url: &str) -> Result<String> { // Simulated async operation sleep(Duration::from_millis(100)).await; Ok(format!("Data from {}", url)) } ``` ## Patterns ### Pattern 1: Concurrent Task Execution ```rust use tokio::task::JoinSet; use anyhow::Result; // Spawn multiple concurrent tasks async fn fetch_all_concurrent(urls: Vec<String>) -> Result<Vec<String>> { let mut set = JoinSet::new(); for url in urls { set.spawn(async move { fetch_data(&url).await }); } let mut results = Vec::new(); while let Some(res) = set.join_next().await { match res { Ok(Ok(data)) => results.push(data), Ok(Err(e)) => tracing::error!("Task failed: {}", e), Err(e) => tracing::error!("Join error: {}", e), } } Ok(results) } // With concurrency limit use futures::stream::{self, StreamExt}; async fn fetch_with_limit(urls: Vec<String>, limit: usize) -> Vec<Result<String>> { stream::iter(urls) .map(|url| async move { fetch_data(&url).await }) .buffer_unordered(limit) // Max concurrent tasks .collect() .await } // Select first to complete use tokio::select; async fn race_requests(url1: &str, url2: &str) -> Result<String> { select! { result = fetch_data(url1) => result, result = fetch_data(url2) => result, } } ``` ### Pattern 2: Channels for Communication ```rust use tokio::sync::{mpsc, broadcast, oneshot, watch}; // Multi-producer, single-consumer async fn mpsc_example() { let (tx, mut rx) = mpsc::channel::<String>(100); // Spawn producer let tx2 = tx.clone(); tokio::spawn(async move { tx2.send("Hello".to_string()).await.unwrap(); }); // Consume while let Some(msg) = rx.recv().await { println!("Got: {}", msg); } } // Broadcast: multi-producer, multi-consumer async fn broadcast_example() { let (tx, _) = broadcast::channel::<String>(100); let mut rx1 = tx.subscribe(); let mut rx2 = tx.subscribe(); tx.send("Event".to_string()).unwrap(); // Both receivers get the message let _ = rx1.recv().await; let _ = rx2.recv().await; } // Oneshot: single value, single use async fn oneshot_example() -> String { let (tx, rx) = oneshot::channel::<String>(); tokio::spawn(async move { tx.send("Result".to_string()).unwrap(); }); rx.await.unwrap() } // Watch: single producer, multi-consumer, latest value async fn watch_example() { let (tx, mut rx) = watch::channel("initial".to_string()); tokio::spawn(async move { loop { // Wait for changes rx.changed().await.unwrap(); println!("New value: {}", *rx.borrow()); } }); tx.send("updated".to_string()).unwrap(); } ``` ### Pattern 3: Async Error Handling ```rust use anyhow::{Context, Result, bail}; use thiserror::Error; #[derive(Error, Debug)] pub enum ServiceError { #[error("Network error: {0}")] Network(#[from] reqwest::Error), #[error("Database error: {0}")] Database(#[from] sqlx::Error), #[error("Not found: {0}")] NotFound(String), #[error("Timeout after {0:?}")] Timeout(std::time::Duration), } // Using anyhow for application errors async fn process_request(id: &str) -> Result<Response> { let data = fetch_data(id) .await .context("Failed to fetch data")?; let parsed = parse_response(&data) .context("Failed to parse response")?; Ok(parsed) } // Using custom errors for library code async fn get_user(id: &str) -> Result<User, ServiceError> { let result = db.query(id).await?; match result { Some(user) => Ok(user), None => Err(ServiceError::NotFound(id.to_string())), } } // Timeout wrapper use tokio::time::timeout; async fn with_timeout<T, F>(duration: Duration, future: F) -> Result<T, ServiceError> where F: std::future::Future<Output = Result<T, ServiceError>>, { timeout(duration, future) .await .map_err(|_| ServiceError::Timeout(duration))? } ``` ### Pattern 4: Graceful Shutdown ```rust use tokio::signal; use tokio::sync::broadcast; use tokio_util::sync::CancellationToken; async fn run_server() -> Result<()> { // Method 1: CancellationToken let token = CancellationToken::new(); let token_clone = token.clone(); // Spawn task that respects cancellation tokio::spawn(async move { loop { tokio::select! { _ = token_clone.cancelled() => { tracing::info!("Task shutting down"); break; } _ = do_work() => {} } } }); // Wait for shutdown signal signal::ctrl_c().await?; tracing::info!("Shutdown signal received"); // Cancel all tasks token.cancel(); // Give tasks time to cleanup tokio::time::sleep(Duration::from_secs(5)).await; Ok(()) } // Method 2: Broadcast channel for shutdown async fn run_with_broadcast() -> Result<()> { let (shutdown_tx, _) = broadcast::channel::<()>(1); let mut rx = shutdown_tx.subscribe(); tokio::spawn(async move { tokio::select! { _ = rx.recv() => { tracing::info!("Received shutdown"); } _ = async { loop { do_work().await } } => {} } }); signal::ctrl_c().await?; let _ = shutdown_tx.send(()); Ok(()) } ``` ### Pattern 5: Async Traits ```rust use async_trait::async_trait; #[async_trait] pub trait Repository { async fn get(&self, id: &str) -> Result<Entity>; async fn save(&self, entity: &Entity) -> Result<()>; async fn delete(&self, id: &str) -> Result<()>; } pub struct PostgresRepository { pool: sqlx::PgPool, } #[async_trait] impl Repository for PostgresRepository { async fn get(&self, id: &str) -> Result<Entity> { sqlx::query_as!(Entity, "SELECT * FROM entities WHERE id = $1", id) .fetch_one(&self.pool) .await .map_err(Into::into) } async fn save(&self, entity: &Entity) -> Result<()> { sqlx::query!( "INSERT INTO entities (id, data) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET data = $2", entity.id, entity.data ) .execute(&self.pool) .await?; Ok(()) } async fn delete(&self, id: &str) -> Result<()> { sqlx::query!("DELETE FROM entities WHERE id = $1", id) .execute(&self.pool) .await?; Ok(()) } } // Trait object usage async fn process(repo: &dyn Repository, id: &str) -> Result<()> { let entity = repo.get(id).await?; // Process... repo.save(&entity).await } ``` ### Pattern 6: Streams and Async Iteration ```rust use futures::stream::{self, Stream, StreamExt}; use async_stream::stream; // Create stream from async iterator fn numbers_stream() -> impl Stream<Item = i32> { stream! { for i in 0..10 { tokio::time::sleep(Duration::from_millis(100)).await; yield i; } } } // Process stream async fn process_stream() { let stream = numbers_stream(); // Map and filter let processed: Vec<_> = stream .filter(|n| futures::future::ready(*n % 2 == 0)) .map(|n| n * 2) .collect() .await; println!("{:?}", processed); } // Chunked processing async fn process_in_chunks() { let stream = numbers_stream(); let mut chunks = stream.chunks(3); while let Some(chunk) = chunks.next().await { println!("Processing chunk: {:?}", chunk); } } // Merge multiple streams async fn merge_streams() { let stream1 = numbers_stream(); let stream2 = numbers_stream(); let merged = stream::select(stream1, stream2); merged .for_each(|n| async move { println!("Got: {}", n); }) .await; } ``` ### Pattern 7: Resource Management ```rust use std::sync::Arc; use tokio::sync::{Mutex, RwLock, Semaphore}; // Shared state with RwLock (prefer for read-heavy) struct Cache { data: RwLock<HashMap<String, String>>, } impl Cache { async fn get(&self, key: &str) -> Option<String> { self.data.read().await.get(key).cloned() } async fn set(&self, key: String, value: String) { self.data.write().await.insert(key, value); } } // Connection pool with semaphore struct Pool { semaphore: Semaphore, connections: Mutex<Vec<Connection>>, } impl Pool { fn new(size: usize) -> Self { Self { semaphore: Semaphore::new(size), connections: Mutex::new((0..size).map(|_| Connection::new()).collect()), } } async fn acquire(&self) -> PooledConnection<'_> { let permit = self.semaphore.acquire().await.unwrap(); let conn = self.connections.lock().await.pop().unwrap(); PooledConnection { pool: self, conn: Some(conn), _permit: permit } } } struct PooledConnection<'a> { pool: &'a Pool, conn: Option<Connection>, _permit: tokio::sync::SemaphorePermit<'a>, } impl Drop for PooledConnection<'_> { fn drop(&mut self) { if let Some(conn) = self.conn.take() { let pool = self.pool; tokio::spawn(async move { pool.connections.lock().await.push(conn); }); } } } ``` ## Debugging Tips ```rust // Enable tokio-console for runtime debugging // Cargo.toml: tokio = { features = ["tracing"] } // Run: RUSTFLAGS="--cfg tokio_unstable" cargo run // Then: tokio-console // Instrument async functions use tracing::instrument; #[instrument(skip(pool))] async fn fetch_user(pool: &PgPool, id: &str) -> Result<User> { tracing::debug!("Fetching user"); // ... } // Track task spawning let span = tracing::info_span!("worker", id = %worker_id); tokio::spawn(async move { // Enters span when polled }.instrument(span)); ``` ## Best Practices ### Do's - **Use `tokio::select!`** - For racing futures - **Prefer channels** - Over shared state when possible - **Use `JoinSet`** - For managing multiple tasks - **Instrument with tracing** - For debugging async code - **Handle cancellation** - Check `CancellationToken` ### Don'ts - **Don't block** - Never use `std::thread::sleep` in async - **Don't hold locks across awaits** - Causes deadlocks - **Don't spawn unboundedly** - Use semaphores for limits - **Don't ignore errors** - Propagate with `?` or log - **Don't forget Send bounds** - For spawned futures ## Resources - [Tokio Tutorial](https://tokio.rs/tokio/tutorial) - [Async Book](https://rust-lang.github.io/async-book/) - [Tokio Console](https://github.com/tokio-rs/console)
πŸ‘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

event-store-design

Design and implement event stores for event-sourced systems. Use

coding
⭐1
# Event Store Design Comprehensive guide to designing event stores for event-sourced applications. ## When to Use This Skill - Designing event sourcing infrastructure - Choosing between event store technologies - Implementing custom event stores - Optimizing event storage and retrieval - Setting up event store schemas - Planning for event store scaling ## Core Concepts ### 1. Event Store Architecture ``` β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Event Store β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ β”‚ β”‚ Stream 1 β”‚ β”‚ Stream 2 β”‚ β”‚ Stream 3 β”‚ β”‚ β”‚ β”‚ (Aggregate) β”‚ β”‚ (Aggregate) β”‚ β”‚ (Aggregate) β”‚ β”‚ β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ β”‚ β”‚ Event 1 β”‚ β”‚ Event 1 β”‚ β”‚ Event 1 β”‚ β”‚ β”‚ β”‚ Event 2 β”‚ β”‚ Event 2 β”‚ β”‚ Event 2 β”‚ β”‚ β”‚ β”‚ Event 3 β”‚ β”‚ ... β”‚ β”‚ Event 3 β”‚ β”‚ β”‚ β”‚ ... β”‚ β”‚ β”‚ β”‚ Event 4 β”‚ β”‚ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β”‚ β”œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€ β”‚ Global Position: 1 β†’ 2 β†’ 3 β†’ 4 β†’ 5 β†’ 6 β†’ ... β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` ### 2. Event Store Requirements | Requirement | Description | | ----------------- | ---------------------------------- | | **Append-only** | Events are immutable, only appends | | **Ordered** | Per-stream and global ordering | | **Versioned** | Optimistic concurrency control | | **Subscriptions** | Real-time event notifications | | **Idempotent** | Handle duplicate writes safely | ## Technology Comparison | Technology | Best For | Limitations | | ---------------- | ------------------------- | -------------------------------- | | **EventStoreDB** | Pure event sourcing | Single-purpose | | **PostgreSQL** | Existing Postgres stack | Manual implementation | | **Kafka** | High-throughput streaming | Not ideal for per-stream queries | | **DynamoDB** | Serverless, AWS-native | Query limitations | | **Marten** | .NET ecosystems | .NET specific | ## Templates ### Template 1: PostgreSQL Event Store Schema ```sql -- Events table CREATE TABLE events ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), stream_id VARCHAR(255) NOT NULL, stream_type VARCHAR(255) NOT NULL, event_type VARCHAR(255) NOT NULL, event_data JSONB NOT NULL, metadata JSONB DEFAULT '{}', version BIGINT NOT NULL, global_position BIGSERIAL, created_at TIMESTAMPTZ DEFAULT NOW(), CONSTRAINT unique_stream_version UNIQUE (stream_id, version) ); -- Index for stream queries CREATE INDEX idx_events_stream_id ON events(stream_id, version); -- Index for global subscription CREATE INDEX idx_events_global_position ON events(global_position); -- Index for event type queries CREATE INDEX idx_events_event_type ON events(event_type); -- Index for time-based queries CREATE INDEX idx_events_created_at ON events(created_at); -- Snapshots table CREATE TABLE snapshots ( stream_id VARCHAR(255) PRIMARY KEY, stream_type VARCHAR(255) NOT NULL, snapshot_data JSONB NOT NULL, version BIGINT NOT NULL, created_at TIMESTAMPTZ DEFAULT NOW() ); -- Subscriptions checkpoint table CREATE TABLE subscription_checkpoints ( subscription_id VARCHAR(255) PRIMARY KEY, last_position BIGINT NOT NULL DEFAULT 0, updated_at TIMESTAMPTZ DEFAULT NOW() ); ``` ### Template 2: Python Event Store Implementation ```python from dataclasses import dataclass, field from datetime import datetime from typing import Any, Optional, List from uuid import UUID, uuid4 import json import asyncpg @dataclass class Event: stream_id: str event_type: str data: dict metadata: dict = field(default_factory=dict) event_id: UUID = field(default_factory=uuid4) version: Optional[int] = None global_position: Optional[int] = None created_at: datetime = field(default_factory=datetime.utcnow) class EventStore: def __init__(self, pool: asyncpg.Pool): self.pool = pool async def append_events( self, stream_id: str, stream_type: str, events: List[Event], expected_version: Optional[int] = None ) -> List[Event]: """Append events to a stream with optimistic concurrency.""" async with self.pool.acquire() as conn: async with conn.transaction(): # Check expected version if expected_version is not None: current = await conn.fetchval( "SELECT MAX(version) FROM events WHERE stream_id = $1", stream_id ) current = current or 0 if current != expected_version: raise ConcurrencyError( f"Expected version {expected_version}, got {current}" ) # Get starting version start_version = await conn.fetchval( "SELECT COALESCE(MAX(version), 0) + 1 FROM events WHERE stream_id = $1", stream_id ) # Insert events saved_events = [] for i, event in enumerate(events): event.version = start_version + i row = await conn.fetchrow( """ INSERT INTO events (id, stream_id, stream_type, event_type, event_data, metadata, version, created_at) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING global_position """, event.event_id, stream_id, stream_type, event.event_type, json.dumps(event.data), json.dumps(event.metadata), event.version, event.created_at ) event.global_position = row['global_position'] saved_events.append(event) return saved_events async def read_stream( self, stream_id: str, from_version: int = 0, limit: int = 1000 ) -> List[Event]: """Read events from a stream.""" async with self.pool.acquire() as conn: rows = await conn.fetch( """ SELECT id, stream_id, event_type, event_data, metadata, version, global_position, created_at FROM events WHERE stream_id = $1 AND version >= $2 ORDER BY version LIMIT $3 """, stream_id, from_version, limit ) return [self._row_to_event(row) for row in rows] async def read_all( self, from_position: int = 0, limit: int = 1000 ) -> List[Event]: """Read all events globally.""" async with self.pool.acquire() as conn: rows = await conn.fetch( """ SELECT id, stream_id, event_type, event_data, metadata, version, global_position, created_at FROM events WHERE global_position > $1 ORDER BY global_position LIMIT $2 """, from_position, limit ) return [self._row_to_event(row) for row in rows] async def subscribe( self, subscription_id: str, handler, from_position: int = 0, batch_size: int = 100 ): """Subscribe to all events from a position.""" # Get checkpoint async with self.pool.acquire() as conn: checkpoint = await conn.fetchval( """ SELECT last_position FROM subscription_checkpoints WHERE subscription_id = $1 """, subscription_id ) position = checkpoint or from_position while True: events = await self.read_all(position, batch_size) if not events: await asyncio.sleep(1) # Poll interval continue for event in events: await handler(event) position = event.global_position # Save checkpoint async with self.pool.acquire() as conn: await conn.execute( """ INSERT INTO subscription_checkpoints (subscription_id, last_position) VALUES ($1, $2) ON CONFLICT (subscription_id) DO UPDATE SET last_position = $2, updated_at = NOW() """, subscription_id, position ) def _row_to_event(self, row) -> Event: return Event( event_id=row['id'], stream_id=row['stream_id'], event_type=row['event_type'], data=json.loads(row['event_data']), metadata=json.loads(row['metadata']), version=row['version'], global_position=row['global_position'], created_at=row['created_at'] ) class ConcurrencyError(Exception): """Raised when optimistic concurrency check fails.""" pass ``` ### Template 3: EventStoreDB Usage ```python from esdbclient import EventStoreDBClient, NewEvent, StreamState import json # Connect client = EventStoreDBClient(uri="esdb://localhost:2113?tls=false") # Append events def append_events(stream_name: str, events: list, expected_revision=None): new_events = [ NewEvent( type=event['type'], data=json.dumps(event['data']).encode(), metadata=json.dumps(event.get('metadata', {})).encode() ) for event in events ] if expected_revision is None: state = StreamState.ANY elif expected_revision == -1: state = StreamState.NO_STREAM else: state = expected_revision return client.append_to_stream( stream_name=stream_name, events=new_events, current_version=state ) # Read stream def read_stream(stream_name: str, from_revision: int = 0): events = client.get_stream( stream_name=stream_name, stream_position=from_revision ) return [ { 'type': event.type, 'data': json.loads(event.data), 'metadata': json.loads(event.metadata) if event.metadata else {}, 'stream_position': event.stream_position, 'commit_position': event.commit_position } for event in events ] # Subscribe to all async def subscribe_to_all(handler, from_position: int = 0): subscription = client.subscribe_to_all(commit_position=from_position) async for event in subscription: await handler({ 'type': event.type, 'data': json.loads(event.data), 'stream_id': event.stream_name, 'position': event.commit_position }) # Category projection ($ce-Category) def read_category(category: str): """Read all events for a category using system projection.""" return read_stream(f"$ce-{category}") ``` ### Template 4: DynamoDB Event Store ```python import boto3 from boto3.dynamodb.conditions import Key from datetime import datetime import json import uuid class DynamoEventStore: def __init__(self, table_name: str): self.dynamodb = boto3.resource('dynamodb') self.table = self.dynamodb.Table(table_name) def append_events(self, stream_id: str, events: list, expected_version: int = None): """Append events with conditional write for concurrency.""" with self.table.batch_writer() as batch: for i, event in enumerate(events): version = (expected_version or 0) + i + 1 item = { 'PK': f"STREAM#{stream_id}", 'SK': f"VERSION#{version:020d}", 'GSI1PK': 'EVENTS', 'GSI1SK': datetime.utcnow().isoformat(), 'event_id': str(uuid.uuid4()), 'stream_id': stream_id, 'event_type': event['type'], 'event_data': json.dumps(event['data']), 'version': version, 'created_at': datetime.utcnow().isoformat() } batch.put_item(Item=item) return events def read_stream(self, stream_id: str, from_version: int = 0): """Read events from a stream.""" response = self.table.query( KeyConditionExpression=Key('PK').eq(f"STREAM#{stream_id}") & Key('SK').gte(f"VERSION#{from_version:020d}") ) return [ { 'event_type': item['event_type'], 'data': json.loads(item['event_data']), 'version': item['version'] } for item in response['Items'] ] # Table definition (CloudFormation/Terraform) """ DynamoDB Table: - PK (Partition Key): String - SK (Sort Key): String - GSI1PK, GSI1SK for global ordering Capacity: On-demand or provisioned based on throughput needs """ ``` ## Best Practices ### Do's - **Use stream IDs that include aggregate type** - `Order-{uuid}` - **Include correlation/causation IDs** - For tracing - **Version events from day one** - Plan for schema evolution - **Implement idempotency** - Use event IDs for deduplication - **Index appropriately** - For your query patterns ### Don'ts - **Don't update or delete events** - They're immutable facts - **Don't store large payloads** - Keep events small - **Don't skip optimistic concurrency** - Prevents data corruption - **Don't ignore backpressure** - Handle slow consumers ## Resources - [EventStoreDB](https://www.eventstore.com/) - [Marten Events](https://martendb.io/events/) - [Event Sourcing Pattern](https://docs.microsoft.com/en-us/azure/architecture/patterns/event-sourcing)
πŸ‘0
πŸ‘οΈ0
πŸ€– Auto-discovered