Skip to content

prefect_aws.deployments.steps

Prefect deployment steps for code storage and retrieval in S3 and S3 compatible services.

PullFromS3Output

Bases: TypedDict

The output of the pull_from_s3 step.

Source code in prefect_aws/deployments/steps.py
31
32
33
34
35
36
37
38
class PullFromS3Output(TypedDict):
    """
    The output of the `pull_from_s3` step.
    """

    bucket: str
    folder: str
    directory: str

PullProjectFromS3Output

Bases: PullFromS3Output

Deprecated. Use PullFromS3Output instead..

Source code in prefect_aws/deployments/steps.py
41
42
43
@deprecated_callable(start_date="Jun 2023", help="Use `PullFromS3Output` instead.")
class PullProjectFromS3Output(PullFromS3Output):
    """Deprecated. Use `PullFromS3Output` instead.."""

PushProjectToS3Output

Bases: PushToS3Output

Deprecated. Use PushToS3Output instead.

Source code in prefect_aws/deployments/steps.py
26
27
28
@deprecated_callable(start_date="Jun 2023", help="Use `PushToS3Output` instead.")
class PushProjectToS3Output(PushToS3Output):
    """Deprecated. Use `PushToS3Output` instead."""

PushToS3Output

Bases: TypedDict

The output of the push_to_s3 step.

Source code in prefect_aws/deployments/steps.py
17
18
19
20
21
22
23
class PushToS3Output(TypedDict):
    """
    The output of the `push_to_s3` step.
    """

    bucket: str
    folder: str

pull_from_s3(bucket, folder, credentials=None, client_parameters=None)

Pulls the contents of an S3 bucket folder to the current working directory.

Parameters:

Name Type Description Default
bucket str

The name of the S3 bucket where files are stored.

required
folder str

The folder in the S3 bucket where files are stored.

required
credentials Optional[Dict]

A dictionary of AWS credentials (aws_access_key_id, aws_secret_access_key, aws_session_token) or MinIO credentials (minio_root_user, minio_root_password).

None
client_parameters Optional[Dict]

A dictionary of additional parameters to pass to the boto3 client.

None

Returns:

Type Description
PullFromS3Output

A dictionary containing the bucket, folder, and local directory where files were downloaded.

Examples:

Pull files from S3 using the default credentials and client parameters:

pull:
    - prefect_aws.deployments.steps.pull_from_s3:
        requires: prefect-aws
        bucket: my-bucket
        folder: my-project

Pull files from S3 using credentials stored in a block:

pull:
    - prefect_aws.deployments.steps.pull_from_s3:
        requires: prefect-aws
        bucket: my-bucket
        folder: my-project
        credentials: "{{ prefect.blocks.aws-credentials.dev-credentials }}"
Source code in prefect_aws/deployments/steps.py
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
190
191
192
193
194
195
196
197
198
def pull_from_s3(
    bucket: str,
    folder: str,
    credentials: Optional[Dict] = None,
    client_parameters: Optional[Dict] = None,
) -> PullFromS3Output:
    """
    Pulls the contents of an S3 bucket folder to the current working directory.

    Args:
        bucket: The name of the S3 bucket where files are stored.
        folder: The folder in the S3 bucket where files are stored.
        credentials: A dictionary of AWS credentials (aws_access_key_id,
            aws_secret_access_key, aws_session_token) or MinIO credentials
            (minio_root_user, minio_root_password).
        client_parameters: A dictionary of additional parameters to pass to the
            boto3 client.

    Returns:
        A dictionary containing the bucket, folder, and local directory where
            files were downloaded.

    Examples:
        Pull files from S3 using the default credentials and client parameters:
        ```yaml
        pull:
            - prefect_aws.deployments.steps.pull_from_s3:
                requires: prefect-aws
                bucket: my-bucket
                folder: my-project
        ```

        Pull files from S3 using credentials stored in a block:
        ```yaml
        pull:
            - prefect_aws.deployments.steps.pull_from_s3:
                requires: prefect-aws
                bucket: my-bucket
                folder: my-project
                credentials: "{{ prefect.blocks.aws-credentials.dev-credentials }}"
        ```
    """
    s3 = get_s3_client(credentials=credentials, client_parameters=client_parameters)

    local_path = Path.cwd()

    paginator = s3.get_paginator("list_objects_v2")
    for result in paginator.paginate(Bucket=bucket, Prefix=folder):
        for obj in result.get("Contents", []):
            remote_key = obj["Key"]

            if remote_key[-1] == "/":
                # object is a folder and will be created if it contains any objects
                continue

            target = PurePosixPath(
                local_path
                / relative_path_to_current_platform(remote_key).relative_to(folder)
            )
            Path.mkdir(Path(target.parent), parents=True, exist_ok=True)
            s3.download_file(bucket, remote_key, str(target))

    return {
        "bucket": bucket,
        "folder": folder,
        "directory": str(local_path),
    }

pull_project_from_s3(*args, **kwargs)

Deprecated. Use pull_from_s3 instead.

Source code in prefect_aws/deployments/steps.py
126
127
128
129
@deprecated_callable(start_date="Jun 2023", help="Use `pull_from_s3` instead.")
def pull_project_from_s3(*args, **kwargs):
    """Deprecated. Use `pull_from_s3` instead."""
    pull_from_s3(*args, **kwargs)

push_project_to_s3(*args, **kwargs)

Deprecated. Use push_to_s3 instead.

Source code in prefect_aws/deployments/steps.py
46
47
48
49
@deprecated_callable(start_date="Jun 2023", help="Use `push_to_s3` instead.")
def push_project_to_s3(*args, **kwargs):
    """Deprecated. Use `push_to_s3` instead."""
    push_to_s3(*args, **kwargs)

push_to_s3(bucket, folder, credentials=None, client_parameters=None, ignore_file='.prefectignore')

Pushes the contents of the current working directory to an S3 bucket, excluding files and folders specified in the ignore_file.

Parameters:

Name Type Description Default
bucket str

The name of the S3 bucket where files will be uploaded.

required
folder str

The folder in the S3 bucket where files will be uploaded.

required
credentials Optional[Dict]

A dictionary of AWS credentials (aws_access_key_id, aws_secret_access_key, aws_session_token) or MinIO credentials (minio_root_user, minio_root_password).

None
client_parameters Optional[Dict]

A dictionary of additional parameters to pass to the boto3 client.

None
ignore_file Optional[str]

The name of the file containing ignore patterns.

'.prefectignore'

Returns:

Type Description
PushToS3Output

A dictionary containing the bucket and folder where files were uploaded.

Examples:

Push files to an S3 bucket:

push:
    - prefect_aws.deployments.steps.push_to_s3:
        requires: prefect-aws
        bucket: my-bucket
        folder: my-project

Push files to an S3 bucket using credentials stored in a block:

push:
    - prefect_aws.deployments.steps.push_to_s3:
        requires: prefect-aws
        bucket: my-bucket
        folder: my-project
        credentials: "{{ prefect.blocks.aws-credentials.dev-credentials }}"
Source code in prefect_aws/deployments/steps.py
 52
 53
 54
 55
 56
 57
 58
 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
def push_to_s3(
    bucket: str,
    folder: str,
    credentials: Optional[Dict] = None,
    client_parameters: Optional[Dict] = None,
    ignore_file: Optional[str] = ".prefectignore",
) -> PushToS3Output:
    """
    Pushes the contents of the current working directory to an S3 bucket,
    excluding files and folders specified in the ignore_file.

    Args:
        bucket: The name of the S3 bucket where files will be uploaded.
        folder: The folder in the S3 bucket where files will be uploaded.
        credentials: A dictionary of AWS credentials (aws_access_key_id,
            aws_secret_access_key, aws_session_token) or MinIO credentials
            (minio_root_user, minio_root_password).
        client_parameters: A dictionary of additional parameters to pass to the boto3
            client.
        ignore_file: The name of the file containing ignore patterns.

    Returns:
        A dictionary containing the bucket and folder where files were uploaded.

    Examples:
        Push files to an S3 bucket:
        ```yaml
        push:
            - prefect_aws.deployments.steps.push_to_s3:
                requires: prefect-aws
                bucket: my-bucket
                folder: my-project
        ```

        Push files to an S3 bucket using credentials stored in a block:
        ```yaml
        push:
            - prefect_aws.deployments.steps.push_to_s3:
                requires: prefect-aws
                bucket: my-bucket
                folder: my-project
                credentials: "{{ prefect.blocks.aws-credentials.dev-credentials }}"
        ```

    """
    s3 = get_s3_client(credentials=credentials, client_parameters=client_parameters)

    local_path = Path.cwd()

    included_files = None
    if ignore_file and Path(ignore_file).exists():
        with open(ignore_file, "r") as f:
            ignore_patterns = f.readlines()

        included_files = filter_files(str(local_path), ignore_patterns)

    for local_file_path in local_path.expanduser().rglob("*"):
        if (
            included_files is not None
            and str(local_file_path.relative_to(local_path)) not in included_files
        ):
            continue
        elif not local_file_path.is_dir():
            remote_file_path = Path(folder) / local_file_path.relative_to(local_path)
            s3.upload_file(
                str(local_file_path), bucket, str(remote_file_path.as_posix())
            )

    return {
        "bucket": bucket,
        "folder": folder,
    }