跳到主内容

Dynamodb

DynamoDBKVStore #

继承自: BaseKVStore

DynamoDB 键值存储。在 DynamoDB 表中存储键值对。DynamoDB 表必须同时包含哈希键 (hash key) 和范围键 (range key)，且其类型必须为字符串。

您可以通过设置 DYNAMODB_URL 环境变量来指定 DynamoDB 的自定义 URL。如果您正在使用 DynamoDB 的本地实例进行开发或测试，这将非常有用。如果未设置 DYNAMODB_URL，应用程序将使用默认的 AWS DynamoDB 服务。

参数

名称 类型 描述 默认值
table Any

DynamoDB 表服务资源

 必需
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
class DynamoDBKVStore(BaseKVStore):
    """
    DynamoDB Key-Value store.
    Stores key-value pairs in a DynamoDB Table.
    The DynamoDB Table must have both a hash key and a range key,
        and their types must be string.

    You can specify a custom URL for DynamoDB by setting the `DYNAMODB_URL`
    environment variable. This is useful if you're using a local instance of
    DynamoDB for development or testing. If `DYNAMODB_URL` is not set, the
    application will use the default AWS DynamoDB service.

    Args:
        table (Any): DynamoDB Table Service Resource

    """

    def __init__(self, table: Any):
        """Init a DynamoDBKVStore."""
        self._table = table
        self._boto3_key = Key
        self._key_hash, self._key_range = parse_schema(table)

    @classmethod
    def from_table_name(cls, table_name: str) -> DynamoDBKVStore:
        """
        Load a DynamoDBKVStore from a DynamoDB table name.

        Args:
            table_name (str): DynamoDB table name

        """
        # Get the DynamoDB URL from environment variable
        dynamodb_url = os.getenv("DYNAMODB_URL")

        # Create a session
        session = boto3.Session()

        # If the DynamoDB URL is set, use it as the endpoint URL
        if dynamodb_url:
            ddb = session.resource("dynamodb", endpoint_url=dynamodb_url)
        else:
            # Otherwise, let boto3 use its default configuration
            ddb = session.resource("dynamodb")
        return cls(table=ddb.Table(table_name))

    def put(self, key: str, val: dict, collection: str = DEFAULT_COLLECTION) -> None:
        """
        Put a key-value pair into the store.

        Args:
            key (str): key
            val (dict): value
            collection (str): collection name

        """
        item = {k: convert_float_to_decimal(v) for k, v in val.items()}
        item[self._key_hash] = collection
        item[self._key_range] = key
        self._table.put_item(Item=item)

    async def aput(
        self, key: str, val: dict, collection: str = DEFAULT_COLLECTION
    ) -> None:
        """
        Put a key-value pair into the store.

        Args:
            key (str): key
            val (dict): value
            collection (str): collection name

        """
        raise NotImplementedError

    def get(self, key: str, collection: str = DEFAULT_COLLECTION) -> dict | None:
        """
        Get a value from the store.

        Args:
            key (str): key
            collection (str): collection name

        """
        resp = self._table.get_item(
            Key={self._key_hash: collection, self._key_range: key}
        )
        if (item := resp.get("Item")) is None:
            return None
        else:
            return {
                k: convert_decimal_to_int_or_float(v)
                for k, v in item.items()
                if k not in {self._key_hash, self._key_range}
            }

    async def aget(self, key: str, collection: str = DEFAULT_COLLECTION) -> dict | None:
        """
        Get a value from the store.

        Args:
            key (str): key
            collection (str): collection name

        """
        raise NotImplementedError

    def get_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
        """
        Get all values from the store.

        Args:
            collection (str): collection name

        """
        result = {}
        last_evaluated_key = None
        is_first = True
        while last_evaluated_key is not None or is_first:
            if is_first:
                is_first = False
            option = {
                "KeyConditionExpression": self._boto3_key(self._key_hash).eq(collection)
            }
            if last_evaluated_key is not None:
                option["ExclusiveStartKey"] = last_evaluated_key
            resp = self._table.query(**option)
            for item in resp.get("Items", []):
                item.pop(self._key_hash)
                key = item.pop(self._key_range)
                result[key] = {
                    k: convert_decimal_to_int_or_float(v) for k, v in item.items()
                }
            last_evaluated_key = resp.get("LastEvaluatedKey")
        return result

    async def aget_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
        """
        Get all values from the store.

        Args:
            collection (str): collection name

        """
        raise NotImplementedError

    def delete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
        """
        Delete a value from the store.

        Args:
            key (str): key
            collection (str): collection name

        """
        resp = self._table.delete_item(
            Key={self._key_hash: collection, self._key_range: key},
            ReturnValues="ALL_OLD",
        )

        if (item := resp.get("Attributes")) is None:
            return False
        else:
            return len(item) > 0

    async def adelete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
        """
        Delete a value from the store.

        Args:
            key (str): key
            collection (str): collection name

        """
        raise NotImplementedError

from_table_name classmethod #

from_table_name(table_name: str) -> DynamoDBKVStore

从 DynamoDB 表名称加载 DynamoDBKVStore。

参数

名称 类型 描述 默认值
table_name str

DynamoDB 表名称

 必需
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
@classmethod
def from_table_name(cls, table_name: str) -> DynamoDBKVStore:
    """
    Load a DynamoDBKVStore from a DynamoDB table name.

    Args:
        table_name (str): DynamoDB table name

    """
    # Get the DynamoDB URL from environment variable
    dynamodb_url = os.getenv("DYNAMODB_URL")

    # Create a session
    session = boto3.Session()

    # If the DynamoDB URL is set, use it as the endpoint URL
    if dynamodb_url:
        ddb = session.resource("dynamodb", endpoint_url=dynamodb_url)
    else:
        # Otherwise, let boto3 use its default configuration
        ddb = session.resource("dynamodb")
    return cls(table=ddb.Table(table_name))

put #

put(key: str, val: dict, collection: str = DEFAULT_COLLECTION) -> None

将键值对放入存储。

参数

名称 类型 描述 默认值
key str

key

 必需
val dict

值

 必需
collection str

集合名称

 DEFAULT_COLLECTION
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def put(self, key: str, val: dict, collection: str = DEFAULT_COLLECTION) -> None:
    """
    Put a key-value pair into the store.

    Args:
        key (str): key
        val (dict): value
        collection (str): collection name

    """
    item = {k: convert_float_to_decimal(v) for k, v in val.items()}
    item[self._key_hash] = collection
    item[self._key_range] = key
    self._table.put_item(Item=item)

aput async #

aput(key: str, val: dict, collection: str = DEFAULT_COLLECTION) -> None

将键值对放入存储。

参数

名称 类型 描述 默认值
key str

key

 必需
val dict

值

 必需
collection str

集合名称

 DEFAULT_COLLECTION
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
async def aput(
    self, key: str, val: dict, collection: str = DEFAULT_COLLECTION
) -> None:
    """
    Put a key-value pair into the store.

    Args:
        key (str): key
        val (dict): value
        collection (str): collection name

    """
    raise NotImplementedError

get #

get(key: str, collection: str = DEFAULT_COLLECTION) -> dict | None

从存储获取值。

参数

名称 类型 描述 默认值
key str

key

 必需
collection str

集合名称

 DEFAULT_COLLECTION
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def get(self, key: str, collection: str = DEFAULT_COLLECTION) -> dict | None:
    """
    Get a value from the store.

    Args:
        key (str): key
        collection (str): collection name

    """
    resp = self._table.get_item(
        Key={self._key_hash: collection, self._key_range: key}
    )
    if (item := resp.get("Item")) is None:
        return None
    else:
        return {
            k: convert_decimal_to_int_or_float(v)
            for k, v in item.items()
            if k not in {self._key_hash, self._key_range}
        }

aget async #

aget(key: str, collection: str = DEFAULT_COLLECTION) -> dict | None

从存储获取值。

参数

名称 类型 描述 默认值
key str

key

 必需
collection str

集合名称

 DEFAULT_COLLECTION
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
150
151
152
153
154
155
156
157
158
159
async def aget(self, key: str, collection: str = DEFAULT_COLLECTION) -> dict | None:
    """
    Get a value from the store.

    Args:
        key (str): key
        collection (str): collection name

    """
    raise NotImplementedError

get_all #

get_all(collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]

从存储获取所有值。

参数

名称 类型 描述 默认值
collection str

集合名称

 DEFAULT_COLLECTION
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def get_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
    """
    Get all values from the store.

    Args:
        collection (str): collection name

    """
    result = {}
    last_evaluated_key = None
    is_first = True
    while last_evaluated_key is not None or is_first:
        if is_first:
            is_first = False
        option = {
            "KeyConditionExpression": self._boto3_key(self._key_hash).eq(collection)
        }
        if last_evaluated_key is not None:
            option["ExclusiveStartKey"] = last_evaluated_key
        resp = self._table.query(**option)
        for item in resp.get("Items", []):
            item.pop(self._key_hash)
            key = item.pop(self._key_range)
            result[key] = {
                k: convert_decimal_to_int_or_float(v) for k, v in item.items()
            }
        last_evaluated_key = resp.get("LastEvaluatedKey")
    return result

aget_all async #

aget_all(collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]

从存储获取所有值。

参数

名称 类型 描述 默认值
collection str

集合名称

 DEFAULT_COLLECTION
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
190
191
192
193
194
195
196
197
198
async def aget_all(self, collection: str = DEFAULT_COLLECTION) -> Dict[str, dict]:
    """
    Get all values from the store.

    Args:
        collection (str): collection name

    """
    raise NotImplementedError

delete #

delete(key: str, collection: str = DEFAULT_COLLECTION) -> bool

从存储删除值。

参数

名称 类型 描述 默认值
key str

key

 必需
collection str

集合名称

 DEFAULT_COLLECTION
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def delete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
    """
    Delete a value from the store.

    Args:
        key (str): key
        collection (str): collection name

    """
    resp = self._table.delete_item(
        Key={self._key_hash: collection, self._key_range: key},
        ReturnValues="ALL_OLD",
    )

    if (item := resp.get("Attributes")) is None:
        return False
    else:
        return len(item) > 0

adelete async #

adelete(key: str, collection: str = DEFAULT_COLLECTION) -> bool

从存储删除值。

参数

名称 类型 描述 默认值
key str

key

 必需
collection str

集合名称

 DEFAULT_COLLECTION
源代码位于 llama-index-integrations/storage/kvstore/llama-index-storage-kvstore-dynamodb/llama_index/storage/kvstore/dynamodb/base.py 
219
220
221
222
223
224
225
226
227
228
async def adelete(self, key: str, collection: str = DEFAULT_COLLECTION) -> bool:
    """
    Delete a value from the store.

    Args:
        key (str): key
        collection (str): collection name

    """
    raise NotImplementedError