Interact with Lakebase Tables

This recipe demonstrates how to build a complete orders management API using Lakebase PostgreSQL database. These endpoints provide CRUD operations and various query patterns for handling orders data synchronized from Databricks Unity Catalog.

Prerequisites

• Lakebase resources must be created using the Create Lakebase Resources endpoint
• The synced table pipeline must be completed and orders data synchronized from samples.tpch.orders
• Database connection must be configured in your application environment

info

In this example, we demonstrate multiple HTTP methods (GET, POST) which are the standard choices for data operations in REST APIs as defined in RFC 7231:

• GET requests are idempotent and cacheable, ideal for data retrieval
• POST requests for updates and modifications

For detailed specifications, refer to RFC 7231 which defines HTTP method semantics.

Code snippet

routes/v1/orders.py

import logging

from config.database import get_async_db
from models.orders import (
    CursorPaginationInfo,
    Order,
    OrderCount,
    OrderListCursorResponse,
    OrderListResponse,
    OrderRead,
    OrderSample,
    OrderStatusUpdate,
    OrderStatusUpdateResponse,
    PaginationInfo,
)
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from fastapi import APIRouter, Depends, HTTPException, Query

logger = logging.getLogger(__name__)
router = APIRouter(tags=["orders"])

# 1. GET ORDERS COUNT
@router.get("/count", response_model=OrderCount, summary="Get total order count")
async def get_order_count(db: AsyncSession = Depends(get_async_db)):
    try:
        stmt = select(func.count(Order.o_orderkey))
        result = await db.execute(stmt)
        count = result.scalar()
        return OrderCount(total_orders=count)
    except Exception as e:
        logger.error(f"Error getting order count: {e}")
        raise HTTPException(status_code=500, detail="Failed to retrieve order count")

# 2. GET SAMPLE ORDERS
@router.get("/sample", response_model=OrderSample, summary="Get 5 random order keys")
async def get_sample_orders(db: AsyncSession = Depends(get_async_db)):
    try:
        stmt = select(Order.o_orderkey).limit(5)
        result = await db.execute(stmt)
        order_keys = result.scalars().all()
        return OrderSample(sample_order_keys=order_keys)
    except Exception as e:
        logger.error(f"Error getting sample orders: {e}")
        raise HTTPException(status_code=500, detail="Failed to retrieve sample orders")

# 3. PAGE-BASED PAGINATION
@router.get("/pages", response_model=OrderListResponse, summary="Get orders with page-based pagination")
async def get_orders_by_page(
    page: int = Query(1, ge=1, description="Page number (1-based)"),
    page_size: int = Query(100, ge=1, le=1000, description="Number of records per page (max 1000)"),
    include_count: bool = Query(True, description="Include total count for pagination info"),
    db: AsyncSession = Depends(get_async_db),
):
    try:
        if include_count:
            count_stmt = select(func.count(Order.o_orderkey))
            count_result = await db.execute(count_stmt)
            total_count = count_result.scalar()
            total_pages = (total_count + page_size - 1) // page_size
        else:
            total_count = -1
            total_pages = -1

        offset = (page - 1) * page_size
        stmt = (
            select(Order)
            .order_by(Order.o_orderkey)
            .offset(offset)
            .limit(page_size + 1)
        )

        result = await db.execute(stmt)
        all_orders = result.scalars().all()

        has_next = len(all_orders) > page_size
        orders = all_orders[:page_size]
        has_previous = page > 1

        pagination_info = PaginationInfo(
            page=page,
            page_size=page_size,
            total_pages=total_pages,
            total_count=total_count,
            has_next=has_next,
            has_previous=has_previous,
        )

        return OrderListResponse(orders=orders, pagination=pagination_info)

    except Exception as e:
        logger.error(f"Error getting page-based orders: {e}")
        raise HTTPException(status_code=500, detail="Failed to retrieve orders")

# 4. CURSOR-BASED PAGINATION
@router.get("/stream", response_model=OrderListCursorResponse, summary="Get orders with cursor-based pagination")
async def get_orders_by_cursor(
    cursor: int = Query(0, ge=0, description="Start after this order key (0 for beginning)"),
    page_size: int = Query(100, ge=1, le=1000, description="Number of records to fetch (max 1000)"),
    db: AsyncSession = Depends(get_async_db),
):
    try:
        stmt = (
            select(Order)
            .where(Order.o_orderkey > cursor)
            .order_by(Order.o_orderkey)
            .limit(page_size + 1)
        )

        result = await db.execute(stmt)
        all_orders = result.scalars().all()

        has_next = len(all_orders) > page_size
        orders = all_orders[:page_size]
        has_previous = cursor > 0

        next_cursor = orders[-1].o_orderkey if orders and has_next else None
        previous_cursor = max(0, cursor - page_size) if has_previous else None

        pagination_info = CursorPaginationInfo(
            page_size=page_size,
            has_next=has_next,
            has_previous=has_previous,
            next_cursor=next_cursor,
            previous_cursor=previous_cursor,
        )

        return OrderListCursorResponse(orders=orders, pagination=pagination_info)

    except Exception as e:
        logger.error(f"Error getting cursor-based orders: {e}")
        raise HTTPException(status_code=500, detail="Failed to retrieve orders")

# 5. GET SPECIFIC ORDER
@router.get("/{order_key}", response_model=OrderRead, summary="Get an order by its key")
async def read_order(order_key: int, db: AsyncSession = Depends(get_async_db)):
    try:
        if order_key <= 0:
            raise HTTPException(status_code=400, detail="Invalid order key provided")

        stmt = select(Order).where(Order.o_orderkey == order_key)
        result = await db.execute(stmt)
        order = result.scalars().first()

        if not order:
            raise HTTPException(status_code=404, detail=f"Order with key '{order_key}' not found")

        return order

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Unexpected error fetching order {order_key}: {e}")
        raise HTTPException(status_code=500, detail="Internal server error occurred")

# 6. UPDATE ORDER STATUS
@router.post("/{order_key}/status", response_model=OrderStatusUpdateResponse, summary="Update order status")
async def update_order_status(
    order_key: int,
    status_data: OrderStatusUpdate,
    db: AsyncSession = Depends(get_async_db),
):
    try:
        if order_key <= 0:
            raise HTTPException(status_code=400, detail="Invalid order key provided")

        check_stmt = select(Order).where(Order.o_orderkey == order_key)
        check_result = await db.execute(check_stmt)
        existing_order = check_result.scalars().first()

        if not existing_order:
            raise HTTPException(status_code=404, detail=f"Order with key '{order_key}' not found")

        existing_order.o_orderstatus = status_data.o_orderstatus
        await db.commit()
        await db.refresh(existing_order)

        return OrderStatusUpdateResponse(
            o_orderkey=order_key,
            o_orderstatus=status_data.o_orderstatus,
            message="Order status updated successfully",
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Error updating status for order {order_key}: {e}")
        raise HTTPException(status_code=500, detail="Failed to update order status")

warning

The above example is shortened for brevity and not suitable for production use. You can find a more advanced sample in the databricks-apps-cookbook GitHub repository.

Example Usage

The orders API provides six main endpoints for different use cases:

Endpoint	Method	Purpose	Best For
/orders/count	GET	Get total orders count	Dashboards, monitoring
/orders/sample	GET	Get 5 sample order keys	Testing, development
/orders/pages	GET	Page-based pagination	Traditional UIs with page numbers
/orders/stream	GET	Cursor-based pagination	Large datasets, infinite scroll
/orders/{order_key}	GET	Get specific order	Order details, lookups
/orders/{order_key}/status	POST	Update order status	Order processing workflows

Get Orders Count

curl -X GET "http://localhost:8000/api/v1/orders/count"

{
  "total_orders": 1500000
}

Get Sample Orders

curl -X GET "http://localhost:8000/api/v1/orders/sample"

{
  "sample_order_keys": [1, 32, 33, 34, 35]
}

Page-Based Pagination

curl -X GET "http://localhost:8000/api/v1/orders/pages?page=1&page_size=2"

{
  "orders": [
    {
      "o_orderkey": 1,
      "o_custkey": 370,
      "o_orderstatus": "O",
      "o_totalprice": 172799.49,
      "o_orderdate": "1996-01-02",
      "o_orderpriority": "5-LOW",
      "o_clerk": "Clerk#000000951",
      "o_shippriority": 0,
      "o_comment": "nstructions sleep furiously among"
    },
    {
      "o_orderkey": 2,
      "o_custkey": 781,
      "o_orderstatus": "O",
      "o_totalprice": 46929.18,
      "o_orderdate": "1996-12-01",
      "o_orderpriority": "1-URGENT",
      "o_clerk": "Clerk#000000880",
      "o_shippriority": 0,
      "o_comment": "foxes. pending accounts at the pending"
    }
  ],
  "pagination": {
    "page": 1,
    "page_size": 2,
    "total_pages": 750000,
    "total_count": 1500000,
    "has_next": true,
    "has_previous": false
  }
}

Update Order Status

curl -X POST "http://localhost:8000/api/v1/orders/1/status" \
  -H "Content-Type: application/json" \
  -d '{"o_orderstatus": "F"}'

{
  "o_orderkey": 1,
  "o_orderstatus": "F",
  "message": "Order status updated successfully"
}

Resources

• Lakebase PostgreSQL
• SQLAlchemy Async
• FastAPI Dependencies

Permissions

Your app service principal needs the following permissions:

• Database instance connection access via OAuth tokens
• SELECT permissions on the orders_synced table in your Lakebase database
• UPDATE permissions on the orders_synced table for status updates
• Database user role with appropriate table access

See Lakebase permissions for more information.

Dependencies

• Databricks SDK for Python - databricks-sdk
• FastAPI - fastapi
• SQLAlchemy - sqlalchemy
• SQLModel - sqlmodel
• asyncpg - asyncpg
• uvicorn - uvicorn

requirements.txt

databricks-sdk>=0.60.0
fastapi
sqlalchemy
sqlmodel
asyncpg
uvicorn