Django with Google Cloud Spanner

Introduction

Google Cloud Spanner is a fully managed, horizontally scalable relational database service. When combined with Django, it provides a powerful solution for applications requiring global distribution and strong consistency. This guide will walk you through setting up Python Django with Cloud Spanner and implementing CRUD operations, culminating in Postman testing.

Prerequisites

Google Cloud account with Spanner API enabled
Python 3.7+ installed
Django 3.2+ installed
Google Cloud SDK installed
Postman for API testing

Setup

1. Install Required Packages

pip install django google-cloud-spanner django-google-spanner

2. Create a Google Cloud Spanner Instance

Step-by-Step:

Go to Google Cloud Console
- Navigate to Google Cloud Console
- Select or create a Google Cloud Project.
Enable Spanner API
- Search for "Spanner API" in the search bar.
- Click "Enable" if not already enabled.
Create a Spanner Instance
- In the left menu, go to Spanner > Instances.
- Click "Create Instance".
- Fill in details:
  - Instance Name: your-instance-name (e.g., my-spanner-instance)
  - Instance ID: Auto-filled (same as name)
  - Configuration: Choose a regional or multi-region (e.g., nam3 for US multi-region)
  - Compute Capacity: Start with 1 node (scalable later)
  - Click "Create".

3. Create a Database in Spanner

Inside the Spanner Instance
- Click on your newly created instance.
- Click "Create Database".
- Database Name: your-database-name (e.g., my-spanner-db).
- Define Schema:
  - You can either:
    - Upload a DDL file (for schema definition).
    - Manually enter SQL (e.g., for Django, you can leave it empty initially—Django will create tables via migrations).
  - Example DDL (if needed):

CREATE TABLE Books (
    id INT64 NOT NULL,
    title STRING(200),
    author STRING(100),
    published_date DATE,
    isbn STRING(13),
    price NUMERIC,
    stock INT64
) PRIMARY KEY (id);

Click "Create".

4. Get Connection Properties for Django

You need these details for Django’s settings.py:

Project ID
- Find it in the Google Cloud Dashboard (top bar, next to the project name).
Instance Name
- The name you gave (e.g., my-spanner-instance).
Database Name
- The name you created (e.g., my-spanner-db).
Service Account Key (JSON)
- Go to IAM & Admin > Service Accounts.
- Click "Create Service Account".
  - Name: spanner-access
  - Role: Cloud Spanner Database Admin (or Cloud Spanner Admin for full access).
  - Click "Create Key" > JSON > Save (this downloads a .json file).
- Store this file securely (e.g., in your Django project’s secrets/ folder).

5. Configure Django with Spanner

In settings.py:

DATABASES = {
    'default': {
        'ENGINE': 'django_google_spanner',
        'PROJECT': 'your-project-id',        # From Cloud Console
        'INSTANCE': 'your-instance-name',    # e.g., my-spanner-instance
        'NAME': 'your-database-name',       # e.g., my-spanner-db
        'OPTIONS': {
            'credentials': 'path/to/service-account-key.json'  # Downloaded JSON
        }
    }
}

Model Definition

Let's create a sample model for demonstration:

# models.py
from django.db import models

class Book(models.Model):
    title = models.CharField(max_length=200)
    author = models.CharField(max_length=100)
    published_date = models.DateField()
    isbn = models.CharField(max_length=13, unique=True)
    price = models.DecimalField(max_digits=6, decimal_places=2)
    stock = models.IntegerField(default=0)
    
    def __str__(self):
        return self.title

Run migrations:

python manage.py makemigrations
python manage.py migrate

CRUD Operations with Django Views

1. Create Views

# views.py
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from .models import Book
import json

@csrf_exempt
def book_list(request):
    if request.method == 'GET':
        books = Book.objects.all()
        data = [{
            'id': book.id,
            'title': book.title,
            'author': book.author,
            'published_date': book.published_date,
            'isbn': book.isbn,
            'price': str(book.price),
            'stock': book.stock
        } for book in books]
        return JsonResponse(data, safe=False)
    
    elif request.method == 'POST':
        try:
            data = json.loads(request.body)
            book = Book.objects.create(
                title=data['title'],
                author=data['author'],
                published_date=data['published_date'],
                isbn=data['isbn'],
                price=data['price'],
                stock=data.get('stock', 0)
            )
            return JsonResponse({
                'id': book.id,
                'message': 'Book created successfully'
            }, status=201)
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=400)

@csrf_exempt
def book_detail(request, pk):
    try:
        book = Book.objects.get(pk=pk)
    except Book.DoesNotExist:
        return JsonResponse({'error': 'Book not found'}, status=404)
    
    if request.method == 'GET':
        data = {
            'id': book.id,
            'title': book.title,
            'author': book.author,
            'published_date': book.published_date,
            'isbn': book.isbn,
            'price': str(book.price),
            'stock': book.stock
        }
        return JsonResponse(data)
    
    elif request.method == 'PUT':
        try:
            data = json.loads(request.body)
            book.title = data.get('title', book.title)
            book.author = data.get('author', book.author)
            book.published_date = data.get('published_date', book.published_date)
            book.isbn = data.get('isbn', book.isbn)
            book.price = data.get('price', book.price)
            book.stock = data.get('stock', book.stock)
            book.save()
            return JsonResponse({'message': 'Book updated successfully'})
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=400)
    
    elif request.method == 'DELETE':
        book.delete()
        return JsonResponse({'message': 'Book deleted successfully'}, status=204)

2. URL Configuration

# urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('books/', views.book_list, name='book_list'),
    path('books/<int:pk>/', views.book_detail, name='book_detail'),
]

Advanced Spanner Features

1. Transaction Management

from django.db import transaction

@csrf_exempt
def update_stock(request, pk):
    if request.method == 'POST':
        try:
            data = json.loads(request.body)
            quantity = data['quantity']
            
            with transaction.atomic():
                book = Book.objects.select_for_update().get(pk=pk)
                if book.stock + quantity < 0:
                    return JsonResponse({'error': 'Insufficient stock'}, status=400)
                book.stock += quantity
                book.save()
                
            return JsonResponse({'message': 'Stock updated successfully'})
            
        except Book.DoesNotExist:
            return JsonResponse({'error': 'Book not found'}, status=404)
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=400)

2. Batch Operations

@csrf_exempt
def batch_create_books(request):
    if request.method == 'POST':
        try:
            data = json.loads(request.body)
            books_data = data['books']
            
            books = [
                Book(
                    title=book['title'],
                    author=book['author'],
                    published_date=book['published_date'],
                    isbn=book['isbn'],
                    price=book['price'],
                    stock=book.get('stock', 0)
                ) for book in books_data
            ]
            
            Book.objects.bulk_create(books)
            return JsonResponse({'message': f'{len(books)} books created successfully'}, status=201)
            
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=400)

Postman Testing

1. Setup

Install Postman
Create a new collection called "Django Spanner Books API"
Set base URL to http://localhost:8000

2. Test Cases

Create a Book (POST /books/)

Method: POST
URL: /books/
Headers:
- Content-Type: application/json
Body (raw JSON):

{
    "title": "The Great Gatsby",
    "author": "F. Scott Fitzgerald",
    "published_date": "1925-04-10",
    "isbn": "9780743273565",
    "price": "12.99",
    "stock": 50
}

Get All Books (GET /books/)

Method: GET
URL: /books/
Expected Response: 200 OK with list of books

Get Single Book (GET /books/{id}/)

Method: GET
URL: /books/1/ (use actual ID from create response)
Expected Response: 200 OK with book details

Update Book (PUT /books/{id}/)

Method: PUT
URL: /books/1/
Headers:
- Content-Type: application/json
Body:

{
    "price": "14.99",
    "stock": 45
}

Expected Response: 200 OK with success message

Delete Book (DELETE /books/{id}/)

Method: DELETE
URL: /books/1/
Expected Response: 204 No Content

Batch Create Books (POST /batch-create/)

Method: POST
URL: /batch-create/ (assuming you added this endpoint)
Headers:
- Content-Type: application/json
Body:

{
    "books": [
        {
            "title": "To Kill a Mockingbird",
            "author": "Harper Lee",
            "published_date": "1960-07-11",
            "isbn": "9780061120084",
            "price": "10.99",
            "stock": 30
        },
        {
            "title": "1984",
            "author": "George Orwell",
            "published_date": "1949-06-08",
            "isbn": "9780451524935",
            "price": "9.99",
            "stock": 40
        }
    ]
}

Expected Response: 201 Created with count of books created

Error Handling

Implement middleware for consistent error responses:

# middleware.py
import json
from django.http import JsonResponse

class SpannerErrorMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self.get_response(request)
        return response

    def process_exception(self, request, exception):
        if hasattr(exception, 'code') and exception.code == 400:
            return JsonResponse({
                'error': str(exception),
                'status_code': 400
            }, status=400)
        return JsonResponse({
            'error': 'Internal server error',
            'status_code': 500
        }, status=500)

Add to MIDDLEWARE in settings.py.

Conclusion

This guide has covered setting up Django with Google Cloud Spanner, implementing comprehensive CRUD operations, and testing with Postman. Key takeaways:

Cloud Spanner provides a horizontally scalable relational database for Django
Transaction management is crucial for data consistency
Batch operations can significantly improve performance
Proper indexing and query optimization are essential

Remember to monitor your Spanner instance's performance and costs in the Google Cloud Console, as Spanner's pricing model is based on node hours, storage, and network usage.

CODE WITH SIBIN

Solving Real Problems with Real Code