1. Getting Started

Basic usage

from pydynamodb import connect

cursor = connect(aws_access_key_id="aws_access_key_id",
                aws_secret_access_key="aws_secret_access_key"
                 region_name="region_name").cursor()
cursor.execute('SELECT * FROM "ddb_table_name"')
print(cursor.fetchall())

Cursor iteration

from pydynamodb import connect

cursor = connect(aws_access_key_id="aws_access_key_id",
                aws_secret_access_key="aws_secret_access_key"
                 region_name="region_name").cursor()
cursor.execute('SELECT * FROM "ddb_table_name"')
rows = cursor.fetchall()
for row in rows:
    print(row)

Query with complex parameters

PyDynamoDB is able to serialize the parameters which passed to DDB.

from pydynamodb import connect
cursor = connect(aws_access_key_id="aws_access_key_id",
                aws_secret_access_key="aws_secret_access_key"
                 region_name="region_name").cursor()
cursor.execute("""INSERT INTO "ddb_table_name" VALUE {
                    'partition_key' = ?, 'sort_key' = ?,
                    'col_str' = ?, 'col_num' = ?, 'col_byte' = ?,
                    'col_ss' = ?, 'col_ns' = ?, 'col_bs' = ?,
                    'col_list' = ?, 'col_map' = ?, 'col_nested' = ?
                }""", ["pkey_value", "skey_value", "str", 100, b"ABC", # String, Number, Bytes
                        {"str", "str"}, {100, 100}, {b"A", b"B"}, # String/Numnber/Bytes Set
                        ["str", 100, b"ABC"],  # List
                        {"key1": "val", "key2": "val"}, # Map
                        ["str", 100, {"key1": "val"}] # Nested Structure
                    ])

cursor.execute('SELECT col_nested FROM "ddb_table_name" WHERE partition_key = ?', ["pkey_value"])
print(cursor.fetchall())

Handle Result Set

PyDynamoDB is able to deserialize the result set to Python built-in data types.

cursor.execute("""
     SELECT col_nested FROM "ddb_table_name" WHERE partition_key = ? AND sort_key = ?
""", ["pkey_value", "skey_value"])
assert cursor.fetchone() == (["str", 100, {"key1": "val"}])

Description of Result Set

DDB is a NoSQL database. That means except key schema, the data in each row may have flexible columns or types. PyDynamoDB cannot get a completed result set description before fetching all result data. So you have to use fetch* method to iterate the whole result set, then call cursor.description to get the full columns description.

from pydynamodb import connect

cursor = connect(aws_access_key_id="aws_access_key_id",
                aws_secret_access_key="aws_secret_access_key"
                 region_name="region_name").cursor()
cursor.execute('SELECT * FROM "ddb_table_name"')
print(cursor.fetchall())
print(cursor.description)

SQLAlchemy

Install SQLAlchemy with pip install "SQLAlchemy>=1.0.0, <2.0.0". Supported SQLAlchemy is 1.0.0 or higher and less than 2.0.0.

The connection string has the following format:

dynamodb://{aws_access_key_id}:{aws_secret_access_key}@dynamodb.{region_name}.amazonaws.com:443?verify=false&...

from pydynamodb import sqlalchemy_dynamodb
from sqlalchemy.engine import create_engine
from sqlalchemy.sql.schema import Column, MetaData, Table

conn_str = (
        "dynamodb://{aws_access_key_id}:{aws_secret_access_key}@dynamodb.{region_name}.amazonaws.com:443"
        + "?verify=false"
    )
conn_str = conn_str.format(
        aws_access_key_id=aws_access_key_id,
        aws_secret_access_key=aws_secret_access_key,
        region_name=region_name,
    )
engine = create_engine(conn_str)
with engine.connect() as connection:
    many_rows = Table("many_rows", MetaData(),
                    Column('key_partition', String, nullable=False),
                    Column('key_sort', Integer),
                    Column('col_str', String),
                    Column('col_num', Numeric)
            )
    rows = conn.execute(many_rows.select()).fetchall()
    print(rows)