boto3 x aws sdk js

client_dynamo.js

import { DynamoDB } from "aws-sdk"

let client = null

export const getClient = (): DynamoDB => {
    if (client) return client
    client = new DynamoDB({
        httpOptions: {
            connectTimeout: 1000,
            timeout: 1000
        }
    })
    return client
}

client_dynamo.py

def dynamo_db_client(localhost: bool = False):
    """Return a DynamoDB Client."""
    config = Config(
        connect_timeout=60,
        read_timeout=60,
        retries={"total_max_attempts": 3},
    )
    return (
        boto3.resource("dynamodb", config=config)
        if not localhost
        else boto3.resource(
            "dynamodb",
            config=config,
            endpoint_url="http://localhost:8000",
        )
    )

file.py

import boto3

@pytest.fixture(scope="session")
def create_stepfunction():
    name = str(uuid4())
    with mock_iam():
        name = str(uuid4())
        role_client = boto3.client("iam")
        document = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": ["sts:AssumeRole"],
                    "Resource": "*",
                }
            ],
        }
        role = role_client.create_role(
            RoleName=name,
            AssumeRolePolicyDocument=json.dumps(document),
        )
        with mock_stepfunctions():
            sfn_client = boto3.client("stepfunctions")
            state = sfn_client.create_state_machine(
                name=name,
                definition=json.dumps({}),
                roleArn=role["Role"]["Arn"],
            )
            yield

step_function.py

import json
import logging
from typing import Dict
from uuid import uuid4

import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger(__name__)


def trigger_step_function(context: Dict, function_name: str) -> None:
    """
    Trigger a step function
    :param context: context to be passed to function
    """
    sfn_client = boto3.client("stepfunctions")
    try:
        sfn_client.start_execution(
            stateMachineArn=function_name,
            input=json.dumps(context),
            name=str(uuid4()),
        )
    except ClientError as err:
        response = err.response.get("Error", None)
        logger.error(response)
        raise Exception()