Skip to main content
EVOKORE// BROWSE
>

./browse/prompts

47 NODES
šŸ¤–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

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

gitlab-ci-patterns

Build GitLab CI/CD pipelines with multi-stage workflows, caching,

coding
⭐1
# GitLab CI Patterns Comprehensive GitLab CI/CD pipeline patterns for automated testing, building, and deployment. ## Purpose Create efficient GitLab CI pipelines with proper stage organization, caching, and deployment strategies. ## When to Use - Automate GitLab-based CI/CD - Implement multi-stage pipelines - Configure GitLab Runners - Deploy to Kubernetes from GitLab - Implement GitOps workflows ## Basic Pipeline Structure ```yaml stages: - build - test - deploy variables: DOCKER_DRIVER: overlay2 DOCKER_TLS_CERTDIR: "/certs" build: stage: build image: node:20 script: - npm ci - npm run build artifacts: paths: - dist/ expire_in: 1 hour cache: key: ${CI_COMMIT_REF_SLUG} paths: - node_modules/ test: stage: test image: node:20 script: - npm ci - npm run lint - npm test coverage: '/Lines\s*:\s*(\d+\.\d+)%/' artifacts: reports: coverage_report: coverage_format: cobertura path: coverage/cobertura-coverage.xml deploy: stage: deploy image: bitnami/kubectl:latest script: - kubectl apply -f k8s/ - kubectl rollout status deployment/my-app only: - main environment: name: production url: https://app.example.com ``` ## Docker Build and Push ```yaml build-docker: stage: build image: docker:24 services: - docker:24-dind before_script: - docker login -u $CI_REGISTRY_USER -p $CI_REGISTRY_PASSWORD $CI_REGISTRY script: - docker build -t $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA . - docker build -t $CI_REGISTRY_IMAGE:latest . - docker push $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA - docker push $CI_REGISTRY_IMAGE:latest only: - main - tags ``` ## Multi-Environment Deployment ```yaml .deploy_template: &deploy_template image: bitnami/kubectl:latest before_script: - kubectl config set-cluster k8s --server="$KUBE_URL" --insecure-skip-tls-verify=true - kubectl config set-credentials admin --token="$KUBE_TOKEN" - kubectl config set-context default --cluster=k8s --user=admin - kubectl config use-context default deploy:staging: <<: *deploy_template stage: deploy script: - kubectl apply -f k8s/ -n staging - kubectl rollout status deployment/my-app -n staging environment: name: staging url: https://staging.example.com only: - develop deploy:production: <<: *deploy_template stage: deploy script: - kubectl apply -f k8s/ -n production - kubectl rollout status deployment/my-app -n production environment: name: production url: https://app.example.com when: manual only: - main ``` ## Terraform Pipeline ```yaml stages: - validate - plan - apply variables: TF_ROOT: ${CI_PROJECT_DIR}/terraform TF_VERSION: "1.6.0" before_script: - cd ${TF_ROOT} - terraform --version validate: stage: validate image: hashicorp/terraform:${TF_VERSION} script: - terraform init -backend=false - terraform validate - terraform fmt -check plan: stage: plan image: hashicorp/terraform:${TF_VERSION} script: - terraform init - terraform plan -out=tfplan artifacts: paths: - ${TF_ROOT}/tfplan expire_in: 1 day apply: stage: apply image: hashicorp/terraform:${TF_VERSION} script: - terraform init - terraform apply -auto-approve tfplan dependencies: - plan when: manual only: - main ``` ## Security Scanning ```yaml include: - template: Security/SAST.gitlab-ci.yml - template: Security/Dependency-Scanning.gitlab-ci.yml - template: Security/Container-Scanning.gitlab-ci.yml trivy-scan: stage: test image: aquasec/trivy:latest script: - trivy image --exit-code 1 --severity HIGH,CRITICAL $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA allow_failure: true ``` ## Caching Strategies ```yaml # Cache node_modules build: cache: key: ${CI_COMMIT_REF_SLUG} paths: - node_modules/ policy: pull-push # Global cache cache: key: ${CI_COMMIT_REF_SLUG} paths: - .cache/ - vendor/ # Separate cache per job job1: cache: key: job1-cache paths: - build/ job2: cache: key: job2-cache paths: - dist/ ``` ## Dynamic Child Pipelines ```yaml generate-pipeline: stage: build script: - python generate_pipeline.py > child-pipeline.yml artifacts: paths: - child-pipeline.yml trigger-child: stage: deploy trigger: include: - artifact: child-pipeline.yml job: generate-pipeline strategy: depend ``` ## Reference Files - `assets/gitlab-ci.yml.template` - Complete pipeline template - `references/pipeline-stages.md` - Stage organization patterns ## Best Practices 1. **Use specific image tags** (node:20, not node:latest) 2. **Cache dependencies** appropriately 3. **Use artifacts** for build outputs 4. **Implement manual gates** for production 5. **Use environments** for deployment tracking 6. **Enable merge request pipelines** 7. **Use pipeline schedules** for recurring jobs 8. **Implement security scanning** 9. **Use CI/CD variables** for secrets 10. **Monitor pipeline performance** ## Related Skills - `github-actions-templates` - For GitHub Actions - `deployment-pipeline-design` - For architecture - `secrets-management` - For secrets handling
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

secrets-management

Implement secure secrets management for CI/CD pipelines using

coding
⭐1
# Secrets Management Secure secrets management practices for CI/CD pipelines using Vault, AWS Secrets Manager, and other tools. ## Purpose Implement secure secrets management in CI/CD pipelines without hardcoding sensitive information. ## When to Use - Store API keys and credentials - Manage database passwords - Handle TLS certificates - Rotate secrets automatically - Implement least-privilege access ## Secrets Management Tools ### HashiCorp Vault - Centralized secrets management - Dynamic secrets generation - Secret rotation - Audit logging - Fine-grained access control ### AWS Secrets Manager - AWS-native solution - Automatic rotation - Integration with RDS - CloudFormation support ### Azure Key Vault - Azure-native solution - HSM-backed keys - Certificate management - RBAC integration ### Google Secret Manager - GCP-native solution - Versioning - IAM integration ## HashiCorp Vault Integration ### Setup Vault ```bash # Start Vault dev server vault server -dev # Set environment export VAULT_ADDR='http://127.0.0.1:8200' export VAULT_TOKEN='root' # Enable secrets engine vault secrets enable -path=secret kv-v2 # Store secret vault kv put secret/database/config username=admin password=secret ``` ### GitHub Actions with Vault ```yaml name: Deploy with Vault Secrets on: [push] jobs: deploy: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Import Secrets from Vault uses: hashicorp/vault-action@v2 with: url: https://vault.example.com:8200 token: ${{ secrets.VAULT_TOKEN }} secrets: | secret/data/database username | DB_USERNAME ; secret/data/database password | DB_PASSWORD ; secret/data/api key | API_KEY - name: Use secrets run: | echo "Connecting to database as $DB_USERNAME" # Use $DB_PASSWORD, $API_KEY ``` ### GitLab CI with Vault ```yaml deploy: image: vault:latest before_script: - export VAULT_ADDR=https://vault.example.com:8200 - export VAULT_TOKEN=$VAULT_TOKEN - apk add curl jq script: - | DB_PASSWORD=$(vault kv get -field=password secret/database/config) API_KEY=$(vault kv get -field=key secret/api/credentials) echo "Deploying with secrets..." # Use $DB_PASSWORD, $API_KEY ``` **Reference:** See `references/vault-setup.md` ## AWS Secrets Manager ### Store Secret ```bash aws secretsmanager create-secret \ --name production/database/password \ --secret-string "super-secret-password" ``` ### Retrieve in GitHub Actions ```yaml - name: Configure AWS credentials uses: aws-actions/configure-aws-credentials@v4 with: aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }} aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} aws-region: us-west-2 - name: Get secret from AWS run: | SECRET=$(aws secretsmanager get-secret-value \ --secret-id production/database/password \ --query SecretString \ --output text) echo "::add-mask::$SECRET" echo "DB_PASSWORD=$SECRET" >> $GITHUB_ENV - name: Use secret run: | # Use $DB_PASSWORD ./deploy.sh ``` ### Terraform with AWS Secrets Manager ```hcl data "aws_secretsmanager_secret_version" "db_password" { secret_id = "production/database/password" } resource "aws_db_instance" "main" { allocated_storage = 100 engine = "postgres" instance_class = "db.t3.large" username = "admin" password = jsondecode(data.aws_secretsmanager_secret_version.db_password.secret_string)["password"] } ``` ## GitHub Secrets ### Organization/Repository Secrets ```yaml - name: Use GitHub secret run: | echo "API Key: ${{ secrets.API_KEY }}" echo "Database URL: ${{ secrets.DATABASE_URL }}" ``` ### Environment Secrets ```yaml deploy: runs-on: ubuntu-latest environment: production steps: - name: Deploy run: | echo "Deploying with ${{ secrets.PROD_API_KEY }}" ``` **Reference:** See `references/github-secrets.md` ## GitLab CI/CD Variables ### Project Variables ```yaml deploy: script: - echo "Deploying with $API_KEY" - echo "Database: $DATABASE_URL" ``` ### Protected and Masked Variables - Protected: Only available in protected branches - Masked: Hidden in job logs - File type: Stored as file ## Best Practices 1. **Never commit secrets** to Git 2. **Use different secrets** per environment 3. **Rotate secrets regularly** 4. **Implement least-privilege access** 5. **Enable audit logging** 6. **Use secret scanning** (GitGuardian, TruffleHog) 7. **Mask secrets in logs** 8. **Encrypt secrets at rest** 9. **Use short-lived tokens** when possible 10. **Document secret requirements** ## Secret Rotation ### Automated Rotation with AWS ```python import boto3 import json def lambda_handler(event, context): client = boto3.client('secretsmanager') # Get current secret response = client.get_secret_value(SecretId='my-secret') current_secret = json.loads(response['SecretString']) # Generate new password new_password = generate_strong_password() # Update database password update_database_password(new_password) # Update secret client.put_secret_value( SecretId='my-secret', SecretString=json.dumps({ 'username': current_secret['username'], 'password': new_password }) ) return {'statusCode': 200} ``` ### Manual Rotation Process 1. Generate new secret 2. Update secret in secret store 3. Update applications to use new secret 4. Verify functionality 5. Revoke old secret ## External Secrets Operator ### Kubernetes Integration ```yaml apiVersion: external-secrets.io/v1beta1 kind: SecretStore metadata: name: vault-backend namespace: production spec: provider: vault: server: "https://vault.example.com:8200" path: "secret" version: "v2" auth: kubernetes: mountPath: "kubernetes" role: "production" --- apiVersion: external-secrets.io/v1beta1 kind: ExternalSecret metadata: name: database-credentials namespace: production spec: refreshInterval: 1h secretStoreRef: name: vault-backend kind: SecretStore target: name: database-credentials creationPolicy: Owner data: - secretKey: username remoteRef: key: database/config property: username - secretKey: password remoteRef: key: database/config property: password ``` ## Secret Scanning ### Pre-commit Hook ```bash #!/bin/bash # .git/hooks/pre-commit # Check for secrets with TruffleHog docker run --rm -v "$(pwd):/repo" \ trufflesecurity/trufflehog:latest \ filesystem --directory=/repo if [ $? -ne 0 ]; then echo "āŒ Secret detected! Commit blocked." exit 1 fi ``` ### CI/CD Secret Scanning ```yaml secret-scan: stage: security image: trufflesecurity/trufflehog:latest script: - trufflehog filesystem . allow_failure: false ``` ## Reference Files - `references/vault-setup.md` - HashiCorp Vault configuration - `references/github-secrets.md` - GitHub Secrets best practices ## Related Skills - `github-actions-templates` - For GitHub Actions integration - `gitlab-ci-patterns` - For GitLab CI integration - `deployment-pipeline-design` - For pipeline architecture
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

cost-optimization

Optimize cloud costs through resource rightsizing, tagging

architecture
⭐1
# Cloud Cost Optimization Strategies and patterns for optimizing cloud costs across AWS, Azure, and GCP. ## Purpose Implement systematic cost optimization strategies to reduce cloud spending while maintaining performance and reliability. ## When to Use - Reduce cloud spending - Right-size resources - Implement cost governance - Optimize multi-cloud costs - Meet budget constraints ## Cost Optimization Framework ### 1. Visibility - Implement cost allocation tags - Use cloud cost management tools - Set up budget alerts - Create cost dashboards ### 2. Right-Sizing - Analyze resource utilization - Downsize over-provisioned resources - Use auto-scaling - Remove idle resources ### 3. Pricing Models - Use reserved capacity - Leverage spot/preemptible instances - Implement savings plans - Use committed use discounts ### 4. Architecture Optimization - Use managed services - Implement caching - Optimize data transfer - Use lifecycle policies ## AWS Cost Optimization ### Reserved Instances ``` Savings: 30-72% vs On-Demand Term: 1 or 3 years Payment: All/Partial/No upfront Flexibility: Standard or Convertible ``` ### Savings Plans ``` Compute Savings Plans: 66% savings EC2 Instance Savings Plans: 72% savings Applies to: EC2, Fargate, Lambda Flexible across: Instance families, regions, OS ``` ### Spot Instances ``` Savings: Up to 90% vs On-Demand Best for: Batch jobs, CI/CD, stateless workloads Risk: 2-minute interruption notice Strategy: Mix with On-Demand for resilience ``` ### S3 Cost Optimization ```hcl resource "aws_s3_bucket_lifecycle_configuration" "example" { bucket = aws_s3_bucket.example.id rule { id = "transition-to-ia" status = "Enabled" transition { days = 30 storage_class = "STANDARD_IA" } transition { days = 90 storage_class = "GLACIER" } expiration { days = 365 } } } ``` ## Azure Cost Optimization ### Reserved VM Instances - 1 or 3 year terms - Up to 72% savings - Flexible sizing - Exchangeable ### Azure Hybrid Benefit - Use existing Windows Server licenses - Up to 80% savings with RI - Available for Windows and SQL Server ### Azure Advisor Recommendations - Right-size VMs - Delete unused resources - Use reserved capacity - Optimize storage ## GCP Cost Optimization ### Committed Use Discounts - 1 or 3 year commitment - Up to 57% savings - Applies to vCPUs and memory - Resource-based or spend-based ### Sustained Use Discounts - Automatic discounts - Up to 30% for running instances - No commitment required - Applies to Compute Engine, GKE ### Preemptible VMs - Up to 80% savings - 24-hour maximum runtime - Best for batch workloads ## Tagging Strategy ### AWS Tagging ```hcl locals { common_tags = { Environment = "production" Project = "my-project" CostCenter = "engineering" Owner = "team@example.com" ManagedBy = "terraform" } } resource "aws_instance" "example" { ami = "ami-12345678" instance_type = "t3.medium" tags = merge( local.common_tags, { Name = "web-server" } ) } ``` **Reference:** See `references/tagging-standards.md` ## Cost Monitoring ### Budget Alerts ```hcl # AWS Budget resource "aws_budgets_budget" "monthly" { name = "monthly-budget" budget_type = "COST" limit_amount = "1000" limit_unit = "USD" time_period_start = "2024-01-01_00:00" time_unit = "MONTHLY" notification { comparison_operator = "GREATER_THAN" threshold = 80 threshold_type = "PERCENTAGE" notification_type = "ACTUAL" subscriber_email_addresses = ["team@example.com"] } } ``` ### Cost Anomaly Detection - AWS Cost Anomaly Detection - Azure Cost Management alerts - GCP Budget alerts ## Architecture Patterns ### Pattern 1: Serverless First - Use Lambda/Functions for event-driven - Pay only for execution time - Auto-scaling included - No idle costs ### Pattern 2: Right-Sized Databases ``` Development: t3.small RDS Staging: t3.large RDS Production: r6g.2xlarge RDS with read replicas ``` ### Pattern 3: Multi-Tier Storage ``` Hot data: S3 Standard Warm data: S3 Standard-IA (30 days) Cold data: S3 Glacier (90 days) Archive: S3 Deep Archive (365 days) ``` ### Pattern 4: Auto-Scaling ```hcl resource "aws_autoscaling_policy" "scale_up" { name = "scale-up" scaling_adjustment = 2 adjustment_type = "ChangeInCapacity" cooldown = 300 autoscaling_group_name = aws_autoscaling_group.main.name } resource "aws_cloudwatch_metric_alarm" "cpu_high" { alarm_name = "cpu-high" comparison_operator = "GreaterThanThreshold" evaluation_periods = "2" metric_name = "CPUUtilization" namespace = "AWS/EC2" period = "60" statistic = "Average" threshold = "80" alarm_actions = [aws_autoscaling_policy.scale_up.arn] } ``` ## Cost Optimization Checklist - [ ] Implement cost allocation tags - [ ] Delete unused resources (EBS, EIPs, snapshots) - [ ] Right-size instances based on utilization - [ ] Use reserved capacity for steady workloads - [ ] Implement auto-scaling - [ ] Optimize storage classes - [ ] Use lifecycle policies - [ ] Enable cost anomaly detection - [ ] Set budget alerts - [ ] Review costs weekly - [ ] Use spot/preemptible instances - [ ] Optimize data transfer costs - [ ] Implement caching layers - [ ] Use managed services - [ ] Monitor and optimize continuously ## Tools - **AWS:** Cost Explorer, Cost Anomaly Detection, Compute Optimizer - **Azure:** Cost Management, Advisor - **GCP:** Cost Management, Recommender - **Multi-cloud:** CloudHealth, Cloudability, Kubecost ## Reference Files - `references/tagging-standards.md` - Tagging conventions - `assets/cost-analysis-template.xlsx` - Cost analysis spreadsheet ## Related Skills - `terraform-module-library` - For resource provisioning - `multi-cloud-architecture` - For cloud selection
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

context-driven-development

Creates and maintains project context artifacts (product.md,

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

workflow-patterns

Use this skill when implementing tasks according to Conductor's TDD

coding
⭐1
# Workflow Patterns Guide for implementing tasks using Conductor's TDD workflow, managing phase checkpoints, handling git commits, and executing the verification protocol that ensures quality throughout implementation. ## When to Use This Skill - Implementing tasks from a track's plan.md - Following TDD red-green-refactor cycle - Completing phase checkpoints - Managing git commits and notes - Understanding quality assurance gates - Handling verification protocols - Recording progress in plan files ## TDD Task Lifecycle Follow these 11 steps for each task: ### Step 1: Select Next Task Read plan.md and identify the next pending `[ ]` task. Select tasks in order within the current phase. Do not skip ahead to later phases. ### Step 2: Mark as In Progress Update plan.md to mark the task as `[~]`: ```markdown - [~] **Task 2.1**: Implement user validation ``` Commit this status change separately from implementation. ### Step 3: RED - Write Failing Tests Write tests that define the expected behavior before writing implementation: - Create test file if needed - Write test cases covering happy path - Write test cases covering edge cases - Write test cases covering error conditions - Run tests - they should FAIL Example: ```python def test_validate_user_email_valid(): user = User(email="test@example.com") assert user.validate_email() is True def test_validate_user_email_invalid(): user = User(email="invalid") assert user.validate_email() is False ``` ### Step 4: GREEN - Implement Minimum Code Write the minimum code necessary to make tests pass: - Focus on making tests green, not perfection - Avoid premature optimization - Keep implementation simple - Run tests - they should PASS ### Step 5: REFACTOR - Improve Clarity With green tests, improve the code: - Extract common patterns - Improve naming - Remove duplication - Simplify logic - Run tests after each change - they should remain GREEN ### Step 6: Verify Coverage Check test coverage meets the 80% target: ```bash pytest --cov=module --cov-report=term-missing ``` If coverage is below 80%: - Identify uncovered lines - Add tests for missing paths - Re-run coverage check ### Step 7: Document Deviations If implementation deviated from plan or introduced new dependencies: - Update tech-stack.md with new dependencies - Note deviations in plan.md task comments - Update spec.md if requirements changed ### Step 8: Commit Implementation Create a focused commit for the task: ```bash git add -A git commit -m "feat(user): implement email validation - Add validate_email method to User class - Handle empty and malformed emails - Add comprehensive test coverage Task: 2.1 Track: user-auth_20250115" ``` Commit message format: - Type: feat, fix, refactor, test, docs, chore - Scope: affected module or component - Summary: imperative, present tense - Body: bullet points of changes - Footer: task and track references ### Step 9: Attach Git Notes Add rich task summary as git note: ```bash git notes add -m "Task 2.1: Implement user validation Summary: - Added email validation using regex pattern - Handles edge cases: empty, no @, no domain - Coverage: 94% on validation module Files changed: - src/models/user.py (modified) - tests/test_user.py (modified) Decisions: - Used simple regex over email-validator library - Reason: No external dependency for basic validation" ``` ### Step 10: Update Plan with SHA Update plan.md to mark task complete with commit SHA: ```markdown - [x] **Task 2.1**: Implement user validation `abc1234` ``` ### Step 11: Commit Plan Update Commit the plan status update: ```bash git add conductor/tracks/*/plan.md git commit -m "docs: update plan - task 2.1 complete Track: user-auth_20250115" ``` ## Phase Completion Protocol When all tasks in a phase are complete, execute the verification protocol: ### Identify Changed Files List all files modified since the last checkpoint: ```bash git diff --name-only <last-checkpoint-sha>..HEAD ``` ### Ensure Test Coverage For each modified file: 1. Identify corresponding test file 2. Verify tests exist for new/changed code 3. Run coverage for modified modules 4. Add tests if coverage < 80% ### Run Full Test Suite Execute complete test suite: ```bash pytest -v --tb=short ``` All tests must pass before proceeding. ### Generate Manual Verification Steps Create checklist of manual verifications: ```markdown ## Phase 1 Verification Checklist - [ ] User can register with valid email - [ ] Invalid email shows appropriate error - [ ] Database stores user correctly - [ ] API returns expected response codes ``` ### WAIT for User Approval Present verification checklist to user: ``` Phase 1 complete. Please verify: 1. [ ] Test suite passes (automated) 2. [ ] Coverage meets target (automated) 3. [ ] Manual verification items (requires human) Respond with 'approved' to continue, or note issues. ``` Do NOT proceed without explicit approval. ### Create Checkpoint Commit After approval, create checkpoint commit: ```bash git add -A git commit -m "checkpoint: phase 1 complete - user-auth_20250115 Verified: - All tests passing - Coverage: 87% - Manual verification approved Phase 1 tasks: - [x] Task 1.1: Setup database schema - [x] Task 1.2: Implement user model - [x] Task 1.3: Add validation logic" ``` ### Record Checkpoint SHA Update plan.md checkpoints table: ```markdown ## Checkpoints | Phase | Checkpoint SHA | Date | Status | | ------- | -------------- | ---------- | -------- | | Phase 1 | def5678 | 2025-01-15 | verified | | Phase 2 | | | pending | ``` ## Quality Assurance Gates Before marking any task complete, verify these gates: ### Passing Tests - All existing tests pass - New tests pass - No test regressions ### Coverage >= 80% - New code has 80%+ coverage - Overall project coverage maintained - Critical paths fully covered ### Style Compliance - Code follows style guides - Linting passes - Formatting correct ### Documentation - Public APIs documented - Complex logic explained - README updated if needed ### Type Safety - Type hints present (if applicable) - Type checker passes - No type: ignore without reason ### No Linting Errors - Zero linter errors - Warnings addressed or justified - Static analysis clean ### Mobile Compatibility If applicable: - Responsive design verified - Touch interactions work - Performance acceptable ### Security Audit - No secrets in code - Input validation present - Authentication/authorization correct - Dependencies vulnerability-free ## Git Integration ### Commit Message Format ``` <type>(<scope>): <subject> <body> <footer> ``` Types: - `feat`: New feature - `fix`: Bug fix - `refactor`: Code change without feature/fix - `test`: Adding tests - `docs`: Documentation - `chore`: Maintenance ### Git Notes for Rich Summaries Attach detailed notes to commits: ```bash git notes add -m "<detailed summary>" ``` View notes: ```bash git log --show-notes ``` Benefits: - Preserves context without cluttering commit message - Enables semantic queries across commits - Supports track-based operations ### SHA Recording in plan.md Always record the commit SHA when completing tasks: ```markdown - [x] **Task 1.1**: Setup schema `abc1234` - [x] **Task 1.2**: Add model `def5678` ``` This enables: - Traceability from plan to code - Semantic revert operations - Progress auditing ## Verification Checkpoints ### Why Checkpoints Matter Checkpoints create restore points for semantic reversion: - Revert to end of any phase - Maintain logical code state - Enable safe experimentation ### When to Create Checkpoints Create checkpoint after: - All phase tasks complete - All phase verifications pass - User approval received ### Checkpoint Commit Content Include in checkpoint commit: - All uncommitted changes - Updated plan.md - Updated metadata.json - Any documentation updates ### How to Use Checkpoints For reverting: ```bash # Revert to end of Phase 1 git revert --no-commit <phase-2-commits>... git commit -m "revert: rollback to phase 1 checkpoint" ``` For review: ```bash # See what changed in Phase 2 git diff <phase-1-sha>..<phase-2-sha> ``` ## Handling Deviations During implementation, deviations from the plan may occur. Handle them systematically: ### Types of Deviations **Scope Addition** Discovered requirement not in original spec. - Document in spec.md as new requirement - Add tasks to plan.md - Note addition in task comments **Scope Reduction** Feature deemed unnecessary during implementation. - Mark tasks as `[-]` (skipped) with reason - Update spec.md scope section - Document decision rationale **Technical Deviation** Different implementation approach than planned. - Note deviation in task completion comment - Update tech-stack.md if dependencies changed - Document why original approach was unsuitable **Requirement Change** Understanding of requirement changes during work. - Update spec.md with corrected requirement - Adjust plan.md tasks if needed - Re-verify acceptance criteria ### Deviation Documentation Format When completing a task with deviation: ```markdown - [x] **Task 2.1**: Implement validation `abc1234` - DEVIATION: Used library instead of custom code - Reason: Better edge case handling - Impact: Added email-validator to dependencies ``` ## Error Recovery ### Failed Tests After GREEN If tests fail after reaching GREEN: 1. Do NOT proceed to REFACTOR 2. Identify which test started failing 3. Check if refactoring broke something 4. Revert to last known GREEN state 5. Re-approach the implementation ### Checkpoint Rejection If user rejects a checkpoint: 1. Note rejection reason in plan.md 2. Create tasks to address issues 3. Complete remediation tasks 4. Request checkpoint approval again ### Blocked by Dependency If task cannot proceed: 1. Mark task as `[!]` with blocker description 2. Check if other tasks can proceed 3. Document expected resolution timeline 4. Consider creating dependency resolution track ## TDD Variations by Task Type ### Data Model Tasks ``` RED: Write test for model creation and validation GREEN: Implement model class with fields REFACTOR: Add computed properties, improve types ``` ### API Endpoint Tasks ``` RED: Write test for request/response contract GREEN: Implement endpoint handler REFACTOR: Extract validation, improve error handling ``` ### Integration Tasks ``` RED: Write test for component interaction GREEN: Wire components together REFACTOR: Improve error propagation, add logging ``` ### Refactoring Tasks ``` RED: Add characterization tests for current behavior GREEN: Apply refactoring (tests should stay green) REFACTOR: Clean up any introduced complexity ``` ## Working with Existing Tests When modifying code with existing tests: ### Extend, Don't Replace - Keep existing tests passing - Add new tests for new behavior - Update tests only when requirements change ### Test Migration When refactoring changes test structure: 1. Run existing tests (should pass) 2. Add new tests for refactored code 3. Migrate test cases to new structure 4. Remove old tests only after new tests pass ### Regression Prevention After any change: 1. Run full test suite 2. Check for unexpected failures 3. Investigate any new failures 4. Fix regressions before proceeding ## Checkpoint Verification Details ### Automated Verification Run before requesting approval: ```bash # Test suite pytest -v --tb=short # Coverage pytest --cov=src --cov-report=term-missing # Linting ruff check src/ tests/ # Type checking (if applicable) mypy src/ ``` ### Manual Verification Guidance For manual items, provide specific instructions: ```markdown ## Manual Verification Steps ### User Registration 1. Navigate to /register 2. Enter valid email: test@example.com 3. Enter password meeting requirements 4. Click Submit 5. Verify success message appears 6. Verify user appears in database ### Error Handling 1. Enter invalid email: "notanemail" 2. Verify error message shows 3. Verify form retains other entered data ``` ## Performance Considerations ### Test Suite Performance Keep test suite fast: - Use fixtures to avoid redundant setup - Mock slow external calls - Run subset during development, full suite at checkpoints ### Commit Performance Keep commits atomic: - One logical change per commit - Complete thought, not work-in-progress - Tests should pass after every commit ## Best Practices 1. **Never skip RED**: Always write failing tests first 2. **Small commits**: One logical change per commit 3. **Immediate updates**: Update plan.md right after task completion 4. **Wait for approval**: Never skip checkpoint verification 5. **Rich git notes**: Include context that helps future understanding 6. **Coverage discipline**: Don't accept coverage below target 7. **Quality gates**: Check all gates before marking complete 8. **Sequential phases**: Complete phases in order 9. **Document deviations**: Note any changes from original plan 10. **Clean state**: Each commit should leave code in working state 11. **Fast feedback**: Run relevant tests frequently during development 12. **Clear blockers**: Address blockers promptly, don't work around them
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

airflow-dag-patterns

Build production Apache Airflow DAGs with best practices for

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

data-quality-frameworks

Implement data quality validation with Great Expectations, dbt

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

dbt-transformation-patterns

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

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

postgresql-table-design

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

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

bazel-build-optimization

Optimize Bazel builds for large-scale monorepos. Use when

coding
⭐1
# Bazel Build Optimization Production patterns for Bazel in large-scale monorepos. ## When to Use This Skill - Setting up Bazel for monorepos - Configuring remote caching/execution - Optimizing build times - Writing custom Bazel rules - Debugging build issues - Migrating to Bazel ## Core Concepts ### 1. Bazel Architecture ``` workspace/ ā”œā”€ā”€ WORKSPACE.bazel # External dependencies ā”œā”€ā”€ .bazelrc # Build configurations ā”œā”€ā”€ .bazelversion # Bazel version ā”œā”€ā”€ BUILD.bazel # Root build file ā”œā”€ā”€ apps/ │ └── web/ │ └── BUILD.bazel ā”œā”€ā”€ libs/ │ └── utils/ │ └── BUILD.bazel └── tools/ └── bazel/ └── rules/ ``` ### 2. Key Concepts | Concept | Description | | ----------- | -------------------------------------- | | **Target** | Buildable unit (library, binary, test) | | **Package** | Directory with BUILD file | | **Label** | Target identifier `//path/to:target` | | **Rule** | Defines how to build a target | | **Aspect** | Cross-cutting build behavior | ## Templates ### Template 1: WORKSPACE Configuration ```python # WORKSPACE.bazel workspace(name = "myproject") load("@bazel_tools//tools/build_defs/repo:http.bzl", "http_archive") # Rules for JavaScript/TypeScript http_archive( name = "aspect_rules_js", sha256 = "...", strip_prefix = "rules_js-1.34.0", url = "https://github.com/aspect-build/rules_js/releases/download/v1.34.0/rules_js-v1.34.0.tar.gz", ) load("@aspect_rules_js//js:repositories.bzl", "rules_js_dependencies") rules_js_dependencies() load("@rules_nodejs//nodejs:repositories.bzl", "nodejs_register_toolchains") nodejs_register_toolchains( name = "nodejs", node_version = "20.9.0", ) load("@aspect_rules_js//npm:repositories.bzl", "npm_translate_lock") npm_translate_lock( name = "npm", pnpm_lock = "//:pnpm-lock.yaml", verify_node_modules_ignored = "//:.bazelignore", ) load("@npm//:repositories.bzl", "npm_repositories") npm_repositories() # Rules for Python http_archive( name = "rules_python", sha256 = "...", strip_prefix = "rules_python-0.27.0", url = "https://github.com/bazelbuild/rules_python/releases/download/0.27.0/rules_python-0.27.0.tar.gz", ) load("@rules_python//python:repositories.bzl", "py_repositories") py_repositories() ``` ### Template 2: .bazelrc Configuration ```bash # .bazelrc # Build settings build --enable_platform_specific_config build --incompatible_enable_cc_toolchain_resolution build --experimental_strict_conflict_checks # Performance build --jobs=auto build --local_cpu_resources=HOST_CPUS*.75 build --local_ram_resources=HOST_RAM*.75 # Caching build --disk_cache=~/.cache/bazel-disk build --repository_cache=~/.cache/bazel-repo # Remote caching (optional) build:remote-cache --remote_cache=grpcs://cache.example.com build:remote-cache --remote_upload_local_results=true build:remote-cache --remote_timeout=3600 # Remote execution (optional) build:remote-exec --remote_executor=grpcs://remote.example.com build:remote-exec --remote_instance_name=projects/myproject/instances/default build:remote-exec --jobs=500 # Platform configurations build:linux --platforms=//platforms:linux_x86_64 build:macos --platforms=//platforms:macos_arm64 # CI configuration build:ci --config=remote-cache build:ci --build_metadata=ROLE=CI build:ci --bes_results_url=https://results.example.com/invocation/ build:ci --bes_backend=grpcs://bes.example.com # Test settings test --test_output=errors test --test_summary=detailed # Coverage coverage --combined_report=lcov coverage --instrumentation_filter="//..." # Convenience aliases build:opt --compilation_mode=opt build:dbg --compilation_mode=dbg # Import user settings try-import %workspace%/user.bazelrc ``` ### Template 3: TypeScript Library BUILD ```python # libs/utils/BUILD.bazel load("@aspect_rules_ts//ts:defs.bzl", "ts_project") load("@aspect_rules_js//js:defs.bzl", "js_library") load("@npm//:defs.bzl", "npm_link_all_packages") npm_link_all_packages(name = "node_modules") ts_project( name = "utils_ts", srcs = glob(["src/**/*.ts"]), declaration = True, source_map = True, tsconfig = "//:tsconfig.json", deps = [ ":node_modules/@types/node", ], ) js_library( name = "utils", srcs = [":utils_ts"], visibility = ["//visibility:public"], ) # Tests load("@aspect_rules_jest//jest:defs.bzl", "jest_test") jest_test( name = "utils_test", config = "//:jest.config.js", data = [ ":utils", "//:node_modules/jest", ], node_modules = "//:node_modules", ) ``` ### Template 4: Python Library BUILD ```python # libs/ml/BUILD.bazel load("@rules_python//python:defs.bzl", "py_library", "py_test", "py_binary") load("@pip//:requirements.bzl", "requirement") py_library( name = "ml", srcs = glob(["src/**/*.py"]), deps = [ requirement("numpy"), requirement("pandas"), requirement("scikit-learn"), "//libs/utils:utils_py", ], visibility = ["//visibility:public"], ) py_test( name = "ml_test", srcs = glob(["tests/**/*.py"]), deps = [ ":ml", requirement("pytest"), ], size = "medium", timeout = "moderate", ) py_binary( name = "train", srcs = ["train.py"], deps = [":ml"], data = ["//data:training_data"], ) ``` ### Template 5: Custom Rule for Docker ```python # tools/bazel/rules/docker.bzl def _docker_image_impl(ctx): dockerfile = ctx.file.dockerfile base_image = ctx.attr.base_image layers = ctx.files.layers # Build the image output = ctx.actions.declare_file(ctx.attr.name + ".tar") args = ctx.actions.args() args.add("--dockerfile", dockerfile) args.add("--output", output) args.add("--base", base_image) args.add_all("--layer", layers) ctx.actions.run( inputs = [dockerfile] + layers, outputs = [output], executable = ctx.executable._builder, arguments = [args], mnemonic = "DockerBuild", progress_message = "Building Docker image %s" % ctx.label, ) return [DefaultInfo(files = depset([output]))] docker_image = rule( implementation = _docker_image_impl, attrs = { "dockerfile": attr.label( allow_single_file = [".dockerfile", "Dockerfile"], mandatory = True, ), "base_image": attr.string(mandatory = True), "layers": attr.label_list(allow_files = True), "_builder": attr.label( default = "//tools/docker:builder", executable = True, cfg = "exec", ), }, ) ``` ### Template 6: Query and Dependency Analysis ```bash # Find all dependencies of a target bazel query "deps(//apps/web:web)" # Find reverse dependencies (what depends on this) bazel query "rdeps(//..., //libs/utils:utils)" # Find all targets in a package bazel query "//libs/..." # Find changed targets since commit bazel query "rdeps(//..., set($(git diff --name-only HEAD~1 | sed 's/.*/"&"/' | tr '\n' ' ')))" # Generate dependency graph bazel query "deps(//apps/web:web)" --output=graph | dot -Tpng > deps.png # Find all test targets bazel query "kind('.*_test', //...)" # Find targets with specific tag bazel query "attr(tags, 'integration', //...)" # Compute build graph size bazel query "deps(//...)" --output=package | wc -l ``` ### Template 7: Remote Execution Setup ```python # platforms/BUILD.bazel platform( name = "linux_x86_64", constraint_values = [ "@platforms//os:linux", "@platforms//cpu:x86_64", ], exec_properties = { "container-image": "docker://gcr.io/myproject/bazel-worker:latest", "OSFamily": "Linux", }, ) platform( name = "remote_linux", parents = [":linux_x86_64"], exec_properties = { "Pool": "default", "dockerNetwork": "standard", }, ) # toolchains/BUILD.bazel toolchain( name = "cc_toolchain_linux", exec_compatible_with = [ "@platforms//os:linux", "@platforms//cpu:x86_64", ], target_compatible_with = [ "@platforms//os:linux", "@platforms//cpu:x86_64", ], toolchain = "@remotejdk11_linux//:jdk", toolchain_type = "@bazel_tools//tools/jdk:runtime_toolchain_type", ) ``` ## Performance Optimization ```bash # Profile build bazel build //... --profile=profile.json bazel analyze-profile profile.json # Identify slow actions bazel build //... --execution_log_json_file=exec_log.json # Memory profiling bazel build //... --memory_profile=memory.json # Skip analysis cache bazel build //... --notrack_incremental_state ``` ## Best Practices ### Do's - **Use fine-grained targets** - Better caching - **Pin dependencies** - Reproducible builds - **Enable remote caching** - Share build artifacts - **Use visibility wisely** - Enforce architecture - **Write BUILD files per directory** - Standard convention ### Don'ts - **Don't use glob for deps** - Explicit is better - **Don't commit bazel-\* dirs** - Add to .gitignore - **Don't skip WORKSPACE setup** - Foundation of build - **Don't ignore build warnings** - Technical debt ## Resources - [Bazel Documentation](https://bazel.build/docs) - [Bazel Remote Execution](https://bazel.build/docs/remote-execution) - [rules_js](https://github.com/aspect-build/rules_js)
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

e2e-testing-patterns

Master end-to-end testing with Playwright and Cypress to build

coding
⭐1
# E2E Testing Patterns Build reliable, fast, and maintainable end-to-end test suites that provide confidence to ship code quickly and catch regressions before users do. ## When to Use This Skill - Implementing end-to-end test automation - Debugging flaky or unreliable tests - Testing critical user workflows - Setting up CI/CD test pipelines - Testing across multiple browsers - Validating accessibility requirements - Testing responsive designs - Establishing E2E testing standards ## Core Concepts ### 1. E2E Testing Fundamentals **What to Test with E2E:** - Critical user journeys (login, checkout, signup) - Complex interactions (drag-and-drop, multi-step forms) - Cross-browser compatibility - Real API integration - Authentication flows **What NOT to Test with E2E:** - Unit-level logic (use unit tests) - API contracts (use integration tests) - Edge cases (too slow) - Internal implementation details ### 2. Test Philosophy **The Testing Pyramid:** ``` /\ /E2E\ ← Few, focused on critical paths /─────\ /Integr\ ← More, test component interactions /────────\ /Unit Tests\ ← Many, fast, isolated /────────────\ ``` **Best Practices:** - Test user behavior, not implementation - Keep tests independent - Make tests deterministic - Optimize for speed - Use data-testid, not CSS selectors ## Playwright Patterns ### Setup and Configuration ```typescript // playwright.config.ts import { defineConfig, devices } from "@playwright/test"; export default defineConfig({ testDir: "./e2e", timeout: 30000, expect: { timeout: 5000, }, fullyParallel: true, forbidOnly: !!process.env.CI, retries: process.env.CI ? 2 : 0, workers: process.env.CI ? 1 : undefined, reporter: [["html"], ["junit", { outputFile: "results.xml" }]], use: { baseURL: "http://localhost:3000", trace: "on-first-retry", screenshot: "only-on-failure", video: "retain-on-failure", }, projects: [ { name: "chromium", use: { ...devices["Desktop Chrome"] } }, { name: "firefox", use: { ...devices["Desktop Firefox"] } }, { name: "webkit", use: { ...devices["Desktop Safari"] } }, { name: "mobile", use: { ...devices["iPhone 13"] } }, ], }); ``` ### Pattern 1: Page Object Model ```typescript // pages/LoginPage.ts import { Page, Locator } from "@playwright/test"; export class LoginPage { readonly page: Page; readonly emailInput: Locator; readonly passwordInput: Locator; readonly loginButton: Locator; readonly errorMessage: Locator; constructor(page: Page) { this.page = page; this.emailInput = page.getByLabel("Email"); this.passwordInput = page.getByLabel("Password"); this.loginButton = page.getByRole("button", { name: "Login" }); this.errorMessage = page.getByRole("alert"); } async goto() { await this.page.goto("/login"); } async login(email: string, password: string) { await this.emailInput.fill(email); await this.passwordInput.fill(password); await this.loginButton.click(); } async getErrorMessage(): Promise<string> { return (await this.errorMessage.textContent()) ?? ""; } } // Test using Page Object import { test, expect } from "@playwright/test"; import { LoginPage } from "./pages/LoginPage"; test("successful login", async ({ page }) => { const loginPage = new LoginPage(page); await loginPage.goto(); await loginPage.login("user@example.com", "password123"); await expect(page).toHaveURL("/dashboard"); await expect(page.getByRole("heading", { name: "Dashboard" })).toBeVisible(); }); test("failed login shows error", async ({ page }) => { const loginPage = new LoginPage(page); await loginPage.goto(); await loginPage.login("invalid@example.com", "wrong"); const error = await loginPage.getErrorMessage(); expect(error).toContain("Invalid credentials"); }); ``` ### Pattern 2: Fixtures for Test Data ```typescript // fixtures/test-data.ts import { test as base } from "@playwright/test"; type TestData = { testUser: { email: string; password: string; name: string; }; adminUser: { email: string; password: string; }; }; export const test = base.extend<TestData>({ testUser: async ({}, use) => { const user = { email: `test-${Date.now()}@example.com`, password: "Test123!@#", name: "Test User", }; // Setup: Create user in database await createTestUser(user); await use(user); // Teardown: Clean up user await deleteTestUser(user.email); }, adminUser: async ({}, use) => { await use({ email: "admin@example.com", password: process.env.ADMIN_PASSWORD!, }); }, }); // Usage in tests import { test } from "./fixtures/test-data"; test("user can update profile", async ({ page, testUser }) => { await page.goto("/login"); await page.getByLabel("Email").fill(testUser.email); await page.getByLabel("Password").fill(testUser.password); await page.getByRole("button", { name: "Login" }).click(); await page.goto("/profile"); await page.getByLabel("Name").fill("Updated Name"); await page.getByRole("button", { name: "Save" }).click(); await expect(page.getByText("Profile updated")).toBeVisible(); }); ``` ### Pattern 3: Waiting Strategies ```typescript // āŒ Bad: Fixed timeouts await page.waitForTimeout(3000); // Flaky! // āœ… Good: Wait for specific conditions await page.waitForLoadState("networkidle"); await page.waitForURL("/dashboard"); await page.waitForSelector('[data-testid="user-profile"]'); // āœ… Better: Auto-waiting with assertions await expect(page.getByText("Welcome")).toBeVisible(); await expect(page.getByRole("button", { name: "Submit" })).toBeEnabled(); // Wait for API response const responsePromise = page.waitForResponse( (response) => response.url().includes("/api/users") && response.status() === 200, ); await page.getByRole("button", { name: "Load Users" }).click(); const response = await responsePromise; const data = await response.json(); expect(data.users).toHaveLength(10); // Wait for multiple conditions await Promise.all([ page.waitForURL("/success"), page.waitForLoadState("networkidle"), expect(page.getByText("Payment successful")).toBeVisible(), ]); ``` ### Pattern 4: Network Mocking and Interception ```typescript // Mock API responses test("displays error when API fails", async ({ page }) => { await page.route("**/api/users", (route) => { route.fulfill({ status: 500, contentType: "application/json", body: JSON.stringify({ error: "Internal Server Error" }), }); }); await page.goto("/users"); await expect(page.getByText("Failed to load users")).toBeVisible(); }); // Intercept and modify requests test("can modify API request", async ({ page }) => { await page.route("**/api/users", async (route) => { const request = route.request(); const postData = JSON.parse(request.postData() || "{}"); // Modify request postData.role = "admin"; await route.continue({ postData: JSON.stringify(postData), }); }); // Test continues... }); // Mock third-party services test("payment flow with mocked Stripe", async ({ page }) => { await page.route("**/api/stripe/**", (route) => { route.fulfill({ status: 200, body: JSON.stringify({ id: "mock_payment_id", status: "succeeded", }), }); }); // Test payment flow with mocked response }); ``` ## Cypress Patterns ### Setup and Configuration ```typescript // cypress.config.ts import { defineConfig } from "cypress"; export default defineConfig({ e2e: { baseUrl: "http://localhost:3000", viewportWidth: 1280, viewportHeight: 720, video: false, screenshotOnRunFailure: true, defaultCommandTimeout: 10000, requestTimeout: 10000, setupNodeEvents(on, config) { // Implement node event listeners }, }, }); ``` ### Pattern 1: Custom Commands ```typescript // cypress/support/commands.ts declare global { namespace Cypress { interface Chainable { login(email: string, password: string): Chainable<void>; createUser(userData: UserData): Chainable<User>; dataCy(value: string): Chainable<JQuery<HTMLElement>>; } } } Cypress.Commands.add("login", (email: string, password: string) => { cy.visit("/login"); cy.get('[data-testid="email"]').type(email); cy.get('[data-testid="password"]').type(password); cy.get('[data-testid="login-button"]').click(); cy.url().should("include", "/dashboard"); }); Cypress.Commands.add("createUser", (userData: UserData) => { return cy.request("POST", "/api/users", userData).its("body"); }); Cypress.Commands.add("dataCy", (value: string) => { return cy.get(`[data-cy="${value}"]`); }); // Usage cy.login("user@example.com", "password"); cy.dataCy("submit-button").click(); ``` ### Pattern 2: Cypress Intercept ```typescript // Mock API calls cy.intercept("GET", "/api/users", { statusCode: 200, body: [ { id: 1, name: "John" }, { id: 2, name: "Jane" }, ], }).as("getUsers"); cy.visit("/users"); cy.wait("@getUsers"); cy.get('[data-testid="user-list"]').children().should("have.length", 2); // Modify responses cy.intercept("GET", "/api/users", (req) => { req.reply((res) => { // Modify response res.body.users = res.body.users.slice(0, 5); res.send(); }); }); // Simulate slow network cy.intercept("GET", "/api/data", (req) => { req.reply((res) => { res.delay(3000); // 3 second delay res.send(); }); }); ``` ## Advanced Patterns ### Pattern 1: Visual Regression Testing ```typescript // With Playwright import { test, expect } from "@playwright/test"; test("homepage looks correct", async ({ page }) => { await page.goto("/"); await expect(page).toHaveScreenshot("homepage.png", { fullPage: true, maxDiffPixels: 100, }); }); test("button in all states", async ({ page }) => { await page.goto("/components"); const button = page.getByRole("button", { name: "Submit" }); // Default state await expect(button).toHaveScreenshot("button-default.png"); // Hover state await button.hover(); await expect(button).toHaveScreenshot("button-hover.png"); // Disabled state await button.evaluate((el) => el.setAttribute("disabled", "true")); await expect(button).toHaveScreenshot("button-disabled.png"); }); ``` ### Pattern 2: Parallel Testing with Sharding ```typescript // playwright.config.ts export default defineConfig({ projects: [ { name: "shard-1", use: { ...devices["Desktop Chrome"] }, grepInvert: /@slow/, shard: { current: 1, total: 4 }, }, { name: "shard-2", use: { ...devices["Desktop Chrome"] }, shard: { current: 2, total: 4 }, }, // ... more shards ], }); // Run in CI // npx playwright test --shard=1/4 // npx playwright test --shard=2/4 ``` ### Pattern 3: Accessibility Testing ```typescript // Install: npm install @axe-core/playwright import { test, expect } from "@playwright/test"; import AxeBuilder from "@axe-core/playwright"; test("page should not have accessibility violations", async ({ page }) => { await page.goto("/"); const accessibilityScanResults = await new AxeBuilder({ page }) .exclude("#third-party-widget") .analyze(); expect(accessibilityScanResults.violations).toEqual([]); }); test("form is accessible", async ({ page }) => { await page.goto("/signup"); const results = await new AxeBuilder({ page }).include("form").analyze(); expect(results.violations).toEqual([]); }); ``` ## Best Practices 1. **Use Data Attributes**: `data-testid` or `data-cy` for stable selectors 2. **Avoid Brittle Selectors**: Don't rely on CSS classes or DOM structure 3. **Test User Behavior**: Click, type, see - not implementation details 4. **Keep Tests Independent**: Each test should run in isolation 5. **Clean Up Test Data**: Create and destroy test data in each test 6. **Use Page Objects**: Encapsulate page logic 7. **Meaningful Assertions**: Check actual user-visible behavior 8. **Optimize for Speed**: Mock when possible, parallel execution ```typescript // āŒ Bad selectors cy.get(".btn.btn-primary.submit-button").click(); cy.get("div > form > div:nth-child(2) > input").type("text"); // āœ… Good selectors cy.getByRole("button", { name: "Submit" }).click(); cy.getByLabel("Email address").type("user@example.com"); cy.get('[data-testid="email-input"]').type("user@example.com"); ``` ## Common Pitfalls - **Flaky Tests**: Use proper waits, not fixed timeouts - **Slow Tests**: Mock external APIs, use parallel execution - **Over-Testing**: Don't test every edge case with E2E - **Coupled Tests**: Tests should not depend on each other - **Poor Selectors**: Avoid CSS classes and nth-child - **No Cleanup**: Clean up test data after each test - **Testing Implementation**: Test user behavior, not internals ## Debugging Failing Tests ```typescript // Playwright debugging // 1. Run in headed mode npx playwright test --headed // 2. Run in debug mode npx playwright test --debug // 3. Use trace viewer await page.screenshot({ path: 'screenshot.png' }); await page.video()?.saveAs('video.webm'); // 4. Add test.step for better reporting test('checkout flow', async ({ page }) => { await test.step('Add item to cart', async () => { await page.goto('/products'); await page.getByRole('button', { name: 'Add to Cart' }).click(); }); await test.step('Proceed to checkout', async () => { await page.goto('/cart'); await page.getByRole('button', { name: 'Checkout' }).click(); }); }); // 5. Inspect page state await page.pause(); // Pauses execution, opens inspector ``` ## Resources - **references/playwright-best-practices.md**: Playwright-specific patterns - **references/cypress-best-practices.md**: Cypress-specific patterns - **references/flaky-test-debugging.md**: Debugging unreliable tests - **assets/e2e-testing-checklist.md**: What to test with E2E - **assets/selector-strategies.md**: Finding reliable selectors - **scripts/test-analyzer.ts**: Analyze test flakiness and duration
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

git-advanced-workflows

Master advanced Git workflows including rebasing, cherry-picking,

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

monorepo-management

Master monorepo management with Turborepo, Nx, and pnpm workspaces

coding
⭐1
# Monorepo Management Build efficient, scalable monorepos that enable code sharing, consistent tooling, and atomic changes across multiple packages and applications. ## When to Use This Skill - Setting up new monorepo projects - Migrating from multi-repo to monorepo - Optimizing build and test performance - Managing shared dependencies - Implementing code sharing strategies - Setting up CI/CD for monorepos - Versioning and publishing packages - Debugging monorepo-specific issues ## Core Concepts ### 1. Why Monorepos? **Advantages:** - Shared code and dependencies - Atomic commits across projects - Consistent tooling and standards - Easier refactoring - Simplified dependency management - Better code visibility **Challenges:** - Build performance at scale - CI/CD complexity - Access control - Large Git repository ### 2. Monorepo Tools **Package Managers:** - pnpm workspaces (recommended) - npm workspaces - Yarn workspaces **Build Systems:** - Turborepo (recommended for most) - Nx (feature-rich, complex) - Lerna (older, maintenance mode) ## Turborepo Setup ### Initial Setup ```bash # Create new monorepo npx create-turbo@latest my-monorepo cd my-monorepo # Structure: # apps/ # web/ - Next.js app # docs/ - Documentation site # packages/ # ui/ - Shared UI components # config/ - Shared configurations # tsconfig/ - Shared TypeScript configs # turbo.json - Turborepo configuration # package.json - Root package.json ``` ### Configuration ```json // turbo.json { "$schema": "https://turbo.build/schema.json", "globalDependencies": ["**/.env.*local"], "pipeline": { "build": { "dependsOn": ["^build"], "outputs": ["dist/**", ".next/**", "!.next/cache/**"] }, "test": { "dependsOn": ["build"], "outputs": ["coverage/**"] }, "lint": { "outputs": [] }, "dev": { "cache": false, "persistent": true }, "type-check": { "dependsOn": ["^build"], "outputs": [] } } } ``` ```json // package.json (root) { "name": "my-monorepo", "private": true, "workspaces": ["apps/*", "packages/*"], "scripts": { "build": "turbo run build", "dev": "turbo run dev", "test": "turbo run test", "lint": "turbo run lint", "format": "prettier --write \"**/*.{ts,tsx,md}\"", "clean": "turbo run clean && rm -rf node_modules" }, "devDependencies": { "turbo": "^1.10.0", "prettier": "^3.0.0", "typescript": "^5.0.0" }, "packageManager": "pnpm@8.0.0" } ``` ### Package Structure ```json // packages/ui/package.json { "name": "@repo/ui", "version": "0.0.0", "private": true, "main": "./dist/index.js", "types": "./dist/index.d.ts", "exports": { ".": { "import": "./dist/index.js", "types": "./dist/index.d.ts" }, "./button": { "import": "./dist/button.js", "types": "./dist/button.d.ts" } }, "scripts": { "build": "tsup src/index.ts --format esm,cjs --dts", "dev": "tsup src/index.ts --format esm,cjs --dts --watch", "lint": "eslint src/", "type-check": "tsc --noEmit" }, "devDependencies": { "@repo/tsconfig": "workspace:*", "tsup": "^7.0.0", "typescript": "^5.0.0" }, "dependencies": { "react": "^18.2.0" } } ``` ## pnpm Workspaces ### Setup ```yaml # pnpm-workspace.yaml packages: - "apps/*" - "packages/*" - "tools/*" ``` ```json // .npmrc # Hoist shared dependencies shamefully-hoist=true # Strict peer dependencies auto-install-peers=true strict-peer-dependencies=true # Performance store-dir=~/.pnpm-store ``` ### Dependency Management ```bash # Install dependency in specific package pnpm add react --filter @repo/ui pnpm add -D typescript --filter @repo/ui # Install workspace dependency pnpm add @repo/ui --filter web # Install in all packages pnpm add -D eslint -w # Update all dependencies pnpm update -r # Remove dependency pnpm remove react --filter @repo/ui ``` ### Scripts ```bash # Run script in specific package pnpm --filter web dev pnpm --filter @repo/ui build # Run in all packages pnpm -r build pnpm -r test # Run in parallel pnpm -r --parallel dev # Filter by pattern pnpm --filter "@repo/*" build pnpm --filter "...web" build # Build web and dependencies ``` ## Nx Monorepo ### Setup ```bash # Create Nx monorepo npx create-nx-workspace@latest my-org # Generate applications nx generate @nx/react:app my-app nx generate @nx/next:app my-next-app # Generate libraries nx generate @nx/react:lib ui-components nx generate @nx/js:lib utils ``` ### Configuration ```json // nx.json { "extends": "nx/presets/npm.json", "$schema": "./node_modules/nx/schemas/nx-schema.json", "targetDefaults": { "build": { "dependsOn": ["^build"], "inputs": ["production", "^production"], "cache": true }, "test": { "inputs": ["default", "^production", "{workspaceRoot}/jest.preset.js"], "cache": true }, "lint": { "inputs": ["default", "{workspaceRoot}/.eslintrc.json"], "cache": true } }, "namedInputs": { "default": ["{projectRoot}/**/*", "sharedGlobals"], "production": [ "default", "!{projectRoot}/**/?(*.)+(spec|test).[jt]s?(x)?(.snap)", "!{projectRoot}/tsconfig.spec.json" ], "sharedGlobals": [] } } ``` ### Running Tasks ```bash # Run task for specific project nx build my-app nx test ui-components nx lint utils # Run for affected projects nx affected:build nx affected:test --base=main # Visualize dependencies nx graph # Run in parallel nx run-many --target=build --all --parallel=3 ``` ## Shared Configurations ### TypeScript Configuration ```json // packages/tsconfig/base.json { "compilerOptions": { "strict": true, "esModuleInterop": true, "skipLibCheck": true, "forceConsistentCasingInFileNames": true, "module": "ESNext", "moduleResolution": "bundler", "resolveJsonModule": true, "isolatedModules": true, "incremental": true, "declaration": true }, "exclude": ["node_modules"] } // packages/tsconfig/react.json { "extends": "./base.json", "compilerOptions": { "jsx": "react-jsx", "lib": ["ES2022", "DOM", "DOM.Iterable"] } } // apps/web/tsconfig.json { "extends": "@repo/tsconfig/react.json", "compilerOptions": { "outDir": "dist", "rootDir": "src" }, "include": ["src"], "exclude": ["node_modules", "dist"] } ``` ### ESLint Configuration ```javascript // packages/config/eslint-preset.js module.exports = { extends: [ "eslint:recommended", "plugin:@typescript-eslint/recommended", "plugin:react/recommended", "plugin:react-hooks/recommended", "prettier", ], plugins: ["@typescript-eslint", "react", "react-hooks"], parser: "@typescript-eslint/parser", parserOptions: { ecmaVersion: 2022, sourceType: "module", ecmaFeatures: { jsx: true, }, }, settings: { react: { version: "detect", }, }, rules: { "@typescript-eslint/no-unused-vars": "error", "react/react-in-jsx-scope": "off", }, }; // apps/web/.eslintrc.js module.exports = { extends: ["@repo/config/eslint-preset"], rules: { // App-specific rules }, }; ``` ## Code Sharing Patterns ### Pattern 1: Shared UI Components ```typescript // packages/ui/src/button.tsx import * as React from 'react'; export interface ButtonProps { variant?: 'primary' | 'secondary'; children: React.ReactNode; onClick?: () => void; } export function Button({ variant = 'primary', children, onClick }: ButtonProps) { return ( <button className={`btn btn-${variant}`} onClick={onClick} > {children} </button> ); } // packages/ui/src/index.ts export { Button, type ButtonProps } from './button'; export { Input, type InputProps } from './input'; // apps/web/src/app.tsx import { Button } from '@repo/ui'; export function App() { return <Button variant="primary">Click me</Button>; } ``` ### Pattern 2: Shared Utilities ```typescript // packages/utils/src/string.ts export function capitalize(str: string): string { return str.charAt(0).toUpperCase() + str.slice(1); } export function truncate(str: string, length: number): string { return str.length > length ? str.slice(0, length) + "..." : str; } // packages/utils/src/index.ts export * from "./string"; export * from "./array"; export * from "./date"; // Usage in apps import { capitalize, truncate } from "@repo/utils"; ``` ### Pattern 3: Shared Types ```typescript // packages/types/src/user.ts export interface User { id: string; email: string; name: string; role: "admin" | "user"; } export interface CreateUserInput { email: string; name: string; password: string; } // Used in both frontend and backend import type { User, CreateUserInput } from "@repo/types"; ``` ## Build Optimization ### Turborepo Caching ```json // turbo.json { "pipeline": { "build": { // Build depends on dependencies being built first "dependsOn": ["^build"], // Cache these outputs "outputs": ["dist/**", ".next/**"], // Cache based on these inputs (default: all files) "inputs": ["src/**/*.tsx", "src/**/*.ts", "package.json"] }, "test": { // Run tests in parallel, don't depend on build "cache": true, "outputs": ["coverage/**"] } } } ``` ### Remote Caching ```bash # Turborepo Remote Cache (Vercel) npx turbo login npx turbo link # Custom remote cache # turbo.json { "remoteCache": { "signature": true, "enabled": true } } ``` ## CI/CD for Monorepos ### GitHub Actions ```yaml # .github/workflows/ci.yml name: CI on: push: branches: [main] pull_request: branches: [main] jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 with: fetch-depth: 0 # For Nx affected commands - uses: pnpm/action-setup@v2 with: version: 8 - uses: actions/setup-node@v3 with: node-version: 18 cache: "pnpm" - name: Install dependencies run: pnpm install --frozen-lockfile - name: Build run: pnpm turbo run build - name: Test run: pnpm turbo run test - name: Lint run: pnpm turbo run lint - name: Type check run: pnpm turbo run type-check ``` ### Deploy Affected Only ```yaml # Deploy only changed apps - name: Deploy affected apps run: | if pnpm nx affected:apps --base=origin/main --head=HEAD | grep -q "web"; then echo "Deploying web app" pnpm --filter web deploy fi ``` ## Best Practices 1. **Consistent Versioning**: Lock dependency versions across workspace 2. **Shared Configs**: Centralize ESLint, TypeScript, Prettier configs 3. **Dependency Graph**: Keep it acyclic, avoid circular dependencies 4. **Cache Effectively**: Configure inputs/outputs correctly 5. **Type Safety**: Share types between frontend/backend 6. **Testing Strategy**: Unit tests in packages, E2E in apps 7. **Documentation**: README in each package 8. **Release Strategy**: Use changesets for versioning ## Common Pitfalls - **Circular Dependencies**: A depends on B, B depends on A - **Phantom Dependencies**: Using deps not in package.json - **Incorrect Cache Inputs**: Missing files in Turborepo inputs - **Over-Sharing**: Sharing code that should be separate - **Under-Sharing**: Duplicating code across packages - **Large Monorepos**: Without proper tooling, builds slow down ## Publishing Packages ```bash # Using Changesets pnpm add -Dw @changesets/cli pnpm changeset init # Create changeset pnpm changeset # Version packages pnpm changeset version # Publish pnpm changeset publish ``` ```yaml # .github/workflows/release.yml - name: Create Release Pull Request or Publish uses: changesets/action@v1 with: publish: pnpm release env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} NPM_TOKEN: ${{ secrets.NPM_TOKEN }} ``` ## Resources - **references/turborepo-guide.md**: Comprehensive Turborepo documentation - **references/nx-guide.md**: Nx monorepo patterns - **references/pnpm-workspaces.md**: pnpm workspace features - **assets/monorepo-checklist.md**: Setup checklist - **assets/migration-guide.md**: Multi-repo to monorepo migration - **scripts/dependency-graph.ts**: Visualize package dependencies
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

nx-workspace-patterns

Configure and optimize Nx monorepo workspaces. Use when setting up

coding
⭐1
# Nx Workspace Patterns Production patterns for Nx monorepo management. ## When to Use This Skill - Setting up new Nx workspaces - Configuring project boundaries - Optimizing CI with affected commands - Implementing remote caching - Managing dependencies between projects - Migrating to Nx ## Core Concepts ### 1. Nx Architecture ``` workspace/ ā”œā”€ā”€ apps/ # Deployable applications │ ā”œā”€ā”€ web/ │ └── api/ ā”œā”€ā”€ libs/ # Shared libraries │ ā”œā”€ā”€ shared/ │ │ ā”œā”€ā”€ ui/ │ │ └── utils/ │ └── feature/ │ ā”œā”€ā”€ auth/ │ └── dashboard/ ā”œā”€ā”€ tools/ # Custom executors/generators ā”œā”€ā”€ nx.json # Nx configuration └── workspace.json # Project configuration ``` ### 2. Library Types | Type | Purpose | Example | | --------------- | -------------------------------- | ------------------- | | **feature** | Smart components, business logic | `feature-auth` | | **ui** | Presentational components | `ui-buttons` | | **data-access** | API calls, state management | `data-access-users` | | **util** | Pure functions, helpers | `util-formatting` | | **shell** | App bootstrapping | `shell-web` | ## Templates ### Template 1: nx.json Configuration ```json { "$schema": "./node_modules/nx/schemas/nx-schema.json", "npmScope": "myorg", "affected": { "defaultBase": "main" }, "tasksRunnerOptions": { "default": { "runner": "nx/tasks-runners/default", "options": { "cacheableOperations": [ "build", "lint", "test", "e2e", "build-storybook" ], "parallel": 3 } } }, "targetDefaults": { "build": { "dependsOn": ["^build"], "inputs": ["production", "^production"], "cache": true }, "test": { "inputs": ["default", "^production", "{workspaceRoot}/jest.preset.js"], "cache": true }, "lint": { "inputs": ["default", "{workspaceRoot}/.eslintrc.json"], "cache": true }, "e2e": { "inputs": ["default", "^production"], "cache": true } }, "namedInputs": { "default": ["{projectRoot}/**/*", "sharedGlobals"], "production": [ "default", "!{projectRoot}/**/?(*.)+(spec|test).[jt]s?(x)?(.snap)", "!{projectRoot}/tsconfig.spec.json", "!{projectRoot}/jest.config.[jt]s", "!{projectRoot}/.eslintrc.json" ], "sharedGlobals": [ "{workspaceRoot}/babel.config.json", "{workspaceRoot}/tsconfig.base.json" ] }, "generators": { "@nx/react": { "application": { "style": "css", "linter": "eslint", "bundler": "webpack" }, "library": { "style": "css", "linter": "eslint" }, "component": { "style": "css" } } } } ``` ### Template 2: Project Configuration ```json // apps/web/project.json { "name": "web", "$schema": "../../node_modules/nx/schemas/project-schema.json", "sourceRoot": "apps/web/src", "projectType": "application", "tags": ["type:app", "scope:web"], "targets": { "build": { "executor": "@nx/webpack:webpack", "outputs": ["{options.outputPath}"], "defaultConfiguration": "production", "options": { "compiler": "babel", "outputPath": "dist/apps/web", "index": "apps/web/src/index.html", "main": "apps/web/src/main.tsx", "tsConfig": "apps/web/tsconfig.app.json", "assets": ["apps/web/src/assets"], "styles": ["apps/web/src/styles.css"] }, "configurations": { "development": { "extractLicenses": false, "optimization": false, "sourceMap": true }, "production": { "optimization": true, "outputHashing": "all", "sourceMap": false, "extractLicenses": true } } }, "serve": { "executor": "@nx/webpack:dev-server", "defaultConfiguration": "development", "options": { "buildTarget": "web:build" }, "configurations": { "development": { "buildTarget": "web:build:development" }, "production": { "buildTarget": "web:build:production" } } }, "test": { "executor": "@nx/jest:jest", "outputs": ["{workspaceRoot}/coverage/{projectRoot}"], "options": { "jestConfig": "apps/web/jest.config.ts", "passWithNoTests": true } }, "lint": { "executor": "@nx/eslint:lint", "outputs": ["{options.outputFile}"], "options": { "lintFilePatterns": ["apps/web/**/*.{ts,tsx,js,jsx}"] } } } } ``` ### Template 3: Module Boundary Rules ```json // .eslintrc.json { "root": true, "ignorePatterns": ["**/*"], "plugins": ["@nx"], "overrides": [ { "files": ["*.ts", "*.tsx", "*.js", "*.jsx"], "rules": { "@nx/enforce-module-boundaries": [ "error", { "enforceBuildableLibDependency": true, "allow": [], "depConstraints": [ { "sourceTag": "type:app", "onlyDependOnLibsWithTags": [ "type:feature", "type:ui", "type:data-access", "type:util" ] }, { "sourceTag": "type:feature", "onlyDependOnLibsWithTags": [ "type:ui", "type:data-access", "type:util" ] }, { "sourceTag": "type:ui", "onlyDependOnLibsWithTags": ["type:ui", "type:util"] }, { "sourceTag": "type:data-access", "onlyDependOnLibsWithTags": ["type:data-access", "type:util"] }, { "sourceTag": "type:util", "onlyDependOnLibsWithTags": ["type:util"] }, { "sourceTag": "scope:web", "onlyDependOnLibsWithTags": ["scope:web", "scope:shared"] }, { "sourceTag": "scope:api", "onlyDependOnLibsWithTags": ["scope:api", "scope:shared"] }, { "sourceTag": "scope:shared", "onlyDependOnLibsWithTags": ["scope:shared"] } ] } ] } } ] } ``` ### Template 4: Custom Generator ```typescript // tools/generators/feature-lib/index.ts import { Tree, formatFiles, generateFiles, joinPathFragments, names, readProjectConfiguration, } from "@nx/devkit"; import { libraryGenerator } from "@nx/react"; interface FeatureLibraryGeneratorSchema { name: string; scope: string; directory?: string; } export default async function featureLibraryGenerator( tree: Tree, options: FeatureLibraryGeneratorSchema, ) { const { name, scope, directory } = options; const projectDirectory = directory ? `${directory}/${name}` : `libs/${scope}/feature-${name}`; // Generate base library await libraryGenerator(tree, { name: `feature-${name}`, directory: projectDirectory, tags: `type:feature,scope:${scope}`, style: "css", skipTsConfig: false, skipFormat: true, unitTestRunner: "jest", linter: "eslint", }); // Add custom files const projectConfig = readProjectConfiguration( tree, `${scope}-feature-${name}`, ); const projectNames = names(name); generateFiles( tree, joinPathFragments(__dirname, "files"), projectConfig.sourceRoot, { ...projectNames, scope, tmpl: "", }, ); await formatFiles(tree); } ``` ### Template 5: CI Configuration with Affected ```yaml # .github/workflows/ci.yml name: CI on: push: branches: [main] pull_request: branches: [main] env: NX_CLOUD_ACCESS_TOKEN: ${{ secrets.NX_CLOUD_ACCESS_TOKEN }} jobs: main: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 with: fetch-depth: 0 - uses: actions/setup-node@v4 with: node-version: 20 cache: "npm" - name: Install dependencies run: npm ci - name: Derive SHAs for affected commands uses: nrwl/nx-set-shas@v4 - name: Run affected lint run: npx nx affected -t lint --parallel=3 - name: Run affected test run: npx nx affected -t test --parallel=3 --configuration=ci - name: Run affected build run: npx nx affected -t build --parallel=3 - name: Run affected e2e run: npx nx affected -t e2e --parallel=1 ``` ### Template 6: Remote Caching Setup ```typescript // nx.json with Nx Cloud { "tasksRunnerOptions": { "default": { "runner": "nx-cloud", "options": { "cacheableOperations": ["build", "lint", "test", "e2e"], "accessToken": "your-nx-cloud-token", "parallel": 3, "cacheDirectory": ".nx/cache" } } }, "nxCloudAccessToken": "your-nx-cloud-token" } // Self-hosted cache with S3 { "tasksRunnerOptions": { "default": { "runner": "@nx-aws-cache/nx-aws-cache", "options": { "cacheableOperations": ["build", "lint", "test"], "awsRegion": "us-east-1", "awsBucket": "my-nx-cache-bucket", "awsProfile": "default" } } } } ``` ## Common Commands ```bash # Generate new library nx g @nx/react:lib feature-auth --directory=libs/web --tags=type:feature,scope:web # Run affected tests nx affected -t test --base=main # View dependency graph nx graph # Run specific project nx build web --configuration=production # Reset cache nx reset # Run migrations nx migrate latest nx migrate --run-migrations ``` ## Best Practices ### Do's - **Use tags consistently** - Enforce with module boundaries - **Enable caching early** - Significant CI savings - **Keep libs focused** - Single responsibility - **Use generators** - Ensure consistency - **Document boundaries** - Help new developers ### Don'ts - **Don't create circular deps** - Graph should be acyclic - **Don't skip affected** - Test only what changed - **Don't ignore boundaries** - Tech debt accumulates - **Don't over-granularize** - Balance lib count ## Resources - [Nx Documentation](https://nx.dev/getting-started/intro) - [Module Boundaries](https://nx.dev/core-features/enforce-module-boundaries) - [Nx Cloud](https://nx.app/)
šŸ‘0
šŸ‘ļø0
šŸ¤– Auto-discovered
šŸ¤–system prompt•7 months ago

turborepo-caching

Configure Turborepo for efficient monorepo builds with local and

coding
⭐1
# Turborepo Caching Production patterns for Turborepo build optimization. ## When to Use This Skill - Setting up new Turborepo projects - Configuring build pipelines - Implementing remote caching - Optimizing CI/CD performance - Migrating from other monorepo tools - Debugging cache misses ## Core Concepts ### 1. Turborepo Architecture ``` Workspace Root/ ā”œā”€ā”€ apps/ │ ā”œā”€ā”€ web/ │ │ └── package.json │ └── docs/ │ └── package.json ā”œā”€ā”€ packages/ │ ā”œā”€ā”€ ui/ │ │ └── package.json │ └── config/ │ └── package.json ā”œā”€ā”€ turbo.json └── package.json ``` ### 2. Pipeline Concepts | Concept | Description | | -------------- | -------------------------------- | | **dependsOn** | Tasks that must complete first | | **cache** | Whether to cache outputs | | **outputs** | Files to cache | | **inputs** | Files that affect cache key | | **persistent** | Long-running tasks (dev servers) | ## Templates ### Template 1: turbo.json Configuration ```json { "$schema": "https://turbo.build/schema.json", "globalDependencies": [".env", ".env.local"], "globalEnv": ["NODE_ENV", "VERCEL_URL"], "pipeline": { "build": { "dependsOn": ["^build"], "outputs": ["dist/**", ".next/**", "!.next/cache/**"], "env": ["API_URL", "NEXT_PUBLIC_*"] }, "test": { "dependsOn": ["build"], "outputs": ["coverage/**"], "inputs": ["src/**/*.tsx", "src/**/*.ts", "test/**/*.ts"] }, "lint": { "outputs": [], "cache": true }, "typecheck": { "dependsOn": ["^build"], "outputs": [] }, "dev": { "cache": false, "persistent": true }, "clean": { "cache": false } } } ``` ### Template 2: Package-Specific Pipeline ```json // apps/web/turbo.json { "$schema": "https://turbo.build/schema.json", "extends": ["//"], "pipeline": { "build": { "outputs": [".next/**", "!.next/cache/**"], "env": ["NEXT_PUBLIC_API_URL", "NEXT_PUBLIC_ANALYTICS_ID"] }, "test": { "outputs": ["coverage/**"], "inputs": ["src/**", "tests/**", "jest.config.js"] } } } ``` ### Template 3: Remote Caching with Vercel ```bash # Login to Vercel npx turbo login # Link to Vercel project npx turbo link # Run with remote cache turbo build --remote-only # CI environment variables TURBO_TOKEN=your-token TURBO_TEAM=your-team ``` ```yaml # .github/workflows/ci.yml name: CI on: push: branches: [main] pull_request: env: TURBO_TOKEN: ${{ secrets.TURBO_TOKEN }} TURBO_TEAM: ${{ vars.TURBO_TEAM }} jobs: build: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - uses: actions/setup-node@v4 with: node-version: 20 cache: "npm" - name: Install dependencies run: npm ci - name: Build run: npx turbo build --filter='...[origin/main]' - name: Test run: npx turbo test --filter='...[origin/main]' ``` ### Template 4: Self-Hosted Remote Cache ```typescript // Custom remote cache server (Express) import express from "express"; import { createReadStream, createWriteStream } from "fs"; import { mkdir } from "fs/promises"; import { join } from "path"; const app = express(); const CACHE_DIR = "./cache"; // Get artifact app.get("/v8/artifacts/:hash", async (req, res) => { const { hash } = req.params; const team = req.query.teamId || "default"; const filePath = join(CACHE_DIR, team, hash); try { const stream = createReadStream(filePath); stream.pipe(res); } catch { res.status(404).send("Not found"); } }); // Put artifact app.put("/v8/artifacts/:hash", async (req, res) => { const { hash } = req.params; const team = req.query.teamId || "default"; const dir = join(CACHE_DIR, team); const filePath = join(dir, hash); await mkdir(dir, { recursive: true }); const stream = createWriteStream(filePath); req.pipe(stream); stream.on("finish", () => { res.json({ urls: [`${req.protocol}://${req.get("host")}/v8/artifacts/${hash}`], }); }); }); // Check artifact exists app.head("/v8/artifacts/:hash", async (req, res) => { const { hash } = req.params; const team = req.query.teamId || "default"; const filePath = join(CACHE_DIR, team, hash); try { await fs.access(filePath); res.status(200).end(); } catch { res.status(404).end(); } }); app.listen(3000); ``` ```json // turbo.json for self-hosted cache { "remoteCache": { "signature": false } } ``` ```bash # Use self-hosted cache turbo build --api="http://localhost:3000" --token="my-token" --team="my-team" ``` ### Template 5: Filtering and Scoping ```bash # Build specific package turbo build --filter=@myorg/web # Build package and its dependencies turbo build --filter=@myorg/web... # Build package and its dependents turbo build --filter=...@myorg/ui # Build changed packages since main turbo build --filter='...[origin/main]' # Build packages in directory turbo build --filter='./apps/*' # Combine filters turbo build --filter=@myorg/web --filter=@myorg/docs # Exclude package turbo build --filter='!@myorg/docs' # Include dependencies of changed turbo build --filter='...[HEAD^1]...' ``` ### Template 6: Advanced Pipeline Configuration ```json { "$schema": "https://turbo.build/schema.json", "pipeline": { "build": { "dependsOn": ["^build"], "outputs": ["dist/**"], "inputs": ["$TURBO_DEFAULT$", "!**/*.md", "!**/*.test.*"] }, "test": { "dependsOn": ["^build"], "outputs": ["coverage/**"], "inputs": ["src/**", "tests/**", "*.config.*"], "env": ["CI", "NODE_ENV"] }, "test:e2e": { "dependsOn": ["build"], "outputs": [], "cache": false }, "deploy": { "dependsOn": ["build", "test", "lint"], "outputs": [], "cache": false }, "db:generate": { "cache": false }, "db:push": { "cache": false, "dependsOn": ["db:generate"] }, "@myorg/web#build": { "dependsOn": ["^build", "@myorg/db#db:generate"], "outputs": [".next/**"], "env": ["NEXT_PUBLIC_*"] } } } ``` ### Template 7: Root package.json Setup ```json { "name": "my-turborepo", "private": true, "workspaces": ["apps/*", "packages/*"], "scripts": { "build": "turbo build", "dev": "turbo dev", "lint": "turbo lint", "test": "turbo test", "clean": "turbo clean && rm -rf node_modules", "format": "prettier --write \"**/*.{ts,tsx,md}\"", "changeset": "changeset", "version-packages": "changeset version", "release": "turbo build --filter=./packages/* && changeset publish" }, "devDependencies": { "turbo": "^1.10.0", "prettier": "^3.0.0", "@changesets/cli": "^2.26.0" }, "packageManager": "npm@10.0.0" } ``` ## Debugging Cache ```bash # Dry run to see what would run turbo build --dry-run # Verbose output with hashes turbo build --verbosity=2 # Show task graph turbo build --graph # Force no cache turbo build --force # Show cache status turbo build --summarize # Debug specific task TURBO_LOG_VERBOSITY=debug turbo build --filter=@myorg/web ``` ## Best Practices ### Do's - **Define explicit inputs** - Avoid cache invalidation - **Use workspace protocol** - `"@myorg/ui": "workspace:*"` - **Enable remote caching** - Share across CI and local - **Filter in CI** - Build only affected packages - **Cache build outputs** - Not source files ### Don'ts - **Don't cache dev servers** - Use `persistent: true` - **Don't include secrets in env** - Use runtime env vars - **Don't ignore dependsOn** - Causes race conditions - **Don't over-filter** - May miss dependencies ## Resources - [Turborepo Documentation](https://turbo.build/repo/docs) - [Caching Guide](https://turbo.build/repo/docs/core-concepts/caching) - [Remote Caching](https://turbo.build/repo/docs/core-concepts/remote-caching)
šŸ‘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