Skip to content

Commit b1f382c

Browse files
committed
feat(dagster-aws): tag S3 user-agent and add S3-compat docs
Signed-off-by: Gonzalo Peña-Castellanos <goanpeca@gmail.com>
1 parent b65621d commit b1f382c

8 files changed

Lines changed: 202 additions & 3 deletions

File tree

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
---
2+
title: Dagster & S3-compatible storage
3+
sidebar_label: S3-compatible storage
4+
sidebar_position: 10
5+
description: Use Dagster's S3 resource with AWS S3 or with S3-compatible object storage such as Backblaze B2 by setting the endpoint URL.
6+
tags: [dagster-supported, storage]
7+
source: https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/dagster-aws
8+
pypi: https://pypi.org/project/dagster-aws/
9+
sidebar_custom_props:
10+
logo: images/integrations/aws-s3.svg
11+
---
12+
13+
<p>{frontMatter.description}</p>
14+
15+
## Installation
16+
17+
<PackageInstallInstructions packageName="dagster-aws" />
18+
19+
## Configuration
20+
21+
`S3Resource` (from `dagster_aws.s3`) talks to AWS S3 by default. It also targets S3-compatible object storage such as Backblaze B2 by setting the `endpoint_url` field to the provider's S3 endpoint and passing the provider's credentials as `aws_access_key_id` and `aws_secret_access_key`.
22+
23+
For Backblaze B2, the endpoint follows the form `https://s3.<region>.backblazeb2.com` (for example, `https://s3.us-west-004.backblazeb2.com`), and the credentials are your B2 application key ID and application key. Refer to your provider's S3-compatible API documentation for the correct endpoint URL, region string, and credentials.
24+
25+
## Example
26+
27+
The example below mirrors the [AWS S3 example](/integrations/libraries/aws/s3) but points `S3Resource` at Backblaze B2 by setting `endpoint_url`:
28+
29+
<CodeExample path="docs_snippets/docs_snippets/integrations/aws-s3-compatible.py" language="python" />
30+
31+
## Drop-in compatibility with `dagster-aws`
32+
33+
`S3Resource` is the same resource consumed by every `dagster-aws` IO manager, file manager, and compute log manager. That means `S3PickleIOManager`, `S3ComputeLogManager`, `S3FileManager`, and the other resources in `dagster_aws.s3` work against AWS S3 or Backblaze B2 through the same `endpoint_url` configuration. You do not need a provider-specific IO manager or file manager.

docs/docs/integrations/libraries/aws/s3.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ Here is an example of how to use the `S3Resource` in a Dagster job to interact w
2323

2424
<CodeExample path="docs_snippets/docs_snippets/integrations/aws-s3.py" language="python" />
2525

26+
`S3Resource` is not limited to AWS S3. It also works with S3-compatible object storage such as Backblaze B2 by setting the `endpoint_url` field. See [Dagster & S3-compatible storage](/integrations/libraries/aws/s3-compatible) for a worked example.
27+
2628
## About AWS S3
2729

2830
**AWS S3** is an object storage service that offers industry-leading scalability, data availability, security, and performance. This means customers of all sizes and industries can use it to store and protect any amount of data for a range of use cases, such as data lakes, websites, mobile applications, backup and restore, archive, enterprise applications, IoT devices, and big data analytics. Amazon S3 provides easy-to-use management features so you can organize your data and configure finely tuned access controls to meet your specific business, organizational, and compliance requirements.

docs/docs/integrations/libraries/aws/secretsmanager.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
---
22
title: Dagster & AWS Secrets Manager
33
sidebar_label: Secrets Manager
4-
sidebar_position: 10
4+
sidebar_position: 11
55
description: This integration allows you to manage, retrieve, and rotate credentials, API keys, and other secrets using AWS Secrets Manager.
66
tags: [dagster-supported]
77
source: https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/dagster-aws

docs/docs/integrations/libraries/aws/ssm.md

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
---
22
title: Dagster & AWS Systems Parameter Store
33
sidebar_label: Systems Parameter Store
4-
sidebar_position: 11
4+
sidebar_position: 12
55
description: The Dagster AWS Systems Manager (SSM) Parameter Store integration allows you to manage and retrieve parameters stored in AWS SSM Parameter Store directly within your Dagster pipelines. This integration provides resources to fetch parameters by name, tags, or paths, and optionally set them as environment variables for your operations.
66
tags: [dagster-supported]
77
source: https://github.com/dagster-io/dagster/tree/master/python_modules/libraries/dagster-aws
Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
import pandas as pd
2+
from dagster_aws.s3 import S3Resource
3+
4+
import dagster as dg
5+
6+
7+
@dg.asset
8+
def my_s3_compatible_asset(s3: S3Resource):
9+
df = pd.DataFrame({"column1": [1, 2, 3], "column2": ["A", "B", "C"]})
10+
11+
csv_data = df.to_csv(index=False)
12+
13+
s3_compatible_client = s3.get_client()
14+
15+
s3_compatible_client.put_object(
16+
Bucket="my-cool-bucket",
17+
Key="path/to/my_dataframe.csv",
18+
Body=csv_data,
19+
)
20+
21+
22+
# ``S3Resource`` works with S3-compatible storage by setting ``endpoint_url``.
23+
# This example targets Backblaze B2.
24+
defs = dg.Definitions(
25+
assets=[my_s3_compatible_asset],
26+
resources={
27+
"s3": S3Resource(
28+
endpoint_url="https://s3.us-west-004.backblazeb2.com",
29+
aws_access_key_id="<your-application-key-id>",
30+
aws_secret_access_key="<your-application-key>",
31+
)
32+
},
33+
)

python_modules/libraries/dagster-aws/dagster_aws/s3/resources.py

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,13 @@ class ResourceWithS3Configuration(ConfigurableResource):
4848
aws_session_token: str | None = Field(
4949
default=None, description="AWS session token to use when creating the boto3 session."
5050
)
51+
user_agent_extra: str | None = Field(
52+
default=None,
53+
description=(
54+
"Optional value appended to the boto3 ``User-Agent`` via "
55+
"``botocore.config.Config(user_agent_extra=...)``."
56+
),
57+
)
5158

5259

5360
class S3Resource(ResourceWithS3Configuration, IAttachDifferentObjectToOpContext):
@@ -97,6 +104,7 @@ def get_client(self) -> Any:
97104
aws_access_key_id=self.aws_access_key_id,
98105
aws_secret_access_key=self.aws_secret_access_key,
99106
aws_session_token=self.aws_session_token,
107+
user_agent_extra=self.user_agent_extra,
100108
)
101109

102110
def get_object_to_set_on_execution_context(self) -> Any:
@@ -196,6 +204,7 @@ def get_client(self) -> S3FileManager:
196204
aws_access_key_id=self.aws_access_key_id,
197205
aws_secret_access_key=self.aws_secret_access_key,
198206
aws_session_token=self.aws_session_token,
207+
user_agent_extra=self.user_agent_extra,
199208
),
200209
s3_bucket=self.s3_bucket,
201210
s3_base_key=self.s3_prefix,

python_modules/libraries/dagster-aws/dagster_aws/s3/utils.py

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,17 @@
1+
from importlib.metadata import PackageNotFoundError, version
2+
13
import boto3
24
import dagster._check as check
5+
from botocore.config import Config
36
from botocore.handlers import disable_signing
47

58
from dagster_aws.utils import construct_boto_client_retry_config
69

10+
try:
11+
_DAGSTER_AWS_PRODUCT = f"dagster-aws/{version('dagster-aws')}"
12+
except PackageNotFoundError:
13+
_DAGSTER_AWS_PRODUCT = "dagster-aws/dev"
14+
715

816
class S3Callback:
917
def __init__(self, logger, bucket, key, filename, size):
@@ -33,6 +41,7 @@ def construct_s3_client(
3341
aws_access_key_id: str | None = None,
3442
aws_secret_access_key: str | None = None,
3543
aws_session_token: str | None = None,
44+
user_agent_extra: str | None = None,
3645
):
3746
check.int_param(max_attempts, "max_attempts")
3847
check.opt_str_param(region_name, "region_name")
@@ -43,6 +52,14 @@ def construct_s3_client(
4352
check.opt_str_param(profile_name, "aws_access_key_id")
4453
check.opt_str_param(profile_name, "aws_secret_access_key")
4554
check.opt_str_param(profile_name, "aws_session_token")
55+
check.opt_str_param(user_agent_extra, "user_agent_extra")
56+
# Append any caller-supplied value after the product token, space-separated.
57+
combined_user_agent_extra = (
58+
f"{_DAGSTER_AWS_PRODUCT} {user_agent_extra}" if user_agent_extra else _DAGSTER_AWS_PRODUCT
59+
)
60+
config = construct_boto_client_retry_config(max_attempts).merge(
61+
Config(user_agent_extra=combined_user_agent_extra)
62+
)
4663

4764
client_session = boto3.session.Session(profile_name=profile_name)
4865
s3_client = client_session.resource(
@@ -54,7 +71,7 @@ def construct_s3_client(
5471
aws_access_key_id=aws_access_key_id,
5572
aws_secret_access_key=aws_secret_access_key,
5673
aws_session_token=aws_session_token,
57-
config=construct_boto_client_retry_config(max_attempts),
74+
config=config,
5875
).meta.client
5976

6077
if use_unsigned_session:
Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
"""Tests for the ``dagster-aws/<version>`` product token and the optional
2+
``user_agent_extra`` knob on ``S3Resource`` / ``S3FileManagerResource``.
3+
4+
Every boto3 S3 client constructed via these resources carries
5+
``dagster-aws/<version>`` in its ``user_agent_extra``, and works against AWS S3
6+
as well as any S3-compatible backend such as Backblaze B2. Any caller-supplied
7+
``user_agent_extra`` is appended after the product token with a space
8+
separator, and never replaces it.
9+
"""
10+
11+
from importlib.metadata import PackageNotFoundError, version
12+
13+
from dagster_aws.s3 import S3Resource
14+
from dagster_aws.s3.resources import S3FileManagerResource
15+
from dagster_aws.s3.utils import construct_s3_client
16+
17+
try:
18+
DAGSTER_AWS_PRODUCT = f"dagster-aws/{version('dagster-aws')}"
19+
except PackageNotFoundError:
20+
DAGSTER_AWS_PRODUCT = "dagster-aws/dev"
21+
22+
23+
def test_construct_s3_client_default_carries_product_token() -> None:
24+
"""With no caller-supplied extra, only ``dagster-aws/<version>`` is appended."""
25+
client = construct_s3_client(max_attempts=5)
26+
extra = client.meta.config.user_agent_extra or ""
27+
assert extra.strip() == DAGSTER_AWS_PRODUCT
28+
# The base SDK token is preserved, not replaced.
29+
assert "Boto3/" in client.meta.config.user_agent
30+
assert DAGSTER_AWS_PRODUCT in client.meta.config.user_agent
31+
32+
33+
def test_construct_s3_client_appends_user_agent_extra_after_product_token() -> None:
34+
"""Caller-supplied tokens are appended after the product token, not in place of it."""
35+
caller_token = "my-platform/1.2.3"
36+
client = construct_s3_client(max_attempts=5, user_agent_extra=caller_token)
37+
extra = client.meta.config.user_agent_extra or ""
38+
assert DAGSTER_AWS_PRODUCT in extra
39+
assert caller_token in extra
40+
# Order: product token first, caller token after, space-separated.
41+
assert extra == f"{DAGSTER_AWS_PRODUCT} {caller_token}"
42+
assert "Boto3/" in client.meta.config.user_agent
43+
assert caller_token in client.meta.config.user_agent
44+
45+
46+
def test_construct_s3_client_empty_user_agent_extra_treated_as_unset() -> None:
47+
"""An empty string should not change behavior: only the product token is appended."""
48+
client = construct_s3_client(max_attempts=5, user_agent_extra="")
49+
extra = client.meta.config.user_agent_extra or ""
50+
assert extra.strip() == DAGSTER_AWS_PRODUCT
51+
52+
53+
def test_construct_s3_client_preserves_retry_config() -> None:
54+
"""Merging the product token must not drop the retry config already in the chain."""
55+
client = construct_s3_client(max_attempts=7, user_agent_extra="my-platform/1.2.3")
56+
retries = client.meta.config.retries
57+
# botocore records the initial call plus retries as ``total_max_attempts``.
58+
assert retries["total_max_attempts"] == 8
59+
assert retries["mode"] == "standard"
60+
extra = client.meta.config.user_agent_extra or ""
61+
assert DAGSTER_AWS_PRODUCT in extra
62+
assert "my-platform/1.2.3" in extra
63+
64+
65+
def test_s3_resource_default_carries_product_token() -> None:
66+
client = S3Resource().get_client()
67+
extra = client.meta.config.user_agent_extra or ""
68+
assert extra.strip() == DAGSTER_AWS_PRODUCT
69+
70+
71+
def test_s3_resource_threads_user_agent_extra_to_client() -> None:
72+
client = S3Resource(user_agent_extra="my-platform/1.2.3").get_client()
73+
extra = client.meta.config.user_agent_extra or ""
74+
assert extra == f"{DAGSTER_AWS_PRODUCT} my-platform/1.2.3"
75+
76+
77+
def test_s3_resource_user_agent_extra_with_b2_endpoint() -> None:
78+
"""End-to-end shape a Backblaze B2 user would write."""
79+
resource = S3Resource(
80+
endpoint_url="https://s3.us-west-004.backblazeb2.com",
81+
aws_access_key_id="00400",
82+
aws_secret_access_key="K004",
83+
user_agent_extra="my-platform/1.2.3",
84+
)
85+
client = resource.get_client()
86+
assert client.meta.endpoint_url == "https://s3.us-west-004.backblazeb2.com"
87+
extra = client.meta.config.user_agent_extra or ""
88+
assert DAGSTER_AWS_PRODUCT in extra
89+
assert "my-platform/1.2.3" in extra
90+
91+
92+
def test_s3_file_manager_resource_threads_user_agent_extra() -> None:
93+
file_manager = S3FileManagerResource(
94+
s3_bucket="my-bucket",
95+
user_agent_extra="my-platform/1.2.3",
96+
).get_client()
97+
# ``S3FileManager`` stores the wrapped boto3 client as ``_s3_session``.
98+
extra = file_manager._s3_session.meta.config.user_agent_extra or "" # noqa: SLF001
99+
assert extra == f"{DAGSTER_AWS_PRODUCT} my-platform/1.2.3"
100+
101+
102+
def test_s3_file_manager_resource_default_carries_product_token() -> None:
103+
file_manager = S3FileManagerResource(s3_bucket="my-bucket").get_client()
104+
extra = file_manager._s3_session.meta.config.user_agent_extra or "" # noqa: SLF001
105+
assert extra.strip() == DAGSTER_AWS_PRODUCT

0 commit comments

Comments
 (0)