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

> Discover how Open-Notebook maps FastAPI routers to services and business logic using a three-layer architecture. Learn how HTTP endpoints delegate tasks for efficient application design.

- Repository: [Luis Novo/open-notebook](https://github.com/lfnovo/open-notebook)
- Tags: how-to-guide
- Published: 2026-06-15

---

**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`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/sources.py) | `api/sources_service.py → SourcesService` | CRUD + async processing of *Source* objects |
| [`api/routers/notebooks.py`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/notebooks.py) | `api/notebook_service.py → NotebookService` | Notebook CRUD and association handling |
| [`api/routers/models.py`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/models.py) | `api/models_service.py → ModelsService` | Managing AI model configuration |
| [`api/routers/search.py`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/search.py) | `api/search_service.py → SearchService` | Vector‑search queries and result aggregation |
| [`api/routers/chat.py`](https://github.com/lfnovo/open-notebook/blob/main/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.

```python

# 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.

```python

# 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`.

```python

# 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:

```python

# 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:

```python

# 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

```

```python

# 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`](https://github.com/lfnovo/open-notebook/blob/main/api/routers/sources.py) | FastAPI router defining `/sources` endpoints |
| [`api/sources_service.py`](https://github.com/lfnovo/open-notebook/blob/main/api/sources_service.py) | Service class implementing source‑related business logic |
| [`api/client.py`](https://github.com/lfnovo/open-notebook/blob/main/api/client.py) | Thin HTTP client used by services to call internal APIs |
| [`open_notebook/domain/source.py`](https://github.com/lfnovo/open-notebook/blob/main/open_notebook/domain/source.py) | Domain entity with methods like `save()`, `add_to_notebook()`, `get_status()` |
| [`api/command_service.py`](https://github.com/lfnovo/open-notebook/blob/main/api/command_service.py) | Central service queuing background commands via `CommandService.submit_command_job()` |
| [`commands/source_commands.py`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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`](https://github.com/lfnovo/open-notebook/blob/main/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.