Hi everyone, first question, please be gentle 🙂 I am sending a parameter called video to my "Process video" workflow. The first task in this workflow is to upload the video to an s3 bucket. The upload task is transcribed in the thread: My main problem seems to be that s3 requires a string filename, but my video is of the type <Parameter: video>, as illustrated by the error message

ValueError: Filename must be a string

So there is clearly some sort of.. unwrapping/unpacking that I'm missing? my flow is likewise in the thread, as is the invocation. So... would someone please tell me in which exact angle I should smack my forehead and the correct octave of the "duh!"... plus maybe what I should be doing instead?

Hi @Manuel Gomes - I think the issue is that you are using the parameter task twice. When you call flow.run you can use video = “abs…..” without using the Parameter task. You can see more in the docs here: https://docs.prefect.io/core/concepts/parameters.html

class UploadTask(Task):
    def run(self, video):
        tgt = uuid.uuid1().__str__()
        print("Video=%s" % video)
        s3 = boto3.resource('s3')
        result = s3.Bucket('MAH_BUCKET').upload_file(
            video,
            tgt
        )
        pprint(result)
        return(result)

with Flow("Process video") as flow:
    uploader = UploadTask()
    video = Parameter('video')
    result = uploader.run(video)

Moving to the thread is great! I think you should be able to declare simply v=“/abs/path…” without using the Parameter task twice. If you see in the docs, when you call flow.run you pass the actual parameter value.

Hmm... I get the feeling that I have not entirely grokked the nature of parameters as tasks, but let me do as you suggest and see if any light shines forth

Sorry I'm on mobile so formatting isn't good but you could do flow.run(parameters =dict(v=“pathtovideo”))

with Flow("Process video") as flow:
    uploader = UploadTask()
    video = Parameter('video')
    result = uploader.run(video)

should be

with Flow("Process video") as flow:
    uploader = UploadTask()
    video = Parameter('video')
    result = uploader(video)

and I think that will work. You can also init the

UploadTask

outside the flow.

uploader = UploadTask()
with Flow("Process video") as flow:
    video = Parameter('video')
    result = uploader(video)

You shouldn’t have to call

run

on the task because the flow block will do it for you.