Python Django with Google Cloud Datastore: In-Depth Guide

This comprehensive guide covers integrating Django with Google Cloud Datastore, including GCP configuration, CRUD operations, Postman testing, and unit testing.

GCP Cloud Configuration
Django Project Setup
Datastore CRUD Operations
Postman Testing
Unit Testing

GCP Cloud Configuration

1. Create a GCP Project

Go to Google Cloud Console
Click "Create Project"
Enter a project name and click "Create"

2. Enable Datastore API

Navigate to "APIs & Services" > "Library"
Search for "Cloud Datastore API"
Click "Enable"

3. Create Service Account and Get Key

Go to "IAM & Admin" > "Service Accounts"
Click "Create Service Account"
Enter name and description
Assign roles: "Cloud Datastore User" and "Cloud Datastore Owner"
Click "Create Key" and select JSON format
Download the key file (keep it secure)

Django Project Setup

1. Install Required Packages

pip install django google-cloud-datastore python-dotenv

2. Create .env file

GOOGLE_APPLICATION_CREDENTIALS=path/to/your/service-account-key.json
GCP_PROJECT_ID=your-project-id

3. Configure settings.py

import os
from dotenv import load_dotenv

load_dotenv()

# Add to your settings.py
GCP_PROJECT_ID = os.getenv('GCP_PROJECT_ID')

Datastore CRUD Operations

1. Initialize Datastore Client

Create datastore_service.py:

from google.cloud import datastore
import os
from django.conf import settings

def get_client():
    return datastore.Client(project=settings.GCP_PROJECT_ID)

2. Model Definition (Example: Book)

class Book:
    def __init__(self, title, author, published_year, id=None):
        self.id = id
        self.title = title
        self.author = author
        self.published_year = published_year

    @classmethod
    def from_entity(cls, entity):
        return cls(
            id=entity.id,
            title=entity['title'],
            author=entity['author'],
            published_year=entity['published_year']
        )

3. CRUD Operations

Create book_service.py:

from .datastore_service import get_client
from .models import Book

def create_book(book):
    client = get_client()
    key = client.key('Book')
    entity = datastore.Entity(key=key)
    entity.update({
        'title': book.title,
        'author': book.author,
        'published_year': book.published_year
    })
    client.put(entity)
    book.id = entity.id
    return book

def get_book(book_id):
    client = get_client()
    key = client.key('Book', int(book_id))
    entity = client.get(key)
    if not entity:
        return None
    return Book.from_entity(entity)

def update_book(book_id, book_data):
    client = get_client()
    key = client.key('Book', int(book_id))
    entity = client.get(key)
    if not entity:
        return None
    entity.update(book_data)
    client.put(entity)
    return Book.from_entity(entity)

def delete_book(book_id):
    client = get_client()
    key = client.key('Book', int(book_id))
    client.delete(key)

def list_books():
    client = get_client()
    query = client.query(kind='Book')
    return [Book.from_entity(entity) for entity in query.fetch()]

4. Django Views

Create views.py:

from django.http import JsonResponse
from .book_service import *
from .models import Book
import json

def book_list(request):
    if request.method == 'GET':
        books = list_books()
        return JsonResponse([book.__dict__ for book in books], safe=False)
    elif request.method == 'POST':
        data = json.loads(request.body)
        book = Book(
            title=data['title'],
            author=data['author'],
            published_year=data['published_year']
        )
        created_book = create_book(book)
        return JsonResponse(created_book.__dict__, status=201)

def book_detail(request, book_id):
    if request.method == 'GET':
        book = get_book(book_id)
        if not book:
            return JsonResponse({'error': 'Book not found'}, status=404)
        return JsonResponse(book.__dict__)
    elif request.method == 'PUT':
        data = json.loads(request.body)
        updated_book = update_book(book_id, data)
        if not updated_book:
            return JsonResponse({'error': 'Book not found'}, status=404)
        return JsonResponse(updated_book.__dict__)
    elif request.method == 'DELETE':
        delete_book(book_id)
        return JsonResponse({}, status=204)

5. URLs Configuration

from django.urls import path
from . import views

urlpatterns = [
    path('books/', views.book_list, name='book-list'),
    path('books/<int:book_id>/', views.book_detail, name='book-detail'),
]

Postman Testing

1. Create Book (POST)

URL: http://localhost:8000/books/
Method: POST
Body (raw JSON):

{
    "title": "Django with Datastore",
    "author": "John Doe",
    "published_year": 2023
}

2. Get All Books (GET)

URL: http://localhost:8000/books/
Method: GET

3. Get Single Book (GET)

URL: http://localhost:8000/books/123456789 (use actual ID)
Method: GET

4. Update Book (PUT)

URL: http://localhost:8000/books/123456789 (use actual ID)
Method: PUT
Body (raw JSON):

{
    "title": "Updated Title",
    "author": "Updated Author",
    "published_year": 2024
}

5. Delete Book (DELETE)

URL: http://localhost:8000/books/123456789 (use actual ID)
Method: DELETE

Unit Testing

1. Setup Test Environment

Create tests.py:

from django.test import TestCase
from google.cloud import datastore
from .models import Book
from .book_service import *
from django.conf import settings
import os

class DatastoreTestCase(TestCase):
    def setUp(self):
        # Use a test namespace to isolate test data
        self.client = datastore.Client(
            project=settings.GCP_PROJECT_ID,
            namespace='test'
        )
        
    def tearDown(self):
        # Clean up all test data
        query = self.client.query(kind='Book', namespace='test')
        for entity in query.fetch():
            self.client.delete(entity.key)

class BookServiceTests(DatastoreTestCase):
    def test_create_and_get_book(self):
        book = Book(title="Test Book", author="Test Author", published_year=2023)
        created_book = create_book(book)
        
        retrieved_book = get_book(created_book.id)
        self.assertIsNotNone(retrieved_book)
        self.assertEqual(retrieved_book.title, "Test Book")
        self.assertEqual(retrieved_book.author, "Test Author")
        self.assertEqual(retrieved_book.published_year, 2023)
    
    def test_update_book(self):
        book = Book(title="Original", author="Original", published_year=2000)
        created_book = create_book(book)
        
        updated_data = {
            'title': 'Updated',
            'author': 'Updated',
            'published_year': 2023
        }
        updated_book = update_book(created_book.id, updated_data)
        
        self.assertEqual(updated_book.title, 'Updated')
        self.assertEqual(updated_book.author, 'Updated')
        self.assertEqual(updated_book.published_year, 2023)
    
    def test_delete_book(self):
        book = Book(title="To Delete", author="Author", published_year=2023)
        created_book = create_book(book)
        
        delete_book(created_book.id)
        deleted_book = get_book(created_book.id)
        self.assertIsNone(deleted_book)
    
    def test_list_books(self):
        book1 = Book(title="Book 1", author="Author 1", published_year=2021)
        book2 = Book(title="Book 2", author="Author 2", published_year=2022)
        create_book(book1)
        create_book(book2)
        
        books = list_books()
        self.assertEqual(len(books), 2)
        titles = [book.title for book in books]
        self.assertIn("Book 1", titles)
        self.assertIn("Book 2", titles)

class ViewTests(DatastoreTestCase):
    def test_book_list_get(self):
        response = self.client.get('/books/')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(len(response.json()), 0)
        
        book = Book(title="Test", author="Test", published_year=2023)
        create_book(book)
        
        response = self.client.get('/books/')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(len(response.json()), 1)
    
    def test_book_detail_get(self):
        book = Book(title="Test", author="Test", published_year=2023)
        created_book = create_book(book)
        
        response = self.client.get(f'/books/{created_book.id}/')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()['title'], "Test")
    
    def test_book_detail_get_not_found(self):
        response = self.client.get('/books/999999/')
        self.assertEqual(response.status_code, 404)

2. Running Tests

python manage.py test

Additional Considerations

Error Handling: Add proper error handling for Datastore operations
Pagination: Implement pagination for list operations
Indexes: Configure appropriate indexes in Datastore for query performance
Transactions: Use Datastore transactions for complex operations
Security: Implement proper authentication and authorization

This guide provides a comprehensive foundation for using Google Cloud Datastore with Django. You can extend it based on your specific application requirements.

CODE WITH SIBIN

Solving Real Problems with Real Code

Table of Contents