Extending SnackBase

from src.snackbase.infrastructure.api.app import app

@app.hook.on_record_after_create("posts")
async def custom_logic(record: dict, context: Context):
    """Custom business logic after post creation."""
    # Your code here
    pass

from fastapi import APIRouter
from src.snackbase.infrastructure.api.app import app

custom_router = APIRouter(prefix="/custom", tags=["custom"])

@custom_router.get("/analytics")
async def get_analytics(context: Context = Depends(get_context)):
    """Custom analytics endpoint."""
    return {"analytics": "data"}

# Register in app.py
app.include_router(custom_router, prefix=API_PREFIX)

# Create migration
# alembic/versions/xxx_add_custom_table.py

def upgrade():
    op.create_table(
        "custom_features",
        sa.Column("id", sa.String(50), primary_key=True),
        sa.Column("account_id", sa.String(10), nullable=False, index=True),
        sa.Column("name", sa.String(255), nullable=False),
        sa.Column("config", sa.JSON, nullable=True),
        sa.ForeignKeyConstraint(["account_id"], ["accounts.id"])
    )

from src.snackbase.infrastructure.api.app import app

@app.middleware("http")
async def custom_middleware(request: Request, call_next):
    """Custom middleware for all requests."""
    # Pre-processing
    start_time = time.time()

    # Call next middleware/route
    response = await call_next(request)

    # Post-processing
    duration = time.time() - start_time
    response.headers["X-Process-Time"] = str(duration)

    return response

# src/snackbase/infrastructure/services/custom_service.py
from typing import Any

class CustomAnalyticsService:
    """Custom analytics service."""

    def __init__(self, db: AsyncSession):
        self._db = db

    async def calculate_metrics(self, account_id: str) -> dict[str, Any]:
        """Calculate custom metrics for account."""
        # Your logic here
        return {"metrics": "data"}

# Use in routes
@router.get("/analytics")
async def get_analytics(
    context: Context = Depends(get_context),
    db: AsyncSession = Depends(get_db)
):
    service = CustomAnalyticsService(db)
    return await service.calculate_metrics(context.account_id)

What do you want to do?
|
+-- Add business logic to existing operations
|   +--> Use Hooks
|
+-- Create new API endpoints
|   +--> Create Custom Router
|
+-- Store custom data structures
|   |
|   +-- Simple, per-record metadata
|   |   +--> Use JSON field in existing collection
|   |
|   +-- Complex, relational data
|       +--> Create Custom Table
|
+-- Modify all requests/responses
|   +--> Use Middleware
|
+-- Integrate external services
    +--> Create Custom Service

# Custom repository
class CustomRepository:
    def __init__(self, session: AsyncSession):
        self._session = session

    async def custom_query(self, account_id: str) -> list[dict]:
        """Custom database query."""
        result = await self._session.execute(
            select(CustomTable).where(
                CustomTable.account_id == account_id
            )
        )
        return [row.__dict__ for row in result.scalars()]

# src/snackbase/domain/services/analytics_service.py
class AnalyticsService:
    """Analytics business logic."""

    async def generate_report(
        self,
        account_id: str,
        date_range: DateRange
    ) -> Report:
        """Generate analytics report."""
        # Business logic here
        pass

# Custom router with authentication
from src.snackbase.infrastructure.api.dependencies import (
    get_context,
    require_permission
)

router = APIRouter(prefix="/analytics", tags=["analytics"])

@router.get("/report")
async def get_report(
    context: Context = Depends(get_context),
    authorized: bool = Depends(require_permission("analytics", "read"))
):
    """Get analytics report (requires permission)."""
    # Your logic
    pass

# Custom auth provider
class CustomAuthProvider:
    async def authenticate(
        self,
        credentials: dict
    ) -> AuthResult:
        """Custom authentication logic."""
        # Integrate with external auth provider
        pass

# Register in auth service
auth_service.register_provider("custom", CustomAuthProvider())

+-----------------------------------------------------+
|                   API Layer                         |
|  (Routers, Controllers, Middleware)                 |
+---------------------+-------------------------------+
                      |
+---------------------+-------------------------------+
|                 Application Layer                   |
|  (Use Cases, Orchestration, Services)               |
+---------------------+-------------------------------+
                      |
+---------------------+-------------------------------+
|                  Domain Layer                       |
|  (Entities, Business Logic, Interfaces)             |
+---------------------+-------------------------------+
                      |
+---------------------+-------------------------------+
|              Infrastructure Layer                   |
|  (Database, External APIs, Storage)                 |
+-----------------------------------------------------+

CORRECT: Dependencies point inward
   Router -> Service -> Repository -> Database

INCORRECT: Dependencies point outward
   Repository -> Service -> Router

src/snackbase/
├── extensions/              # Your extensions
│   ├── analytics/          # Analytics feature
│   │   ├── router.py       # API endpoints
│   │   ├── service.py      # Business logic
│   │   └── repository.py   # Data access
│   └── integrations/       # External integrations
│       └── slack.py        # Slack integration

# BAD: Modifying core files
# src/snackbase/infrastructure/api/routes/users_router.py
# (Adding custom logic here will conflict with updates)

# GOOD: Separate extension file
# src/snackbase/extensions/custom_users.py
# (Separate file survives updates)

# .env
ENABLE_ANALYTICS_FEATURE=true
SLACK_WEBHOOK_URL=https://hooks.slack.com/...
CUSTOM_API_KEY=your-key-here

# config.py
from pydantic import Settings

class ExtensionSettings(BaseSettings):
    enable_analytics: bool = False
    slack_webhook_url: str | None = None
    custom_api_key: str | None = None

# 1. Create custom table (migration)
op.create_table(
    "page_views",
    sa.Column("id", sa.String(50), primary_key=True),
    sa.Column("account_id", sa.String(10)),
    sa.Column("path", sa.String(255)),
    sa.Column("views", sa.Integer),
    sa.Column("date", sa.Date)
)

# 2. Create service
class AnalyticsService:
    async def get_page_views(self, account_id: str, days: int = 30):
        """Get page views for last N days."""
        # Query and aggregate
        pass

# 3. Create router
@router.get("/analytics/page-views")
async def page_views(
    days: int = 30,
    context: Context = Depends(get_context)
):
    service = AnalyticsService(db)
    return await service.get_page_views(context.account_id, days)

# 1. Create hook
@app.hook.on_record_after_create("posts")
async def notify_slack(record: dict, context: Context):
    """Send Slack notification on post creation."""
    if record.get("status") == "published":
        await slack_service.send_notification(
            webhook_url=settings.slack_webhook_url,
            message=f"New post: {record['title']}"
        )

# 2. Create service
class SlackService:
    async def send_notification(self, webhook_url: str, message: str):
        """Send notification to Slack."""
        async with httpx.AsyncClient() as client:
            await client.post(webhook_url, json={"text": message})

# 1. Create hook
@app.hook.on_record_before_create("posts")
async def validate_post_content(record: dict, context: Context):
    """Validate post content before creation."""
    content = record.get("content", "")

    # Custom validation
    if len(content) < 50:
        raise HookAbortException(
            message="Content must be at least 50 characters",
            status_code=400
        )

    # Check for prohibited words
    prohibited = ["spam", "advertisement"]
    if any(word in content.lower() for word in prohibited):
        raise HookAbortException(
            message="Content contains prohibited words",
            status_code=400
        )

# Custom router with permissions
router = APIRouter(prefix="/reports", tags=["reports"])

@router.get("/sales")
async def sales_report(
    context: Context = Depends(get_context),
    authorized: bool = Depends(require_permission("reports", "read"))
):
    """Generate sales report (requires reports:read permission)."""
    # Generate report
    report = await report_service.generate_sales_report(context.account_id)
    return report

# Register permission
# Via UI or API:
# {
#   "role": "manager",
#   "collection": "reports",
#   "read": true,
#   "create": false,
#   "update": false,
#   "delete": false
# }