How FastAPI Routers Map to Services and Business Logic in Open‑Notebook

Open‑Notebook implements a clean three‑layer architecture where FastAPI routers act as thin HTTP endpoints that delegate all business logic to dedicated service classes, which then coordinate with domain models and background job systems.

The lfnovo/open-notebook repository demonstrates enterprise‑grade separation of concerns by explicitly mapping FastAPI routers to service layers. This architectural pattern ensures that HTTP handling remains separate from business rules, making the codebase testable, reusable, and async‑safe.

Three‑Layer Architecture Overview

Open‑Notebook organizes its backend into three distinct layers:

1. FastAPI routers – Thin HTTP endpoints that parse request data, perform minimal validation, and delegate work.
2. Service layer – Python classes that encapsulate business logic, call domain models, and interact with the database or background job system.
3. Domain layer – Pydantic models and SurrealDB entities (Source, Notebook, Asset) that represent persistent objects and contain core behaviors (e.g., Source.get_status(), Source.add_to_notebook()).

Explicit Router‑to‑Service Mapping

The mapping from a router to its service is explicit and consistent across the codebase:

Router file	Service class	Primary responsibility
api/routers/sources.py	api/sources_service.py → SourcesService	CRUD + async processing of Source objects
api/routers/notebooks.py	api/notebook_service.py → NotebookService	Notebook CRUD and association handling
api/routers/models.py	api/models_service.py → ModelsService	Managing AI model configuration
api/routers/search.py	api/search_service.py → SearchService	Vector‑search queries and result aggregation
api/routers/chat.py	api/chat_service.py → ChatService	Conversational flow orchestration

The Request Lifecycle

Understanding how FastAPI routers map to services requires tracing a request through the entire stack.

Step 1: Router Receives HTTP Request

The router is created with APIRouter() and registers routes using decorators like @router.post. The router only extracts the payload and forwards it to the service.

# api/routers/sources.py

from api.sources_service import sources_service

@router.post("/sources", response_model=SourceResponse)
async def create_source(form_data: tuple[SourceCreate, Optional[UploadFile]] = Depends(parse_source_form_data)):
    source_data, upload_file = form_data
    result = await sources_service.create_source_async(
        notebooks=source_data.notebooks,
        source_type=source_data.type,
        url=source_data.url,
        file_path=source_data.file_path,
        content=source_data.content,
        title=source_data.title,
        transformations=source_data.transformations,
        embed=source_data.embed,
        delete_source=source_data.delete_source,
    )
    return result

Step 2: Service Implements Business Rules

The service layer contains the actual business logic. SourcesService.create_source_async() builds the request payload for the API client, decides whether the operation is synchronous or asynchronous, and returns a thin data container.

# api/sources_service.py

class SourcesService:
    async def create_source_async(self, *, notebooks: List[str] = None, source_type: str = "text",
                                  url: str = None, file_path: str = None, content: str = None,
                                  title: str = None, transformations: List[str] = None,
                                  embed: bool = False, delete_source: bool = False) -> SourceProcessingResult:
        # Call the low‑level client that talks to the FastAPI server

        payload = api_client.create_source(
            notebooks=notebooks,
            source_type=source_type,
            url=url,
            file_path=file_path,
            content=content,
            title=title,
            transformations=transformations,
            embed=embed,
            delete_source=delete_source,
            async_processing=True,
        )
        # Build a domain object from the raw JSON

        source = Source(title=payload["title"], topics=payload.get("topics", []), asset=Asset(...))
        return SourceProcessingResult(
            source=source,
            is_async=True,
            command_id=payload.get("command_id"),
            status=payload.get("status"),
            processing_info=payload.get("processing_info"),
        )

Step 3: Domain Layer Handles Persistence

The service ultimately invokes methods on domain entities or submits background jobs through CommandService.

# Service method continues...

source = Source(title=..., topics=...)
await source.save()
await source.add_to_notebook(notebook_id)
command_id = await CommandService.submit_command_job(...)
source.command = ensure_record_id(command_id)
await source.save()

Step 4: Response Returns to Client

The service returns the data container, the router translates it into the declared response model (SourceResponse), and FastAPI serializes it to JSON.

Complete Implementation Examples

Creating a Source (Router to Service)

This example demonstrates the thin router pattern where all logic lives in the service:

# api/routers/sources.py

from api.sources_service import sources_service

@router.post("/sources", response_model=SourceResponse)
async def create_source(form_data: tuple[SourceCreate, Optional[UploadFile]] = Depends(parse_source_form_data)):
    source_data, upload_file = form_data
    # The router only passes the parsed data; all logic lives in the service

    result = await sources_service.create_source_async(
        notebooks=source_data.notebooks,
        source_type=source_data.type,
        url=source_data.url,
        file_path=source_data.file_path,
        content=source_data.content,
        title=source_data.title,
        transformations=source_data.transformations,
        embed=source_data.embed,
        delete_source=source_data.delete_source,
    )
    return result  # FastAPI serializes SourceProcessingResult as SourceResponse

Updating a Source (Service to Domain)

For updates, the service validates existence, applies patches, and persists the entity:

# api/routers/sources.py

@router.put("/sources/{source_id}", response_model=SourceResponse)
async def update_source(source_id: str, source_update: SourceUpdate):
    updated_source = await sources_service.update_source_id(source_id, source_update)
    return updated_source

# api/sources_service.py

class SourcesService:
    async def update_source_id(self, source_id: str, update: SourceUpdate) -> Source:
        source = await Source.get(source_id)
        if not source:
            raise HTTPException(status_code=404, detail="Source not found")
        if update.title is not None:
            source.title = update.title
        if update.topics is not None:
            source.topics = update.topics
        await source.save()
        return source

Key Files in the Architecture

File	Role
api/routers/sources.py	FastAPI router defining /sources endpoints
api/sources_service.py	Service class implementing source‑related business logic
api/client.py	Thin HTTP client used by services to call internal APIs
open_notebook/domain/source.py	Domain entity with methods like save(), add_to_notebook(), get_status()
api/command_service.py	Central service queuing background commands via CommandService.submit_command_job()
commands/source_commands.py	Command definitions for async processing pipelines

Summary

• FastAPI routers in Open‑Notebook remain thin by delegating all business logic to service classes.
• Services such as SourcesService and NotebookService encapsulate complex operations, database interactions, and async job submission.
• Domain entities like Source and Notebook handle persistence through SurrealDB and contain core business behaviors.
• This separation enables unit testing without server startup, reusability across UI clients and background jobs, and async safety through proper thread pool management.

Frequently Asked Questions

Why does Open‑Notebook separate routers from services?

Separating FastAPI routers from services allows the codebase to maintain a clear separation of concerns. Routers handle HTTP‑specific tasks like parsing multipart form data and serializing JSON responses, while services contain pure business logic that can be tested independently without spinning up a FastAPI server. This architecture also enables the same service methods to be reused by the frontend SDK and internal background workers.

How does the service layer handle asynchronous operations?

The service layer handles async operations by accepting async_processing parameters and offloading blocking work to background jobs. In api/sources_service.py, methods like create_source_async() submit tasks via CommandService.submit_command_job() and return immediately with a command_id, allowing the FastAPI router to return an HTTP 202 response while the heavy processing continues in the background.

What is the role of the domain layer versus the service layer?

The domain layer (found in open_notebook/domain/) contains Pydantic models and SurrealDB entities that represent persistent data and enforce business invariants (e.g., Source.save(), Source.add_to_notebook()). The service layer coordinates these domain objects, manages transactions, calls external APIs via api/client.py, and decides when to queue background commands. Services orchestrate; domain objects execute.

Can service methods be reused outside of FastAPI endpoints?

Yes, service methods are designed for reuse across the codebase. The same SourcesService methods used by api/routers/sources.py are also consumed by the UI client in the frontend/ directory via the Open‑Notebook SDK and by internal background jobs in the commands/ directory. This eliminates code duplication and ensures consistent business logic across sync and async execution paths.

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