Skip to content

prefect_aws.credentials

Module handling AWS credentials

Classes

AwsCredentials (CredentialsBlock) pydantic-model

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.

Examples:

Load stored AWS credentials: 

from prefect_aws import AwsCredentials

aws_credentials_block = AwsCredentials.load("BLOCK_NAME")

Source code in prefect_aws/credentials.py 
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)

Attributes

aws_access_key_id: str pydantic-field

A specific AWS access key ID.

aws_client_parameters: AwsClientParameters pydantic-field

Extra parameters to initialize the Client.

aws_secret_access_key: SecretStr pydantic-field

A specific AWS secret access key.

aws_session_token: str pydantic-field

The session key for your AWS account. This is only needed when you are using temporary credentials.

profile_name: str pydantic-field

The profile to use when creating your session.

region_name: str pydantic-field

The AWS Region where you want to create new connections.

Classes

Config

Config class for pydantic model.

Source code in prefect_aws/credentials.py 
class Config:
    """Config class for pydantic model."""

    arbitrary_types_allowed = True

Methods

__hash__ special

Return hash(self).

Source code in prefect_aws/credentials.py 
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)
__json_encoder__ special staticmethod

partial(func, args, *keywords) - new function with partial application of the given arguments and keywords.

get_boto3_session

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

Examples:

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 
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, prefect_aws.credentials.ClientType]

The client's service name.

 required

Returns:

Type Description

An authenticated client.

Exceptions:

Type Description
ValueError

if the client is not supported.
Source code in prefect_aws/credentials.py 
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 
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 
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 (Enum)

The supported boto3 clients.

Source code in prefect_aws/credentials.py 
class ClientType(Enum):
    """The supported boto3 clients."""

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

MinIOCredentials (CredentialsBlock) pydantic-model

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".

Examples:

Load stored MinIO credentials: 

from prefect_aws import MinIOCredentials

minio_credentials_block = MinIOCredentials.load("BLOCK_NAME")

Source code in prefect_aws/credentials.py 
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)

Attributes

aws_client_parameters: AwsClientParameters pydantic-field

Extra parameters to initialize the Client.

minio_root_password: SecretStr pydantic-field required

Admin or root password.

minio_root_user: str pydantic-field required

Admin or root user.

region_name: str pydantic-field

The AWS Region where you want to create new connections.

Classes

Config

Config class for pydantic model.

Source code in prefect_aws/credentials.py 
class Config:
    """Config class for pydantic model."""

    arbitrary_types_allowed = True

Methods

__hash__ special

Return hash(self).

Source code in prefect_aws/credentials.py 
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())),
        )
    )
__json_encoder__ special staticmethod

partial(func, args, *keywords) - new function with partial application of the given arguments and keywords.

get_boto3_session

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

Examples:

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 
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, prefect_aws.credentials.ClientType]

The client's service name.

 required

Returns:

Type Description

An authenticated client.

Exceptions:

Type Description
ValueError

if the client is not supported.
Source code in prefect_aws/credentials.py 
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 
def get_s3_client(self) -> S3Client:
    """
    Gets an authenticated S3 client.

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