hennge / aiodynamo Goto Github PK

We could perhaps benchmark https://pypi.org/project/pysimdjson/ as a starting point...

As seen by dynamodb-admin:

{
  [key omitted]
  "user_count": 4126,
  "foo_count": 0,
  "bar_count": 0,
  "mt": "1590462268"
}

Read by aiodynamo:

{[key omitted], 'user_count': 4126.0, 'foo_count': 0.0, 'bar_count': 0.0, 'mt': '1590462268'}

Sends StreamSpecification=None when none is supplied by the caller.
Perhaps aiodynamo should not inject this key if the value is None.

It seems that's OK for AWS SaaS dynamo, dynamodb-local and dynalite, but not ScyllaDB:

scylladb/scylladb#5796

Benchmarks broken due to issue supplying invalid base64 byte dummy data.

E       binascii.Error: Invalid base64-encoded string: number of data characters (5) cannot be 1 more than a multiple of 4

Seems like this issue was fixed in unit tests but not propagated to benchmarks

I propose we just copy the implementation over since there's no easy way to share test utils with benchmark because of the folder benchmarks are expected to run from.

there is a typo:

includes #86

Read operations in Dynamo support a ConsistentRead boolean parameter. Would be great if this was exposed through the API.

When HTTP client wrappers fail for any reason, they raise a aiodynamo.http.base.RequestFailed error to signal to the rest of aiodynamo that this was not a DynamoDB side issue and that a request should likely be retried. However, right now it doesn't do a good job at reporting why the HTTP Client failed, making diagnosing these errors hard/impossible.

These points are not in the documentations, so it might be confusing to those with minimum knowledge of Dynamo

• Operations that can be done with client.Table (create, get_item, put_item, update_item, delete_item, query)
• Usage example of KeyCondition and Condition/Filter Expressions
• Exceptions that will be thrown

I will try to update it later.

There's an aiohttp example in quickstart, I think it would be nice to have httpx example too.

After all, it's not like one transport is favoured over another, is it?

For example if pass limit "foobar" to limit in count it gets throttled. At least in tests against DynamoDb Local

____________________________________________________________________________________________________________ test_count_with_limit[httpx-True] _____________________________________________________________________________________________________________

client = Client(http=HTTPX(client=<httpx.AsyncClient object at 0x105f3f4f0>), credentials=<aiodynamo.credentials.ChainCredentia...class 'float'>, throttle_config=ExponentialBackoffThrottling(time_limit_secs=60, base_delay_secs=2, max_delay_secs=20))
prefilled_table = '429ba920-a8ec-44b1-bd2b-44c17b4ee4b4', consistent_read = True

    @pytest.mark.parametrize("consistent_read", [None, True, False])
    async def test_count_with_limit(
        client: Client, prefilled_table: TableName, consistent_read: Optional[bool]
    ) -> None:
>       assert (
            await client.count(
                prefilled_table,
                HashKey("h", "h"),
                limit="foobar", #90,
                consistent_read=consistent_read,
            )
            == 90
        )

tests/integration/test_client.py:150: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/aiodynamo/client.py:824: in count
    async for result in self._depaginate("Query", payload, limit):
src/aiodynamo/client.py:999: in _depaginate
    result = await task
src/aiodynamo/client.py:945: in send_request
    async for _ in self.throttle_config.attempts():
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = ExponentialBackoffThrottling(time_limit_secs=60, base_delay_secs=2, max_delay_secs=20)

    async def attempts(self) -> AsyncIterable[None]:
        deadline = time.monotonic() + self.time_limit_secs
        for delay in self.delays():
            yield
            if time.monotonic() > deadline:
>               raise Throttled()
E               aiodynamo.errors.Throttled

src/aiodynamo/models.py:238: Throttled

For example, ResourceInUseException (e.g. trying to create a table that already exists)

Today the error is aiodynamo.errors.UnknownError: b'{"__type":"com.amazonaws.dynamodb.v20120810#ResourceInUseException","message":""}'

I can't figure out what the difference between Condition and KeyCondition is 😢

AWS just released Fargate platform 1.4.0 with new task metadata endpoint v4
https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-metadata-endpoint-v4.html

botocore introduces quite significant slowdowns, especially when querying large amounts of data. the dynamodb api is simple enough that we should be able to bypass it and talk to dynamodb directly with an http client.

Running the recommended python bench_aiodynamo_aiohttp.py -o aiodynamo_aiohttp.json --rigorous --inherit-env BENCH_TABLE_NAME,BENCH_KEY_FIELD,BENCH_KEY_VALUE,BENCH_REGION_NAME,AWS_ACCESS_KEY_ID,AWS_SECRET_ACCESS_KEY,AWS_SESSION_TOKEN

[snip]
    "KeyConditionExpression": key_condition.encode(params),
AttributeError: 'Equals' object has no attribute 'encode'

When I was adding credentials handler to assume role with web identity which we need for our EKS setup I came across few issues why I am for now implementing it as custom code in our own project instead of providing it for the library.

Http wrappers assume response is json, https://github.com/HENNGE/aiodynamo/blob/master/src/aiodynamo/http/base.py but as endpoint response is xml (and at least I can't find any mention of possibility to request it as json, but you know AWS docs 💩 ). This is blocker and thought that perhaps there needs to be some refactoring around http handling which takes too long for us.

Some additional things I noticed

1. in MetadataCredentials there is fetch_with_retry which couldn't use because it assumes GET but endpoint for assume role is POST. Was thinking should/could this retry option be in base http classes?

2. Missing timeout in POST, would prefer to have timeout also for this credentials call. As mentioned in #45

3. http.post(..) requires body as bytes even when it can in aiohttp and httpx also be None. https://github.com/HENNGE/aiodynamo/blob/master/src/aiodynamo/http/base.py#L28-L30

For example, test_ttl contains:

async def test_ttl(client: Client, table: TableName):
    try:
        desc = await client.describe_time_to_live(table)
    except UnknownOperation:
        raise pytest.skip("TTL not supported by database")

Yet, when ran against newest dynalite, test fails with:

tests/integration/test_client.py::test_ttl[httpx] FAILED                                                                                                                                                     [ 50%]

===================================================================================================== FAILURES =====================================================================================================
_________________________________________________________________________________________________ test_ttl[httpx] __________________________________________________________________________________________________

client = Client(http=HTTPX(client=<httpx.client.AsyncClient object at 0x7f83600fb1c0>), credentials=ChainCredentials(candidates...onfig=ThrottleConfig(max_attempts=5, delay_func=<function ThrottleConfig.default.<locals>.<lambda> at 0x7f83584f2c10>))
table = '40588f16-bdec-43f4-bc2a-72bce590d83d'

    async def test_ttl(client: Client, table: TableName):
        try:
            desc = await client.describe_time_to_live(table)
        except UnknownOperation:
            raise pytest.skip("TTL not supported by database")
        assert desc.status == TimeToLiveStatus.disabled
        assert desc.attribute == None
>       await client.enable_time_to_live(table, "ttl")

tests/integration/test_client.py:222:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/aiodynamo/client.py:287: in enable_time_to_live
    await self.set_time_to_live(
src/aiodynamo/client.py:322: in set_time_to_live
    await self.send_request(
src/aiodynamo/client.py:650: in send_request
    return await self.http.post(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = HTTPX(client=<httpx.client.AsyncClient object at 0x7f83600fb1c0>)

    async def post(
        self, *, url: URL, body: bytes, headers: Optional[Headers] = None
    ) -> Dict[str, Any]:
        response = await self.client.post(str(url), data=body, headers=headers)
        if response.status_code >= 400:
>           raise exception_from_response(response.status_code, await response.aread())
E           aiodynamo.errors.UnknownOperation: {'__type': 'com.amazon.coral.service#UnknownOperationException'}

src/aiodynamo/http/httpx.py:32: UnknownOperation

update_item(..., F("foo").set(0) & F("bar").set(False)) incorrectly encodes to SET #N0 = :V0, #N1 =: V0 with #N0 = {"S": "foo"}, #N1 = {"S": "bar"} and V0 = {"N": "0"}

These 2 should be retried:

    if status == 500:
        return InternalDynamoError()
    elif status == 503:
        raise ServiceUnavailable()

Possibly some other errors too.

ref: microsoft/pyright#2282

Just wondering if these were left out on purpose 🤔

Hey

I found another problem while using this library.

When using InstanceMetadataCredentials there is an error thrown from httpx:

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/aiodynamo/http/httpx.py", line 18, in wrapper
    return await coro(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/aiodynamo/http/httpx.py", line 33, in get
    response = await self.client.get(str(url), headers=headers, timeout=timeout)
  File "/usr/local/lib/python3.8/site-packages/httpx/client.py", line 1236, in get
    return await self.request(
  File "/usr/local/lib/python3.8/site-packages/httpx/client.py", line 1086, in request
    request = self.build_request(
  File "/usr/local/lib/python3.8/site-packages/httpx/client.py", line 200, in build_request
    url = self.merge_url(url)
  File "/usr/local/lib/python3.8/site-packages/httpx/client.py", line 220, in merge_url
    url = self.base_url.join(relative_url=url)
  File "/usr/local/lib/python3.8/site-packages/httpx/models.py", line 229, in join
    return URL(relative_url)
  File "/usr/local/lib/python3.8/site-packages/httpx/models.py", line 108, in __init__
    raise InvalidURL("No host included in URL.")
httpx.exceptions.InvalidURL: No host included in URL.

I could track the problem down to this line

>>> role_url = URL("http://169.254.169.254/latest/meta-data/iam/security-credentials/")
>>> role = "test-role"
>>> role_url.join(URL(role))
URL('http://169.254.169.254/latest/meta-data/iam/security-credentials/test-role')
>>> role = "arn:aws:iam::1234567890:role/test-role"
>>> role_url.join(URL(role))
URL('arn:aws:iam::1234567890:role/test-role') # expected would be URL('http://169.254.169.254/latest/meta-data/iam/security-credentials/arn:aws:iam::1234567890:role/test-role')

My use case is that the request gets intercepted with KIAM and my code is supposed to assume this role. I tried the urls with curl and they worked as expected.

Hey
while experimenting with this library and one of my existing dynamodbs i encountered an issue when I use the query api.

id = "1"
async for item in table.query(
    key_condition=HashKey("id", id) # this doesn't work
):
    print(item)

I got this exception

aiodynamo.errors.ValidationException: {'__type': 'com.amazon.coral.validate#ValidationException', 'message': 'KeyConditionExpressions cannot have conditions on nested attributes'}

I didn't use anything nested so I tried calling encode on the hashkey directly to find this

params = Parameters()
HashKey("id", "1").encode(params)  # output: '#n0.#n1 = :v0'
params.names  # output: {'#n0': 'i', '#n1': 'd'}
params.values  # output: {':v0': {'S': '1'}}

It seems to build a query "i.d = 1" instead of "id = 1"

I was able to query it correctly after wrapping the column name into a array but that goes against what the type of the name parameter for HashKey should be.

working example

id = "1"
async for item in table.query(
    key_condition=HashKey(["id"], id)  # this works
):
    print(item)

Including:

• #50 empty strings
• #49 retry on internal error

errors like failing to connect to a host, doing dns lookup and connnection resets should probably retried.

when I tried

from aiodynamo.client import Client
from aiodynamo.credentials import Credentials
from aiodynamo.http.aiohttp import AIOHTTP
from aiohttp import ClientSession
import asyncio

async def main():
    async with ClientSession() as session:
        client = Client(AIOHTTP(session), Credentials.auto(), "ap-northeast-1")
        table = client.table("table")
        await table.put_item(
          {
                'id':0,
                'data':"000"
            }
        )
        
asyncio.run(main())

and terminal showed below

Traceback (most recent call last):
File "C:\Users\user\AppData\Local\Programs\Python\Python39\lib\site-packages\aiodynamo\credentials.py", line 101, in get_key
key = await candidate.get_key(http)
File "C:\Users\user\AppData\Local\Programs\Python\Python39\lib\site-packages\aiodynamo\credentials.py", line 215, in get_key
await self._refresher
File "C:\Users\user\AppData\Local\Programs\Python\Python39\lib\site-packages\aiodynamo\credentials.py", line 239, in _refresh
self._metadata = await self.fetch_metadata(http)
File "C:\Users\user\AppData\Local\Programs\Python\Python39\lib\site-packages\aiodynamo\credentials.py", line 338, in fetch_metadata
await self.fetch_with_retry(
File "C:\Users\user\AppData\Local\Programs\Python\Python39\lib\site-packages\aiodynamo\credentials.py", line 196, in fetch_with_retry
raise TooManyRetries()
aiodynamo.credentials.TooManyRetries
Candidate InstanceMetadataCredentials(timeout=1, max_attempts=1, base_url=URL('http://169.254.169.254'), disabled=False) failed

I think this problem because of my networking

ping 169.254.169.254

Pinging 169.254.169.254 with 32 bytes of data:
PING: transmit failed.
PING: transmit failed.
PING: transmit failed.
PING: transmit failed.

Ping statistics for 169.254.169.254:
Packets: Sent = 4, Received = 0, Lost = 4 (100% loss),

But according to boto3 available, I think it doesn't need this process.

When dealing with large items/large number of items, deserialize quickly becomes a bottleneck. The current version is already an optimized version over what botocore does (which is a major reason why aiodynamo is faster than botocore), but there is probably still room for improvement.

One option would be to include an (optional) cython version of the function.

Ref: scylladb/scylladb#9542 (comment)

@freedomofkeima says s/ScanForward/ScanIndexForward?

Empty strings (and bytes) are now allowed except in key/index fields. aiodynamo should support this.

Outstanding questions are:

• Should aiodynamo try to check key fields are not empty? Currently keys are not handled special (eg their types are not checked even though only S, B and N are allowed)
• How would aiodynamo know which fields are keys? In update_item that's trivial (though indices cannot be detected) but put_item the user would have to supply the keys somehow.
• Maybe this should be something only handled on Table, not Client, by having the user pass a list of key/index fields to the constructor.

Outstanding problems:

• dynalite does not support this yet

I'm wondering if there are any plans to support TransactWriteItems or just Transactions in general.

I can probably work on this if/when I have more time 🤔

Python has a canonical str.starswith
Meanwhile Dynamo has begins_with

WDYT about F gaining F.startswith = F.begins_with?

😢

Not sure if this is correct channel for this kind of request.

But I wish there was faster release cycle, currently of course interested in having released version which would include #101. Last release has been October 7th and after that has been 2 functional improvement PRs, and that consistent read support was merged 23 days ago.

I am now using git dependency but as there are also no tags have to rely on random rev value which is not obviously clear what we are using.

It seems that the core only uses yarl.URL as a type: src/aiodynamo/http/base.py

Transport only does str(url): src/aiodynamo/http/httpx.py

Credentials code has base_url.with_path and role_url.join(URL(role))

Signature code has URL.build(scheme="https", host=f"{SERVICE}.{region}.amazonaws.com", path="/")

It seems the use is relatively minor 🤔

otherwise I'm not sure what we're testing against...

The [snip] are mine because well, I don't want to share keys on github ;-)

① key refresh debug

DEBUG:aiodynamo:fetchhed metadata b'{\\n \"Code\" : \"Success\",\\n \"LastUpdated\" : \"2020-04-02T04:16:57Z\",\\n \"Type\" : \"AWS-HMAC\",\\n \"AccessKeyId\" : \"A[snip]\",\\n \"SecretAccessKey\" : \"x[snip]\",\\n \"Token\" : \"IQ[snip]==\",\\n \"Expiration\" : \"2020-04-02T10:39:09Z\"\\n}'",

② request debug

DEBUG:aiodynamo:sending request Request(url=URL('https://dyna[snip].com/'), headers={'Content-Type': 'application/x-amz-json-1.0', 'X-Amz-Date': '20200402T045338Z', 'X-Amz-Target': 'DynamoDB_20120810.UpdateItem', 'Authorization': 'AWS4-HMAC-SHA256 Credential=A[snip]/20200402/ap-northeast-1/dynamodb/aws4_request, SignedHeaders=content-type;host;x-amz-date;x-amz-target, Signature=db[snip]', 'X-Amz-Security-Token': 'IQ[snip]=='}, body=b'{\"TableName\":\"is[snip]fault\",\"Key\":{\"H\":{\"S\":\"DELETION_REQUEST\"},\"R\":{\"S\":\"pur[snip]com\"}},\"UpdateExpression\":\"SET #n0 = :v0, #n1 = :v1, #n2 = :v2\",\"ReturnValues\":\"NONE\",\"ExpressionAttributeNames\":{\"#n0\":\"not_before\",\"#n1\":\"request_by\",\"#n2\":\"request_time\"},\"ExpressionAttributeValues\":{\":v0\":{\"N\":\"1585803518.478671\"},\":v1\":{\"S\":\"X2/si4Ej[snip]\"},\":v2\":{\"N\":\"1585803218.794926\"}}}')",

    async def test_query_with_limit(client: Client, fast_table: TableName):
        big = "x" * 20_000

>       await asyncio.gather(
            *(
                client.put_item(fast_table, {"h": "h", "r": str(i), "big": big})
                for i in range(100)
            )
        )

tests/integration/test_client.py:239:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/aiodynamo/client.py:483: in put_item
    resp = await self.send_request(action="PutItem", payload=payload)
src/aiodynamo/client.py:650: in send_request
    return await self.http.post(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = HTTPX(client=<httpx.client.AsyncClient object at 0x7fd9914d37f0>)

    async def post(
        self, *, url: URL, body: bytes, headers: Optional[Headers] = None
    ) -> Dict[str, Any]:
        response = await self.client.post(str(url), data=body, headers=headers)
        if response.status_code >= 400:
>           raise exception_from_response(response.status_code, await response.aread())
E           aiodynamo.errors.ProvisionedThroughputExceeded: {'__type': 'com.amazonaws.dynamodb.v20120810#ProvisionedThroughputExceededException', 'message': 'The level of configured provisioned throughput for the table was exceeded. Consider increasing your provisioning level with the UpdateTable API.'}

I've hacked it to create tables with 1000/1000 rcu (?) and yet, test cannot insert 100 items concurrently.
😠

And while on the subject, if the endpoint is passed to the Client, is the region still strictly necessary?

This happened when I was using conditions expressions with update_item

This used to work in botoland:
condition=Attr("files").size().lt(Attr("max_num_files"))

Now when I do this
condition=F("files").size().lt(F("max_num_files"))

it looks like it can't do attribute values?

|   File "/usr/local/lib/python3.8/site-packages/aiodynamo/utils.py", line 127, in low_level_serialize
|     raise TypeError(f"Unsupported type {type(value)}: {value!r}")
| TypeError: Unsupported type <class 'aiodynamo.expressions.F'>: <aiodynamo.expressions.F object at 0x7fbcac2ea130>

Described in https://dev.to/matthewvielkind/updating-values-in-dyanmodb-map-attributes-can

UpdateExpression="SET players.#player_id.score = :score_val",  
ExpressionAttributeNames={"#player_id": player_id},  

I'm not sure what the API is in this case...
Would I need to overload F?

Was trying to migrate from botocore which provides exceptions module, is there an equivalent of botocore.exceptions.ClientError with error response message?

Hello

I noticed that aiodynamo still requires a old version of httpx from begining of 2020.
The current constrain httpx = {version = "^0.11.1", optional = true} limits the available versions to >=0.11.1,<0.12 according to the poetry documentation on version constrains.

Would you mind trying it out with newer versions and relaxing the strict version constrain a bit?

Thank you

How come get has a timeout= mandatory keyword argument, but post does not have a timeout= argument?

Exception text: ClientConnectorError: Cannot connect to host dynamodb.<snip>.amazonaws.com:443 ssl:default [None]

Exception type: aiohttp.client_exceptions.ClientConnectorError

Stack:

aiohttp/connector.py in _wrap_create_connection at line 943
aiohttp/connector.py in _create_direct_connection at line 980
aiohttp/connector.py in _create_direct_connection at line 1004
aiohttp/connector.py in _create_connection at line 858
aiohttp/connector.py in connect at line 523
aiohttp/client.py in _request at line 480
aiohttp/client.py in __aenter__ at line 1012
aiodynamo/http/aiohttp.py in post at line 31
aiodynamo/client.py in send_request at line 651
aiodynamo/client.py in get_item at line 454
aiodynamo/client.py in get_item at line 124

scylladb/scylladb#5796 (comment)

ScyllaDB is a fast reimplementation of Cassandra, and they have a dynamodb compatibility layer called alternator.

I've ran some basic tests against their docker image. It would be awesome to run a performance test now that aiodynamo is so much faster :)