Skip to content

add small fixes/improvements to Firehose.CreateDeliveryStream #12656

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

add small fixes/improvements to Firehose.CreateDeliveryStream #12656

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1c3b0ef
add small fixes/improvements to Firehose.CreateDeliveryStream
pinzon May 22, 2025
ca8214f
fix linting
pinzon May 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 18 additions & 7 deletions localstack-core/localstack/services/firehose/provider.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@
RedshiftDestinationConfiguration,
RedshiftDestinationDescription,
RedshiftDestinationUpdate,
ResourceInUseException,
ResourceNotFoundException,
S3DestinationConfiguration,
S3DestinationDescription,
Expand Down Expand Up @@ -282,6 +283,18 @@ def create_delivery_stream(
) -> CreateDeliveryStreamOutput:
# TODO add support for database_source_configuration and direct_put_source_configuration
store = self.get_store(context.account_id, context.region)
delivery_stream_type = delivery_stream_type or DeliveryStreamType.DirectPut

delivery_stream_arn = firehose_stream_arn(
stream_name=delivery_stream_name,
account_id=context.account_id,
region_name=context.region,
)

if delivery_stream_name in store.delivery_streams.keys():
raise ResourceInUseException(
f"Firehose {delivery_stream_name} under accountId {context.account_id} already exists"
)

destinations: DestinationDescriptionList = []
if elasticsearch_destination_configuration:
Expand Down Expand Up @@ -344,11 +357,7 @@ def create_delivery_stream(

stream = DeliveryStreamDescription(
DeliveryStreamName=delivery_stream_name,
DeliveryStreamARN=firehose_stream_arn(
stream_name=delivery_stream_name,
account_id=context.account_id,
region_name=context.region,
),
DeliveryStreamARN=delivery_stream_arn,
DeliveryStreamStatus=DeliveryStreamStatus.ACTIVE,
DeliveryStreamType=delivery_stream_type,
HasMoreDestinations=False,
Expand All @@ -358,8 +367,6 @@ def create_delivery_stream(
Source=convert_source_config_to_desc(kinesis_stream_source_configuration),
)
delivery_stream_arn = stream["DeliveryStreamARN"]
store.TAGS.tag_resource(delivery_stream_arn, tags)
store.delivery_streams[delivery_stream_name] = stream

if delivery_stream_type == DeliveryStreamType.KinesisStreamAsSource:
if not kinesis_stream_source_configuration:
Expand Down Expand Up @@ -396,6 +403,10 @@ def _startup():
stream["DeliveryStreamStatus"] = DeliveryStreamStatus.CREATING_FAILED

run_for_max_seconds(25, _startup)

store.TAGS.tag_resource(delivery_stream_arn, tags)
store.delivery_streams[delivery_stream_name] = stream

return CreateDeliveryStreamOutput(DeliveryStreamARN=stream["DeliveryStreamARN"])

def delete_delivery_stream(
Expand Down
Loading