How to Add a New API Endpoint Using the Service Router Pattern in Open Notebook

To add a new API endpoint in Open Notebook, you create a service class for business logic, define Pydantic schemas for validation, implement an APIRouter for HTTP handling, and register the router in api/main.py alongside the existing routes.

Open Notebook uses FastAPI with a strict service-router pattern that separates HTTP concerns from business logic. This architecture, found in the lfnovo/open-notebook repository, ensures that routes remain thin while domain operations live in reusable, testable service classes.

Understanding the Service-Router Architecture

The codebase organizes API functionality into four distinct layers. First, router modules in api/routers/*.py declare HTTP routes using FastAPI's APIRouter and handle request/response serialization. Second, service classes in api/*_service.py encapsulate all business logic, including database calls via domain models and async operations. Third, domain models in open_notebook/domain/*.py provide the data layer that services interact with directly. Finally, error mapping occurs at the router level, where domain-specific exceptions like InvalidInputError or NotFoundError translate into appropriate HTTPException status codes.

This separation allows the existing notes endpoint to serve as a reference implementation. In api/routers/notes.py, routes like GET /notes delegate immediately to NotesService methods defined in api/notes_service.py, which manipulate the Note class from open_notebook/domain/notebook.py.

Step-by-Step Implementation Guide

Step 1: Define Pydantic Schemas in api/models.py

Create request and response models that validate incoming data and standardize outgoing payloads. Add these to the existing api/models.py file to keep schema definitions centralized.

# api/models.py

from pydantic import BaseModel, Field
from typing import Optional

class TagCreate(BaseModel):
    name: str = Field(..., description="Tag name")
    color: Optional[str] = Field(None, description="Hex colour, e.g. #ffcc00")

class TagResponse(BaseModel):
    id: str
    name: str
    color: Optional[str]
    created: str
    updated: str

Step 2: Create the Service Layer in api/tag_service.py

Implement a service class that handles validation, database operations, and domain logic. Services raise domain-specific exceptions rather than HTTP exceptions, keeping them framework-agnostic.

# api/tag_service.py

from loguru import logger
from open_notebook.domain.notebook import Tag
from open_notebook.exceptions import InvalidInputError, NotFoundError

class TagService:
    async def create_tag(self, name: str, color: str | None = None) -> Tag:
        if not name:
            raise InvalidInputError("Tag name cannot be empty")
        tag = Tag(name=name, color=color)
        await tag.save()
        return tag

    async def get_tag(self, tag_id: str) -> Tag:
        return await Tag.get(tag_id)

    async def list_tags(self) -> list[Tag]:
        return await Tag.get_all(order_by="created desc")

Step 3: Build the Router in api/routers/tags.py

Define the APIRouter instance and path operations that call service methods. The router catches domain exceptions and converts them to appropriate HTTP responses using HTTPException.

# api/routers/tags.py

from fastapi import APIRouter, HTTPException
from api.models import TagCreate, TagResponse
from api.tag_service import TagService

router = APIRouter()
service = TagService()

@router.post("/tags", response_model=TagResponse)
async def create_tag(payload: TagCreate):
    try:
        tag = await service.create_tag(payload.name, payload.color)
        return TagResponse(
            id=tag.id or "",
            name=tag.name,
            color=tag.color,
            created=str(tag.created),
            updated=str(tag.updated),
        )
    except InvalidInputError as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/tags/{tag_id}", response_model=TagResponse)
async def get_tag(tag_id: str):
    try:
        tag = await service.get_tag(tag_id)
        return TagResponse(
            id=tag.id or "",
            name=tag.name,
            color=tag.color,
            created=str(tag.created),
            updated=str(tag.updated),
        )
    except NotFoundError:
        raise HTTPException(status_code=404, detail="Tag not found")

@router.get("/tags", response_model=list[TagResponse])
async def list_tags():
    tags = await service.list_tags()
    return [
        TagResponse(
            id=t.id or "",
            name=t.name,
            color=t.color,
            created=str(t.created),
            updated=str(t.updated),
        )
        for t in tags
    ]

Step 4: Register the Router in api/main.py

Import the new router module and add it to the existing router imports. The FastAPI application factory automatically includes all imported routers.

# api/main.py

from api.routers import (
    # ... existing routers ...

    tags,          # <-- newly added

)

Step 5: Write Tests in tests/

Add async tests that verify both the endpoint integration and service logic. Use httpx.AsyncClient to test the full request lifecycle.

# tests/test_tags_api.py

import pytest
from httpx import AsyncClient
from api.main import app

@pytest.mark.asyncio
async def test_create_and_get_tag():
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Create

        resp = await client.post("/tags", json={"name": "demo", "color": "#123456"})
        assert resp.status_code == 200
        tag_id = resp.json()["id"]

        # Retrieve

        get_resp = await client.get(f"/tags/{tag_id}")
        assert get_resp.status_code == 200
        assert get_resp.json()["name"] == "demo"

Run the full test suite to ensure compatibility with existing endpoints:

uv run pytest

Key Design Principles and Benefits

Separation of concerns keeps HTTP handling logic decoupled from business rules, making the codebase easier to navigate and modify. Reusability allows service methods to be called from CLI commands, background jobs, or other routers without duplicating logic. Testability improves because services are pure async functions that can be unit-tested without instantiating FastAPI dependencies or making HTTP requests. Consistent error handling emerges from the pattern where services raise domain exceptions and routers translate these to HTTP status codes, as seen throughout the Open Notebook codebase.

Summary

• Service classes live in api/*_service.py and encapsulate all business logic, database calls, and validation
• Routers in api/routers/*.py handle HTTP concerns and exception mapping using APIRouter
• Schemas in api/models.py define Pydantic models for request/validation and response serialization
• Registration occurs in api/main.py by importing the router module alongside existing ones
• Testing covers both service unit tests and full integration tests using httpx.AsyncClient

Frequently Asked Questions

What is the service router pattern in FastAPI?

The service router pattern is an architectural approach where FastAPI routers handle only HTTP-specific concerns like routing, request parsing, and response formatting, while delegating all business logic to dedicated service classes. In Open Notebook, this pattern appears in files like api/routers/notes.py and api/notes_service.py, ensuring that domain operations remain isolated from web framework details.

Where should I place domain-specific exceptions in Open Notebook?

Domain-specific exceptions such as InvalidInputError and NotFoundError should be defined in the exceptions module (referenced as open_notebook.exceptions) and raised within your service classes. The router then catches these exceptions and converts them to HTTPException instances with appropriate status codes, maintaining a clean separation between domain logic and HTTP protocol details.

Can I reuse service methods across different routers?

Yes, the service layer is designed for reusability. Since services are Python classes with no FastAPI dependencies, methods like TagService.create_tag() can be imported and called from other routers, CLI commands, or background task workers without modification. This avoids code duplication and ensures consistent business logic across different entry points.

How do I handle database operations when adding a new endpoint?

Database operations should occur within the service class methods, not the router. The service interacts with domain models located in open_notebook/domain/*.py (such as Tag or Note classes) which handle the actual persistence logic. This keeps the service focused on business rules while the domain models manage data access patterns.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}