Dp 2534 aws s3 by davezuckerman · Pull Request #91 · BerkeleyLibrary/mokelumne

davezuckerman · 2026-07-22T18:45:35Z

No description provided.

awilfox

Found a few things to fix so far, but a nice architecture.

anarchivist

a few comments, mostly around using Airflow's native operators to do this.

anarchivist · 2026-07-22T22:58:22Z

+    file_extension = "{{ params.file_extension }}"
+
+    validate_destination = validate_destination(dest_directory) 
+    bucket_filenames = get_bucket_file_names(bucket, file_prefix, file_extension)


i wonder if you can do this using the S3ListOperator and simplify the custom method to filter.

I think using S3hook in this case might be better since we also need to filter out the extension. With S3hook I can do that directly on the list in memory. It looks like the S3ListOperator would need to get the whole list first and put it in the DB first and then I'd need to retrieve it to strip out based on the extension.

anarchivist · 2026-07-22T23:00:48Z

+    def get_bucket_file_names(bucket: str, file_prefix: str | None = None, file_extension: str | None = None) -> list:
+        """Get a list of files for a given bucket. Can be filterd by prefix and extension"""
+        prefix = file_prefix.strip() if file_prefix else None
+        extension = file_extension.strip() if file_extension else None
+
+        file_names = list_bucket_files(bucket_name=bucket, file_prefix=prefix, file_extension=extension)
+
+        if file_names:
+            for name in file_names:
+                logger.info("Found file: %s", name)
+        else:
+            logger.info("The bucket is empty!")
+
+        return file_names


see comment on line 90.

anarchivist · 2026-07-22T23:30:31Z

+
+    validate_destination = validate_destination(dest_directory) 
+    bucket_filenames = get_bucket_file_names(bucket, file_prefix, file_extension)
+    retrieved_files = retrieve_files_from_bucket(bucket, bucket_filenames, dest_directory)


similarly for this you might be able to use the FileTransferOperator here.

Actually I was thinking about this part some more. I've just been testing with tiny example files and not taking into account these files could be pretty darn big (wav files etc.). I'm going to redo how this is done so it won't crash.

I looked at this for a while. I'm thinking dynamic task mapping using S3hook is the way to go since we can have hundreds of files and they could be really large. I changed it up to do it this way.

Each file will be downloaded as it's own task and if there's an issue with a particular task the issue could be resolved and the task could be resumed. It also looks like performance wise using S3hook is preferable since it's optimized for large S3 downloads and is more configurable.

We're not doing further processing of the downloaded file but if we wanted to we could run into a FileNotFound issue using FileTransferOperator since a subsequent task could start on a different node and possibly not see the file.

@awilfox Since Maria is out for a little bit I'm wondering if you could take a peak at the way downloading is now handled? I 'think' this should work well. I tested it locally with several large files and it handled them nicely

anarchivist

r+wc. i think we're close; have some questions about how we pass in connection ids and adding a description to make the purpose of this Dag clear.

anarchivist · 2026-07-27T20:57:00Z

+    dag_id="s3_download",
+    schedule=None,
+    catchup=False,
+    params={


do we need to parameterize the connection ID to use here? i'm also a little curious about whether we need to identify a specific connection for this since it's credentials specifically for city arts.

We don't need to, it could be hardcoded to cityartsmedia. I did it that way for the future assuming we'd be using it for other buckets and connections. That's why I chose a general fetch_from_s3.py for the name as well. Should I remove it and hardcode it to cityartsmedia?

I actually do see there is another bucket for them that we have access to cityarts_sf and that uses the same connection as cityartsmedia

no, i don't think you should hardcode it. if the intent is to use it with other jobs to copy from S3, then we'll need to support other connection IDs. my question is about how those connections get passed in to the Dag.

I'm not sure the best way of handling multiple connection id's. I was under the assumption the scope of this was for City Arts but could be enhanced for other connections when and if the need arises (via a separate ticket). Should I look into how we'd have this handle multiple connections before merging this? As it is now it will default to the cityartsmedia bucket but the connection should also work for the cityarts-sf which has the same connection string.

yes, i think we should look into that before we merge it.

I don't think Airflow handles conditional parameters so if a user chose a particular connection from a dropdown (S3 connections) list I don't think I could dynamically populate another (bucket) parameter with a list of buckets available for that connection.

I can think of a few ways of handling this.

I don't imagine connections would have that many buckets. It could be a single dropdown with both connection and bucket listed in the same dropdown. e.g.

[
"conn: cityarts | bucket: cityartsmedia",
"conn: cityarts | bucket: cityarts-sf",
"conn: somevendor | bucket: somebucket",
etc. etc.
]

I could parse out the connection and bucket pretty easily.

It could have two drop downs. One populated with the available connections and another for the user to type in the bucket. They'd need to know the bucket for a given connection though.

Looks like a multi-step dag where Dag 1 triggers Dag2 with supplied values and pauses at the first step would maybe work but that seems kind of messy and overkill for this. I'd probably have to redo a lot of this if we went that route

Of the three I like option 1 the best. Open to other suggestions

confirming we discussed an option to provide a string param (without dropdown) for connection_id and bucket name.

Just pushed but forget to change the tests to accommodate the new param. Will fix that now

anarchivist

great, let's ship it!

Adding option to retrieve files based on prefix and extension minor formatting fixes renamed s3_utils.py using dynamic task mapping for retrieving files, some refactoring for S3 Dag and associatled s3 util removing fail fast from Dag so downstream failures aren't skipped. Added a bit of cleanup for spaces in bucket name Added test for s3 utility and Dag, changed case name for fetch_from_s3.py Added description for S3 Dag Making s3 connection configurable, Adding test seed data for multiple buckets for localstack testing Changed tests to account for s3 connection_id being a parameter only reading in neccessary environment variables for localstack instead of whole env file

davezuckerman requested review from anarchivist and awilfox July 22, 2026 18:45

awilfox reviewed Jul 22, 2026

View reviewed changes

anarchivist reviewed Jul 22, 2026

View reviewed changes

anarchivist approved these changes Jul 27, 2026

View reviewed changes

davezuckerman marked this pull request as ready for review July 27, 2026 22:38

anarchivist approved these changes Jul 28, 2026

View reviewed changes

davezuckerman force-pushed the DP-2534-AWS-s3 branch from dc132da to a3befb7 Compare July 28, 2026 18:22

davezuckerman merged commit e9abe28 into main Jul 28, 2026
5 checks passed

davezuckerman deleted the DP-2534-AWS-s3 branch July 28, 2026 18:36

Uh oh!

Conversation

davezuckerman commented Jul 22, 2026

Uh oh!

awilfox left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

anarchivist left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

davezuckerman Jul 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anarchivist left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

davezuckerman Jul 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

davezuckerman Jul 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anarchivist Jul 28, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

anarchivist left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

davezuckerman Jul 23, 2026 •

edited

Loading

davezuckerman Jul 27, 2026 •

edited

Loading

davezuckerman Jul 28, 2026 •

edited

Loading

anarchivist Jul 28, 2026 •

edited

Loading