En utilisant exactement le même hook (endpoint_url), on va utiliser notre faux cloud pour créer une base de données et insérer des données et les lires.

Le code

import boto3

# All clients use the same endpoint
def client(service):
    return boto3.client(
        service,
        endpoint_url="http://awslocal:4566",
        aws_access_key_id="test",
        aws_secret_access_key="test",
        region_name="us-east-1",
    )

dyn_db = client("dynamodb")

# 1. Create Users table
try:
    dyn_db.create_table(
        TableName='Users',
        KeySchema=[{'AttributeName': 'UserId', 'KeyType': 'HASH'}],
        AttributeDefinitions=[{'AttributeName': 'UserId', 'AttributeType': 'S'}],
        BillingMode='PAY_PER_REQUEST'
    )
    print("'Users' Table created.")
except dyn_db.exceptions.ResourceInUseException:
    print("'Users' Table already exists, skip creation...")

# 2. Insert data (put_item)
print("Insertion de données...")
users_to_add = [
    {'UserId': 'john', 'LastName': 'Doe', 'FirstName': 'John', 'Role': 'CIO'},
    {'UserId': 'jane', 'LastName': 'Doe', 'FirstName': 'Jane', 'Role': 'HR Director'},
    {'UserId': 'vladimir', 'LastName': 'Popov', 'FirstName': 'Vladimir', 'Role': 'Lead Tech'}
]

for user in users_to_add:
    dyn_db.put_item(
        TableName='Users',
        Item={
            'UserId': {'S': user['UserId']},
            'FirstName': {'S': user['FirstName']},
            'LastName': {'S': user['LastName']},
            'Role': {'S': user['Role']}
        }
    )

# 3. Read specific entry (get_item)
print("\nRead specific user (jane) :")
response = dyn_db.get_item(
    TableName='Users',
    Key={'UserId': {'S': 'jane'}}
)
if 'Item' in response:
    print(f"Found : {response['Item']['LastName']['S']}, {response['Item']['FirstName']['S']}")

# 4. Read all entries of the Users table (scan)
print("\nList all users :")
all_users = dyn_db.scan(TableName='Users')
for item in all_users['Items']:
    print(f"- ID: {item['UserId']['S']}, Nom: {item['LastName']['S']}, {item['FirstName']['S']}")

Le test

docker compose exec cli sh -c "cd tests ; poetry run python src/tuto-02.py"

'Users' Table already exists, skip creation...
Insertion de données...

Read specific user (jane) :
Found : Doe, Jane

List all users :
- ID: jane, Nom: Doe, Jane
- ID: john, Nom: Doe, John
- ID: vladimir, Nom: Popov, Vladimir