DEV Community

Cover image for Building a CRM with AWS SAM, part 2: Creating a contact
Claudio Guerra
Claudio Guerra

Posted on

Building a CRM with AWS SAM, part 2: Creating a contact

#python #webdev #aws #serverless

In the previous post we stood up a basic /ping health‑check endpoint to confirm our SAM project was wired correctly.

Today we will build our first real piece of CRM functionality: a POST /contacts endpoint that stores a contact record in Amazon DynamoDB, now powered by AWS Lambda Powertools for structured logging, tracing, and metrics.

The API Contract

Request POST /contacts

{
  "email": "jane.doe@example.com",   // required
  "name": "Jane Doe",                // optional
  "phone": "+11234567890",           // optional
}
Enter fullscreen mode Exit fullscreen mode

Response 201 Created

{
  "id": "c8a5c3d1-2bfe-45d0-b7e3-6d1fb25ce9a7",
  "email": "jane.doe@example.com",
  "name": "Jane Doe",
  "phone": "+11234567890",
  "created_at": 1721340000,
  "updated_at": 1721340000
}
Enter fullscreen mode Exit fullscreen mode

We return the full record, including generated fields such as id, created_at, and updated_at

Designing the Data Model

For now we will use a DynamoDB table named ContactsTable with:

Attribute Purpose Key type
pk contact ID (UUID) PartitionKey
sk static value METADATA SortKey
other attributes email, name, etc (projected)

In DynamoDB, the partition key (pk) decides which storage partition the item belongs to, while the sort key (sk) lets us order or group multiple items that share that same partition. By using a fixed sk value like METADATA for the main contact record today, we reserve the option to add related entities—such as notes or tags—under the same pk but with different sk values later on.

Dependency management

We will create a python virtual environment per lambda function, so that we can manage their package versions in isolation.

Create and activate the venv

python -m venv venvs/contacts
source venvs/contacts/bin/activate
Enter fullscreen mode Exit fullscreen mode

Install the dependencies and lock them:

pip install "aws-lambda-powertools[all]"
pip freeze > src/functions/contacts/requirements.txt
Enter fullscreen mode Exit fullscreen mode

sam build will detect requirements.txt and bundle these exact versions into the deployment artifact.

Contacts Lambda function

The lambda function will recieve a payload with the contact data which will be validated with pydantic, then we store the contact data in DynamoDB where we're abstracting away all the data access logic by following the repository pattern, and finally we return the contact in the response, including system generated attributes such as id, created_at, and updated_at.

File structure:

src/functions/contacts/
├─ app/
|  ├─ __init__.py
|  ├─ main.py                # Lambda entry‑point
|  ├─ models.py              # Pydantic models
|  └─ repository.py          # Data‑access abstraction
├─ __init__.py
├─ requirements.txt          # Locked package dependencies
└─ events/
   └─ create_contact.json    # API Gateway payload for testing
Enter fullscreen mode Exit fullscreen mode

models.py

from pydantic import BaseModel, EmailStr
from typing import Optional

class CreateContactRequest(BaseModel):
    email: EmailStr
    name: Optional[str] = None
    phone: Optional[str] = None

class Contact(BaseModel):
    id: str
    email: EmailStr
    name: Optional[str] = None
    phone: Optional[str] = None
    created_at: str
    updated_at: str
Enter fullscreen mode Exit fullscreen mode

repository.py

import os, uuid
from datetime import datetime, timezone
import boto3
from .models import CreateContactRequest, Contact

class ContactRepository:
    def __init__(self):
        self.contacts_table_name: str = os.environ.get("CONTACTS_TABLE_NAME")
        self.contacts_table = boto3.resource("dynamodb").Table(self.contacts_table_name)

    def create(self, create_contact_request: CreateContactRequest) -> Contact:
        now = datetime.now(timezone.utc).isoformat()
        contact_id = str(uuid.uuid4())

        item = {
            "pk": contact_id,
            "sk": "METADATA",
            **create_contact_request.model_dump(),
            "created_at": now,
            "updated_at": now,
        }

        self.contacts_table.put_item(Item=item, ConditionExpression="attribute_not_exists(pk)")

        return Contact(
            id=contact_id,
            **create_contact_request.model_dump(),
            created_at=now,
            updated_at=now
        )
Enter fullscreen mode Exit fullscreen mode

main.py

from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, CORSConfig
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext
from typing import Dict, Any
from .models import CreateContactRequest, Contact
from .repository import ContactRepository

tracer = Tracer()
logger = Logger()
metrics = Metrics()

contact_repository = ContactRepository()

cors_config = CORSConfig(allow_origin="*", max_age=300)
app = APIGatewayRestResolver(cors=cors_config, enable_validation=True)

@app.post("/contacts")
@tracer.capture_method
def create_contact(create_contact_request: CreateContactRequest) -> Contact:
    contact = contact_repository.create(create_contact_request)
    metrics.add_metric(name="ContactsCreated", unit=MetricUnit.Count, value=1)
    tracer.put_annotation("contact_id", contact.id)
    return contact, 201

@logger.inject_lambda_context(correlation_id_path=correlation_paths.API_GATEWAY_REST)
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start_metric=True)
def lambda_handler(event: Dict[str, Any], context: LambdaContext) -> Dict[str, Any]:
    return app.resolve(event, context)

@app.exception_handler(Exception)
@tracer.capture_method
def handle_exception(ex: Exception):
    logger.exception("Internal server error", extra={"error": str(ex)})
    return {"message": "Internal server error"}, 500
Enter fullscreen mode Exit fullscreen mode

Build, test, deploy

In addition to testing the endpoint with curl, we’ll run the Lambda function locally. That requires passing it a mock event. Because the function is invoked by API Gateway in production, the mock event must replicate the JSON payload that API Gateway sends to the Lambda.

// src/functions/contacts/events/create_contact.json
{
    "httpMethod": "POST",
    "path": "/contacts",
    "headers": {
        "Content-Type": "application/json"
    },
    "requestContext": {
        "httpMethod": "POST"
    },
    "body": "{\"email\": \"jane.doe@example.com\", \"name\": \"Jane Doe\", \"phone\": \"+11234567890\"}",
    "isBase64Encoded": false
}
Enter fullscreen mode Exit fullscreen mode

we also need a .env file with the values of the lambda function's environment variables for when we run it locally

// .env
{
    "ContactsFunction": {
        "CONTACTS_TABLE_NAME": "crm-sam-contacts-table"
    }
}
Enter fullscreen mode Exit fullscreen mode

and now we can proceed to build

sam build
Enter fullscreen mode Exit fullscreen mode

run the lambda function locally

sam local invoke ContactsFunction \
  -e src/functions/contacts/events/create_contact.json
  -n .env

{"statusCode": 201, "body": "{\"id\":\"788c7d86-30a6-46da-9550-27e42f7686cd\",\"email\":\"jane.doe@example.com\",\"name\":\"Jane Doe\",\"phone\":\"+11234567890\",\"created_at\":\"2025-07-20T14:59:28.811723+00:00\",\"updated_at\":\"2025-07-20T14:59:28.811723+00:00\"}", "isBase64Encoded": false, "multiValueHeaders": {"Content-Type": ["application/json"]}}
Enter fullscreen mode Exit fullscreen mode

run the API locally

sam local start-api
curl -X POST http://127.0.0.1:3000/contacts \
     -H 'Content-Type: application/json' \
     -d '{"email":"alice@example.com","name":"Alice"}'

{"id":"ac1c51b7-2fe6-495f-8578-70c123c8927f","email":"alice@example.com","name":"Alice","phone":null,"created_at":"2025-07-20T15:11:21.174541+00:00","updated_at":"2025-07-20T15:11:21.174541+00:00"}
Enter fullscreen mode Exit fullscreen mode

and deploy it

sam deploy
curl -X POST https://uywf3xcoej.execute-api.us-east-1.amazonaws.com/prod/contacts \
     -H 'Content-Type: application/json' \
     -d '{"email":"bob@example.com","name":"Bob"}'

{"id":"a7ea3b2c-6b7c-422a-b8eb-36795f39e537","email":"bob@example.com","name":"Bob","phone":null,"created_at":"2025-07-20T15:15:29.276161+00:00","updated_at":"2025-07-20T15:15:29.276161+00:00"}
Enter fullscreen mode Exit fullscreen mode

What did we learn?

  • Use of AWS Lambda Powertools to simplify the lambda function code
  • Use of pydantic for data validation
  • Use of the repository pattern to abstract data access logic
  • Dependency management for lambda functions

And with that, we already have the first CRM feature. If you want to see the code of the project, you can find it here, and if you have and suggestion or feedback, let me know in the comments.

Coming next

Now that we can create contacts, in part 3 we will retrieve them (GET /contacts), add pagination, and explore query patterns.

Top comments (1)

Collapse
 
cashoefman profile image
Cas Hoefman •

Looking forward to part 3.