Skip to content

Commit 75f7a01

Browse files
committed
Add archive restructuring
1 parent 7c2d1e3 commit 75f7a01

File tree

2 files changed

+35
-27
lines changed

2 files changed

+35
-27
lines changed

localstack-core/localstack/services/events/archive.py

Lines changed: 32 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import json
22
import logging
33
from datetime import datetime, timezone
4+
from typing import Self
45

56
from botocore.client import BaseClient
67

@@ -42,33 +43,42 @@ class ArchiveService:
4243
rule_name: RuleName
4344
target_id: TargetId
4445

45-
def __init__(
46-
self,
46+
def __init__(self, archive: Archive):
47+
self.archive = archive
48+
self.set_state(ArchiveState.CREATING)
49+
self.set_creation_time()
50+
self.client: BaseClient = self._initialize_client()
51+
self.event_bus_name: EventBusName = extract_event_bus_name(archive.event_source_arn)
52+
self.set_state(ArchiveState.ENABLED)
53+
self.rule_name = f"Events-Archive-{self.archive_name}"
54+
self.target_id = f"Events-Archive-{self.archive_name}"
55+
56+
@classmethod
57+
def create_archive_service(
58+
cls,
4759
archive_name: ArchiveName,
4860
region: str,
4961
account_id: str,
5062
event_source_arn: Arn,
5163
description: ArchiveDescription,
5264
event_pattern: EventPattern,
5365
retention_days: RetentionDays,
54-
):
55-
self.archive = Archive(
56-
archive_name,
57-
region,
58-
account_id,
59-
event_source_arn,
60-
description,
61-
event_pattern,
62-
retention_days,
66+
) -> Self:
67+
return cls(
68+
Archive(
69+
archive_name,
70+
region,
71+
account_id,
72+
event_source_arn,
73+
description,
74+
event_pattern,
75+
retention_days,
76+
)
6377
)
64-
self.set_state(ArchiveState.CREATING)
65-
self.set_creation_time()
66-
self.client: BaseClient = self._initialize_client()
67-
self.event_bus_name: EventBusName = extract_event_bus_name(event_source_arn)
6878

69-
self.rule_name: RuleName = self._create_archive_rule()
70-
self.target_id: TargetId = self._create_archive_target()
71-
self.set_state(ArchiveState.ENABLED)
79+
def register_archive_rule_and_targets(self):
80+
self._create_archive_rule()
81+
self._create_archive_target()
7282

7383
def __getattr__(self, name):
7484
return getattr(self.archive, name)
@@ -133,8 +143,7 @@ def _initialize_client(self) -> BaseClient:
133143

134144
def _create_archive_rule(
135145
self,
136-
) -> RuleName:
137-
rule_name = f"Events-Archive-{self.name}"
146+
):
138147
default_event_pattern = {
139148
"replay-name": [{"exists": False}],
140149
}
@@ -144,25 +153,22 @@ def _create_archive_rule(
144153
else:
145154
updated_event_pattern = default_event_pattern
146155
self.client.put_rule(
147-
Name=rule_name,
156+
Name=self.rule_name,
148157
EventBusName=self.event_bus_name,
149158
EventPattern=json.dumps(updated_event_pattern),
150159
)
151-
return rule_name
152160

153161
def _create_archive_target(
154162
self,
155-
) -> TargetId:
163+
):
156164
"""Creates a target for the archive rule. The target is required for accessing parameters
157165
from the provider during sending of events to the target but it is not invoked
158166
because events are put to the archive directly to not overload the gateway"""
159-
target_id = f"Events-Archive-{self.name}"
160167
self.client.put_targets(
161168
Rule=self.rule_name,
162169
EventBusName=self.event_bus_name,
163-
Targets=[{"Id": target_id, "Arn": self.arn}],
170+
Targets=[{"Id": self.target_id, "Arn": self.arn}],
164171
)
165-
return target_id
166172

167173
def _normalize_datetime(self, dt: datetime) -> datetime:
168174
return dt.replace(second=0, microsecond=0)

localstack-core/localstack/services/events/provider.py

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1559,6 +1559,7 @@ def start_replay(
15591559
re_formatted_event_to_replay = replay_service.re_format_events_from_archive(
15601560
events_to_replay, replay_name
15611561
)
1562+
# TODO should this really be run synchronously within the request?
15621563
self._process_entries(context, re_formatted_event_to_replay)
15631564
replay_service.finish()
15641565

@@ -1733,7 +1734,7 @@ def create_archive_service(
17331734
event_pattern: EventPattern,
17341735
retention_days: RetentionDays,
17351736
) -> ArchiveService:
1736-
archive_service = ArchiveService(
1737+
archive_service = ArchiveService.create_archive_service(
17371738
archive_name,
17381739
region,
17391740
account_id,
@@ -1742,6 +1743,7 @@ def create_archive_service(
17421743
event_pattern,
17431744
retention_days,
17441745
)
1746+
archive_service.register_archive_rule_and_targets()
17451747
self._archive_service_store[archive_service.arn] = archive_service
17461748
return archive_service
17471749

0 commit comments

Comments
 (0)