FastAPI & Databricks Apps: Things I Wish I Knew

Yesterday someone asked me about my experience with Databricks Apps and FastAPI. I went looking for a recap post to share and realized I didn’t have one, so here it is.

Additionally, I’m about to join another project with a similar stack, so I needed a refresher, which is why I generated ~90% of this post using AI.

The article reflects real project experience with actual mistakes and fixes, and it is a useful quickstarter for developers planning to use a similar stack.

Once again, ~90% of what you will read is written by AI, but prompted by me.

FastAPI

Dependency Injection

The most important pattern in FastAPI. Everything runs through it — database sessions, auth, shared config. Once you get this, the rest of the framework clicks.

def get_db():
    db = SessionLocal()
    try:
        yield db  # yield = cleanup happens after the request finishes
    finally:
        db.close()

def get_current_user(db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)):
    return decode_and_validate(token, db)

@app.get("/me")
def read_me(user: User = Depends(get_current_user)):
    return user

Key things to remember:

• Depends() builds a DAG — FastAPI resolves the chain automatically
• yield dependencies act as context managers: setup → handler runs → cleanup
• Apply Depends at the router level to protect a whole group of routes at once (e.g. all /admin routes require auth)

Full docs on dependencies

Middleware

Middleware runs on every request. Use it for cross-cutting concerns — logging, request IDs, CORS, timing.

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
    logger.info(f"{request.method} {request.url.path} - {response.status_code} - {duration:.2f}s")
    return response

CORS in particular will bite you if you forget it — the frontend simply can’t call the API without it.

Middleware docs

Background Tasks

For fire-and-forget work (light processing, sending notifications):

@app.post("/upload")
async def upload(file: UploadFile, background_tasks: BackgroundTasks):
    data = await file.read()
    background_tasks.add_task(process_file, data)  # runs after response is sent
    return {"status": "accepted"}

Important limitation: these run in the same process. If the app restarts, the task is lost. Good enough for non-critical work.

For long-running tasks, use a proper task queue instead — Celery (with Redis or RabbitMQ as broker) is the most common choice. The pattern: route returns a job ID immediately, worker picks up the task in the background, client polls for status. ARQ is a lighter async alternative if you’re already on Redis.

Background tasks docs

Project Structure & Service-Controller-Repository Pattern

The architecture behind a well-structured FastAPI project is the Controller → Service → Repository pattern. Routes should be thin — validate input, call a service, return output. Business logic lives in services/, DB access in repositories/.

app/
  main.py                  # app setup, middleware, startup events
  api/
    v1/
      routes/              # Controller — thin, just routing and HTTP in/out
      dependencies.py      # shared deps: get_db, get_current_user
  models/                  # SQLAlchemy ORM models
  schemas/                 # Pydantic request/response models
  services/                # Service — business logic, orchestration
  repositories/            # Repository — all DB queries live here
  core/
    config.py              # settings via pydantic-settings
    security.py            # auth helpers

In practice, a request flows like this:

# routes/users.py — Controller: only HTTP concerns
@router.get("/users/{user_id}", response_model=UserSchema)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    return await user_service.get_user(db, user_id)

# services/user_service.py — Service: business logic
async def get_user(db: AsyncSession, user_id: int) -> User:
    user = await user_repo.get_by_id(db, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

# repositories/user_repo.py — Repository: DB access only
async def get_by_id(db: AsyncSession, user_id: int) -> User | None:
    result = await db.execute(select(User).where(User.id == user_id))
    return result.scalar_one_or_none()

Why this matters: routes stay readable, business logic is testable in isolation, and DB queries are in one place.

Resources

• awesome-fastapi — curated list of FastAPI libraries, tools, and examples. Worth skimming before starting a project.
• FastAPI automatically generates an OpenAPI spec — available at /docs (Swagger UI) and /redoc. Useful for exploring and testing the API during development; frontend devs will use it to build their client.

SQLAlchemy 2.0

Async Session Management

One session per request, managed through a FastAPI dependency.

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://...")
async_session = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        yield session

expire_on_commit=False is worth noting — without it, accessing attributes after a commit triggers additional queries, which can cause subtle bugs (and was a source of greenlet errors when we migrated from sync to async on a past project).

The N+1 Problem

The most common performance killer, and easy to miss:

# BAD — 1 query for users, then 1 query per user to load their orders
users = await session.execute(select(User))
for user in users.scalars():
    print(user.orders)  # each access fires a separate query

# GOOD — load everything in one query
from sqlalchemy.orm import selectinload
users = await session.execute(
    select(User).options(selectinload(User.orders))
)

Use selectinload for collections (one-to-many), joinedload for single relationships (many-to-one). When in doubt, check your query count in dev with logging enabled.

Relationships and Lazy vs Eager Loading

How you define a relationship controls when SQLAlchemy fetches related data:

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    orders: Mapped[list["Order"]] = relationship("Order", back_populates="user")

class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    user: Mapped["User"] = relationship("User", back_populates="orders")

Loading strategies:

In async SQLAlchemy, lazy loading is broken by default. Accessing a relationship without eager loading raises a MissingGreenlet error. Always use selectinload or joinedload in async code.

Sync vs Async SQLAlchemy — The 4 Key Differences

1. Imports and classes — create_engine / Session become create_async_engine / AsyncSession. They are not interchangeable.

2. Connection string — needs an async-compatible driver: postgresql+asyncpg://... instead of postgresql://.... The driver (asyncpg, aiomysql) must be installed separately.

3. Everything needs await — session.execute(...) becomes await session.execute(...), session.commit() becomes await session.commit(), etc. Forgetting await is a silent bug — it returns a coroutine object instead of the result.

4. Lazy loading does not work — sync SQLAlchemy will fire a lazy query automatically when you access a relationship. Async SQLAlchemy raises MissingGreenlet instead. You must explicitly eager load everything you need.

Common Mistakes

Returning ORM objects directly from routes. Always use Pydantic response schemas. ORM objects trigger lazy loads during serialization, which causes errors in async context and leaks internal model structure.

Greenlet errors in async context. These happen when SQLAlchemy tries to do synchronous I/O inside an async context — usually caused by lazy-loaded relationships being accessed after the session closes. Fix: eager load what you need, or use expire_on_commit=False.

Forgetting to commit. session.add() stages the object. await session.commit() is what actually writes it. Easy to miss when refactoring.

SQLAlchemy 2.0 tutorial

Auth (JWT)

The standard pattern for API auth. Access tokens are short-lived (~15-60 min), refresh tokens are long-lived and stored server-side.

from fastapi.security import OAuth2PasswordBearer
from jose import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    return payload

Things to remember:

• Hash passwords with bcrypt via passlib — never roll your own
• Access tokens: client keeps in memory. Refresh tokens: server stores in DB, client keeps in an httpOnly cookie
• A sharing system (e.g. users sharing resources with teams) lives on top of this — auth tells you who the user is, authorization logic tells you what they can access

FastAPI security tutorial — work through all sub-pages including the JWT one.

Databricks Apps

What It Is

A serverless platform to deploy Python web apps (FastAPI, Dash, Streamlit, Gradio) directly inside your Databricks workspace. No separate infra — your app gets automatic access to Unity Catalog, SQL warehouses, and the Databricks SDK.

Supported frameworks: FastAPI, Dash, Gradio, Streamlit, React, Express.

app.yaml

Every Databricks App is configured via app.yaml:

command: ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
env:
  - name: ENVIRONMENT
    value: "production"
  - name: DB_SECRET
    valueFrom:
      secretRef:
        name: my-databricks-secret

This defines the entry point, compute size, environment variables, and resource permissions. It’s the equivalent of a Dockerfile + environment config in one.

Key Things to Know

• Apps run as a service principal — permissions are scoped to what that principal has access to in Unity Catalog
• Environment variables can reference Databricks secrets (avoids hardcoding credentials)
• The Databricks SDK works natively inside an App without extra credential setup
• Local development uses databricks apps run-local — worth setting up early
• In a GenAI architecture, Databricks Apps fits naturally as the stable application layer, while Model Serving and Lakeflow Jobs handle the AI/intelligence layer separately. This separation lets you iterate on AI fast without touching the stable frontend-facing app. A colleague of mine who knows this space well wrote about this pattern in detail: GenAI App Architecture: Bimodal Approach for Stable Apps and Fast AI

Resources

• Databricks Apps overview
• App development & app.yaml reference
• Deployment
• FastAPI examples on Apps Cookbook
• Databricks Apps FastAPI starter project

Python Concurrency in Short

This comes up constantly in async FastAPI code and is worth having a clear mental model for.

async/await — coroutines running on a single-threaded event loop. When you await something (a DB query, an HTTP call), the event loop can run other coroutines while waiting. This is how FastAPI handles thousands of concurrent requests without threads. It only helps with I/O-bound work.

Threads — actual OS threads. Python can run multiple threads, but the GIL (Global Interpreter Lock) prevents more than one thread from executing Python bytecode at the same time. Threads still help for I/O-bound work (the GIL is released during I/O), but provide no speedup for CPU-bound work.

Thread pool — if you have blocking (synchronous) code that you must call inside an async context without blocking the event loop, offload it to a thread pool:

import asyncio
from concurrent.futures import ThreadPoolExecutor

executor = ThreadPoolExecutor()

async def call_blocking_library():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, some_blocking_function, arg)
    return result

FastAPI does this automatically for synchronous route functions (declared with def instead of async def) — it runs them in a thread pool so they don’t block the event loop.

When to use what:

The GIL means Python threads won’t speed up CPU-bound work — for that you need separate processes.

Performance Testing with Locust

Locust is a Python-based load testing tool. Define your test as a Python class, run it, and get a real-time dashboard showing requests/sec, response times, and failure rates.

# locustfile.py
from locust import HttpUser, task, between

class APIUser(HttpUser):
    wait_time = between(1, 3)  # simulated wait between requests

    @task
    def get_users(self):
        self.client.get("/api/v1/users")

    @task(3)  # weight 3 — runs 3x more often than get_users
    def get_user_by_id(self):
        self.client.get("/api/v1/users/1")

locust -f locustfile.py --host=http://localhost:8000
# then open http://localhost:8089 for the dashboard

Worth running before going to production to find bottlenecks — especially useful after adding async or changing DB query patterns.

API Design Quick Rules

• Resource names are nouns, not verbs: GET /users/{id}/orders not GET /getUserOrders
• Use plural nouns: /users, /orders, not /user, /order
• PUT/DELETE should be safe to retry (idempotent)
• Nest resources only one level deep: /users/{id}/orders is fine, /users/{id}/orders/{id}/items/{id} is a smell
• Version your API from day one: /api/v1/...

REST API best practices (Stack Overflow blog)

Suggested Tech Stack for a FastAPI Backend

A starting point for a production-ready FastAPI project. Swap out as needed.