DynamoDB with boto3 (Python)

Uses boto3 1.43 with endpoint_url. The code is an ordinary application for the original database; nothing in it knows about MachNoSQL.

The only change

https://dynamodb.eu-central-1.amazonaws.com
https://machnosql.example.com:8000

The address comes from the environment variable DYNAMODB_ENDPOINT, so switching to MachNoSQL changes configuration only.

Run

export DYNAMODB_ENDPOINT='https://machnosql.example.com:8000'
pip install -r requirements.txt
python app.py

Credentials are the usual AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY (an access key created in MachNoSQL) and AWS_REGION. Instead of passing the endpoint in code, current AWS SDKs and the AWS CLI also read AWS_ENDPOINT_URL_DYNAMODB from the environment; then the application needs no change at all.

Source

The program as it is kept and tested in the Mach Database NoSQL repository; highlighted lines read the connection setting.

app.py

"""Orders table on DynamoDB, unchanged on MachNoSQL."""

import os

import boto3
from boto3.dynamodb.conditions import Key
from botocore.exceptions import ClientError

dynamodb = boto3.resource("dynamodb", endpoint_url=os.environ.get("DYNAMODB_ENDPOINT"))

try:
    table = dynamodb.create_table(
        TableName="Orders",
        KeySchema=[
            {"AttributeName": "customer", "KeyType": "HASH"},
            {"AttributeName": "orderId", "KeyType": "RANGE"},
        ],
        AttributeDefinitions=[
            {"AttributeName": "customer", "AttributeType": "S"},
            {"AttributeName": "orderId", "AttributeType": "S"},
        ],
        BillingMode="PAY_PER_REQUEST",
    )
    table.wait_until_exists()
except ClientError as error:
    if error.response["Error"]["Code"] != "ResourceInUseException":
        raise
    table = dynamodb.Table("Orders")

with table.batch_writer() as batch:
    batch.put_item(Item={"customer": "ada", "orderId": "1001", "item": "keyboard", "total": 49})
    batch.put_item(Item={"customer": "ada", "orderId": "1002", "item": "mouse", "total": 39})
    batch.put_item(Item={"customer": "linus", "orderId": "1003", "item": "monitor", "total": 199})

print(table.get_item(Key={"customer": "ada", "orderId": "1001"})["Item"])
print(table.query(KeyConditionExpression=Key("customer").eq("ada"))["Items"])

table.update_item(
    Key={"customer": "ada", "orderId": "1002"},
    UpdateExpression="SET #s = :shipped",
    ConditionExpression="attribute_not_exists(#s)",
    ExpressionAttributeNames={"#s": "status"},
    ExpressionAttributeValues={":shipped": "shipped"},
)

requirements.txt

boto3==1.43.122