agent_delta/docs/adding_features.md

Adding New Features

Adding a New API Endpoint

This example adds a "Notification" feature end-to-end, following the repository + service pattern used throughout the codebase. Routes never contain direct database calls — all data access goes through a service, which delegates to a repository.

1. Create schema in schemas/

   # schemas/notification.py
   from datetime import datetime
   from uuid import UUID
   
   from pydantic import BaseModel
   
   
   class NotificationCreate(BaseModel):
       title: str
       body: str
       channel: str = "email"
   
   
   class NotificationResponse(BaseModel):
       id: UUID
       title: str
       body: str
       channel: str
       is_read: bool
       created_at: datetime
   

2. Create model in db/models/

   # db/models/notification.py
   from uuid import uuid4
   
   from sqlalchemy import Boolean, DateTime, String, func
   from sqlalchemy.dialects.postgresql import UUID
   from sqlalchemy.orm import Mapped, mapped_column
   
   from app.db.base import Base
   
   
   class Notification(Base):
       __tablename__ = "notifications"
   
       id: Mapped[UUID] = mapped_column(
           UUID(as_uuid=True), primary_key=True, default=uuid4
       )
       title: Mapped[str] = mapped_column(String(255))
       body: Mapped[str] = mapped_column(String)
       channel: Mapped[str] = mapped_column(String(50), default="email")
       is_read: Mapped[bool] = mapped_column(Boolean, default=False)
       created_at: Mapped[DateTime] = mapped_column(
           DateTime(timezone=True), server_default=func.now()
       )
   

   Don't forget to import it in db/models/__init__.py.

3. Create repository in repositories/

   # repositories/notification.py
   from uuid import UUID
   
   from sqlalchemy import select
   from sqlalchemy.ext.asyncio import AsyncSession
   
   from app.db.models.notification import Notification
   
   
   class NotificationRepository:
       async def create(self, db: AsyncSession, **kwargs) -> Notification:
           notification = Notification(**kwargs)
           db.add(notification)
           await db.flush()
           await db.refresh(notification)
           return notification
   
       async def get_by_id(self, db: AsyncSession, notification_id: UUID) -> Notification | None:
           return await db.get(Notification, notification_id)
   
       async def list_unread(self, db: AsyncSession, limit: int = 50) -> list[Notification]:
           result = await db.execute(
               select(Notification)
               .where(Notification.is_read.is_(False))
               .order_by(Notification.created_at.desc())
               .limit(limit)
           )
           return list(result.scalars().all())
   

4. Create service in services/

   # services/notification.py
   from uuid import UUID
   
   from sqlalchemy.ext.asyncio import AsyncSession
   
   from app.core.exceptions import NotFoundError
   from app.repositories.notification import NotificationRepository
   from app.schemas.notification import NotificationCreate
   
   
   class NotificationService:
       def __init__(self, db: AsyncSession):
           self.db = db
           self.repo = NotificationRepository()
   
       async def create(self, data: NotificationCreate) -> "Notification":
           return await self.repo.create(self.db, **data.model_dump())
   
       async def get_or_raise(self, notification_id: UUID) -> "Notification":
           notification = await self.repo.get_by_id(self.db, notification_id)
           if not notification:
               raise NotFoundError(
                   message="Notification not found",
                   details={"id": str(notification_id)},
               )
           return notification
   
       async def list_unread(self) -> list["Notification"]:
           return await self.repo.list_unread(self.db)
   

5. Register dependency in api/deps.py

   from app.services.notification import NotificationService
   
   
   def get_notification_service(db: DBSession) -> NotificationService:
       """Create NotificationService instance with database session."""
       return NotificationService(db)
   
   
   NotificationSvc = Annotated[NotificationService, Depends(get_notification_service)]
   

6. Create route in api/routes/v1/

   # api/routes/v1/notifications.py
   from fastapi import APIRouter, status
   
   from app.api.deps import CurrentUser, NotificationSvc
   from app.schemas.notification import NotificationCreate, NotificationResponse
   
   router = APIRouter()
   
   
   @router.post("/", response_model=NotificationResponse, status_code=status.HTTP_201_CREATED)
   async def create_notification(
       data: NotificationCreate,
       current_user: CurrentUser,
       service: NotificationSvc,
   ):
       return await service.create(data)
   
   
   @router.get("/", response_model=list[NotificationResponse])
   async def list_unread(
       current_user: CurrentUser,
       service: NotificationSvc,
   ):
       return await service.list_unread()
   

7. Register router in api/routes/v1/__init__.py

   from app.api.routes.v1 import notifications
   
   v1_router.include_router(
       notifications.router, prefix="/notifications", tags=["notifications"]
   )
   

Adding a Custom CLI Command

Commands are auto-discovered from app/commands/.

# app/commands/my_command.py
from app.commands import command, success, error
import click

@command("my-command", help="Description of what this does")
@click.option("--name", "-n", required=True, help="Name parameter")
def my_command(name: str):
    # Your logic here
    success(f"Done: {name}")

Run with: uv run agent_delta cmd my-command --name test

Adding a Database Migration

# Create migration
uv run alembic revision --autogenerate -m "Add notifications table"

# Apply migration
uv run alembic upgrade head

# Or use CLI
uv run agent_delta db migrate -m "Add notifications table"
uv run agent_delta db upgrade