Add Persistence and Run the Service

This page wires the SQLAlchemy models from page 4 into the FastAPI app from page 5, then runs it. You finish with a service that reads and writes a database and serves its own docs.

Create the engine and a session dependency

Spring Boot configures a DataSource and hands each request a transactional EntityManager. In FastAPI you create the engine once and yield a Session per request through a dependency:

# src/my_service/database.py
from typing import Annotated
from fastapi import Depends
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from my_service.config import settings

engine = create_engine(settings.database_url)

def get_session():
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]

The with block closes the session after the response, the way Spring closes the persistence context at the end of a request. A handler that declares session: SessionDep gets a live session.

Create tables at startup with lifespan

Spring runs @PostConstruct hooks and schema setup as the context loads. FastAPI runs startup and shutdown work in a lifespan handler, an async context manager passed to the app:

# src/my_service/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from my_service.database import engine
from my_service.models import Base
from my_service.users.router import router as users_router

@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(engine)   # for real schemas, use Alembic migrations
    yield

app = FastAPI(lifespan=lifespan)
app.include_router(users_router)

include_router mounts the users routes, the way component scanning registers a @RestController. For real schema management, reach for Alembic, the SQLAlchemy counterpart to Flyway or Liquibase, rather than create_all.

Use the session in a route

from fastapi import HTTPException
from my_service.database import SessionDep
from my_service.models import User

@router.get("/{user_id}", response_model=UserOut)
def get_user(user_id: int, session: SessionDep):
    user = session.get(User, user_id)
    if user is None:
        raise HTTPException(status_code=404, detail="user not found")
    return user

@router.post("", response_model=UserOut, status_code=201)
def create_user(body: CreateUser, session: SessionDep):
    user = User(email=body.email, display_name=body.display_name)
    session.add(user)
    session.commit()
    session.refresh(user)
    return user

response_model=UserOut with from_attributes turns the ORM object into the response shape, so you return the User directly.

Sync, async, and where the loop fits

A Spring WebFlux handler returns Mono or Flux and runs on an event loop. FastAPI supports both styles. Declare a handler async def to run it on the event loop, or def to run it in a worker thread pool. FastAPI picks the path from the keyword:

@router.get("/health")
async def health():
    return {"status": "ok"}

A blocking call inside an async def handler freezes the loop. Keep blocking database work in a def handler, or push it off the loop with asyncio.to_thread. The synchronous SQLAlchemy session above belongs in a def handler.

Run it

Spring Boot boots an embedded Tomcat through SpringApplication.run. FastAPI runs on uvicorn, driven by the FastAPI CLI that arrived with the standard extras:

uv run fastapi dev src/my_service/main.py   # development, auto-reload
uv run fastapi run src/my_service/main.py   # production

fastapi dev reloads on every change, the way spring-boot-devtools restarts the context. Open http://127.0.0.1:8000/docs for the interactive OpenAPI page.

Test the API

Spring uses MockMvc and @MockBean to drive a controller without a live server. FastAPI uses TestClient, and dependency_overrides swaps a real provider for a fake:

# tests/test_users_api.py
import pytest
from fastapi.testclient import TestClient
from my_service.main import app
from my_service.database import get_session

def fake_session():
    yield FakeSession()   # an in-memory or fixture-backed stand-in

@pytest.fixture
def client():
    app.dependency_overrides[get_session] = fake_session
    yield TestClient(app)
    app.dependency_overrides.clear()

def test_get_missing_user_returns_404(client):
    response = client.get("/users/999")
    assert response.status_code == 404

The client fixture installs the override and clears it on teardown, so each test starts clean. dependency_overrides[get_session] is your @MockBean, swapped in for the test and removed afterward. You now have a typed REST service with validation, persistence, OpenAPI docs, and tests, assembled from a few libraries you wired yourself.

The fastest way to make this concrete: take one @RestController from a service you maintain and rebuild that single endpoint in FastAPI this week, from request model to database row.