Basic CRUD Application

This example demonstrates a simple CRUD (Create, Read, Update, Delete) application using Da Vinci and DynamoDB.

Overview

We’ll build a task management application where users can:

  • Create tasks

  • Read task details

  • Update task status

  • Delete tasks

  • List all tasks

Step 1: Define the Table

Create tables.py:

from datetime import UTC, datetime
from da_vinci.core.orm.table_object import (
    TableObject,
    TableObjectAttribute,
    TableObjectAttributeType,
)

class TaskTable(TableObject):
    """Table for storing tasks"""

    table_name = "tasks"
    partition_key_attribute = "task_id"

    attributes = [
        TableObjectAttribute(
            name="task_id",
            attribute_type=TableObjectAttributeType.STRING,
            description="Unique task identifier",
        ),
        TableObjectAttribute(
            name="title",
            attribute_type=TableObjectAttributeType.STRING,
            description="Task title",
        ),
        TableObjectAttribute(
            name="description",
            attribute_type=TableObjectAttributeType.STRING,
            description="Task description",
            optional=True,
        ),
        TableObjectAttribute(
            name="status",
            attribute_type=TableObjectAttributeType.STRING,
            description="Task status (pending, in_progress, completed)",
            default="pending",
        ),
        TableObjectAttribute(
            name="priority",
            attribute_type=TableObjectAttributeType.NUMBER,
            description="Priority level (1-5)",
            default=3,
        ),
        TableObjectAttribute(
            name="created_at",
            attribute_type=TableObjectAttributeType.DATETIME,
            description="When the task was created",
        ),
        TableObjectAttribute(
            name="updated_at",
            attribute_type=TableObjectAttributeType.DATETIME,
            description="When the task was last updated",
        ),
    ]

Step 2: Create Service Layer

Create task_service.py:

import uuid
from datetime import UTC, datetime
from typing import Optional

from da_vinci.core.orm.client import TableClient, TableScanDefinition
from da_vinci.core.exceptions import ResourceNotFoundError
from tables import TaskTable


class TaskService:
    """Service for managing tasks"""

    def __init__(self):
        self.client = TableClient(TaskTable)

    def create_task(
        self,
        title: str,
        description: Optional[str] = None,
        priority: int = 3,
    ) -> TaskTable:
        """Create a new task"""
        now = datetime.now(UTC)
        task_id = str(uuid.uuid4())

        task = TaskTable(
            task_id=task_id,
            title=title,
            description=description,
            status="pending",
            priority=priority,
            created_at=now,
            updated_at=now,
        )

        self.client.put(task)
        return task

    def get_task(self, task_id: str) -> TaskTable:
        """Get a task by ID"""
        task = self.client.get(task_id)
        if not task:
            raise ResourceNotFoundError(f"Task {task_id} not found")
        return task

    def update_task_status(self, task_id: str, status: str) -> TaskTable:
        """Update task status"""
        task = self.get_task(task_id)
        task.status = status
        task.updated_at = datetime.now(UTC)
        self.client.put(task)
        return task

    def update_task(
        self,
        task_id: str,
        title: Optional[str] = None,
        description: Optional[str] = None,
        priority: Optional[int] = None,
    ) -> TaskTable:
        """Update task details"""
        task = self.get_task(task_id)

        if title is not None:
            task.title = title
        if description is not None:
            task.description = description
        if priority is not None:
            task.priority = priority

        task.updated_at = datetime.now(UTC)
        self.client.put(task)
        return task

    def delete_task(self, task_id: str) -> None:
        """Delete a task"""
        self.client.delete(task_id)

    def list_tasks(self, status: Optional[str] = None) -> list[TaskTable]:
        """List all tasks, optionally filtered by status"""
        if status:
            scan_def = TableScanDefinition(TaskTable)
            scan_def.add("status", "equal", status)
            return list(self.client.scan(scan_definition=scan_def))
        else:
            return list(self.client.scan())

    def list_tasks_by_priority(self, min_priority: int) -> list[TaskTable]:
        """List tasks with priority >= min_priority"""
        scan_def = TableScanDefinition(TaskTable)
        scan_def.add("priority", "greater_than_or_equal", min_priority)
        return list(self.client.scan(scan_definition=scan_def))

Step 3: Deploy Infrastructure

Create table_stack.py:

from constructs import Construct
from da_vinci_cdk.stack import Stack
from da_vinci_cdk.constructs.dynamodb import DynamoDBTable
from tables import TaskTable

class TaskTableStack(Stack):
    """Stack that provisions the Task DynamoDB Table"""

    def __init__(
        self,
        app_name: str,
        deployment_id: str,
        scope: Construct,
        stack_name: str
    ) -> None:
        super().__init__(
            app_name=app_name,
            deployment_id=deployment_id,
            scope=scope,
            stack_name=stack_name,
        )

        self.table = DynamoDBTable.from_orm_table_object(
            table_object=TaskTable,
            scope=self,
        )

Create app.py:

from os.path import dirname, abspath
from da_vinci_cdk.application import Application
from table_stack import TaskTableStack

# Create application
app = Application(
    app_name="task-manager",
    deployment_id="dev",
    app_entry=abspath(dirname(__file__)),
)

# Add table stack
app.add_uninitialized_stack(TaskTableStack)

# Synthesize CDK
app.synth()

Deploy:

cdk bootstrap  # First time only
cdk deploy --all

Step 4: Use the Application

from task_service import TaskService

# Initialize service
service = TaskService()

# Create tasks
task1 = service.create_task(
    title="Write documentation",
    description="Complete the user guide",
    priority=5
)

task2 = service.create_task(
    title="Review pull requests",
    priority=4
)

task3 = service.create_task(
    title="Fix bug in login",
    description="Users can't login on mobile",
    priority=5
)

# Get a task
task = service.get_task(task1.task_id)
print(f"Task: {task.title} - Status: {task.status}")

# Update task status
service.update_task_status(task1.task_id, "in_progress")
service.update_task_status(task2.task_id, "completed")

# Update task details
service.update_task(
    task3.task_id,
    description="Users can't login on mobile - login button not responding"
)

# List all tasks
all_tasks = service.list_tasks()
print(f"\nAll tasks ({len(all_tasks)}):")
for task in all_tasks:
    print(f"  - {task.title} [{task.status}]")

# List pending tasks
pending = service.list_tasks(status="pending")
print(f"\nPending tasks ({len(pending)}):")
for task in pending:
    print(f"  - {task.title}")

# List high priority tasks
high_priority = service.list_tasks_by_priority(min_priority=4)
print(f"\nHigh priority tasks ({len(high_priority)}):")
for task in high_priority:
    print(f"  - {task.title} (Priority: {task.priority})")

# Delete a task
service.delete_task(task2.task_id)

Expected Output:

Task: Write documentation - Status: pending

All tasks (3):
  - Write documentation [in_progress]
  - Review pull requests [completed]
  - Fix bug in login [pending]

Pending tasks (1):
  - Fix bug in login

High priority tasks (3):
  - Write documentation (Priority: 5)
  - Review pull requests (Priority: 4)
  - Fix bug in login (Priority: 5)

Key Concepts

Table Definition

The TaskTable class defines the table schema with typed attributes.

Service Layer

The TaskService class encapsulates business logic and table operations.

TableClient

Provides methods for CRUD operations (get, put, delete, scan).

Scan Filters

Use TableScanDefinition to filter results when scanning the table.

Type Safety

Table objects provide type-safe access to attributes.

Variations

Add GSI for Status Queries

class TaskTable(TableObject):
    # ... existing code ...

    global_secondary_indexes = [
        {
            "index_name": "status_priority_index",
            "partition_key": "status",
            "sort_key": "priority",
        }
    ]

# Query instead of scan
tasks = client.query(
    index_name="status_priority_index",
    partition_key_value="pending"
)

Add Batch Operations

def create_tasks_batch(self, tasks_data: list[dict]) -> list[TaskTable]:
    """Create multiple tasks at once"""
    tasks = []
    for data in tasks_data:
        task = self.create_task(**data)
        tasks.append(task)
    return tasks

Next Steps