Skip to content

prefect_aws.credentials

Module handling AWS credentials

AwsCredentials

Bases: CredentialsBlock

Block used to manage authentication with AWS. AWS authentication is handled via the boto3 module. Refer to the boto3 docs for more info about the possible credential configurations.

Example

Load stored AWS credentials: 

from prefect_aws import AwsCredentials

aws_credentials_block = AwsCredentials.load("BLOCK_NAME")

Source code in prefect_aws/credentials.py 
 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
class AwsCredentials(CredentialsBlock):
    """
    Block used to manage authentication with AWS. AWS authentication is
    handled via the `boto3` module. Refer to the
    [boto3 docs](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html)
    for more info about the possible credential configurations.

    Example:
        Load stored AWS credentials:
        ```python
        from prefect_aws import AwsCredentials

        aws_credentials_block = AwsCredentials.load("BLOCK_NAME")
        ```
    """  # noqa E501

    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/d74b16fe84ce626345adf235a47008fea2869a60-225x225.png"  # noqa
    _block_type_name = "AWS Credentials"
    _documentation_url = "https://prefecthq.github.io/prefect-aws/credentials/#prefect_aws.credentials.AwsCredentials"  # noqa

    aws_access_key_id: Optional[str] = Field(
        default=None,
        description="A specific AWS access key ID.",
        title="AWS Access Key ID",
    )
    aws_secret_access_key: Optional[SecretStr] = Field(
        default=None,
        description="A specific AWS secret access key.",
        title="AWS Access Key Secret",
    )
    aws_session_token: Optional[str] = Field(
        default=None,
        description=(
            "The session key for your AWS account. "
            "This is only needed when you are using temporary credentials."
        ),
        title="AWS Session Token",
    )
    profile_name: Optional[str] = Field(
        default=None, description="The profile to use when creating your session."
    )
    region_name: Optional[str] = Field(
        default=None,
        description="The AWS Region where you want to create new connections.",
    )
    aws_client_parameters: AwsClientParameters = Field(
        default_factory=AwsClientParameters,
        description="Extra parameters to initialize the Client.",
        title="AWS Client Parameters",
    )

    class Config:
        """Config class for pydantic model."""

        arbitrary_types_allowed = True

    def __hash__(self):
        field_hashes = (
            hash(self.aws_access_key_id),
            hash(self.aws_secret_access_key),
            hash(self.aws_session_token),
            hash(self.profile_name),
            hash(self.region_name),
            hash(self.aws_client_parameters),
        )
        return hash(field_hashes)

    def get_boto3_session(self) -> boto3.Session:
        """
        Returns an authenticated boto3 session that can be used to create clients
        for AWS services

        Example:
            Create an S3 client from an authorized boto3 session:
            ```python
            aws_credentials = AwsCredentials(
                aws_access_key_id = "access_key_id",
                aws_secret_access_key = "secret_access_key"
                )
            s3_client = aws_credentials.get_boto3_session().client("s3")
            ```
        """

        if self.aws_secret_access_key:
            aws_secret_access_key = self.aws_secret_access_key.get_secret_value()
        else:
            aws_secret_access_key = None

        return boto3.Session(
            aws_access_key_id=self.aws_access_key_id,
            aws_secret_access_key=aws_secret_access_key,
            aws_session_token=self.aws_session_token,
            profile_name=self.profile_name,
            region_name=self.region_name,
        )

    def get_client(self, client_type: Union[str, ClientType]):
        """
        Helper method to dynamically get a client type.

        Args:
            client_type: The client's service name.

        Returns:
            An authenticated client.

        Raises:
            ValueError: if the client is not supported.
        """
        if isinstance(client_type, ClientType):
            client_type = client_type.value

        return _get_client_cached(ctx=self, client_type=client_type)

    def get_s3_client(self) -> S3Client:
        """
        Gets an authenticated S3 client.

        Returns:
            An authenticated S3 client.
        """
        return self.get_client(client_type=ClientType.S3)

    def get_secrets_manager_client(self) -> SecretsManagerClient:
        """
        Gets an authenticated Secrets Manager client.

        Returns:
            An authenticated Secrets Manager client.
        """
        return self.get_client(client_type=ClientType.SECRETS_MANAGER)

Config

Config class for pydantic model.

Source code in prefect_aws/credentials.py 
110
111
112
113
class Config:
    """Config class for pydantic model."""

    arbitrary_types_allowed = True

get_boto3_session

Returns an authenticated boto3 session that can be used to create clients for AWS services

Example

Create an S3 client from an authorized boto3 session: 

aws_credentials = AwsCredentials(
    aws_access_key_id = "access_key_id",
    aws_secret_access_key = "secret_access_key"
    )
s3_client = aws_credentials.get_boto3_session().client("s3")

Source code in prefect_aws/credentials.py 
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
def get_boto3_session(self) -> boto3.Session:
    """
    Returns an authenticated boto3 session that can be used to create clients
    for AWS services

    Example:
        Create an S3 client from an authorized boto3 session:
        ```python
        aws_credentials = AwsCredentials(
            aws_access_key_id = "access_key_id",
            aws_secret_access_key = "secret_access_key"
            )
        s3_client = aws_credentials.get_boto3_session().client("s3")
        ```
    """

    if self.aws_secret_access_key:
        aws_secret_access_key = self.aws_secret_access_key.get_secret_value()
    else:
        aws_secret_access_key = None

    return boto3.Session(
        aws_access_key_id=self.aws_access_key_id,
        aws_secret_access_key=aws_secret_access_key,
        aws_session_token=self.aws_session_token,
        profile_name=self.profile_name,
        region_name=self.region_name,
    )

get_client

Helper method to dynamically get a client type.

Parameters:

Name Type Description Default
client_type Union[str, ClientType]

The client's service name.

 required

Returns:

Type Description

An authenticated client.

Raises:

Type Description
ValueError

if the client is not supported.
Source code in prefect_aws/credentials.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def get_client(self, client_type: Union[str, ClientType]):
    """
    Helper method to dynamically get a client type.

    Args:
        client_type: The client's service name.

    Returns:
        An authenticated client.

    Raises:
        ValueError: if the client is not supported.
    """
    if isinstance(client_type, ClientType):
        client_type = client_type.value

    return _get_client_cached(ctx=self, client_type=client_type)

get_s3_client

Gets an authenticated S3 client.

Returns:

Type Description
S3Client

An authenticated S3 client.
Source code in prefect_aws/credentials.py 
173
174
175
176
177
178
179
180
def get_s3_client(self) -> S3Client:
    """
    Gets an authenticated S3 client.

    Returns:
        An authenticated S3 client.
    """
    return self.get_client(client_type=ClientType.S3)

get_secrets_manager_client

Gets an authenticated Secrets Manager client.

Returns:

Type Description
SecretsManagerClient

An authenticated Secrets Manager client.
Source code in prefect_aws/credentials.py 
182
183
184
185
186
187
188
189
def get_secrets_manager_client(self) -> SecretsManagerClient:
    """
    Gets an authenticated Secrets Manager client.

    Returns:
        An authenticated Secrets Manager client.
    """
    return self.get_client(client_type=ClientType.SECRETS_MANAGER)

ClientType

Bases: Enum

The supported boto3 clients.

Source code in prefect_aws/credentials.py 
25
26
27
28
29
30
31
class ClientType(Enum):
    """The supported boto3 clients."""

    S3 = "s3"
    ECS = "ecs"
    BATCH = "batch"
    SECRETS_MANAGER = "secretsmanager"

MinIOCredentials

Bases: CredentialsBlock

Block used to manage authentication with MinIO. Refer to the MinIO docs for more info about the possible credential configurations.

Attributes:

Name Type Description
minio_root_user str

Admin or root user.
minio_root_password SecretStr

Admin or root password.
region_name Optional[str]

Location of server, e.g. "us-east-1".
Example

Load stored MinIO credentials: 

from prefect_aws import MinIOCredentials

minio_credentials_block = MinIOCredentials.load("BLOCK_NAME")

Source code in prefect_aws/credentials.py 
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
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
class MinIOCredentials(CredentialsBlock):
    """
    Block used to manage authentication with MinIO. Refer to the
    [MinIO docs](https://docs.min.io/docs/minio-server-configuration-guide.html)
    for more info about the possible credential configurations.

    Attributes:
        minio_root_user: Admin or root user.
        minio_root_password: Admin or root password.
        region_name: Location of server, e.g. "us-east-1".

    Example:
        Load stored MinIO credentials:
        ```python
        from prefect_aws import MinIOCredentials

        minio_credentials_block = MinIOCredentials.load("BLOCK_NAME")
        ```
    """  # noqa E501

    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/676cb17bcbdff601f97e0a02ff8bcb480e91ff40-250x250.png"  # noqa
    _block_type_name = "MinIO Credentials"
    _description = (
        "Block used to manage authentication with MinIO. Refer to the MinIO "
        "docs: https://docs.min.io/docs/minio-server-configuration-guide.html "
        "for more info about the possible credential configurations."
    )
    _documentation_url = "https://prefecthq.github.io/prefect-aws/credentials/#prefect_aws.credentials.MinIOCredentials"  # noqa

    minio_root_user: str = Field(default=..., description="Admin or root user.")
    minio_root_password: SecretStr = Field(
        default=..., description="Admin or root password."
    )
    region_name: Optional[str] = Field(
        default=None,
        description="The AWS Region where you want to create new connections.",
    )
    aws_client_parameters: AwsClientParameters = Field(
        default_factory=AwsClientParameters,
        description="Extra parameters to initialize the Client.",
    )

    class Config:
        """Config class for pydantic model."""

        arbitrary_types_allowed = True

    def __hash__(self):
        return hash(
            (
                hash(self.minio_root_user),
                hash(self.minio_root_password),
                hash(self.region_name),
                hash(frozenset(self.aws_client_parameters.dict().items())),
            )
        )

    def get_boto3_session(self) -> boto3.Session:
        """
        Returns an authenticated boto3 session that can be used to create clients
        and perform object operations on MinIO server.

        Example:
            Create an S3 client from an authorized boto3 session

            ```python
            minio_credentials = MinIOCredentials(
                minio_root_user = "minio_root_user",
                minio_root_password = "minio_root_password"
            )
            s3_client = minio_credentials.get_boto3_session().client(
                service="s3",
                endpoint_url="http://localhost:9000"
            )
            ```
        """

        minio_root_password = (
            self.minio_root_password.get_secret_value()
            if self.minio_root_password
            else None
        )

        return boto3.Session(
            aws_access_key_id=self.minio_root_user,
            aws_secret_access_key=minio_root_password,
            region_name=self.region_name,
        )

    def get_client(self, client_type: Union[str, ClientType]):
        """
        Helper method to dynamically get a client type.

        Args:
            client_type: The client's service name.

        Returns:
            An authenticated client.

        Raises:
            ValueError: if the client is not supported.
        """
        if isinstance(client_type, ClientType):
            client_type = client_type.value

        return _get_client_cached(ctx=self, client_type=client_type)

    def get_s3_client(self) -> S3Client:
        """
        Gets an authenticated S3 client.

        Returns:
            An authenticated S3 client.
        """
        return self.get_client(client_type=ClientType.S3)

Config

Config class for pydantic model.

Source code in prefect_aws/credentials.py 
234
235
236
237
class Config:
    """Config class for pydantic model."""

    arbitrary_types_allowed = True

get_boto3_session

Returns an authenticated boto3 session that can be used to create clients and perform object operations on MinIO server.

Example

Create an S3 client from an authorized boto3 session

minio_credentials = MinIOCredentials(
    minio_root_user = "minio_root_user",
    minio_root_password = "minio_root_password"
)
s3_client = minio_credentials.get_boto3_session().client(
    service="s3",
    endpoint_url="http://localhost:9000"
)
Source code in prefect_aws/credentials.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def get_boto3_session(self) -> boto3.Session:
    """
    Returns an authenticated boto3 session that can be used to create clients
    and perform object operations on MinIO server.

    Example:
        Create an S3 client from an authorized boto3 session

        ```python
        minio_credentials = MinIOCredentials(
            minio_root_user = "minio_root_user",
            minio_root_password = "minio_root_password"
        )
        s3_client = minio_credentials.get_boto3_session().client(
            service="s3",
            endpoint_url="http://localhost:9000"
        )
        ```
    """

    minio_root_password = (
        self.minio_root_password.get_secret_value()
        if self.minio_root_password
        else None
    )

    return boto3.Session(
        aws_access_key_id=self.minio_root_user,
        aws_secret_access_key=minio_root_password,
        region_name=self.region_name,
    )

get_client

Helper method to dynamically get a client type.

Parameters:

Name Type Description Default
client_type Union[str, ClientType]

The client's service name.

 required

Returns:

Type Description

An authenticated client.

Raises:

Type Description
ValueError

if the client is not supported.
Source code in prefect_aws/credentials.py 
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def get_client(self, client_type: Union[str, ClientType]):
    """
    Helper method to dynamically get a client type.

    Args:
        client_type: The client's service name.

    Returns:
        An authenticated client.

    Raises:
        ValueError: if the client is not supported.
    """
    if isinstance(client_type, ClientType):
        client_type = client_type.value

    return _get_client_cached(ctx=self, client_type=client_type)

get_s3_client

Gets an authenticated S3 client.

Returns:

Type Description
S3Client

An authenticated S3 client.
Source code in prefect_aws/credentials.py 
299
300
301
302
303
304
305
306
def get_s3_client(self) -> S3Client:
    """
    Gets an authenticated S3 client.

    Returns:
        An authenticated S3 client.
    """
    return self.get_client(client_type=ClientType.S3)
Was this page helpful?