Skip to content
This repository was archived by the owner on Mar 23, 2026. It is now read-only.

Commit 13584a5

Browse files
giogranogiograno
committed
running test
1 parent 8cda50b commit 13584a5

5 files changed

Lines changed: 81 additions & 41 deletions

File tree

‎localstack-core/localstack/services/dynamodb/models.py‎

Lines changed: 20 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,14 @@
33

44
from localstack.aws.api.dynamodb import (
55
AttributeMap,
6+
BackupDetails,
7+
ContinuousBackupsDescription,
8+
GlobalTableDescription,
69
Key,
710
RegionName,
811
ReplicaDescription,
912
StreamViewType,
13+
TableDescription,
1014
TableName,
1115
TimeToLiveSpecification,
1216
)
@@ -91,9 +95,20 @@ class TableRecords(TypedDict):
9195
RecordsMap = dict[TableName, TableRecords]
9296

9397

98+
class TableProperties(TypedDict, total=False):
99+
ContinuousBackupsDescription: ContinuousBackupsDescription
100+
101+
102+
@dataclasses.dataclass
103+
class Backup:
104+
details: BackupDetails
105+
backup_file: str
106+
table_name: str
107+
108+
94109
class DynamoDBStore(BaseStore):
95110
# maps global table names to configurations (for the legacy v.2017 tables)
96-
GLOBAL_TABLES: dict[str, dict] = CrossRegionAttribute(default=dict)
111+
GLOBAL_TABLES: dict[str, GlobalTableDescription] = CrossRegionAttribute(default=dict)
97112

98113
# Maps table name to the region they exist in on DDBLocal (for v.2019 global tables)
99114
TABLE_REGION: dict[TableName, RegionName] = CrossRegionAttribute(default=dict)
@@ -104,19 +119,19 @@ class DynamoDBStore(BaseStore):
104119
)
105120

106121
# cache table taggings - maps table ARN to tags dict
107-
TABLE_TAGS: dict[str, dict] = CrossRegionAttribute(default=dict)
122+
TABLE_TAGS: dict[str, dict[str, str]] = CrossRegionAttribute(default=dict)
108123

109124
# maps table names to cached table definitions
110-
table_definitions: dict[str, dict] = LocalAttribute(default=dict)
125+
table_definitions: dict[str, TableDescription] = LocalAttribute(default=dict)
111126

112127
# maps table names to additional table properties that are not stored upstream (e.g., ReplicaUpdates)
113-
table_properties: dict[str, dict] = LocalAttribute(default=dict)
128+
table_properties: dict[str, TableProperties] = LocalAttribute(default=dict)
114129

115130
# maps table names to TTL specifications
116131
ttl_specifications: dict[str, TimeToLiveSpecification] = LocalAttribute(default=dict)
117132

118133
# maps backups
119-
backups: dict[str, dict] = LocalAttribute(default=dict)
134+
backups: dict[str, Backup] = LocalAttribute(default=dict)
120135

121136

122137
dynamodb_stores = AccountRegionBundle("dynamodb", DynamoDBStore)

‎localstack-core/localstack/services/dynamodb/v2/provider.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1625,7 +1625,7 @@ def is_index_query_valid(account_id: str, region_name: str, query_data: dict) ->
16251625
return True
16261626

16271627

1628-
def kinesis_stream_exists(stream_arn):
1628+
def kinesis_stream_exists(stream_arn: str) -> bool:
16291629
account_id = extract_account_id_from_arn(stream_arn)
16301630
region_name = extract_region_from_arn(stream_arn)
16311631

‎localstack-core/localstack/services/dynamodbstreams/dynamodbstreams_api.py‎

Lines changed: 38 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -6,10 +6,19 @@
66

77
from localstack import config
88
from localstack.aws.api import RequestContext
9-
from localstack.aws.api.dynamodbstreams import StreamStatus, StreamViewType, TableName
9+
from localstack.aws.api.dynamodbstreams import (
10+
StreamDescription,
11+
StreamStatus,
12+
StreamViewType,
13+
TableName,
14+
)
1015
from localstack.aws.connect import connect_to
1116
from localstack.services.dynamodb.v2.provider import DynamoDBProvider
12-
from localstack.services.dynamodbstreams.models import DynamoDbStreamsStore, dynamodbstreams_stores
17+
from localstack.services.dynamodbstreams.models import (
18+
DynamoDbStreamsStore,
19+
Stream,
20+
dynamodbstreams_stores,
21+
)
1322
from localstack.utils.aws import arns, resources
1423
from localstack.utils.common import now_utc
1524
from localstack.utils.threads import FuncThread
@@ -65,28 +74,32 @@ def add_dynamodb_stream(
6574
stream_name=stream_name,
6675
)
6776
latest_stream_label = latest_stream_label or "latest"
68-
stream = {
69-
"StreamArn": arns.dynamodb_stream_arn(
70-
table_name=table_name,
71-
latest_stream_label=latest_stream_label,
72-
account_id=account_id,
73-
region_name=region_name,
74-
),
75-
"TableName": table_name,
76-
"StreamLabel": latest_stream_label,
77-
"StreamStatus": StreamStatus.ENABLING,
78-
"KeySchema": [],
79-
"Shards": [],
80-
"StreamViewType": view_type,
81-
"shards_id_map": {},
82-
}
83-
store.ddb_streams[table_name] = stream
84-
85-
86-
def get_stream_for_table(account_id: str, region_name: str, table_arn: str) -> dict:
77+
stream_arn = arns.dynamodb_stream_arn(
78+
table_name=table_name,
79+
latest_stream_label=latest_stream_label,
80+
account_id=account_id,
81+
region_name=region_name,
82+
)
83+
stream = StreamDescription(
84+
TableName=table_name,
85+
StreamArn=stream_arn,
86+
StreamLabel=latest_stream_label,
87+
StreamStatus=StreamStatus.ENABLING,
88+
KeySchema=[],
89+
Shards=[],
90+
StreamViewType=view_type,
91+
)
92+
store.ddb_streams[table_name] = Stream(StreamDescription=stream)
93+
94+
95+
def get_stream_for_table(
96+
account_id: str, region_name: str, table_arn: str
97+
) -> StreamDescription | None:
8798
store = get_dynamodbstreams_store(account_id, region_name)
8899
table_name = table_name_from_stream_arn(table_arn)
89-
return store.ddb_streams.get(table_name)
100+
if stream := store.ddb_streams.get(table_name):
101+
return stream.StreamDescription
102+
return None
90103

91104

92105
def _process_forwarded_records(
@@ -206,11 +219,11 @@ def kinesis_shard_id(dynamodbstream_shard_id: str) -> str:
206219
return f"{shard_params[0]}-{shard_params[-1]}"
207220

208221

209-
def get_shard_id(stream: dict, kinesis_shard_id: str) -> str:
210-
ddb_stream_shard_id = stream.get("shards_id_map", {}).get(kinesis_shard_id)
222+
def get_shard_id(stream: Stream, kinesis_shard_id: str) -> str:
223+
ddb_stream_shard_id = stream.shards_id_map.get(kinesis_shard_id)
211224
if not ddb_stream_shard_id:
212225
ddb_stream_shard_id = shard_id(kinesis_shard_id)
213-
stream["shards_id_map"][kinesis_shard_id] = ddb_stream_shard_id
226+
stream.shards_id_map[kinesis_shard_id] = ddb_stream_shard_id
214227

215228
return ddb_stream_shard_id
216229

‎localstack-core/localstack/services/dynamodbstreams/models.py‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,19 @@
1+
import dataclasses
2+
3+
from localstack.aws.api.dynamodbstreams import StreamDescription
14
from localstack.services.stores import AccountRegionBundle, BaseStore, LocalAttribute
25

36

7+
@dataclasses.dataclass
8+
class Stream:
9+
"""Wrapper for the API stub and additional information about a store"""
10+
11+
StreamDescription: StreamDescription
12+
shards_id_map: dict[str, str] = dataclasses.field(default_factory=dict)
13+
14+
415
class DynamoDbStreamsStore(BaseStore):
5-
# maps table names to DynamoDB stream descriptions
6-
ddb_streams: dict[str, dict] = LocalAttribute(default=dict)
16+
ddb_streams: dict[str, Stream] = LocalAttribute(default=dict)
717

818

919
dynamodbstreams_stores = AccountRegionBundle("dynamodbstreams", DynamoDbStreamsStore)

‎localstack-core/localstack/services/dynamodbstreams/provider.py‎

Lines changed: 10 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
ShardIteratorType,
2121
Stream,
2222
StreamArn,
23-
StreamDescription,
2423
StreamStatus,
2524
TableName,
2625
)
@@ -77,20 +76,21 @@ def describe_stream(
7776
store = get_dynamodbstreams_store(context.account_id, og_region)
7877
kinesis = get_kinesis_client(account_id=context.account_id, region_name=og_region)
7978
for stream in store.ddb_streams.values():
79+
stream_description = stream.StreamDescription
8080
_stream_arn = stream_arn
8181
if context.region != og_region:
8282
_stream_arn = change_region_in_ddb_stream_arn(_stream_arn, og_region)
83-
if stream["StreamArn"] == _stream_arn:
83+
if stream_description["StreamArn"] == _stream_arn:
8484
# get stream details
8585
dynamodb = connect_to(
8686
aws_access_key_id=context.account_id, region_name=og_region
8787
).dynamodb
88-
table_name = table_name_from_stream_arn(stream["StreamArn"])
88+
table_name = table_name_from_stream_arn(stream_description["StreamArn"])
8989
stream_name = get_kinesis_stream_name(table_name)
9090
stream_details = kinesis.describe_stream(StreamName=stream_name)
9191
table_details = dynamodb.describe_table(TableName=table_name)
92-
stream["KeySchema"] = table_details["Table"]["KeySchema"]
93-
stream["StreamStatus"] = STREAM_STATUS_MAP.get(
92+
stream_description["KeySchema"] = table_details["Table"]["KeySchema"]
93+
stream_description["StreamStatus"] = STREAM_STATUS_MAP.get(
9494
stream_details["StreamDescription"]["StreamStatus"]
9595
)
9696

@@ -110,8 +110,7 @@ def describe_stream(
110110
# slicing the resulting shards after the exclusive_start_shard_id parameters
111111
stream_shards = stream_shards[start_index + 1 :]
112112

113-
stream["Shards"] = stream_shards
114-
stream_description = select_from_typed_dict(StreamDescription, stream)
113+
stream_description["Shards"] = stream_shards
115114
stream_description["StreamArn"] = _stream_arn
116115
return DescribeStreamOutput(StreamDescription=stream_description)
117116

@@ -190,7 +189,10 @@ def list_streams(
190189
) -> ListStreamsOutput:
191190
og_region = get_original_region(context=context, table_name=table_name)
192191
store = get_dynamodbstreams_store(context.account_id, og_region)
193-
result = [select_from_typed_dict(Stream, res) for res in store.ddb_streams.values()]
192+
result = [
193+
select_from_typed_dict(Stream, _s.StreamDescription)
194+
for _s in store.ddb_streams.values()
195+
]
194196
if table_name:
195197
result: list[Stream] = [res for res in result if res["TableName"] == table_name]
196198
# If this is a stream from a table replica, we need to change the region in the stream ARN, as LocalStack

0 commit comments

Comments
 (0)