Usage

Client instantiation

You should try to re-use clients, and especially HTTP clients, as much as possible. Don’t create a new one for each action.

The aiodynamo.client.Client class takes three required and three optional arguments:

  1. An HTTP client adaptor, conforming to the aiodynamo.http.base.HTTP interface.
  2. An instance of aiodynamo.credentials.Credentials to authenticate with DynamoDB. You may use Credentials.auto() to use the default loading strategy.
  3. The region your DynamoDB is in.
  4. An optional endpoint URL of your DynamoDB, as a yarl.URL instance. Useful when using a local DynamoDB implementation such as dynalite or dynamodb-local.
  5. Which numeric type to use. This should be a callable which accepts a string as input and returns your numeric type as output. Defaults to float.
  6. The throttling configuration to use. An instance of aiodynamo.models.ThrottleConfig. By default, if the DynamoDB rate limit is exceeded, aiodynamo will retry up for up to one minute with increasing delays.

Credentials

In most cases, Credentials.auto() will load the credentials as you’d expect. Specifically, it will try multiple credentials providers in this order: aiodynamo.credentials.EnvironmentCredentials, aiodynamo.credentials.FileCredentials, aiodynamo.credentials.ContainerMetadataCredentials and aiodynamo.credentials.InstanceMetadataCredentials.

In case you want to explicitly pass the credentials from Python, use aiodynamo.credentials.StaticCredentials.

classmethod Credentials.auto() → aiodynamo.credentials.ChainCredentials

Return the default credentials loader chain.


class aiodynamo.credentials.EnvironmentCredentials

Loads the credentials from the environment.


class aiodynamo.credentials.ContainerMetadataCredentials(timeout: Union[float, int] = 2, max_attempts: int = 3, base_url: yarl.URL = URL('http://169.254.170.2'), relative_uri: Optional[str] = <factory>, full_uri: Optional[yarl.URL] = <factory>, auth_token: Optional[str] = <factory>)

Loads credentials from the ECS container metadata endpoint.


class aiodynamo.credentials.InstanceMetadataCredentials(timeout: Union[float, int] = 1, max_attempts: int = 1, base_url: yarl.URL = URL('http://169.254.169.254'), disabled: bool = <factory>)

Loads credentials from the EC2 instance metadata endpoint.


class aiodynamo.credentials.ChainCredentials(candidates: Sequence[aiodynamo.credentials.Credentials])

Chains multiple credentials providers together, trying them in order. Returns the first key found. Exceptions are suppressed.


class aiodynamo.credentials.StaticCredentials(key: Optional[aiodynamo.credentials.Key])

Static credentials provided in Python.


class aiodynamo.credentials.FileCredentials(*, path: Optional[pathlib.Path] = None, profile_name: str = 'default')

Loads the credentials from an AWS credentials file


class aiodynamo.credentials.Key(id: 'str', secret: 'str', token: 'Optional[str]' = None)
token = None

The Client class

class aiodynamo.client.Client
Client.table(name: str) → aiodynamo.client.Table
Client.table_exists(name: str) → bool
Client.create_table(name: str, throughput: aiodynamo.models.Throughput, keys: aiodynamo.models.KeySchema, *, lsis: Optional[List[aiodynamo.models.LocalSecondaryIndex]] = None, gsis: Optional[List[aiodynamo.models.GlobalSecondaryIndex]] = None, stream: Optional[aiodynamo.models.StreamSpecification] = None, wait_for_active: Union[bool, aiodynamo.models.WaitConfig] = False) → None

If wait_for_active is set to True, it will wait until the table status changed into Active. If after the defined wait time the table is not active, an exception will be raised.

Client.delete_table(table: str, *, wait_for_disabled: Union[bool, aiodynamo.models.WaitConfig] = False) → None

If wait_for_disabled is set to True, it will wait until the table status changed into Disabled. If after the defined wait time the table is not disabled, an exception will be raised.

Client.put_item(table: str, item: Dict[str, Any], *, return_values: aiodynamo.models.ReturnValues = <ReturnValues.none: 'NONE'>, condition: Optional[aiodynamo.expressions.Condition] = None) → Union[None, Dict[str, Any]]
Client.update_item(table: str, key: Dict[str, Any], update_expression: aiodynamo.expressions.UpdateExpression, *, return_values: aiodynamo.models.ReturnValues = <ReturnValues.none: 'NONE'>, condition: Optional[aiodynamo.expressions.Condition] = None) → Optional[Dict[str, Any]]
Client.delete_item(table: str, key: Dict[str, Any], *, return_values: aiodynamo.models.ReturnValues = <ReturnValues.none: 'NONE'>, condition: Optional[aiodynamo.expressions.Condition] = None) → Union[None, Dict[str, Any]]
Client.get_item(table: str, key: Dict[str, Any], *, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None) → Dict[str, Any]
Client.query(table: str, key_condition: aiodynamo.expressions.KeyCondition, *, start_key: Optional[Dict[str, Any]] = None, filter_expression: Optional[aiodynamo.expressions.Condition] = None, scan_forward: bool = True, index: Optional[str] = None, limit: Optional[int] = None, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None, select: aiodynamo.models.Select = <Select.all_attributes: 'ALL_ATTRIBUTES'>) → AsyncIterator[Dict[str, Any]]

Query the table. Unlike the DynamoDB API, the results are automatically de-paginated and a single stream of items is returned. For manual pagination, use query_single_page(…) instead.

Aiodynamo handles pagination automatically, so this method returns an asynchronous iterator of items.

To only retrieve a single page, use aiodynamo.client.Client.query_single_page()

Client.query_single_page(table: str, key_condition: aiodynamo.expressions.KeyCondition, *, start_key: Optional[Dict[str, Any]] = None, filter_expression: Optional[aiodynamo.expressions.Condition] = None, scan_forward: bool = True, index: Optional[str] = None, limit: Optional[int] = None, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None, select: aiodynamo.models.Select = <Select.all_attributes: 'ALL_ATTRIBUTES'>) → aiodynamo.models.Page

Query a single DynamoDB page. To automatically handle pagination, uses query(…) instead.

Queries a single page from DynamoDB. To automatically handle pagination, use aiodynamo.client.Client.query()

Client.scan(table: str, *, index: Optional[str] = None, limit: Optional[int] = None, start_key: Optional[Dict[str, Any]] = None, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None, filter_expression: Optional[aiodynamo.expressions.Condition] = None) → AsyncIterator[Dict[str, Any]]

Scan the table. Unlike the DynamoDB API, the results are automatically de-paginated and a single stream of items is returned. For manual pagination, use scan_single_page(…) instead.

Aiodynamo handles pagination automatically, so this method returns an asynchronous iterator of items.

To only retrieve a single page, use aiodynamo.client.Client.scan_single_page()

Client.scan_single_page(table: str, *, index: Optional[str] = None, limit: Optional[int] = None, start_key: Optional[Dict[str, Any]] = None, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None, filter_expression: Optional[aiodynamo.expressions.Condition] = None) → aiodynamo.models.Page

Scan a single DynamoDB page. To automatically handle pagination, uses scan(…) instead.

Scans a single page from DynamoDB. To automatically handle pagination, use aiodynamo.client.Client.scan()

Client.count(table: str, key_condition: aiodynamo.expressions.KeyCondition, *, start_key: Optional[Dict[str, Any]] = None, filter_expression: Optional[aiodynamo.expressions.Condition] = None, index: Optional[str] = None, limit: Optional[int] = None) → int

Aiodynamo handles pagination automatically, so this method returns the number of items.

Queries DynamoDB and returns number of matching items, optionally bounded by limit keyword argument.

Client.batch_get(request: Dict[str, aiodynamo.models.BatchGetRequest]) → aiodynamo.models.BatchGetResponse
Client.batch_write(request: Dict[str, aiodynamo.models.BatchWriteRequest]) → Dict[str, aiodynamo.models.BatchWriteResult]

The Table class

The aiodynamo.client.Table class wraps all methods on aiodynamo.client.Client so you don’t have to provide the table name each time.

This class should not be instantiated directly. Instead, create it by calling aiodynamo.client.Client.table().

Methods

class aiodynamo.client.Table
Table.exists() → bool
Table.create(throughput: aiodynamo.models.Throughput, keys: aiodynamo.models.KeySchema, *, lsis: Optional[List[aiodynamo.models.LocalSecondaryIndex]] = None, gsis: Optional[List[aiodynamo.models.GlobalSecondaryIndex]] = None, stream: Optional[aiodynamo.models.StreamSpecification] = None, wait_for_active: Union[bool, aiodynamo.models.WaitConfig] = False) → None

If wait_for_active is set to True, it will wait until the table status changed into Active. If after the defined wait time the table is not active, an exception will be raised.

Table.delete(*, wait_for_disabled: Union[bool, aiodynamo.models.WaitConfig] = False) → None

If wait_for_disabled is set to True, it will wait until the table status changed into Disabled. If after the defined wait time the table is not disabled, an exception will be raised.

Table.put_item(item: Dict[str, Any], *, return_values: aiodynamo.models.ReturnValues = <ReturnValues.none: 'NONE'>, condition: Optional[aiodynamo.expressions.Condition] = None) → Union[None, Dict[str, Any]]

Create a new item or replace it if it already exists. This will overwrite all attributes in an item.

Table.update_item(key: Dict[str, Any], update_expression: aiodynamo.expressions.UpdateExpression, *, return_values: aiodynamo.models.ReturnValues = <ReturnValues.none: 'NONE'>, condition: Optional[aiodynamo.expressions.Condition] = None) → Optional[Dict[str, Any]]

Edit an item’s attribute or create a new item if it does not exist. This will edit only the passed attributes.

Table.delete_item(key: Dict[str, Any], *, return_values: aiodynamo.models.ReturnValues = <ReturnValues.none: 'NONE'>, condition: Optional[aiodynamo.expressions.Condition] = None) → Union[None, Dict[str, Any]]
Table.get_item(key: Dict[str, Any], *, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None) → Dict[str, Any]

Returns the attributes of an item from table. This will return all attributes by default. To get only some attributes, use a projection expression.

Table.query(key_condition: aiodynamo.expressions.KeyCondition, *, start_key: Optional[Dict[str, Any]] = None, filter_expression: Optional[aiodynamo.expressions.Condition] = None, scan_forward: bool = True, index: Optional[str] = None, limit: Optional[int] = None, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None, select: aiodynamo.models.Select = <Select.all_attributes: 'ALL_ATTRIBUTES'>) → AsyncIterator[Dict[str, Any]]

Query the table.

Unlike the DynamoDB API, the results are automatically de-paginated and a single stream of items is returned. For manual pagination, use query_single_page(…) instead.

To filter the result, use a filter expression. This will return all attributes by default. To get only some attributes, use a projection expression.

Aiodynamo handles pagination automatically, so this method returns an asynchronous iterator of items.

To only retrieve a single page, use aiodynamo.client.Table.query_single_page()

Table.query_single_page(key_condition: aiodynamo.expressions.KeyCondition, *, start_key: Optional[Dict[str, Any]] = None, filter_expression: Optional[aiodynamo.expressions.Condition] = None, scan_forward: bool = True, index: Optional[str] = None, limit: Optional[int] = None, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None, select: aiodynamo.models.Select = <Select.all_attributes: 'ALL_ATTRIBUTES'>) → aiodynamo.models.Page

Query a single DynamoDB page. To automatically handle pagination, uses query(…) instead.

Queries a single page from DynamoDB. To automatically handle pagination, use aiodynamo.client.Table.query()

Table.scan(*, index: Optional[str] = None, limit: Optional[int] = None, start_key: Optional[Dict[str, Any]] = None, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None, filter_expression: Optional[aiodynamo.expressions.Condition] = None) → AsyncIterator[Dict[str, Any]]

Scan the table.

Unlike the DynamoDB API, the results are automatically de-paginated and a single stream of items is returned. For manual pagination, use scan_single_page(…) instead.

To filter the result, use a filter expression. This will return all attributes by default. To get only some attributes, use a projection expression.

Aiodynamo handles pagination automatically, so this method returns an asynchronous iterator of items.

To only retrieve a single page, use aiodynamo.client.Table.scan_single_page()

Table.scan_single_page(*, index: Optional[str] = None, limit: Optional[int] = None, start_key: Optional[Dict[str, Any]] = None, projection: Optional[aiodynamo.expressions.ProjectionExpression] = None, filter_expression: Optional[aiodynamo.expressions.Condition] = None) → aiodynamo.models.Page

Scan a single DynamoDB page. To automatically handle pagination, uses scan(…) instead.

Scans a single page from DynamoDB. To automatically handle pagination, use aiodynamo.client.Table.scan()

The F class

The aiodynamo.expressions.F class is used when building expressions that refer to fields in your DynamoDB items. It is used to build four different type of expressions: Projection Expression, Update Expression, Filter Expression and Condition.

To refer to a top-level field, simply pass the name of the field to the class constructor. To refer to a nested field, pass the path as multiple arguments. To refer to indices in a list, pass an integer.

For example, to refer to the bar field in the second element of the foo field, use F("foo", 1, "bar").

Projection Expressions

class aiodynamo.expressions.ProjectionExpression

Abstract base class to represent projection expresssions.

Projection expressions are built using the & operator. An instance of aiodynamo.expressions.F is a valid Projection expression too.

For example, to get the field foo and bar, you would use F("foo") & F("bar").

Update Expressions

class aiodynamo.expressions.UpdateExpression

Update expressions are created by calling methods on instances of aiodynamo.expressions.F and combining the return values of those method calls with the & operator.

F.set(value: Any) → aiodynamo.expressions.UpdateExpression

Set a field to a value.

F.set_if_not_exists(value: Any) → aiodynamo.expressions.UpdateExpression

Set a field to a value if the field does not exist in the item yet.

F.change(diff: Union[float, int, decimal.Decimal]) → aiodynamo.expressions.UpdateExpression

Change a numeric field by a given value.

F.append(value: List[Any]) → aiodynamo.expressions.UpdateExpression

Add items to a list field. Note that the value passed in should be a list, not an individual item.

F.remove() → aiodynamo.expressions.UpdateExpression

Remove a field.

F.add(value: Union[float, int, decimal.Decimal, Set[bytes], Set[str], Set[Union[float, int, decimal.Decimal]]]) → aiodynamo.expressions.UpdateExpression

Add a value to a field. Only allowed for top level fields.

For numeric fields add a numeric value. For set fields, this will set the field to the union of the existing set and the set provided.

F.delete(value: Set[Any]) → aiodynamo.expressions.UpdateExpression

Deletes all items in the set provided from the set stored in this field.

Filter Expression and Condition Expressions

class aiodynamo.expressions.Condition

Abstract base class of conditions and filters.

Filter and Condition expressions have the same syntax and they are created by calling methods on instances of aiodynamo.expressions.F and combining the return values of those method calls with the & or | operators. To negate a condition, use the ~ infix operator.

F.does_not_exist() → aiodynamo.expressions.Condition

Checks that a field does not exist.

F.exists() → aiodynamo.expressions.Condition

Checks that a field exist.

F.attribute_type(attribute_type: aiodynamo.types.AttributeType) → aiodynamo.expressions.Condition

Checks the attribute type of a field.

F.begins_with(substr: str) → aiodynamo.expressions.Condition

Checks that a field begins with a substring. The substring must not be empty. Fields matching the substring provided completely are not returned.

F.between(low: Any, high: Any) → aiodynamo.expressions.Condition

Checks that a field is between two given values.

F.contains(value: Union[str, bytes, int, float, decimal.Decimal]) → aiodynamo.expressions.Condition

Checks if a set or list contains a certain value. If a string or bytes object is used as a value, they must not be empty.

F.is_in(values: Sequence[Any]) → aiodynamo.expressions.Condition

Checks if the field is in a sequence of values. Between one and one hundred values may be provided.

F.gt(other: Any) → aiodynamo.expressions.Condition

Checks if a field is greater than a value.

F.gte(other: Any) → aiodynamo.expressions.Condition

Checks if a field is greater than a value.

F.lt(other: Any) → aiodynamo.expressions.Condition

Checks if a field is less than a value.

F.lte(other: Any) → aiodynamo.expressions.Condition

Checks if a field is less or equal than a value.

F.equals(other: Any) → aiodynamo.expressions.Condition

Checks if a field is equal to value.

F.not_equals(other: Any) → aiodynamo.expressions.Condition

Checks if a field is not equal to a value.

F.size() → aiodynamo.expressions.Size

Allows checking for the item size. Does not return a condition, to return a condition, call the appropriate method on the Size class instance returned.

class aiodynamo.expressions.Size(field: 'F')
equals(other: Any) → aiodynamo.expressions.Condition
gt(other: Any) → aiodynamo.expressions.Condition
gte(other: Any) → aiodynamo.expressions.Condition
lt(other: Any) → aiodynamo.expressions.Condition
lte(other: Any) → aiodynamo.expressions.Condition
not_equals(other: Any) → aiodynamo.expressions.Condition

Key conditions

Key conditions are created using the aiodynamo.expressions.HashKey and optionally the aiodynamo.expressions.RangeKey classes.

class aiodynamo.expressions.HashKey(name: str, value: Any)

Used for Key Conditions. To also constrain the Key Condition by the range key, create an instance of RangeKey, call a method on it and combine the HashKey with the return value of that method call using the & operator.


class aiodynamo.expressions.RangeKey(name: str)

Can be used to further constrain a Key Condition. Must be used together with a HashKey instance.

The provided methods behave the same as their counterparts in the F class.

Models

class aiodynamo.models.Throughput(read: 'int', write: 'int')
class aiodynamo.models.KeySchema(hash_key: 'KeySpec', range_key: 'Optional[KeySpec]' = None)
range_key = None
class aiodynamo.models.LocalSecondaryIndex(name: 'str', schema: 'KeySchema', projection: 'Projection')
class aiodynamo.models.GlobalSecondaryIndex(name: 'str', schema: 'KeySchema', projection: 'Projection', throughput: 'Throughput')
class aiodynamo.models.StreamSpecification(enabled: 'bool' = False, view_type: 'StreamViewType' = <StreamViewType.new_and_old_images: 'NEW_AND_OLD_IMAGES'>)
enabled = False
view_type = 'NEW_AND_OLD_IMAGES'
class aiodynamo.models.StreamViewType

An enumeration.

keys_only = 'KEYS_ONLY'
new_and_old_images = 'NEW_AND_OLD_IMAGES'
new_image = 'NEW_IMAGE'
old_image = 'OLD_IMAGE'
class aiodynamo.models.WaitConfig(max_attempts: 'int', retry_delay: 'Timespan')
class aiodynamo.models.ReturnValues

An enumeration.

all_new = 'ALL_NEW'
all_old = 'ALL_OLD'
none = 'NONE'
updated_new = 'UPDATED_NEW'
updated_old = 'UPDATED_OLD'
class aiodynamo.models.Projection(type: 'ProjectionType', attrs: 'Optional[List[str]]' = None)
attrs = None
class aiodynamo.models.ProjectionType

An enumeration.

all = 'ALL'
include = 'INCLUDE'
keys_only = 'KEYS_ONLY'
class aiodynamo.models.BatchGetRequest(keys: 'List[Item]', projection: 'Optional[ProjectionExpression]' = None)
projection = None
class aiodynamo.models.BatchGetResponse(items: 'Dict[TableName, List[Item]]', unprocessed_keys: 'Dict[TableName, List[Item]]')
class aiodynamo.models.BatchWriteRequest(keys_to_delete: 'Optional[List[Item]]' = None, items_to_put: 'Optional[List[Item]]' = None)
items_to_put = None
keys_to_delete = None
class aiodynamo.models.BatchWriteResult(undeleted_keys: 'List[Item]', unput_items: 'List[Item]')