Have you ever wanted to run Couchbase on Docker to freely build and test your app?

Here’s a simple guide to help you set up Couchbase in Docker and build a sample TODO app using Python and FastAPI.

Developer Cluster Setup

First, let’s create a minimal docker-compose.yml file to spin up Couchbase:

services:
  couchbase:
    image: couchbase:latest
    container_name: couchbase
    ports:
      - "8091:8091"  # Couchbase Web Console
      - "8092:8092"  # Query Service
      - "8093:8093"  # Full Text Search
      - "11210:11210"  # Data Service
    environment:
      COUCHBASE_ADMINISTRATOR_USERNAME: admin
      COUCHBASE_ADMINISTRATOR_PASSWORD: password
    volumes:
      - couchbase_data:/opt/couchbase/var
      - ./init_bucket.sh:/init_bucket.sh
      - ./start-couchbase.sh:/start-couchbase.sh
    # get the image entry point using docker inspect -f '{{.Config.Entrypoint}}' couchbase
    # for the current image is /entrypoint.sh
    command: ["/bin/bash", "/start-couchbase.sh"]
    mem_limit: 1024m  # Limit memory usage to 3100 MB for full 

volumes:
  couchbase_data:

To make it work, we need two shell scripts: start-couchbase.sh and init_bucket.sh.

The first script (start-couchbase.sh) starts Couchbase in the background and then runs the cluster and pool initialization script (init_bucket.sh).

Here is start-couchbase.sh:

#!/bin/bash

# Start Couchbase Server in the background
/entrypoint.sh couchbase-server &

sleep 5

# Wait for Couchbase to be available
echo "Waiting for Couchbase to be ready..."
until curl -s http://localhost:8091/pools > /dev/null; do
  echo "Waiting for Couchbase server to start..."
  sleep 5
done

# Run the initialization script
echo "Couchbase is up! Running init script..."
./init_bucket.sh

# Wait for Couchbase Server to keep the container running
wait

And here is init_bucket.sh:

#!/bin/bash

echo "Starting init..."

# Wait until Couchbase server is ready to accept requests
echo "Waiting for Couchbase to be ready..."
until curl -s http://localhost:8091/pools > /dev/null; do
  echo "Waiting for Couchbase server to start..."
  sleep 5
done

# Create the cluster
echo "Creating the cluster..."

curl -v -X POST http://localhost:8091/clusterInit \
-d "indexPath=%2Fopt%2Fcouchbase%2Fvar%2Flib%2Fcouchbase%2Fdata" \
-d "eventingPath=%2Fopt%2Fcouchbase%2Fvar%2Flib%2Fcouchbase%2Fdata" \
-d "analyticsPath=%2Fopt%2Fcouchbase%2Fvar%2Flib%2Fcouchbase%2Fdata" \
-d "javaHome=" \
-d "sendStats=false" \
-d "clusterName=DockerCluster" \
-d "services=kv%2Cn1ql%2Cindex" \
-d "memoryQuota=512" \
-d "afamily=ipv4" \
-d "afamilyOnly=false" \
-d "nodeEncryption=off" \
-d "username=admin" \
-d "password=password" \
-d "port=SAME" \
-d "indexMemoryQuota=256" 

# For Full Add:

#-d "services=kv%2Cn1ql%2Cindex%2Cfts%2Ceventing%2Ccbas%2Cbackup" \
#-d "eventingMemoryQuota=256" \
#-d "ftsMemoryQuota=256" 

echo "Creating the pool..."
curl  -X POST http://localhost:8091/pools/default \
  -u admin:password \
  -d memoryQuota=512 \
  -d indexMemoryQuota=256 

  # FOR Full add:
  #-d eventingMemoryQuota=256 \
  #-d ftsMemoryQuota=256 \

echo "done"

# Create the bucket
BUCKET_NAME="myBucketTest"
BUCKET_QUOTA=100  # Size in MB

# Create the bucket using Couchbase REST API
curl  -X POST http://localhost:8091/pools/default/buckets \
  -u admin:password \
  -d name=$BUCKET_NAME \
  -d ramQuota=$BUCKET_QUOTA \
  -d authType=sasl \
  -d bucketType=couchbase

echo "Bucket '$BUCKET_NAME' created successfully."

These settings initialize a developer Couchbase cluster with only the KV (Data Service), N1QL (Query Service), and Index (Index Service) enabled.

Sample python app

Now for the fun part: let’s build a sample TODO app using Python and FastAPI.

Creating a bucket admin user

To create a new user and give a user admin rights on the bucket, add the following to init_bucket.sh:

USER_NAME="todoUser"
USER_PASSWORD="todoPassword"
# BUCKET_NAME="myBucket"

# Create the user using Couchbase REST API
curl -v -X PUT http://localhost:8091/settings/rbac/users/local/$USER_NAME \
-u admin:password \
-d password=$USER_PASSWORD \
-d name="Todo User" \
-d roles="bucket_admin[$BUCKET_NAME],bucket_full_access[$BUCKET_NAME]"

echo "User '$USER_NAME' created with bucket_admin and bucket_full_access rights on bucket '$BUCKET_NAME'."

The Python app

First, let’s install the required packages:

pip install fastapi uvicorn couchbase

Next, create the Pydantic model (models.py) :

from pydantic import BaseModel
from typing import Optional

class TodoItem(BaseModel):
    id: Optional[str]
    title: str
    description: Optional[str]
    completed: bool = False

Now, build the FastAPI app (app.py):

import uuid

from couchbase.auth import PasswordAuthenticator
from couchbase.cluster import Cluster
from couchbase.options import ClusterOptions
from fastapi import FastAPI, HTTPException

from .models import TodoItem

app = FastAPI()

# Couchbase connection setup
cluster = Cluster(
    'couchbase://localhost',
    ClusterOptions(PasswordAuthenticator('todoUser', 'todoPassword'))
)
bucket = cluster.bucket('myBucket')
collection = bucket.default_collection()

# In-memory store for TODOs
todos = {}

# CRUD Operations

@app.post("/todos/", response_model=TodoItem)
def create_todo(todo: TodoItem):
    todo_id = str(uuid.uuid4())  # Generate a unique ID
    todo.id = todo_id

    # Insert the TODO into Couchbase
    collection.upsert(todo_id, todo.model_dump())

    return todo

@app.get("/todos/", response_model=list[TodoItem])
def get_all_todos():
    try:
        # Use N1QL query to get all TODOs from Couchbase bucket
        query = "SELECT meta().id, title, description, completed FROM myBucket"
        rows = cluster.query(query)

        todo_list = [TodoItem(**row) for row in rows]
        return todo_list
    except Exception as e:
        print("Error retrieving TODOs: " + str(e))
        raise HTTPException(status_code=500, detail="Error retrieving TODOs: " + str(e))

@app.get("/todos/{todo_id}", response_model=TodoItem)
def read_todo(todo_id: str):
    try:
        # Get the TODO from Couchbase
        result = collection.get(todo_id)
        todo = result.content_as[TodoItem]
        return todo
    except Exception:
        raise HTTPException(status_code=404, detail="TODO not found")

@app.put("/todos/{todo_id}", response_model=TodoItem)
def update_todo(todo_id: str, todo: TodoItem):
    try:
        # Update the TODO in Couchbase
        collection.upsert(todo_id, todo.dict())
        return todo
    except Exception:
        raise HTTPException(status_code=404, detail="TODO not found")

@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: str):
    try:
        # Delete the TODO from Couchbase
        collection.remove(todo_id)
        return {"detail": "TODO deleted"}
    except Exception:
        raise HTTPException(status_code=404, detail="TODO not found")

Run it all

Now we can finally run the app.

First, start the Couchbase server with:

docker compose up --build

Then, run the FastAPI app with:

uvicorn app:app --reload

And then open the browser to http://127.0.0.1:8000/docs and try it out.

If everything is set up correctly, you should be able to make API calls via FastAPI:

fastapi

And see the results in the Couchbase Admin UI:

couchdb_admin