Reputation: 415
I want to make a class representing a dynamodb connection which I can use locally to mimic the behavior of DynamoDB without having to actually contact the AWS service.
I want to use it during development (not just for running tests) and use that class for creating a table which exists during program execution, and make queries to it just as I would a real dynamo table with a real connection
I am using boto3 v1.34.162
and moto v5.0.15
So far, here is what I have come up with:
import boto3
from moto import mock_aws
@mock_aws
class MyDbClient:
def __init__(self):
self.dynamodb = boto3.client('dynamodb')
self.table_name = 'ExampleTable'
self.create_table()
def create_table(self):
table_params = {
'TableName': self.table_name,
'KeySchema': [
{'AttributeName': 'id', 'KeyType': 'HASH'},
],
'AttributeDefinitions': [
{'AttributeName': 'id', 'AttributeType': 'S'},
],
'ProvisionedThroughput': {
'ReadCapacityUnits': 5,
'WriteCapacityUnits': 5,
}
}
print(f"Creating table {self.table_name}...")
self.dynamodb.create_table(**table_params)
print(f"Waiting for table {self.table_name} to exist...")
waiter = self.dynamodb.get_waiter('table_exists')
waiter.wait(TableName=self.table_name)
print(f"Table {self.table_name} is now active.")
tables = self.dynamodb.list_tables()['TableNames']
print("Existing tables:", tables)
def put_item_in_table(self, item):
self.dynamodb.put_item(
TableName=self.table_name,
Item={
'id': {'S': item['id']},
'name': {'S': item['name']},
'description': {'S': item['description']}
}
)
def get_item_from_table(self, key):
response = self.dynamodb.get_item(
TableName=self.table_name,
Key={
'id': {'S': key['id']}
}
)
print(response)
if __name__ == "__main__":
db_client = MyDbClient()
item_to_put = {
'id': '123',
'name': 'ExampleName',
'description': 'This is a sample item'
}
db_client.put_item_in_table(item_to_put)
key_to_get = {'id': '123'}
db_client.get_item_from_table(key_to_get)
The @mock_aws
decorator should be what I need, but when executing this I am getting the following error: botocore.exceptions.ClientError: An error occurred (404) when calling the GetRoleCredentials operation: Not yet implemented
when the program execution comes to the create_table
call.
Here is the full stack trace as well:
(venv) ⏺ 15:19:16 ~/Dev/sandbox $ python main.py
Creating table ExampleTable...
Traceback (most recent call last):
File "/Users/jovan/Dev/sandbox/main.py", line 60, in <module>
db_client = MyDbClient()
^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/main.py", line 9, in __init__
self.create_table()
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/moto/core/models.py", line 122, in wrapper
result = func(*args, **kwargs)
^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/main.py", line 27, in create_table
self.dynamodb.create_table(**table_params)
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/client.py", line 569, in _api_call
return self._make_api_call(operation_name, kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/client.py", line 1005, in _make_api_call
http, parsed_response = self._make_request(
^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/client.py", line 1029, in _make_request
return self._endpoint.make_request(operation_model, request_dict)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/endpoint.py", line 119, in make_request
return self._send_request(request_dict, operation_model)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/endpoint.py", line 196, in _send_request
request = self.create_request(request_dict, operation_model)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/endpoint.py", line 132, in create_request
self._event_emitter.emit(
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/hooks.py", line 412, in emit
return self._emitter.emit(aliased_event_name, **kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/hooks.py", line 256, in emit
return self._emit(event_name, kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/hooks.py", line 239, in _emit
response = handler(**kwargs)
^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/signers.py", line 105, in handler
return self.sign(operation_name, request)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/signers.py", line 188, in sign
auth = self.get_auth_instance(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/signers.py", line 306, in get_auth_instance
frozen_credentials = credentials.get_frozen_credentials()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/credentials.py", line 634, in get_frozen_credentials
self._refresh()
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/credentials.py", line 522, in _refresh
self._protected_refresh(is_mandatory=is_mandatory_refresh)
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/credentials.py", line 538, in _protected_refresh
metadata = self._refresh_using()
^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/credentials.py", line 685, in fetch_credentials
return self._get_cached_credentials()
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/credentials.py", line 695, in _get_cached_credentials
response = self._get_credentials()
^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/credentials.py", line 2160, in _get_credentials
response = client.get_role_credentials(**kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/client.py", line 569, in _api_call
return self._make_api_call(operation_name, kwargs)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/jovan/Dev/sandbox/venv/lib/python3.12/site-packages/botocore/client.py", line 1023, in _make_api_call
raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (404) when calling the GetRoleCredentials operation: Not yet implemented
Another thing that I have tried is to individually wrap the __init__
, put_item_in_table
and get_item_from_table
with the @mock_aws
decorator, however that does NOT work, since in that case the the aws mock environment is created anew every time any of those functions is called, and therefore the created dynamo table does not persist between invocations (which is the complete opposite of what I need)
If possible I would like to just rely on the boto3
and moto
libraries. I know that aws localstack exists as a tool, but that is overkill for what I am trying to achieve and I would like to not use it
Upvotes: 0
Views: 238
Reputation: 2093
It sounds like the system has SSO-based credentials configured. All requests to AWS are intercepted, Moto just doesn't know how to respond to the SSO auth request.
Two workarounds for that:
~/.aws/credentials
to something else)AWS_ACCESS_KEY_ID=test AWS_SECRET_ACCESS_KEY=test pytest main`
That should override the SSO-based credentials that are configured on your system.
(FWIW - the code works on my machine! So it looks like you're pretty close.)
Upvotes: 1