Skip to main content
EVOKORE// BROWSE / CODING & DEVELOPMENT / SYSTEM PROMPTS

Coding & Development / System prompts

System prompts for Coding & Development

Browse curated system prompts for assistants, agents, tools, and durable AI workflows for coding & development teams.

107 indexed promptsISR: 300s
>

./browse/prompts

107 NODES
🤖system prompt•7 months ago

screen-reader-testing

Test web applications with screen readers including VoiceOver,

coding
⭐1
# Screen Reader Testing Practical guide to testing web applications with screen readers for comprehensive accessibility validation. ## When to Use This Skill - Validating screen reader compatibility - Testing ARIA implementations - Debugging assistive technology issues - Verifying form accessibility - Testing dynamic content announcements - Ensuring navigation accessibility ## Core Concepts ### 1. Major Screen Readers | Screen Reader | Platform | Browser | Usage | | ------------- | --------- | -------------- | ----- | | **VoiceOver** | macOS/iOS | Safari | ~15% | | **NVDA** | Windows | Firefox/Chrome | ~31% | | **JAWS** | Windows | Chrome/IE | ~40% | | **TalkBack** | Android | Chrome | ~10% | | **Narrator** | Windows | Edge | ~4% | ### 2. Testing Priority ``` Minimum Coverage: 1. NVDA + Firefox (Windows) 2. VoiceOver + Safari (macOS) 3. VoiceOver + Safari (iOS) Comprehensive Coverage: + JAWS + Chrome (Windows) + TalkBack + Chrome (Android) + Narrator + Edge (Windows) ``` ### 3. Screen Reader Modes | Mode | Purpose | When Used | | ------------------ | ---------------------- | ----------------- | | **Browse/Virtual** | Read content | Default reading | | **Focus/Forms** | Interact with controls | Filling forms | | **Application** | Custom widgets | ARIA applications | ## VoiceOver (macOS) ### Setup ``` Enable: System Preferences → Accessibility → VoiceOver Toggle: Cmd + F5 Quick Toggle: Triple-press Touch ID ``` ### Essential Commands ``` Navigation: VO = Ctrl + Option (VoiceOver modifier) VO + Right Arrow Next element VO + Left Arrow Previous element VO + Shift + Down Enter group VO + Shift + Up Exit group Reading: VO + A Read all from cursor Ctrl Stop speaking VO + B Read current paragraph Interaction: VO + Space Activate element VO + Shift + M Open menu Tab Next focusable element Shift + Tab Previous focusable element Rotor (VO + U): Navigate by: Headings, Links, Forms, Landmarks Left/Right Arrow Change rotor category Up/Down Arrow Navigate within category Enter Go to item Web Specific: VO + Cmd + H Next heading VO + Cmd + J Next form control VO + Cmd + L Next link VO + Cmd + T Next table ``` ### Testing Checklist ```markdown ## VoiceOver Testing Checklist ### Page Load - [ ] Page title announced - [ ] Main landmark found - [ ] Skip link works ### Navigation - [ ] All headings discoverable via rotor - [ ] Heading levels logical (H1 → H2 → H3) - [ ] Landmarks properly labeled - [ ] Skip links functional ### Links & Buttons - [ ] Link purpose clear - [ ] Button actions described - [ ] New window/tab announced ### Forms - [ ] All labels read with inputs - [ ] Required fields announced - [ ] Error messages read - [ ] Instructions available - [ ] Focus moves to errors ### Dynamic Content - [ ] Alerts announced immediately - [ ] Loading states communicated - [ ] Content updates announced - [ ] Modals trap focus correctly ### Tables - [ ] Headers associated with cells - [ ] Table navigation works - [ ] Complex tables have captions ``` ### Common Issues & Fixes ```html <!-- Issue: Button not announcing purpose --> <button><svg>...</svg></button> <!-- Fix --> <button aria-label="Close dialog"><svg aria-hidden="true">...</svg></button> <!-- Issue: Dynamic content not announced --> <div id="results">New results loaded</div> <!-- Fix --> <div id="results" role="status" aria-live="polite">New results loaded</div> <!-- Issue: Form error not read --> <input type="email" /> <span class="error">Invalid email</span> <!-- Fix --> <input type="email" aria-invalid="true" aria-describedby="email-error" /> <span id="email-error" role="alert">Invalid email</span> ``` ## NVDA (Windows) ### Setup ``` Download: nvaccess.org Start: Ctrl + Alt + N Stop: Insert + Q ``` ### Essential Commands ``` Navigation: Insert = NVDA modifier Down Arrow Next line Up Arrow Previous line Tab Next focusable Shift + Tab Previous focusable Reading: NVDA + Down Arrow Say all Ctrl Stop speech NVDA + Up Arrow Current line Headings: H Next heading Shift + H Previous heading 1-6 Heading level 1-6 Forms: F Next form field B Next button E Next edit field X Next checkbox C Next combo box Links: K Next link U Next unvisited link V Next visited link Landmarks: D Next landmark Shift + D Previous landmark Tables: T Next table Ctrl + Alt + Arrows Navigate cells Elements List (NVDA + F7): Shows all links, headings, form fields, landmarks ``` ### Browse vs Focus Mode ``` NVDA automatically switches modes: - Browse Mode: Arrow keys navigate content - Focus Mode: Arrow keys control interactive elements Manual switch: NVDA + Space Watch for: - "Browse mode" announcement when navigating - "Focus mode" when entering form fields - Application role forces forms mode ``` ### Testing Script ```markdown ## NVDA Test Script ### Initial Load 1. Navigate to page 2. Let page finish loading 3. Press Insert + Down to read all 4. Note: Page title, main content identified? ### Landmark Navigation 1. Press D repeatedly 2. Check: All main areas reachable? 3. Check: Landmarks properly labeled? ### Heading Navigation 1. Press Insert + F7 → Headings 2. Check: Logical heading structure? 3. Press H to navigate headings 4. Check: All sections discoverable? ### Form Testing 1. Press F to find first form field 2. Check: Label read? 3. Fill in invalid data 4. Submit form 5. Check: Errors announced? 6. Check: Focus moved to error? ### Interactive Elements 1. Tab through all interactive elements 2. Check: Each announces role and state 3. Activate buttons with Enter/Space 4. Check: Result announced? ### Dynamic Content 1. Trigger content update 2. Check: Change announced? 3. Open modal 4. Check: Focus trapped? 5. Close modal 6. Check: Focus returns? ``` ## JAWS (Windows) ### Essential Commands ``` Start: Desktop shortcut or Ctrl + Alt + J Virtual Cursor: Auto-enabled in browsers Navigation: Arrow keys Navigate content Tab Next focusable Insert + Down Read all Ctrl Stop speech Quick Keys: H Next heading T Next table F Next form field B Next button G Next graphic L Next list ; Next landmark Forms Mode: Enter Enter forms mode Numpad + Exit forms mode F5 List form fields Lists: Insert + F7 Link list Insert + F6 Heading list Insert + F5 Form field list Tables: Ctrl + Alt + Arrows Table navigation ``` ## TalkBack (Android) ### Setup ``` Enable: Settings → Accessibility → TalkBack Toggle: Hold both volume buttons 3 seconds ``` ### Gestures ``` Explore: Drag finger across screen Next: Swipe right Previous: Swipe left Activate: Double tap Scroll: Two finger swipe Reading Controls (swipe up then right): - Headings - Links - Controls - Characters - Words - Lines - Paragraphs ``` ## Common Test Scenarios ### 1. Modal Dialog ```html <!-- Accessible modal structure --> <div role="dialog" aria-modal="true" aria-labelledby="dialog-title" aria-describedby="dialog-desc" > <h2 id="dialog-title">Confirm Delete</h2> <p id="dialog-desc">This action cannot be undone.</p> <button>Cancel</button> <button>Delete</button> </div> ``` ```javascript // Focus management function openModal(modal) { // Store last focused element lastFocus = document.activeElement; // Move focus to modal modal.querySelector("h2").focus(); // Trap focus modal.addEventListener("keydown", trapFocus); } function closeModal(modal) { // Return focus lastFocus.focus(); } function trapFocus(e) { if (e.key === "Tab") { const focusable = modal.querySelectorAll( 'button, [href], input, select, textarea, [tabindex]:not([tabindex="-1"])', ); const first = focusable[0]; const last = focusable[focusable.length - 1]; if (e.shiftKey && document.activeElement === first) { last.focus(); e.preventDefault(); } else if (!e.shiftKey && document.activeElement === last) { first.focus(); e.preventDefault(); } } if (e.key === "Escape") { closeModal(modal); } } ``` ### 2. Live Regions ```html <!-- Status messages (polite) --> <div role="status" aria-live="polite" aria-atomic="true"> <!-- Content updates will be announced after current speech --> </div> <!-- Alerts (assertive) --> <div role="alert" aria-live="assertive"> <!-- Content updates interrupt current speech --> </div> <!-- Progress updates --> <div role="progressbar" aria-valuenow="75" aria-valuemin="0" aria-valuemax="100" aria-label="Upload progress" ></div> <!-- Log (additions only) --> <div role="log" aria-live="polite" aria-relevant="additions"> <!-- New messages announced, removals not --> </div> ``` ### 3. Tab Interface ```html <div role="tablist" aria-label="Product information"> <button role="tab" id="tab-1" aria-selected="true" aria-controls="panel-1"> Description </button> <button role="tab" id="tab-2" aria-selected="false" aria-controls="panel-2" tabindex="-1" > Reviews </button> </div> <div role="tabpanel" id="panel-1" aria-labelledby="tab-1"> Product description content... </div> <div role="tabpanel" id="panel-2" aria-labelledby="tab-2" hidden> Reviews content... </div> ``` ```javascript // Tab keyboard navigation tablist.addEventListener("keydown", (e) => { const tabs = [...tablist.querySelectorAll('[role="tab"]')]; const index = tabs.indexOf(document.activeElement); let newIndex; switch (e.key) { case "ArrowRight": newIndex = (index + 1) % tabs.length; break; case "ArrowLeft": newIndex = (index - 1 + tabs.length) % tabs.length; break; case "Home": newIndex = 0; break; case "End": newIndex = tabs.length - 1; break; default: return; } tabs[newIndex].focus(); activateTab(tabs[newIndex]); e.preventDefault(); }); ``` ## Debugging Tips ```javascript // Log what screen reader sees function logAccessibleName(element) { const computed = window.getComputedStyle(element); console.log({ role: element.getAttribute("role") || element.tagName, name: element.getAttribute("aria-label") || element.getAttribute("aria-labelledby") || element.textContent, state: { expanded: element.getAttribute("aria-expanded"), selected: element.getAttribute("aria-selected"), checked: element.getAttribute("aria-checked"), disabled: element.disabled, }, visible: computed.display !== "none" && computed.visibility !== "hidden", }); } ``` ## Best Practices ### Do's - **Test with actual screen readers** - Not just simulators - **Use semantic HTML first** - ARIA is supplemental - **Test in browse and focus modes** - Different experiences - **Verify focus management** - Especially for SPAs - **Test keyboard only first** - Foundation for SR testing ### Don'ts - **Don't assume one SR is enough** - Test multiple - **Don't ignore mobile** - Growing user base - **Don't test only happy path** - Test error states - **Don't skip dynamic content** - Most common issues - **Don't rely on visual testing** - Different experience ## Resources - [VoiceOver User Guide](https://support.apple.com/guide/voiceover/welcome/mac) - [NVDA User Guide](https://www.nvaccess.org/files/nvda/documentation/userGuide.html) - [JAWS Documentation](https://support.freedomscientific.com/Products/Blindness/JAWS) - [WebAIM Screen Reader Survey](https://webaim.org/projects/screenreadersurvey/)
👍1
👁️1
🤖 Auto-discovered
🤖system prompt•7 months ago

wcag-audit-patterns

Conduct WCAG 2.2 accessibility audits with automated testing,

coding
⭐1
# WCAG Audit Patterns Comprehensive guide to auditing web content against WCAG 2.2 guidelines with actionable remediation strategies. ## When to Use This Skill - Conducting accessibility audits - Fixing WCAG violations - Implementing accessible components - Preparing for accessibility lawsuits - Meeting ADA/Section 508 requirements - Achieving VPAT compliance ## Core Concepts ### 1. WCAG Conformance Levels | Level | Description | Required For | | ------- | ---------------------- | ----------------- | | **A** | Minimum accessibility | Legal baseline | | **AA** | Standard conformance | Most regulations | | **AAA** | Enhanced accessibility | Specialized needs | ### 2. POUR Principles ``` Perceivable: Can users perceive the content? Operable: Can users operate the interface? Understandable: Can users understand the content? Robust: Does it work with assistive tech? ``` ### 3. Common Violations by Impact ``` Critical (Blockers): ├── Missing alt text for functional images ├── No keyboard access to interactive elements ├── Missing form labels └── Auto-playing media without controls Serious: ├── Insufficient color contrast ├── Missing skip links ├── Inaccessible custom widgets └── Missing page titles Moderate: ├── Missing language attribute ├── Unclear link text ├── Missing landmarks └── Improper heading hierarchy ``` ## Audit Checklist ### Perceivable (Principle 1) ````markdown ## 1.1 Text Alternatives ### 1.1.1 Non-text Content (Level A) - [ ] All images have alt text - [ ] Decorative images have alt="" - [ ] Complex images have long descriptions - [ ] Icons with meaning have accessible names - [ ] CAPTCHAs have alternatives Check: ```html <!-- Good --> <img src="chart.png" alt="Sales increased 25% from Q1 to Q2" /> <img src="decorative-line.png" alt="" /> <!-- Bad --> <img src="chart.png" /> <img src="decorative-line.png" alt="decorative line" /> ``` ```` ## 1.2 Time-based Media ### 1.2.1 Audio-only and Video-only (Level A) - [ ] Audio has text transcript - [ ] Video has audio description or transcript ### 1.2.2 Captions (Level A) - [ ] All video has synchronized captions - [ ] Captions are accurate and complete - [ ] Speaker identification included ### 1.2.3 Audio Description (Level A) - [ ] Video has audio description for visual content ## 1.3 Adaptable ### 1.3.1 Info and Relationships (Level A) - [ ] Headings use proper tags (h1-h6) - [ ] Lists use ul/ol/dl - [ ] Tables have headers - [ ] Form inputs have labels - [ ] ARIA landmarks present Check: ```html <!-- Heading hierarchy --> <h1>Page Title</h1> <h2>Section</h2> <h3>Subsection</h3> <h2>Another Section</h2> <!-- Table headers --> <table> <thead> <tr> <th scope="col">Name</th> <th scope="col">Price</th> </tr> </thead> </table> ``` ### 1.3.2 Meaningful Sequence (Level A) - [ ] Reading order is logical - [ ] CSS positioning doesn't break order - [ ] Focus order matches visual order ### 1.3.3 Sensory Characteristics (Level A) - [ ] Instructions don't rely on shape/color alone - [ ] "Click the red button" → "Click Submit (red button)" ## 1.4 Distinguishable ### 1.4.1 Use of Color (Level A) - [ ] Color is not only means of conveying info - [ ] Links distinguishable without color - [ ] Error states not color-only ### 1.4.3 Contrast (Minimum) (Level AA) - [ ] Text: 4.5:1 contrast ratio - [ ] Large text (18pt+): 3:1 ratio - [ ] UI components: 3:1 ratio Tools: WebAIM Contrast Checker, axe DevTools ### 1.4.4 Resize Text (Level AA) - [ ] Text resizes to 200% without loss - [ ] No horizontal scrolling at 320px - [ ] Content reflows properly ### 1.4.10 Reflow (Level AA) - [ ] Content reflows at 400% zoom - [ ] No two-dimensional scrolling - [ ] All content accessible at 320px width ### 1.4.11 Non-text Contrast (Level AA) - [ ] UI components have 3:1 contrast - [ ] Focus indicators visible - [ ] Graphical objects distinguishable ### 1.4.12 Text Spacing (Level AA) - [ ] No content loss with increased spacing - [ ] Line height 1.5x font size - [ ] Paragraph spacing 2x font size - [ ] Letter spacing 0.12x font size - [ ] Word spacing 0.16x font size ```` ### Operable (Principle 2) ```markdown ## 2.1 Keyboard Accessible ### 2.1.1 Keyboard (Level A) - [ ] All functionality keyboard accessible - [ ] No keyboard traps - [ ] Tab order is logical - [ ] Custom widgets are keyboard operable Check: ```javascript // Custom button must be keyboard accessible <div role="button" tabindex="0" onkeydown="if(event.key === 'Enter' || event.key === ' ') activate()"> ```` ### 2.1.2 No Keyboard Trap (Level A) - [ ] Focus can move away from all components - [ ] Modal dialogs trap focus correctly - [ ] Focus returns after modal closes ## 2.2 Enough Time ### 2.2.1 Timing Adjustable (Level A) - [ ] Session timeouts can be extended - [ ] User warned before timeout - [ ] Option to disable auto-refresh ### 2.2.2 Pause, Stop, Hide (Level A) - [ ] Moving content can be paused - [ ] Auto-updating content can be paused - [ ] Animations respect prefers-reduced-motion ```css @media (prefers-reduced-motion: reduce) { * { animation: none !important; transition: none !important; } } ``` ## 2.3 Seizures and Physical Reactions ### 2.3.1 Three Flashes (Level A) - [ ] No content flashes more than 3 times/second - [ ] Flashing area is small (<25% viewport) ## 2.4 Navigable ### 2.4.1 Bypass Blocks (Level A) - [ ] Skip to main content link present - [ ] Landmark regions defined - [ ] Proper heading structure ```html <a href="#main" class="skip-link">Skip to main content</a> <main id="main">...</main> ``` ### 2.4.2 Page Titled (Level A) - [ ] Unique, descriptive page titles - [ ] Title reflects page content ### 2.4.3 Focus Order (Level A) - [ ] Focus order matches visual order - [ ] tabindex used correctly ### 2.4.4 Link Purpose (In Context) (Level A) - [ ] Links make sense out of context - [ ] No "click here" or "read more" alone ```html <!-- Bad --> <a href="report.pdf">Click here</a> <!-- Good --> <a href="report.pdf">Download Q4 Sales Report (PDF)</a> ``` ### 2.4.6 Headings and Labels (Level AA) - [ ] Headings describe content - [ ] Labels describe purpose ### 2.4.7 Focus Visible (Level AA) - [ ] Focus indicator visible on all elements - [ ] Custom focus styles meet contrast ```css :focus { outline: 3px solid #005fcc; outline-offset: 2px; } ``` ### 2.4.11 Focus Not Obscured (Level AA) - WCAG 2.2 - [ ] Focused element not fully hidden - [ ] Sticky headers don't obscure focus ```` ### Understandable (Principle 3) ```markdown ## 3.1 Readable ### 3.1.1 Language of Page (Level A) - [ ] HTML lang attribute set - [ ] Language correct for content ```html <html lang="en"> ```` ### 3.1.2 Language of Parts (Level AA) - [ ] Language changes marked ```html <p>The French word <span lang="fr">bonjour</span> means hello.</p> ``` ## 3.2 Predictable ### 3.2.1 On Focus (Level A) - [ ] No context change on focus alone - [ ] No unexpected popups on focus ### 3.2.2 On Input (Level A) - [ ] No automatic form submission - [ ] User warned before context change ### 3.2.3 Consistent Navigation (Level AA) - [ ] Navigation consistent across pages - [ ] Repeated components same order ### 3.2.4 Consistent Identification (Level AA) - [ ] Same functionality = same label - [ ] Icons used consistently ## 3.3 Input Assistance ### 3.3.1 Error Identification (Level A) - [ ] Errors clearly identified - [ ] Error message describes problem - [ ] Error linked to field ```html <input aria-describedby="email-error" aria-invalid="true" /> <span id="email-error" role="alert">Please enter valid email</span> ``` ### 3.3.2 Labels or Instructions (Level A) - [ ] All inputs have visible labels - [ ] Required fields indicated - [ ] Format hints provided ### 3.3.3 Error Suggestion (Level AA) - [ ] Errors include correction suggestion - [ ] Suggestions are specific ### 3.3.4 Error Prevention (Level AA) - [ ] Legal/financial forms reversible - [ ] Data checked before submission - [ ] User can review before submit ```` ### Robust (Principle 4) ```markdown ## 4.1 Compatible ### 4.1.1 Parsing (Level A) - Obsolete in WCAG 2.2 - [ ] Valid HTML (good practice) - [ ] No duplicate IDs - [ ] Complete start/end tags ### 4.1.2 Name, Role, Value (Level A) - [ ] Custom widgets have accessible names - [ ] ARIA roles correct - [ ] State changes announced ```html <!-- Accessible custom checkbox --> <div role="checkbox" aria-checked="false" tabindex="0" aria-labelledby="label"> </div> <span id="label">Accept terms</span> ```` ### 4.1.3 Status Messages (Level AA) - [ ] Status updates announced - [ ] Live regions used correctly ```html <div role="status" aria-live="polite">3 items added to cart</div> <div role="alert" aria-live="assertive">Error: Form submission failed</div> ``` ```` ## Automated Testing ```javascript // axe-core integration const axe = require('axe-core'); async function runAccessibilityAudit(page) { await page.addScriptTag({ path: require.resolve('axe-core') }); const results = await page.evaluate(async () => { return await axe.run(document, { runOnly: { type: 'tag', values: ['wcag2a', 'wcag2aa', 'wcag21aa', 'wcag22aa'] } }); }); return { violations: results.violations, passes: results.passes, incomplete: results.incomplete }; } // Playwright test example test('should have no accessibility violations', async ({ page }) => { await page.goto('/'); const results = await runAccessibilityAudit(page); expect(results.violations).toHaveLength(0); }); ```` ```bash # CLI tools npx @axe-core/cli https://example.com npx pa11y https://example.com lighthouse https://example.com --only-categories=accessibility ``` ## Remediation Patterns ### Fix: Missing Form Labels ```html <!-- Before --> <input type="email" placeholder="Email" /> <!-- After: Option 1 - Visible label --> <label for="email">Email address</label> <input id="email" type="email" /> <!-- After: Option 2 - aria-label --> <input type="email" aria-label="Email address" /> <!-- After: Option 3 - aria-labelledby --> <span id="email-label">Email</span> <input type="email" aria-labelledby="email-label" /> ``` ### Fix: Insufficient Color Contrast ```css /* Before: 2.5:1 contrast */ .text { color: #767676; } /* After: 4.5:1 contrast */ .text { color: #595959; } /* Or add background */ .text { color: #767676; background: #000; } ``` ### Fix: Keyboard Navigation ```javascript // Make custom element keyboard accessible class AccessibleDropdown extends HTMLElement { connectedCallback() { this.setAttribute("tabindex", "0"); this.setAttribute("role", "combobox"); this.setAttribute("aria-expanded", "false"); this.addEventListener("keydown", (e) => { switch (e.key) { case "Enter": case " ": this.toggle(); e.preventDefault(); break; case "Escape": this.close(); break; case "ArrowDown": this.focusNext(); e.preventDefault(); break; case "ArrowUp": this.focusPrevious(); e.preventDefault(); break; } }); } } ``` ## Best Practices ### Do's - **Start early** - Accessibility from design phase - **Test with real users** - Disabled users provide best feedback - **Automate what you can** - 30-50% issues detectable - **Use semantic HTML** - Reduces ARIA needs - **Document patterns** - Build accessible component library ### Don'ts - **Don't rely only on automated testing** - Manual testing required - **Don't use ARIA as first solution** - Native HTML first - **Don't hide focus outlines** - Keyboard users need them - **Don't disable zoom** - Users need to resize - **Don't use color alone** - Multiple indicators needed ## Resources - [WCAG 2.2 Guidelines](https://www.w3.org/TR/WCAG22/) - [WebAIM](https://webaim.org/) - [A11y Project Checklist](https://www.a11yproject.com/checklist/) - [axe DevTools](https://www.deque.com/axe/)
👍1
👁️1
🤖 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

parallel-feature-development

Coordinate parallel feature development with file ownership

coding
⭐1
# Parallel Feature Development Strategies for decomposing features into parallel work streams, establishing file ownership boundaries, avoiding conflicts, and integrating results from multiple implementer agents. ## When to Use This Skill - Decomposing a feature for parallel implementation - Establishing file ownership boundaries between agents - Designing interface contracts between parallel work streams - Choosing integration strategies (vertical slice vs horizontal layer) - Managing branch and merge workflows for parallel development ## File Ownership Strategies ### By Directory Assign each implementer ownership of specific directories: ``` implementer-1: src/components/auth/ implementer-2: src/api/auth/ implementer-3: tests/auth/ ``` **Best for**: Well-organized codebases with clear directory boundaries. ### By Module Assign ownership of logical modules (which may span directories): ``` implementer-1: Authentication module (login, register, logout) implementer-2: Authorization module (roles, permissions, guards) ``` **Best for**: Feature-oriented architectures, domain-driven design. ### By Layer Assign ownership of architectural layers: ``` implementer-1: UI layer (components, styles, layouts) implementer-2: Business logic layer (services, validators) implementer-3: Data layer (models, repositories, migrations) ``` **Best for**: Traditional MVC/layered architectures. ## Conflict Avoidance Rules ### The Cardinal Rule **One owner per file.** No file should be assigned to multiple implementers. ### When Files Must Be Shared If a file genuinely needs changes from multiple implementers: 1. **Designate a single owner** — One implementer owns the file 2. **Other implementers request changes** — Message the owner with specific change requests 3. **Owner applies changes sequentially** — Prevents merge conflicts 4. **Alternative: Extract interfaces** — Create a separate interface file that the non-owner can import without modifying ### Interface Contracts When implementers need to coordinate at boundaries: ```typescript // src/types/auth-contract.ts (owned by team-lead, read-only for implementers) export interface AuthResponse { token: string; user: UserProfile; expiresAt: number; } export interface AuthService { login(email: string, password: string): Promise<AuthResponse>; register(data: RegisterData): Promise<AuthResponse>; } ``` Both implementers import from the contract file but neither modifies it. ## Integration Patterns ### Vertical Slice Each implementer builds a complete feature slice (UI + API + tests): ``` implementer-1: Login feature (login form + login API + login tests) implementer-2: Register feature (register form + register API + register tests) ``` **Pros**: Each slice is independently testable, minimal integration needed. **Cons**: May duplicate shared utilities, harder with tightly coupled features. ### Horizontal Layer Each implementer builds one layer across all features: ``` implementer-1: All UI components (login form, register form, profile page) implementer-2: All API endpoints (login, register, profile) implementer-3: All tests (unit, integration, e2e) ``` **Pros**: Consistent patterns within each layer, natural specialization. **Cons**: More integration points, layer 3 depends on layers 1 and 2. ### Hybrid Mix vertical and horizontal based on coupling: ``` implementer-1: Login feature (vertical slice — UI + API + tests) implementer-2: Shared auth infrastructure (horizontal — middleware, JWT utils, types) ``` **Best for**: Most real-world features with some shared infrastructure. ## Branch Management ### Single Branch Strategy All implementers work on the same feature branch: - Simple setup, no merge overhead - Requires strict file ownership to avoid conflicts - Best for: small teams (2-3), well-defined boundaries ### Multi-Branch Strategy Each implementer works on a sub-branch: ``` feature/auth ├── feature/auth-login (implementer-1) ├── feature/auth-register (implementer-2) └── feature/auth-tests (implementer-3) ``` - More isolation, explicit merge points - Higher overhead, merge conflicts still possible in shared files - Best for: larger teams (4+), complex features
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

task-coordination-strategies

Decompose complex tasks, design dependency graphs, and coordinate

coding
⭐1
# Task Coordination Strategies Strategies for decomposing complex tasks into parallelizable units, designing dependency graphs, writing effective task descriptions, and monitoring workload across agent teams. ## When to Use This Skill - Breaking down a complex task for parallel execution - Designing task dependency relationships (blockedBy/blocks) - Writing task descriptions with clear acceptance criteria - Monitoring and rebalancing workload across teammates - Identifying the critical path in a multi-task workflow ## Task Decomposition Strategies ### By Layer Split work by architectural layer: - Frontend components - Backend API endpoints - Database migrations/models - Test suites **Best for**: Full-stack features, vertical slices ### By Component Split work by functional component: - Authentication module - User profile module - Notification module **Best for**: Microservices, modular architectures ### By Concern Split work by cross-cutting concern: - Security review - Performance review - Architecture review **Best for**: Code reviews, audits ### By File Ownership Split work by file/directory boundaries: - `src/components/` — Implementer 1 - `src/api/` — Implementer 2 - `src/utils/` — Implementer 3 **Best for**: Parallel implementation, conflict avoidance ## Dependency Graph Design ### Principles 1. **Minimize chain depth** — Prefer wide, shallow graphs over deep chains 2. **Identify the critical path** — The longest chain determines minimum completion time 3. **Use blockedBy sparingly** — Only add dependencies that are truly required 4. **Avoid circular dependencies** — Task A blocks B blocks A is a deadlock ### Patterns **Independent (Best parallelism)**: ``` Task A ─┐ Task B ─┼─→ Integration Task C ─┘ ``` **Sequential (Necessary dependencies)**: ``` Task A → Task B → Task C ``` **Diamond (Mixed)**: ``` ┌→ Task B ─┐ Task A ─┤ ├→ Task D └→ Task C ─┘ ``` ### Using blockedBy/blocks ``` TaskCreate: { subject: "Build API endpoints" } → Task #1 TaskCreate: { subject: "Build frontend components" } → Task #2 TaskCreate: { subject: "Integration testing" } → Task #3 TaskUpdate: { taskId: "3", addBlockedBy: ["1", "2"] } → #3 waits for #1 and #2 ``` ## Task Description Best Practices Every task should include: 1. **Objective** — What needs to be accomplished (1-2 sentences) 2. **Owned Files** — Explicit list of files/directories this teammate may modify 3. **Requirements** — Specific deliverables or behaviors expected 4. **Interface Contracts** — How this work connects to other teammates' work 5. **Acceptance Criteria** — How to verify the task is done correctly 6. **Scope Boundaries** — What is explicitly out of scope ### Template ``` ## Objective Build the user authentication API endpoints. ## Owned Files - src/api/auth.ts - src/api/middleware/auth-middleware.ts - src/types/auth.ts (shared — read only, do not modify) ## Requirements - POST /api/login — accepts email/password, returns JWT - POST /api/register — creates new user, returns JWT - GET /api/me — returns current user profile (requires auth) ## Interface Contract - Import User type from src/types/auth.ts (owned by implementer-1) - Export AuthResponse type for frontend consumption ## Acceptance Criteria - All endpoints return proper HTTP status codes - JWT tokens expire after 24 hours - Passwords are hashed with bcrypt ## Out of Scope - OAuth/social login - Password reset flow - Rate limiting ``` ## Workload Monitoring ### Indicators of Imbalance | Signal | Meaning | Action | | -------------------------- | ------------------- | --------------------------- | | Teammate idle, others busy | Uneven distribution | Reassign pending tasks | | Teammate stuck on one task | Possible blocker | Check in, offer help | | All tasks blocked | Dependency issue | Resolve critical path first | | One teammate has 3x others | Overloaded | Split tasks or reassign | ### Rebalancing Steps 1. Call `TaskList` to assess current state 2. Identify idle or overloaded teammates 3. Use `TaskUpdate` to reassign tasks 4. Use `SendMessage` to notify affected teammates 5. Monitor for improved throughput
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

team-communication-protocols

Structured messaging protocols for agent team communication

coding
⭐1
# Team Communication Protocols Protocols for effective communication between agent teammates, including message type selection, plan approval workflows, shutdown procedures, and common anti-patterns to avoid. ## When to Use This Skill - Establishing communication norms for a new team - Choosing between message types (message, broadcast, shutdown_request) - Handling plan approval workflows - Managing graceful team shutdown - Discovering teammate identities and capabilities ## Message Type Selection ### `message` (Direct Message) — Default Choice Send to a single specific teammate: ```json { "type": "message", "recipient": "implementer-1", "content": "Your API endpoint is ready. You can now build the frontend form.", "summary": "API endpoint ready for frontend" } ``` **Use for**: Task updates, coordination, questions, integration notifications. ### `broadcast` — Use Sparingly Send to ALL teammates simultaneously: ```json { "type": "broadcast", "content": "Critical: shared types file has been updated. Pull latest before continuing.", "summary": "Shared types updated" } ``` **Use ONLY for**: Critical blockers affecting everyone, major changes to shared resources. **Why sparingly?**: Each broadcast sends N separate messages (one per teammate), consuming API resources proportional to team size. ### `shutdown_request` — Graceful Termination Request a teammate to shut down: ```json { "type": "shutdown_request", "recipient": "reviewer-1", "content": "Review complete, shutting down team." } ``` The teammate responds with `shutdown_response` (approve or reject with reason). ## Communication Anti-Patterns | Anti-Pattern | Problem | Better Approach | | --------------------------------------- | ---------------------------------------- | -------------------------------------- | | Broadcasting routine updates | Wastes resources, noise | Direct message to affected teammate | | Sending JSON status messages | Not designed for structured data | Use TaskUpdate to update task status | | Not communicating at integration points | Teammates build against stale interfaces | Message when your interface is ready | | Micromanaging via messages | Overwhelms teammates, slows work | Check in at milestones, not every step | | Using UUIDs instead of names | Hard to read, error-prone | Always use teammate names | | Ignoring idle teammates | Wasted capacity | Assign new work or shut down | ## Plan Approval Workflow When a teammate is spawned with `plan_mode_required`: 1. Teammate creates a plan using read-only exploration tools 2. Teammate calls `ExitPlanMode` which sends a `plan_approval_request` to the lead 3. Lead reviews the plan 4. Lead responds with `plan_approval_response`: **Approve**: ```json { "type": "plan_approval_response", "request_id": "abc-123", "recipient": "implementer-1", "approve": true } ``` **Reject with feedback**: ```json { "type": "plan_approval_response", "request_id": "abc-123", "recipient": "implementer-1", "approve": false, "content": "Please add error handling for the API calls" } ``` ## Shutdown Protocol ### Graceful Shutdown Sequence 1. **Lead sends shutdown_request** to each teammate 2. **Teammate receives request** as a JSON message with `type: "shutdown_request"` 3. **Teammate responds** with `shutdown_response`: - `approve: true` — Teammate saves state and exits - `approve: false` + reason — Teammate continues working 4. **Lead handles rejections** — Wait for teammate to finish, then retry 5. **After all teammates shut down** — Call `Teammate` cleanup ### Handling Rejections If a teammate rejects shutdown: - Check their reason (usually "still working on task") - Wait for their current task to complete - Retry shutdown request - If urgent, user can force shutdown ## Teammate Discovery Find team members by reading the config file: **Location**: `~/.claude/teams/{team-name}/config.json` **Structure**: ```json { "members": [ { "name": "security-reviewer", "agentId": "uuid-here", "agentType": "team-reviewer" }, { "name": "perf-reviewer", "agentId": "uuid-here", "agentType": "team-reviewer" } ] } ``` **Always use `name`** for messaging and task assignment. Never use `agentId` directly.
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

team-composition-patterns

Design optimal agent team compositions with sizing heuristics,

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

fastapi-templates

Create production-ready FastAPI projects with async patterns,

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

api-design-principles

Master REST and GraphQL API design principles to build intuitive,

coding
⭐1
# API Design Principles Master REST and GraphQL API design principles to build intuitive, scalable, and maintainable APIs that delight developers and stand the test of time. ## When to Use This Skill - Designing new REST or GraphQL APIs - Refactoring existing APIs for better usability - Establishing API design standards for your team - Reviewing API specifications before implementation - Migrating between API paradigms (REST to GraphQL, etc.) - Creating developer-friendly API documentation - Optimizing APIs for specific use cases (mobile, third-party integrations) ## Core Concepts ### 1. RESTful Design Principles **Resource-Oriented Architecture** - Resources are nouns (users, orders, products), not verbs - Use HTTP methods for actions (GET, POST, PUT, PATCH, DELETE) - URLs represent resource hierarchies - Consistent naming conventions **HTTP Methods Semantics:** - `GET`: Retrieve resources (idempotent, safe) - `POST`: Create new resources - `PUT`: Replace entire resource (idempotent) - `PATCH`: Partial resource updates - `DELETE`: Remove resources (idempotent) ### 2. GraphQL Design Principles **Schema-First Development** - Types define your domain model - Queries for reading data - Mutations for modifying data - Subscriptions for real-time updates **Query Structure:** - Clients request exactly what they need - Single endpoint, multiple operations - Strongly typed schema - Introspection built-in ### 3. API Versioning Strategies **URL Versioning:** ``` /api/v1/users /api/v2/users ``` **Header Versioning:** ``` Accept: application/vnd.api+json; version=1 ``` **Query Parameter Versioning:** ``` /api/users?version=1 ``` ## REST API Design Patterns ### Pattern 1: Resource Collection Design ```python # Good: Resource-oriented endpoints GET /api/users # List users (with pagination) POST /api/users # Create user GET /api/users/{id} # Get specific user PUT /api/users/{id} # Replace user PATCH /api/users/{id} # Update user fields DELETE /api/users/{id} # Delete user # Nested resources GET /api/users/{id}/orders # Get user's orders POST /api/users/{id}/orders # Create order for user # Bad: Action-oriented endpoints (avoid) POST /api/createUser POST /api/getUserById POST /api/deleteUser ``` ### Pattern 2: Pagination and Filtering ```python from typing import List, Optional from pydantic import BaseModel, Field class PaginationParams(BaseModel): page: int = Field(1, ge=1, description="Page number") page_size: int = Field(20, ge=1, le=100, description="Items per page") class FilterParams(BaseModel): status: Optional[str] = None created_after: Optional[str] = None search: Optional[str] = None class PaginatedResponse(BaseModel): items: List[dict] total: int page: int page_size: int pages: int @property def has_next(self) -> bool: return self.page < self.pages @property def has_prev(self) -> bool: return self.page > 1 # FastAPI endpoint example from fastapi import FastAPI, Query, Depends app = FastAPI() @app.get("/api/users", response_model=PaginatedResponse) async def list_users( page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), status: Optional[str] = Query(None), search: Optional[str] = Query(None) ): # Apply filters query = build_query(status=status, search=search) # Count total total = await count_users(query) # Fetch page offset = (page - 1) * page_size users = await fetch_users(query, limit=page_size, offset=offset) return PaginatedResponse( items=users, total=total, page=page, page_size=page_size, pages=(total + page_size - 1) // page_size ) ``` ### Pattern 3: Error Handling and Status Codes ```python from fastapi import HTTPException, status from pydantic import BaseModel class ErrorResponse(BaseModel): error: str message: str details: Optional[dict] = None timestamp: str path: str class ValidationErrorDetail(BaseModel): field: str message: str value: Any # Consistent error responses STATUS_CODES = { "success": 200, "created": 201, "no_content": 204, "bad_request": 400, "unauthorized": 401, "forbidden": 403, "not_found": 404, "conflict": 409, "unprocessable": 422, "internal_error": 500 } def raise_not_found(resource: str, id: str): raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail={ "error": "NotFound", "message": f"{resource} not found", "details": {"id": id} } ) def raise_validation_error(errors: List[ValidationErrorDetail]): raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={ "error": "ValidationError", "message": "Request validation failed", "details": {"errors": [e.dict() for e in errors]} } ) # Example usage @app.get("/api/users/{user_id}") async def get_user(user_id: str): user = await fetch_user(user_id) if not user: raise_not_found("User", user_id) return user ``` ### Pattern 4: HATEOAS (Hypermedia as the Engine of Application State) ```python class UserResponse(BaseModel): id: str name: str email: str _links: dict @classmethod def from_user(cls, user: User, base_url: str): return cls( id=user.id, name=user.name, email=user.email, _links={ "self": {"href": f"{base_url}/api/users/{user.id}"}, "orders": {"href": f"{base_url}/api/users/{user.id}/orders"}, "update": { "href": f"{base_url}/api/users/{user.id}", "method": "PATCH" }, "delete": { "href": f"{base_url}/api/users/{user.id}", "method": "DELETE" } } ) ``` ## GraphQL Design Patterns ### Pattern 1: Schema Design ```graphql # schema.graphql # Clear type definitions type User { id: ID! email: String! name: String! createdAt: DateTime! # Relationships orders(first: Int = 20, after: String, status: OrderStatus): OrderConnection! profile: UserProfile } type Order { id: ID! status: OrderStatus! total: Money! items: [OrderItem!]! createdAt: DateTime! # Back-reference user: User! } # Pagination pattern (Relay-style) type OrderConnection { edges: [OrderEdge!]! pageInfo: PageInfo! totalCount: Int! } type OrderEdge { node: Order! cursor: String! } type PageInfo { hasNextPage: Boolean! hasPreviousPage: Boolean! startCursor: String endCursor: String } # Enums for type safety enum OrderStatus { PENDING CONFIRMED SHIPPED DELIVERED CANCELLED } # Custom scalars scalar DateTime scalar Money # Query root type Query { user(id: ID!): User users(first: Int = 20, after: String, search: String): UserConnection! order(id: ID!): Order } # Mutation root type Mutation { createUser(input: CreateUserInput!): CreateUserPayload! updateUser(input: UpdateUserInput!): UpdateUserPayload! deleteUser(id: ID!): DeleteUserPayload! createOrder(input: CreateOrderInput!): CreateOrderPayload! } # Input types for mutations input CreateUserInput { email: String! name: String! password: String! } # Payload types for mutations type CreateUserPayload { user: User errors: [Error!] } type Error { field: String message: String! } ``` ### Pattern 2: Resolver Design ```python from typing import Optional, List from ariadne import QueryType, MutationType, ObjectType from dataclasses import dataclass query = QueryType() mutation = MutationType() user_type = ObjectType("User") @query.field("user") async def resolve_user(obj, info, id: str) -> Optional[dict]: """Resolve single user by ID.""" return await fetch_user_by_id(id) @query.field("users") async def resolve_users( obj, info, first: int = 20, after: Optional[str] = None, search: Optional[str] = None ) -> dict: """Resolve paginated user list.""" # Decode cursor offset = decode_cursor(after) if after else 0 # Fetch users users = await fetch_users( limit=first + 1, # Fetch one extra to check hasNextPage offset=offset, search=search ) # Pagination has_next = len(users) > first if has_next: users = users[:first] edges = [ { "node": user, "cursor": encode_cursor(offset + i) } for i, user in enumerate(users) ] return { "edges": edges, "pageInfo": { "hasNextPage": has_next, "hasPreviousPage": offset > 0, "startCursor": edges[0]["cursor"] if edges else None, "endCursor": edges[-1]["cursor"] if edges else None }, "totalCount": await count_users(search=search) } @user_type.field("orders") async def resolve_user_orders(user: dict, info, first: int = 20) -> dict: """Resolve user's orders (N+1 prevention with DataLoader).""" # Use DataLoader to batch requests loader = info.context["loaders"]["orders_by_user"] orders = await loader.load(user["id"]) return paginate_orders(orders, first) @mutation.field("createUser") async def resolve_create_user(obj, info, input: dict) -> dict: """Create new user.""" try: # Validate input validate_user_input(input) # Create user user = await create_user( email=input["email"], name=input["name"], password=hash_password(input["password"]) ) return { "user": user, "errors": [] } except ValidationError as e: return { "user": None, "errors": [{"field": e.field, "message": e.message}] } ``` ### Pattern 3: DataLoader (N+1 Problem Prevention) ```python from aiodataloader import DataLoader from typing import List, Optional class UserLoader(DataLoader): """Batch load users by ID.""" async def batch_load_fn(self, user_ids: List[str]) -> List[Optional[dict]]: """Load multiple users in single query.""" users = await fetch_users_by_ids(user_ids) # Map results back to input order user_map = {user["id"]: user for user in users} return [user_map.get(user_id) for user_id in user_ids] class OrdersByUserLoader(DataLoader): """Batch load orders by user ID.""" async def batch_load_fn(self, user_ids: List[str]) -> List[List[dict]]: """Load orders for multiple users in single query.""" orders = await fetch_orders_by_user_ids(user_ids) # Group orders by user_id orders_by_user = {} for order in orders: user_id = order["user_id"] if user_id not in orders_by_user: orders_by_user[user_id] = [] orders_by_user[user_id].append(order) # Return in input order return [orders_by_user.get(user_id, []) for user_id in user_ids] # Context setup def create_context(): return { "loaders": { "user": UserLoader(), "orders_by_user": OrdersByUserLoader() } } ``` ## Best Practices ### REST APIs 1. **Consistent Naming**: Use plural nouns for collections (`/users`, not `/user`) 2. **Stateless**: Each request contains all necessary information 3. **Use HTTP Status Codes Correctly**: 2xx success, 4xx client errors, 5xx server errors 4. **Version Your API**: Plan for breaking changes from day one 5. **Pagination**: Always paginate large collections 6. **Rate Limiting**: Protect your API with rate limits 7. **Documentation**: Use OpenAPI/Swagger for interactive docs ### GraphQL APIs 1. **Schema First**: Design schema before writing resolvers 2. **Avoid N+1**: Use DataLoaders for efficient data fetching 3. **Input Validation**: Validate at schema and resolver levels 4. **Error Handling**: Return structured errors in mutation payloads 5. **Pagination**: Use cursor-based pagination (Relay spec) 6. **Deprecation**: Use `@deprecated` directive for gradual migration 7. **Monitoring**: Track query complexity and execution time ## Common Pitfalls - **Over-fetching/Under-fetching (REST)**: Fixed in GraphQL but requires DataLoaders - **Breaking Changes**: Version APIs or use deprecation strategies - **Inconsistent Error Formats**: Standardize error responses - **Missing Rate Limits**: APIs without limits are vulnerable to abuse - **Poor Documentation**: Undocumented APIs frustrate developers - **Ignoring HTTP Semantics**: POST for idempotent operations breaks expectations - **Tight Coupling**: API structure shouldn't mirror database schema ## Resources - **references/rest-best-practices.md**: Comprehensive REST API design guide - **references/graphql-schema-design.md**: GraphQL schema patterns and anti-patterns - **references/api-versioning-strategies.md**: Versioning approaches and migration paths - **assets/rest-api-template.py**: FastAPI REST API template - **assets/graphql-schema-template.graphql**: Complete GraphQL schema example - **assets/api-design-checklist.md**: Pre-implementation review checklist - **scripts/openapi-generator.py**: Generate OpenAPI specs from code
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

architecture-patterns

Implement proven backend architecture patterns including Clean

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

cqrs-implementation

Implement Command Query Responsibility Segregation for scalable

coding
⭐1
# CQRS Implementation Comprehensive guide to implementing CQRS (Command Query Responsibility Segregation) patterns. ## When to Use This Skill - Separating read and write concerns - Scaling reads independently from writes - Building event-sourced systems - Optimizing complex query scenarios - Different read/write data models needed - High-performance reporting requirements ## Core Concepts ### 1. CQRS Architecture ``` ┌─────────────┐ │ Client │ └──────┬──────┘ │ ┌────────────┴────────────┐ │ │ ▼ ▼ ┌─────────────┐ ┌─────────────┐ │ Commands │ │ Queries │ │ API │ │ API │ └──────┬──────┘ └──────┬──────┘ │ │ ▼ ▼ ┌─────────────┐ ┌─────────────┐ │ Command │ │ Query │ │ Handlers │ │ Handlers │ └──────┬──────┘ └──────┬──────┘ │ │ ▼ ▼ ┌─────────────┐ ┌─────────────┐ │ Write │─────────►│ Read │ │ Model │ Events │ Model │ └─────────────┘ └─────────────┘ ``` ### 2. Key Components | Component | Responsibility | | ------------------- | ------------------------------- | | **Command** | Intent to change state | | **Command Handler** | Validates and executes commands | | **Event** | Record of state change | | **Query** | Request for data | | **Query Handler** | Retrieves data from read model | | **Projector** | Updates read model from events | ## Templates ### Template 1: Command Infrastructure ```python from abc import ABC, abstractmethod from dataclasses import dataclass from typing import TypeVar, Generic, Dict, Any, Type from datetime import datetime import uuid # Command base @dataclass class Command: command_id: str = None timestamp: datetime = None def __post_init__(self): self.command_id = self.command_id or str(uuid.uuid4()) self.timestamp = self.timestamp or datetime.utcnow() # Concrete commands @dataclass class CreateOrder(Command): customer_id: str items: list shipping_address: dict @dataclass class AddOrderItem(Command): order_id: str product_id: str quantity: int price: float @dataclass class CancelOrder(Command): order_id: str reason: str # Command handler base T = TypeVar('T', bound=Command) class CommandHandler(ABC, Generic[T]): @abstractmethod async def handle(self, command: T) -> Any: pass # Command bus class CommandBus: def __init__(self): self._handlers: Dict[Type[Command], CommandHandler] = {} def register(self, command_type: Type[Command], handler: CommandHandler): self._handlers[command_type] = handler async def dispatch(self, command: Command) -> Any: handler = self._handlers.get(type(command)) if not handler: raise ValueError(f"No handler for {type(command).__name__}") return await handler.handle(command) # Command handler implementation class CreateOrderHandler(CommandHandler[CreateOrder]): def __init__(self, order_repository, event_store): self.order_repository = order_repository self.event_store = event_store async def handle(self, command: CreateOrder) -> str: # Validate if not command.items: raise ValueError("Order must have at least one item") # Create aggregate order = Order.create( customer_id=command.customer_id, items=command.items, shipping_address=command.shipping_address ) # Persist events await self.event_store.append_events( stream_id=f"Order-{order.id}", stream_type="Order", events=order.uncommitted_events ) return order.id ``` ### Template 2: Query Infrastructure ```python from abc import ABC, abstractmethod from dataclasses import dataclass from typing import TypeVar, Generic, List, Optional # Query base @dataclass class Query: pass # Concrete queries @dataclass class GetOrderById(Query): order_id: str @dataclass class GetCustomerOrders(Query): customer_id: str status: Optional[str] = None page: int = 1 page_size: int = 20 @dataclass class SearchOrders(Query): query: str filters: dict = None sort_by: str = "created_at" sort_order: str = "desc" # Query result types @dataclass class OrderView: order_id: str customer_id: str status: str total_amount: float item_count: int created_at: datetime shipped_at: Optional[datetime] = None @dataclass class PaginatedResult(Generic[T]): items: List[T] total: int page: int page_size: int @property def total_pages(self) -> int: return (self.total + self.page_size - 1) // self.page_size # Query handler base T = TypeVar('T', bound=Query) R = TypeVar('R') class QueryHandler(ABC, Generic[T, R]): @abstractmethod async def handle(self, query: T) -> R: pass # Query bus class QueryBus: def __init__(self): self._handlers: Dict[Type[Query], QueryHandler] = {} def register(self, query_type: Type[Query], handler: QueryHandler): self._handlers[query_type] = handler async def dispatch(self, query: Query) -> Any: handler = self._handlers.get(type(query)) if not handler: raise ValueError(f"No handler for {type(query).__name__}") return await handler.handle(query) # Query handler implementation class GetOrderByIdHandler(QueryHandler[GetOrderById, Optional[OrderView]]): def __init__(self, read_db): self.read_db = read_db async def handle(self, query: GetOrderById) -> Optional[OrderView]: async with self.read_db.acquire() as conn: row = await conn.fetchrow( """ SELECT order_id, customer_id, status, total_amount, item_count, created_at, shipped_at FROM order_views WHERE order_id = $1 """, query.order_id ) if row: return OrderView(**dict(row)) return None class GetCustomerOrdersHandler(QueryHandler[GetCustomerOrders, PaginatedResult[OrderView]]): def __init__(self, read_db): self.read_db = read_db async def handle(self, query: GetCustomerOrders) -> PaginatedResult[OrderView]: async with self.read_db.acquire() as conn: # Build query with optional status filter where_clause = "customer_id = $1" params = [query.customer_id] if query.status: where_clause += " AND status = $2" params.append(query.status) # Get total count total = await conn.fetchval( f"SELECT COUNT(*) FROM order_views WHERE {where_clause}", *params ) # Get paginated results offset = (query.page - 1) * query.page_size rows = await conn.fetch( f""" SELECT order_id, customer_id, status, total_amount, item_count, created_at, shipped_at FROM order_views WHERE {where_clause} ORDER BY created_at DESC LIMIT ${len(params) + 1} OFFSET ${len(params) + 2} """, *params, query.page_size, offset ) return PaginatedResult( items=[OrderView(**dict(row)) for row in rows], total=total, page=query.page, page_size=query.page_size ) ``` ### Template 3: FastAPI CQRS Application ```python from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from typing import List, Optional app = FastAPI() # Request/Response models class CreateOrderRequest(BaseModel): customer_id: str items: List[dict] shipping_address: dict class OrderResponse(BaseModel): order_id: str customer_id: str status: str total_amount: float item_count: int created_at: datetime # Dependency injection def get_command_bus() -> CommandBus: return app.state.command_bus def get_query_bus() -> QueryBus: return app.state.query_bus # Command endpoints (POST, PUT, DELETE) @app.post("/orders", response_model=dict) async def create_order( request: CreateOrderRequest, command_bus: CommandBus = Depends(get_command_bus) ): command = CreateOrder( customer_id=request.customer_id, items=request.items, shipping_address=request.shipping_address ) order_id = await command_bus.dispatch(command) return {"order_id": order_id} @app.post("/orders/{order_id}/items") async def add_item( order_id: str, product_id: str, quantity: int, price: float, command_bus: CommandBus = Depends(get_command_bus) ): command = AddOrderItem( order_id=order_id, product_id=product_id, quantity=quantity, price=price ) await command_bus.dispatch(command) return {"status": "item_added"} @app.delete("/orders/{order_id}") async def cancel_order( order_id: str, reason: str, command_bus: CommandBus = Depends(get_command_bus) ): command = CancelOrder(order_id=order_id, reason=reason) await command_bus.dispatch(command) return {"status": "cancelled"} # Query endpoints (GET) @app.get("/orders/{order_id}", response_model=OrderResponse) async def get_order( order_id: str, query_bus: QueryBus = Depends(get_query_bus) ): query = GetOrderById(order_id=order_id) result = await query_bus.dispatch(query) if not result: raise HTTPException(status_code=404, detail="Order not found") return result @app.get("/customers/{customer_id}/orders") async def get_customer_orders( customer_id: str, status: Optional[str] = None, page: int = 1, page_size: int = 20, query_bus: QueryBus = Depends(get_query_bus) ): query = GetCustomerOrders( customer_id=customer_id, status=status, page=page, page_size=page_size ) return await query_bus.dispatch(query) @app.get("/orders/search") async def search_orders( q: str, sort_by: str = "created_at", query_bus: QueryBus = Depends(get_query_bus) ): query = SearchOrders(query=q, sort_by=sort_by) return await query_bus.dispatch(query) ``` ### Template 4: Read Model Synchronization ```python class ReadModelSynchronizer: """Keeps read models in sync with events.""" def __init__(self, event_store, read_db, projections: List[Projection]): self.event_store = event_store self.read_db = read_db self.projections = {p.name: p for p in projections} async def run(self): """Continuously sync read models.""" while True: for name, projection in self.projections.items(): await self._sync_projection(projection) await asyncio.sleep(0.1) async def _sync_projection(self, projection: Projection): checkpoint = await self._get_checkpoint(projection.name) events = await self.event_store.read_all( from_position=checkpoint, limit=100 ) for event in events: if event.event_type in projection.handles(): try: await projection.apply(event) except Exception as e: # Log error, possibly retry or skip logger.error(f"Projection error: {e}") continue await self._save_checkpoint(projection.name, event.global_position) async def rebuild_projection(self, projection_name: str): """Rebuild a projection from scratch.""" projection = self.projections[projection_name] # Clear existing data await projection.clear() # Reset checkpoint await self._save_checkpoint(projection_name, 0) # Rebuild while True: checkpoint = await self._get_checkpoint(projection_name) events = await self.event_store.read_all(checkpoint, 1000) if not events: break for event in events: if event.event_type in projection.handles(): await projection.apply(event) await self._save_checkpoint( projection_name, events[-1].global_position ) ``` ### Template 5: Eventual Consistency Handling ```python class ConsistentQueryHandler: """Query handler that can wait for consistency.""" def __init__(self, read_db, event_store): self.read_db = read_db self.event_store = event_store async def query_after_command( self, query: Query, expected_version: int, stream_id: str, timeout: float = 5.0 ): """ Execute query, ensuring read model is at expected version. Used for read-your-writes consistency. """ start_time = time.time() while time.time() - start_time < timeout: # Check if read model is caught up projection_version = await self._get_projection_version(stream_id) if projection_version >= expected_version: return await self.execute_query(query) # Wait a bit and retry await asyncio.sleep(0.1) # Timeout - return stale data with warning return { "data": await self.execute_query(query), "_warning": "Data may be stale" } async def _get_projection_version(self, stream_id: str) -> int: """Get the last processed event version for a stream.""" async with self.read_db.acquire() as conn: return await conn.fetchval( "SELECT last_event_version FROM projection_state WHERE stream_id = $1", stream_id ) or 0 ``` ## Best Practices ### Do's - **Separate command and query models** - Different needs - **Use eventual consistency** - Accept propagation delay - **Validate in command handlers** - Before state change - **Denormalize read models** - Optimize for queries - **Version your events** - For schema evolution ### Don'ts - **Don't query in commands** - Use only for writes - **Don't couple read/write schemas** - Independent evolution - **Don't over-engineer** - Start simple - **Don't ignore consistency SLAs** - Define acceptable lag ## Resources - [CQRS Pattern](https://martinfowler.com/bliki/CQRS.html) - [Microsoft CQRS Guidance](https://docs.microsoft.com/en-us/azure/architecture/patterns/cqrs)
👍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
🤖system prompt•7 months ago

microservices-patterns

Design microservices architectures with service boundaries,

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

projection-patterns

Build read models and projections from event streams. Use when

coding
⭐1
# Projection Patterns Comprehensive guide to building projections and read models for event-sourced systems. ## When to Use This Skill - Building CQRS read models - Creating materialized views from events - Optimizing query performance - Implementing real-time dashboards - Building search indexes from events - Aggregating data across streams ## Core Concepts ### 1. Projection Architecture ``` ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ Event Store │────►│ Projector │────►│ Read Model │ │ │ │ │ │ (Database) │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ ┌─────────┐ │ │ │ Events │ │ │ │ Handler │ │ │ │ Tables │ │ │ └─────────┘ │ │ │ Logic │ │ │ │ Views │ │ │ │ │ └─────────┘ │ │ │ Cache │ │ └─────────────┘ └─────────────┘ └─────────────┘ ``` ### 2. Projection Types | Type | Description | Use Case | | -------------- | --------------------------- | ---------------------- | | **Live** | Real-time from subscription | Current state queries | | **Catchup** | Process historical events | Rebuilding read models | | **Persistent** | Stores checkpoint | Resume after restart | | **Inline** | Same transaction as write | Strong consistency | ## Templates ### Template 1: Basic Projector ```python from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Dict, Any, Callable, List import asyncpg @dataclass class Event: stream_id: str event_type: str data: dict version: int global_position: int class Projection(ABC): """Base class for projections.""" @property @abstractmethod def name(self) -> str: """Unique projection name for checkpointing.""" pass @abstractmethod def handles(self) -> List[str]: """List of event types this projection handles.""" pass @abstractmethod async def apply(self, event: Event) -> None: """Apply event to the read model.""" pass class Projector: """Runs projections from event store.""" def __init__(self, event_store, checkpoint_store): self.event_store = event_store self.checkpoint_store = checkpoint_store self.projections: List[Projection] = [] def register(self, projection: Projection): self.projections.append(projection) async def run(self, batch_size: int = 100): """Run all projections continuously.""" while True: for projection in self.projections: await self._run_projection(projection, batch_size) await asyncio.sleep(0.1) async def _run_projection(self, projection: Projection, batch_size: int): checkpoint = await self.checkpoint_store.get(projection.name) position = checkpoint or 0 events = await self.event_store.read_all(position, batch_size) for event in events: if event.event_type in projection.handles(): await projection.apply(event) await self.checkpoint_store.save( projection.name, event.global_position ) async def rebuild(self, projection: Projection): """Rebuild a projection from scratch.""" await self.checkpoint_store.delete(projection.name) # Optionally clear read model tables await self._run_projection(projection, batch_size=1000) ``` ### Template 2: Order Summary Projection ```python class OrderSummaryProjection(Projection): """Projects order events to a summary read model.""" def __init__(self, db_pool: asyncpg.Pool): self.pool = db_pool @property def name(self) -> str: return "order_summary" def handles(self) -> List[str]: return [ "OrderCreated", "OrderItemAdded", "OrderItemRemoved", "OrderShipped", "OrderCompleted", "OrderCancelled" ] async def apply(self, event: Event) -> None: handlers = { "OrderCreated": self._handle_created, "OrderItemAdded": self._handle_item_added, "OrderItemRemoved": self._handle_item_removed, "OrderShipped": self._handle_shipped, "OrderCompleted": self._handle_completed, "OrderCancelled": self._handle_cancelled, } handler = handlers.get(event.event_type) if handler: await handler(event) async def _handle_created(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ INSERT INTO order_summaries (order_id, customer_id, status, total_amount, item_count, created_at) VALUES ($1, $2, $3, $4, $5, $6) """, event.data['order_id'], event.data['customer_id'], 'pending', 0, 0, event.data['created_at'] ) async def _handle_item_added(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET total_amount = total_amount + $2, item_count = item_count + 1, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['price'] * event.data['quantity'] ) async def _handle_item_removed(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET total_amount = total_amount - $2, item_count = item_count - 1, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['price'] * event.data['quantity'] ) async def _handle_shipped(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET status = 'shipped', shipped_at = $2, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['shipped_at'] ) async def _handle_completed(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET status = 'completed', completed_at = $2, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['completed_at'] ) async def _handle_cancelled(self, event: Event): async with self.pool.acquire() as conn: await conn.execute( """ UPDATE order_summaries SET status = 'cancelled', cancelled_at = $2, cancellation_reason = $3, updated_at = NOW() WHERE order_id = $1 """, event.data['order_id'], event.data['cancelled_at'], event.data.get('reason') ) ``` ### Template 3: Elasticsearch Search Projection ```python from elasticsearch import AsyncElasticsearch class ProductSearchProjection(Projection): """Projects product events to Elasticsearch for full-text search.""" def __init__(self, es_client: AsyncElasticsearch): self.es = es_client self.index = "products" @property def name(self) -> str: return "product_search" def handles(self) -> List[str]: return [ "ProductCreated", "ProductUpdated", "ProductPriceChanged", "ProductDeleted" ] async def apply(self, event: Event) -> None: if event.event_type == "ProductCreated": await self.es.index( index=self.index, id=event.data['product_id'], document={ 'name': event.data['name'], 'description': event.data['description'], 'category': event.data['category'], 'price': event.data['price'], 'tags': event.data.get('tags', []), 'created_at': event.data['created_at'] } ) elif event.event_type == "ProductUpdated": await self.es.update( index=self.index, id=event.data['product_id'], doc={ 'name': event.data['name'], 'description': event.data['description'], 'category': event.data['category'], 'tags': event.data.get('tags', []), 'updated_at': event.data['updated_at'] } ) elif event.event_type == "ProductPriceChanged": await self.es.update( index=self.index, id=event.data['product_id'], doc={ 'price': event.data['new_price'], 'price_updated_at': event.data['changed_at'] } ) elif event.event_type == "ProductDeleted": await self.es.delete( index=self.index, id=event.data['product_id'] ) ``` ### Template 4: Aggregating Projection ```python class DailySalesProjection(Projection): """Aggregates sales data by day for reporting.""" def __init__(self, db_pool: asyncpg.Pool): self.pool = db_pool @property def name(self) -> str: return "daily_sales" def handles(self) -> List[str]: return ["OrderCompleted", "OrderRefunded"] async def apply(self, event: Event) -> None: if event.event_type == "OrderCompleted": await self._increment_sales(event) elif event.event_type == "OrderRefunded": await self._decrement_sales(event) async def _increment_sales(self, event: Event): date = event.data['completed_at'][:10] # YYYY-MM-DD async with self.pool.acquire() as conn: await conn.execute( """ INSERT INTO daily_sales (date, total_orders, total_revenue, total_items) VALUES ($1, 1, $2, $3) ON CONFLICT (date) DO UPDATE SET total_orders = daily_sales.total_orders + 1, total_revenue = daily_sales.total_revenue + $2, total_items = daily_sales.total_items + $3, updated_at = NOW() """, date, event.data['total_amount'], event.data['item_count'] ) async def _decrement_sales(self, event: Event): date = event.data['original_completed_at'][:10] async with self.pool.acquire() as conn: await conn.execute( """ UPDATE daily_sales SET total_orders = total_orders - 1, total_revenue = total_revenue - $2, total_refunds = total_refunds + $2, updated_at = NOW() WHERE date = $1 """, date, event.data['refund_amount'] ) ``` ### Template 5: Multi-Table Projection ```python class CustomerActivityProjection(Projection): """Projects customer activity across multiple tables.""" def __init__(self, db_pool: asyncpg.Pool): self.pool = db_pool @property def name(self) -> str: return "customer_activity" def handles(self) -> List[str]: return [ "CustomerCreated", "OrderCompleted", "ReviewSubmitted", "CustomerTierChanged" ] async def apply(self, event: Event) -> None: async with self.pool.acquire() as conn: async with conn.transaction(): if event.event_type == "CustomerCreated": # Insert into customers table await conn.execute( """ INSERT INTO customers (customer_id, email, name, tier, created_at) VALUES ($1, $2, $3, 'bronze', $4) """, event.data['customer_id'], event.data['email'], event.data['name'], event.data['created_at'] ) # Initialize activity summary await conn.execute( """ INSERT INTO customer_activity_summary (customer_id, total_orders, total_spent, total_reviews) VALUES ($1, 0, 0, 0) """, event.data['customer_id'] ) elif event.event_type == "OrderCompleted": # Update activity summary await conn.execute( """ UPDATE customer_activity_summary SET total_orders = total_orders + 1, total_spent = total_spent + $2, last_order_at = $3 WHERE customer_id = $1 """, event.data['customer_id'], event.data['total_amount'], event.data['completed_at'] ) # Insert into order history await conn.execute( """ INSERT INTO customer_order_history (customer_id, order_id, amount, completed_at) VALUES ($1, $2, $3, $4) """, event.data['customer_id'], event.data['order_id'], event.data['total_amount'], event.data['completed_at'] ) elif event.event_type == "ReviewSubmitted": await conn.execute( """ UPDATE customer_activity_summary SET total_reviews = total_reviews + 1, last_review_at = $2 WHERE customer_id = $1 """, event.data['customer_id'], event.data['submitted_at'] ) elif event.event_type == "CustomerTierChanged": await conn.execute( """ UPDATE customers SET tier = $2, updated_at = NOW() WHERE customer_id = $1 """, event.data['customer_id'], event.data['new_tier'] ) ``` ## Best Practices ### Do's - **Make projections idempotent** - Safe to replay - **Use transactions** - For multi-table updates - **Store checkpoints** - Resume after failures - **Monitor lag** - Alert on projection delays - **Plan for rebuilds** - Design for reconstruction ### Don'ts - **Don't couple projections** - Each is independent - **Don't skip error handling** - Log and alert on failures - **Don't ignore ordering** - Events must be processed in order - **Don't over-normalize** - Denormalize for query patterns ## Resources - [CQRS Pattern](https://docs.microsoft.com/en-us/azure/architecture/patterns/cqrs) - [Projection Building Blocks](https://zimarev.com/blog/event-sourcing/projections/)
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

saga-orchestration

Implement saga patterns for distributed transactions and

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

temporal-python-testing

Test Temporal workflows with pytest, time-skipping, and mocking

coding
⭐1
# Temporal Python Testing Strategies Comprehensive testing approaches for Temporal workflows using pytest, progressive disclosure resources for specific testing scenarios. ## When to Use This Skill - **Unit testing workflows** - Fast tests with time-skipping - **Integration testing** - Workflows with mocked activities - **Replay testing** - Validate determinism against production histories - **Local development** - Set up Temporal server and pytest - **CI/CD integration** - Automated testing pipelines - **Coverage strategies** - Achieve ≥80% test coverage ## Testing Philosophy **Recommended Approach** (Source: docs.temporal.io/develop/python/testing-suite): - Write majority as integration tests - Use pytest with async fixtures - Time-skipping enables fast feedback (month-long workflows → seconds) - Mock activities to isolate workflow logic - Validate determinism with replay testing **Three Test Types**: 1. **Unit**: Workflows with time-skipping, activities with ActivityEnvironment 2. **Integration**: Workers with mocked activities 3. **End-to-end**: Full Temporal server with real activities (use sparingly) ## Available Resources This skill provides detailed guidance through progressive disclosure. Load specific resources based on your testing needs: ### Unit Testing Resources **File**: `resources/unit-testing.md` **When to load**: Testing individual workflows or activities in isolation **Contains**: - WorkflowEnvironment with time-skipping - ActivityEnvironment for activity testing - Fast execution of long-running workflows - Manual time advancement patterns - pytest fixtures and patterns ### Integration Testing Resources **File**: `resources/integration-testing.md` **When to load**: Testing workflows with mocked external dependencies **Contains**: - Activity mocking strategies - Error injection patterns - Multi-activity workflow testing - Signal and query testing - Coverage strategies ### Replay Testing Resources **File**: `resources/replay-testing.md` **When to load**: Validating determinism or deploying workflow changes **Contains**: - Determinism validation - Production history replay - CI/CD integration patterns - Version compatibility testing ### Local Development Resources **File**: `resources/local-setup.md` **When to load**: Setting up development environment **Contains**: - Docker Compose configuration - pytest setup and configuration - Coverage tool integration - Development workflow ## Quick Start Guide ### Basic Workflow Test ```python import pytest from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker @pytest.fixture async def workflow_env(): env = await WorkflowEnvironment.start_time_skipping() yield env await env.shutdown() @pytest.mark.asyncio async def test_workflow(workflow_env): async with Worker( workflow_env.client, task_queue="test-queue", workflows=[YourWorkflow], activities=[your_activity], ): result = await workflow_env.client.execute_workflow( YourWorkflow.run, args, id="test-wf-id", task_queue="test-queue", ) assert result == expected ``` ### Basic Activity Test ```python from temporalio.testing import ActivityEnvironment async def test_activity(): env = ActivityEnvironment() result = await env.run(your_activity, "test-input") assert result == expected_output ``` ## Coverage Targets **Recommended Coverage** (Source: docs.temporal.io best practices): - **Workflows**: ≥80% logic coverage - **Activities**: ≥80% logic coverage - **Integration**: Critical paths with mocked activities - **Replay**: All workflow versions before deployment ## Key Testing Principles 1. **Time-Skipping** - Month-long workflows test in seconds 2. **Mock Activities** - Isolate workflow logic from external dependencies 3. **Replay Testing** - Validate determinism before deployment 4. **High Coverage** - ≥80% target for production workflows 5. **Fast Feedback** - Unit tests run in milliseconds ## How to Use Resources **Load specific resource when needed**: - "Show me unit testing patterns" → Load `resources/unit-testing.md` - "How do I mock activities?" → Load `resources/integration-testing.md` - "Setup local Temporal server" → Load `resources/local-setup.md` - "Validate determinism" → Load `resources/replay-testing.md` ## Additional References - Python SDK Testing: docs.temporal.io/develop/python/testing-suite - Testing Patterns: github.com/temporalio/temporal/blob/main/docs/development/testing.md - Python Samples: github.com/temporalio/samples-python
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

workflow-orchestration-patterns

Design durable workflows with Temporal for distributed systems.

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

defi-protocol-templates

Implement DeFi protocols with production-ready templates for

coding
⭐1
# DeFi Protocol Templates Production-ready templates for common DeFi protocols including staking, AMMs, governance, lending, and flash loans. ## When to Use This Skill - Building staking platforms with reward distribution - Implementing AMM (Automated Market Maker) protocols - Creating governance token systems - Developing lending/borrowing protocols - Integrating flash loan functionality - Launching yield farming platforms ## Staking Contract ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/IERC20.sol"; import "@openzeppelin/contracts/security/ReentrancyGuard.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; contract StakingRewards is ReentrancyGuard, Ownable { IERC20 public stakingToken; IERC20 public rewardsToken; uint256 public rewardRate = 100; // Rewards per second uint256 public lastUpdateTime; uint256 public rewardPerTokenStored; mapping(address => uint256) public userRewardPerTokenPaid; mapping(address => uint256) public rewards; mapping(address => uint256) public balances; uint256 private _totalSupply; event Staked(address indexed user, uint256 amount); event Withdrawn(address indexed user, uint256 amount); event RewardPaid(address indexed user, uint256 reward); constructor(address _stakingToken, address _rewardsToken) { stakingToken = IERC20(_stakingToken); rewardsToken = IERC20(_rewardsToken); } modifier updateReward(address account) { rewardPerTokenStored = rewardPerToken(); lastUpdateTime = block.timestamp; if (account != address(0)) { rewards[account] = earned(account); userRewardPerTokenPaid[account] = rewardPerTokenStored; } _; } function rewardPerToken() public view returns (uint256) { if (_totalSupply == 0) { return rewardPerTokenStored; } return rewardPerTokenStored + ((block.timestamp - lastUpdateTime) * rewardRate * 1e18) / _totalSupply; } function earned(address account) public view returns (uint256) { return (balances[account] * (rewardPerToken() - userRewardPerTokenPaid[account])) / 1e18 + rewards[account]; } function stake(uint256 amount) external nonReentrant updateReward(msg.sender) { require(amount > 0, "Cannot stake 0"); _totalSupply += amount; balances[msg.sender] += amount; stakingToken.transferFrom(msg.sender, address(this), amount); emit Staked(msg.sender, amount); } function withdraw(uint256 amount) public nonReentrant updateReward(msg.sender) { require(amount > 0, "Cannot withdraw 0"); _totalSupply -= amount; balances[msg.sender] -= amount; stakingToken.transfer(msg.sender, amount); emit Withdrawn(msg.sender, amount); } function getReward() public nonReentrant updateReward(msg.sender) { uint256 reward = rewards[msg.sender]; if (reward > 0) { rewards[msg.sender] = 0; rewardsToken.transfer(msg.sender, reward); emit RewardPaid(msg.sender, reward); } } function exit() external { withdraw(balances[msg.sender]); getReward(); } } ``` ## AMM (Automated Market Maker) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/IERC20.sol"; contract SimpleAMM { IERC20 public token0; IERC20 public token1; uint256 public reserve0; uint256 public reserve1; uint256 public totalSupply; mapping(address => uint256) public balanceOf; event Mint(address indexed to, uint256 amount); event Burn(address indexed from, uint256 amount); event Swap(address indexed trader, uint256 amount0In, uint256 amount1In, uint256 amount0Out, uint256 amount1Out); constructor(address _token0, address _token1) { token0 = IERC20(_token0); token1 = IERC20(_token1); } function addLiquidity(uint256 amount0, uint256 amount1) external returns (uint256 shares) { token0.transferFrom(msg.sender, address(this), amount0); token1.transferFrom(msg.sender, address(this), amount1); if (totalSupply == 0) { shares = sqrt(amount0 * amount1); } else { shares = min( (amount0 * totalSupply) / reserve0, (amount1 * totalSupply) / reserve1 ); } require(shares > 0, "Shares = 0"); _mint(msg.sender, shares); _update( token0.balanceOf(address(this)), token1.balanceOf(address(this)) ); emit Mint(msg.sender, shares); } function removeLiquidity(uint256 shares) external returns (uint256 amount0, uint256 amount1) { uint256 bal0 = token0.balanceOf(address(this)); uint256 bal1 = token1.balanceOf(address(this)); amount0 = (shares * bal0) / totalSupply; amount1 = (shares * bal1) / totalSupply; require(amount0 > 0 && amount1 > 0, "Amount0 or amount1 = 0"); _burn(msg.sender, shares); _update(bal0 - amount0, bal1 - amount1); token0.transfer(msg.sender, amount0); token1.transfer(msg.sender, amount1); emit Burn(msg.sender, shares); } function swap(address tokenIn, uint256 amountIn) external returns (uint256 amountOut) { require(tokenIn == address(token0) || tokenIn == address(token1), "Invalid token"); bool isToken0 = tokenIn == address(token0); (IERC20 tokenIn_, IERC20 tokenOut, uint256 resIn, uint256 resOut) = isToken0 ? (token0, token1, reserve0, reserve1) : (token1, token0, reserve1, reserve0); tokenIn_.transferFrom(msg.sender, address(this), amountIn); // 0.3% fee uint256 amountInWithFee = (amountIn * 997) / 1000; amountOut = (resOut * amountInWithFee) / (resIn + amountInWithFee); tokenOut.transfer(msg.sender, amountOut); _update( token0.balanceOf(address(this)), token1.balanceOf(address(this)) ); emit Swap(msg.sender, isToken0 ? amountIn : 0, isToken0 ? 0 : amountIn, isToken0 ? 0 : amountOut, isToken0 ? amountOut : 0); } function _mint(address to, uint256 amount) private { balanceOf[to] += amount; totalSupply += amount; } function _burn(address from, uint256 amount) private { balanceOf[from] -= amount; totalSupply -= amount; } function _update(uint256 res0, uint256 res1) private { reserve0 = res0; reserve1 = res1; } function sqrt(uint256 y) private pure returns (uint256 z) { if (y > 3) { z = y; uint256 x = y / 2 + 1; while (x < z) { z = x; x = (y / x + x) / 2; } } else if (y != 0) { z = 1; } } function min(uint256 x, uint256 y) private pure returns (uint256) { return x <= y ? x : y; } } ``` ## Governance Token ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/extensions/ERC20Votes.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; contract GovernanceToken is ERC20Votes, Ownable { constructor() ERC20("Governance Token", "GOV") ERC20Permit("Governance Token") { _mint(msg.sender, 1000000 * 10**decimals()); } function _afterTokenTransfer( address from, address to, uint256 amount ) internal override(ERC20Votes) { super._afterTokenTransfer(from, to, amount); } function _mint(address to, uint256 amount) internal override(ERC20Votes) { super._mint(to, amount); } function _burn(address account, uint256 amount) internal override(ERC20Votes) { super._burn(account, amount); } } contract Governor is Ownable { GovernanceToken public governanceToken; struct Proposal { uint256 id; address proposer; string description; uint256 forVotes; uint256 againstVotes; uint256 startBlock; uint256 endBlock; bool executed; mapping(address => bool) hasVoted; } uint256 public proposalCount; mapping(uint256 => Proposal) public proposals; uint256 public votingPeriod = 17280; // ~3 days in blocks uint256 public proposalThreshold = 100000 * 10**18; event ProposalCreated(uint256 indexed proposalId, address proposer, string description); event VoteCast(address indexed voter, uint256 indexed proposalId, bool support, uint256 weight); event ProposalExecuted(uint256 indexed proposalId); constructor(address _governanceToken) { governanceToken = GovernanceToken(_governanceToken); } function propose(string memory description) external returns (uint256) { require( governanceToken.getPastVotes(msg.sender, block.number - 1) >= proposalThreshold, "Proposer votes below threshold" ); proposalCount++; Proposal storage newProposal = proposals[proposalCount]; newProposal.id = proposalCount; newProposal.proposer = msg.sender; newProposal.description = description; newProposal.startBlock = block.number; newProposal.endBlock = block.number + votingPeriod; emit ProposalCreated(proposalCount, msg.sender, description); return proposalCount; } function vote(uint256 proposalId, bool support) external { Proposal storage proposal = proposals[proposalId]; require(block.number >= proposal.startBlock, "Voting not started"); require(block.number <= proposal.endBlock, "Voting ended"); require(!proposal.hasVoted[msg.sender], "Already voted"); uint256 weight = governanceToken.getPastVotes(msg.sender, proposal.startBlock); require(weight > 0, "No voting power"); proposal.hasVoted[msg.sender] = true; if (support) { proposal.forVotes += weight; } else { proposal.againstVotes += weight; } emit VoteCast(msg.sender, proposalId, support, weight); } function execute(uint256 proposalId) external { Proposal storage proposal = proposals[proposalId]; require(block.number > proposal.endBlock, "Voting not ended"); require(!proposal.executed, "Already executed"); require(proposal.forVotes > proposal.againstVotes, "Proposal failed"); proposal.executed = true; // Execute proposal logic here emit ProposalExecuted(proposalId); } } ``` ## Flash Loan ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC20/IERC20.sol"; interface IFlashLoanReceiver { function executeOperation( address asset, uint256 amount, uint256 fee, bytes calldata params ) external returns (bool); } contract FlashLoanProvider { IERC20 public token; uint256 public feePercentage = 9; // 0.09% fee event FlashLoan(address indexed borrower, uint256 amount, uint256 fee); constructor(address _token) { token = IERC20(_token); } function flashLoan( address receiver, uint256 amount, bytes calldata params ) external { uint256 balanceBefore = token.balanceOf(address(this)); require(balanceBefore >= amount, "Insufficient liquidity"); uint256 fee = (amount * feePercentage) / 10000; // Send tokens to receiver token.transfer(receiver, amount); // Execute callback require( IFlashLoanReceiver(receiver).executeOperation( address(token), amount, fee, params ), "Flash loan failed" ); // Verify repayment uint256 balanceAfter = token.balanceOf(address(this)); require(balanceAfter >= balanceBefore + fee, "Flash loan not repaid"); emit FlashLoan(receiver, amount, fee); } } // Example flash loan receiver contract FlashLoanReceiver is IFlashLoanReceiver { function executeOperation( address asset, uint256 amount, uint256 fee, bytes calldata params ) external override returns (bool) { // Decode params and execute arbitrage, liquidation, etc. // ... // Approve repayment IERC20(asset).approve(msg.sender, amount + fee); return true; } } ``` ## Resources - **references/staking.md**: Staking mechanics and reward distribution - **references/liquidity-pools.md**: AMM mathematics and pricing - **references/governance-tokens.md**: Governance and voting systems - **references/lending-protocols.md**: Lending/borrowing implementation - **references/flash-loans.md**: Flash loan security and use cases - **assets/staking-contract.sol**: Production staking template - **assets/amm-contract.sol**: Full AMM implementation - **assets/governance-token.sol**: Governance system - **assets/lending-protocol.sol**: Lending platform template ## Best Practices 1. **Use Established Libraries**: OpenZeppelin, Solmate 2. **Test Thoroughly**: Unit tests, integration tests, fuzzing 3. **Audit Before Launch**: Professional security audits 4. **Start Simple**: MVP first, add features incrementally 5. **Monitor**: Track contract health and user activity 6. **Upgradability**: Consider proxy patterns for upgrades 7. **Emergency Controls**: Pause mechanisms for critical issues ## Common DeFi Patterns - **Time-Weighted Average Price (TWAP)**: Price oracle resistance - **Liquidity Mining**: Incentivize liquidity provision - **Vesting**: Lock tokens with gradual release - **Multisig**: Require multiple signatures for critical operations - **Timelocks**: Delay execution of governance decisions
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

nft-standards

Implement NFT standards (ERC-721, ERC-1155) with proper metadata

coding
⭐1
# NFT Standards Master ERC-721 and ERC-1155 NFT standards, metadata best practices, and advanced NFT features. ## When to Use This Skill - Creating NFT collections (art, gaming, collectibles) - Implementing marketplace functionality - Building on-chain or off-chain metadata - Creating soulbound tokens (non-transferable) - Implementing royalties and revenue sharing - Developing dynamic/evolving NFTs ## ERC-721 (Non-Fungible Token Standard) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC721/extensions/ERC721URIStorage.sol"; import "@openzeppelin/contracts/token/ERC721/extensions/ERC721Enumerable.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; import "@openzeppelin/contracts/utils/Counters.sol"; contract MyNFT is ERC721URIStorage, ERC721Enumerable, Ownable { using Counters for Counters.Counter; Counters.Counter private _tokenIds; uint256 public constant MAX_SUPPLY = 10000; uint256 public constant MINT_PRICE = 0.08 ether; uint256 public constant MAX_PER_MINT = 20; constructor() ERC721("MyNFT", "MNFT") {} function mint(uint256 quantity) external payable { require(quantity > 0 && quantity <= MAX_PER_MINT, "Invalid quantity"); require(_tokenIds.current() + quantity <= MAX_SUPPLY, "Exceeds max supply"); require(msg.value >= MINT_PRICE * quantity, "Insufficient payment"); for (uint256 i = 0; i < quantity; i++) { _tokenIds.increment(); uint256 newTokenId = _tokenIds.current(); _safeMint(msg.sender, newTokenId); _setTokenURI(newTokenId, generateTokenURI(newTokenId)); } } function generateTokenURI(uint256 tokenId) internal pure returns (string memory) { // Return IPFS URI or on-chain metadata return string(abi.encodePacked("ipfs://QmHash/", Strings.toString(tokenId), ".json")); } // Required overrides function _beforeTokenTransfer( address from, address to, uint256 tokenId, uint256 batchSize ) internal override(ERC721, ERC721Enumerable) { super._beforeTokenTransfer(from, to, tokenId, batchSize); } function _burn(uint256 tokenId) internal override(ERC721, ERC721URIStorage) { super._burn(tokenId); } function tokenURI(uint256 tokenId) public view override(ERC721, ERC721URIStorage) returns (string memory) { return super.tokenURI(tokenId); } function supportsInterface(bytes4 interfaceId) public view override(ERC721, ERC721Enumerable) returns (bool) { return super.supportsInterface(interfaceId); } function withdraw() external onlyOwner { payable(owner()).transfer(address(this).balance); } } ``` ## ERC-1155 (Multi-Token Standard) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "@openzeppelin/contracts/token/ERC1155/ERC1155.sol"; import "@openzeppelin/contracts/access/Ownable.sol"; contract GameItems is ERC1155, Ownable { uint256 public constant SWORD = 1; uint256 public constant SHIELD = 2; uint256 public constant POTION = 3; mapping(uint256 => uint256) public tokenSupply; mapping(uint256 => uint256) public maxSupply; constructor() ERC1155("ipfs://QmBaseHash/{id}.json") { maxSupply[SWORD] = 1000; maxSupply[SHIELD] = 500; maxSupply[POTION] = 10000; } function mint( address to, uint256 id, uint256 amount ) external onlyOwner { require(tokenSupply[id] + amount <= maxSupply[id], "Exceeds max supply"); _mint(to, id, amount, ""); tokenSupply[id] += amount; } function mintBatch( address to, uint256[] memory ids, uint256[] memory amounts ) external onlyOwner { for (uint256 i = 0; i < ids.length; i++) { require(tokenSupply[ids[i]] + amounts[i] <= maxSupply[ids[i]], "Exceeds max supply"); tokenSupply[ids[i]] += amounts[i]; } _mintBatch(to, ids, amounts, ""); } function burn( address from, uint256 id, uint256 amount ) external { require(from == msg.sender || isApprovedForAll(from, msg.sender), "Not authorized"); _burn(from, id, amount); tokenSupply[id] -= amount; } } ``` ## Metadata Standards ### Off-Chain Metadata (IPFS) ```json { "name": "NFT #1", "description": "Description of the NFT", "image": "ipfs://QmImageHash", "attributes": [ { "trait_type": "Background", "value": "Blue" }, { "trait_type": "Rarity", "value": "Legendary" }, { "trait_type": "Power", "value": 95, "display_type": "number", "max_value": 100 } ] } ``` ### On-Chain Metadata ```solidity contract OnChainNFT is ERC721 { struct Traits { uint8 background; uint8 body; uint8 head; uint8 rarity; } mapping(uint256 => Traits) public tokenTraits; function tokenURI(uint256 tokenId) public view override returns (string memory) { Traits memory traits = tokenTraits[tokenId]; string memory json = Base64.encode( bytes( string( abi.encodePacked( '{"name": "NFT #', Strings.toString(tokenId), '",', '"description": "On-chain NFT",', '"image": "data:image/svg+xml;base64,', generateSVG(traits), '",', '"attributes": [', '{"trait_type": "Background", "value": "', Strings.toString(traits.background), '"},', '{"trait_type": "Rarity", "value": "', getRarityName(traits.rarity), '"}', ']}' ) ) ) ); return string(abi.encodePacked("data:application/json;base64,", json)); } function generateSVG(Traits memory traits) internal pure returns (string memory) { // Generate SVG based on traits return "..."; } } ``` ## Royalties (EIP-2981) ```solidity import "@openzeppelin/contracts/interfaces/IERC2981.sol"; contract NFTWithRoyalties is ERC721, IERC2981 { address public royaltyRecipient; uint96 public royaltyFee = 500; // 5% constructor() ERC721("Royalty NFT", "RNFT") { royaltyRecipient = msg.sender; } function royaltyInfo(uint256 tokenId, uint256 salePrice) external view override returns (address receiver, uint256 royaltyAmount) { return (royaltyRecipient, (salePrice * royaltyFee) / 10000); } function setRoyalty(address recipient, uint96 fee) external onlyOwner { require(fee <= 1000, "Royalty fee too high"); // Max 10% royaltyRecipient = recipient; royaltyFee = fee; } function supportsInterface(bytes4 interfaceId) public view override(ERC721, IERC165) returns (bool) { return interfaceId == type(IERC2981).interfaceId || super.supportsInterface(interfaceId); } } ``` ## Soulbound Tokens (Non-Transferable) ```solidity contract SoulboundToken is ERC721 { constructor() ERC721("Soulbound", "SBT") {} function _beforeTokenTransfer( address from, address to, uint256 tokenId, uint256 batchSize ) internal virtual override { require(from == address(0) || to == address(0), "Token is soulbound"); super._beforeTokenTransfer(from, to, tokenId, batchSize); } function mint(address to) external { uint256 tokenId = totalSupply() + 1; _safeMint(to, tokenId); } // Burn is allowed (user can destroy their SBT) function burn(uint256 tokenId) external { require(ownerOf(tokenId) == msg.sender, "Not token owner"); _burn(tokenId); } } ``` ## Dynamic NFTs ```solidity contract DynamicNFT is ERC721 { struct TokenState { uint256 level; uint256 experience; uint256 lastUpdated; } mapping(uint256 => TokenState) public tokenStates; function gainExperience(uint256 tokenId, uint256 exp) external { require(ownerOf(tokenId) == msg.sender, "Not token owner"); TokenState storage state = tokenStates[tokenId]; state.experience += exp; // Level up logic if (state.experience >= state.level * 100) { state.level++; } state.lastUpdated = block.timestamp; } function tokenURI(uint256 tokenId) public view override returns (string memory) { TokenState memory state = tokenStates[tokenId]; // Generate metadata based on current state return generateMetadata(tokenId, state); } function generateMetadata(uint256 tokenId, TokenState memory state) internal pure returns (string memory) { // Dynamic metadata generation return ""; } } ``` ## Gas-Optimized Minting (ERC721A) ```solidity import "erc721a/contracts/ERC721A.sol"; contract OptimizedNFT is ERC721A { uint256 public constant MAX_SUPPLY = 10000; uint256 public constant MINT_PRICE = 0.05 ether; constructor() ERC721A("Optimized NFT", "ONFT") {} function mint(uint256 quantity) external payable { require(_totalMinted() + quantity <= MAX_SUPPLY, "Exceeds max supply"); require(msg.value >= MINT_PRICE * quantity, "Insufficient payment"); _mint(msg.sender, quantity); } function _baseURI() internal pure override returns (string memory) { return "ipfs://QmBaseHash/"; } } ``` ## Resources - **references/erc721.md**: ERC-721 specification details - **references/erc1155.md**: ERC-1155 multi-token standard - **references/metadata-standards.md**: Metadata best practices - **references/enumeration.md**: Token enumeration patterns - **assets/erc721-contract.sol**: Production ERC-721 template - **assets/erc1155-contract.sol**: Production ERC-1155 template - **assets/metadata-schema.json**: Standard metadata format - **assets/metadata-uploader.py**: IPFS upload utility ## Best Practices 1. **Use OpenZeppelin**: Battle-tested implementations 2. **Pin Metadata**: Use IPFS with pinning service 3. **Implement Royalties**: EIP-2981 for marketplace compatibility 4. **Gas Optimization**: Use ERC721A for batch minting 5. **Reveal Mechanism**: Placeholder → reveal pattern 6. **Enumeration**: Support walletOfOwner for marketplaces 7. **Whitelist**: Merkle trees for efficient whitelisting ## Marketplace Integration - OpenSea: ERC-721/1155, metadata standards - LooksRare: Royalty enforcement - Rarible: Protocol fees, lazy minting - Blur: Gas-optimized trading
👍0
👁️0
🤖 Auto-discovered
🤖system prompt•7 months ago

web3-testing

Test smart contracts comprehensively using Hardhat and Foundry with

coding
⭐1
# Web3 Smart Contract Testing Master comprehensive testing strategies for smart contracts using Hardhat, Foundry, and advanced testing patterns. ## When to Use This Skill - Writing unit tests for smart contracts - Setting up integration test suites - Performing gas optimization testing - Fuzzing for edge cases - Forking mainnet for realistic testing - Automating test coverage reporting - Verifying contracts on Etherscan ## Hardhat Testing Setup ```javascript // hardhat.config.js require("@nomicfoundation/hardhat-toolbox"); require("@nomiclabs/hardhat-etherscan"); require("hardhat-gas-reporter"); require("solidity-coverage"); module.exports = { solidity: { version: "0.8.19", settings: { optimizer: { enabled: true, runs: 200, }, }, }, networks: { hardhat: { forking: { url: process.env.MAINNET_RPC_URL, blockNumber: 15000000, }, }, goerli: { url: process.env.GOERLI_RPC_URL, accounts: [process.env.PRIVATE_KEY], }, }, gasReporter: { enabled: true, currency: "USD", coinmarketcap: process.env.COINMARKETCAP_API_KEY, }, etherscan: { apiKey: process.env.ETHERSCAN_API_KEY, }, }; ``` ## Unit Testing Patterns ```javascript const { expect } = require("chai"); const { ethers } = require("hardhat"); const { loadFixture, time, } = require("@nomicfoundation/hardhat-network-helpers"); describe("Token Contract", function () { // Fixture for test setup async function deployTokenFixture() { const [owner, addr1, addr2] = await ethers.getSigners(); const Token = await ethers.getContractFactory("Token"); const token = await Token.deploy(); return { token, owner, addr1, addr2 }; } describe("Deployment", function () { it("Should set the right owner", async function () { const { token, owner } = await loadFixture(deployTokenFixture); expect(await token.owner()).to.equal(owner.address); }); it("Should assign total supply to owner", async function () { const { token, owner } = await loadFixture(deployTokenFixture); const ownerBalance = await token.balanceOf(owner.address); expect(await token.totalSupply()).to.equal(ownerBalance); }); }); describe("Transactions", function () { it("Should transfer tokens between accounts", async function () { const { token, owner, addr1 } = await loadFixture(deployTokenFixture); await expect(token.transfer(addr1.address, 50)).to.changeTokenBalances( token, [owner, addr1], [-50, 50], ); }); it("Should fail if sender doesn't have enough tokens", async function () { const { token, addr1 } = await loadFixture(deployTokenFixture); const initialBalance = await token.balanceOf(addr1.address); await expect( token.connect(addr1).transfer(owner.address, 1), ).to.be.revertedWith("Insufficient balance"); }); it("Should emit Transfer event", async function () { const { token, owner, addr1 } = await loadFixture(deployTokenFixture); await expect(token.transfer(addr1.address, 50)) .to.emit(token, "Transfer") .withArgs(owner.address, addr1.address, 50); }); }); describe("Time-based tests", function () { it("Should handle time-locked operations", async function () { const { token } = await loadFixture(deployTokenFixture); // Increase time by 1 day await time.increase(86400); // Test time-dependent functionality }); }); describe("Gas optimization", function () { it("Should use gas efficiently", async function () { const { token } = await loadFixture(deployTokenFixture); const tx = await token.transfer(addr1.address, 100); const receipt = await tx.wait(); expect(receipt.gasUsed).to.be.lessThan(50000); }); }); }); ``` ## Foundry Testing (Forge) ```solidity // SPDX-License-Identifier: MIT pragma solidity ^0.8.0; import "forge-std/Test.sol"; import "../src/Token.sol"; contract TokenTest is Test { Token token; address owner = address(1); address user1 = address(2); address user2 = address(3); function setUp() public { vm.prank(owner); token = new Token(); } function testInitialSupply() public { assertEq(token.totalSupply(), 1000000 * 10**18); } function testTransfer() public { vm.prank(owner); token.transfer(user1, 100); assertEq(token.balanceOf(user1), 100); assertEq(token.balanceOf(owner), token.totalSupply() - 100); } function testFailTransferInsufficientBalance() public { vm.prank(user1); token.transfer(user2, 100); // Should fail } function testCannotTransferToZeroAddress() public { vm.prank(owner); vm.expectRevert("Invalid recipient"); token.transfer(address(0), 100); } // Fuzzing test function testFuzzTransfer(uint256 amount) public { vm.assume(amount > 0 && amount <= token.totalSupply()); vm.prank(owner); token.transfer(user1, amount); assertEq(token.balanceOf(user1), amount); } // Test with cheatcodes function testDealAndPrank() public { // Give ETH to address vm.deal(user1, 10 ether); // Impersonate address vm.prank(user1); // Test functionality assertEq(user1.balance, 10 ether); } // Mainnet fork test function testForkMainnet() public { vm.createSelectFork("https://eth-mainnet.alchemyapi.io/v2/..."); // Interact with mainnet contracts address dai = 0x6B175474E89094C44Da98b954EedeAC495271d0F; assertEq(IERC20(dai).symbol(), "DAI"); } } ``` ## Advanced Testing Patterns ### Snapshot and Revert ```javascript describe("Complex State Changes", function () { let snapshotId; beforeEach(async function () { snapshotId = await network.provider.send("evm_snapshot"); }); afterEach(async function () { await network.provider.send("evm_revert", [snapshotId]); }); it("Test 1", async function () { // Make state changes }); it("Test 2", async function () { // State reverted, clean slate }); }); ``` ### Mainnet Forking ```javascript describe("Mainnet Fork Tests", function () { let uniswapRouter, dai, usdc; before(async function () { await network.provider.request({ method: "hardhat_reset", params: [ { forking: { jsonRpcUrl: process.env.MAINNET_RPC_URL, blockNumber: 15000000, }, }, ], }); // Connect to existing mainnet contracts uniswapRouter = await ethers.getContractAt( "IUniswapV2Router", "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", ); dai = await ethers.getContractAt( "IERC20", "0x6B175474E89094C44Da98b954EedeAC495271d0F", ); }); it("Should swap on Uniswap", async function () { // Test with real Uniswap contracts }); }); ``` ### Impersonating Accounts ```javascript it("Should impersonate whale account", async function () { const whaleAddress = "0x..."; await network.provider.request({ method: "hardhat_impersonateAccount", params: [whaleAddress], }); const whale = await ethers.getSigner(whaleAddress); // Use whale's tokens await dai .connect(whale) .transfer(addr1.address, ethers.utils.parseEther("1000")); }); ``` ## Gas Optimization Testing ```javascript const { expect } = require("chai"); describe("Gas Optimization", function () { it("Compare gas usage between implementations", async function () { const Implementation1 = await ethers.getContractFactory("OptimizedContract"); const Implementation2 = await ethers.getContractFactory( "UnoptimizedContract", ); const contract1 = await Implementation1.deploy(); const contract2 = await Implementation2.deploy(); const tx1 = await contract1.doSomething(); const receipt1 = await tx1.wait(); const tx2 = await contract2.doSomething(); const receipt2 = await tx2.wait(); console.log("Optimized gas:", receipt1.gasUsed.toString()); console.log("Unoptimized gas:", receipt2.gasUsed.toString()); expect(receipt1.gasUsed).to.be.lessThan(receipt2.gasUsed); }); }); ``` ## Coverage Reporting ```bash # Generate coverage report npx hardhat coverage # Output shows: # File | % Stmts | % Branch | % Funcs | % Lines | # -------------------|---------|----------|---------|---------| # contracts/Token.sol | 100 | 90 | 100 | 95 | ``` ## Contract Verification ```javascript // Verify on Etherscan await hre.run("verify:verify", { address: contractAddress, constructorArguments: [arg1, arg2], }); ``` ```bash # Or via CLI npx hardhat verify --network mainnet CONTRACT_ADDRESS "Constructor arg1" "arg2" ``` ## CI/CD Integration ```yaml # .github/workflows/test.yml name: Tests on: [push, pull_request] jobs: test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v2 - uses: actions/setup-node@v2 with: node-version: "16" - run: npm install - run: npx hardhat compile - run: npx hardhat test - run: npx hardhat coverage - name: Upload coverage to Codecov uses: codecov/codecov-action@v2 ``` ## Resources - **references/hardhat-setup.md**: Hardhat configuration guide - **references/foundry-setup.md**: Foundry testing framework - **references/test-patterns.md**: Testing best practices - **references/mainnet-forking.md**: Fork testing strategies - **references/contract-verification.md**: Etherscan verification - **assets/hardhat-config.js**: Complete Hardhat configuration - **assets/test-suite.js**: Comprehensive test examples - **assets/foundry.toml**: Foundry configuration - **scripts/test-contract.sh**: Automated testing script ## Best Practices 1. **Test Coverage**: Aim for >90% coverage 2. **Edge Cases**: Test boundary conditions 3. **Gas Limits**: Verify functions don't hit block gas limit 4. **Reentrancy**: Test for reentrancy vulnerabilities 5. **Access Control**: Test unauthorized access attempts 6. **Events**: Verify event emissions 7. **Fixtures**: Use fixtures to avoid code duplication 8. **Mainnet Fork**: Test with real contracts 9. **Fuzzing**: Use property-based testing 10. **CI/CD**: Automate testing on every commit
👍0
👁️0
🤖 Auto-discovered