Hi I m trying to write a workflow that takes in video as inp

Question

Hi I m trying to write a workflow that takes in video as input and simply writes it Input can either be in uri variable which is a downloadable link to video or it can be in data which is base64 string It is known that if input is an uri then value of data is going to be 1 Thus I check if value of data is 1 and if it is then call download task which downloads video from uri and saves content in data This data is then passed to write video task I m new to flyte and this code is what I have so far ```from typing import List from flytekit import task workflow conditional import base64 import requests import cv2 task def is base64 data bytes gt bool try return base64 b64encode base64 b64decode data == data except Exception return False task def download uri str gt bytes code to download from url return data task def write video data bytes gt None if is base64 data=data data = base64 b64decode data with open input mp4 wb as out file out file write data workflow def wf data str uri str job id str gt None write video data=conditional input if data == 1 then download uri=uri else then data ``` however I am running into errors and would really appreciate some help

Pulkit Mishra · Answer

One clear issue is the way I am handling bytes type what is the correct way of doing that

Samhita Alla · Answer

Hi < Pulkit Mishra> `bytes` isn t a native Flyte type and hence it gets pickled automatically That should work But the recommended way of usage is to either send bytes as string or handle bytes conversion using a custom If the bytes string is large you can also store that in a file and return that as a

Pulkit Mishra · Answer

thanks for the response < Samhita Alla> I understand will change bytes to FlyteFile but if that is not the reason for failure then can you help me understand what is This is the stacktrace I am getting ```Traceback most recent call last File opt conda envs venv app lib python3 8 site packages flytekit core promise py line 940 in create and link node binding from python std File opt conda envs venv app lib python3 8 site packages flytekit core promise py line 660 in binding from python std binding data = binding data from python std ctx expected literal type t value t value type File opt conda envs venv app lib python3 8 site packages flytekit core promise py line 599 in binding data from python std raise AssertionError AssertionError Cannot pass output from task n0 that produces no outputs to a downstream task The above exception was the direct cause of the following exception Traceback most recent call last File opt conda envs venv app bin pyflyte line 8 in lt module gt sys exit main File opt conda envs venv app lib python3 8 site packages click core py line 1130 in call return self main args kwargs File opt conda envs venv app lib python3 8 site packages click core py line 1055 in main rv = self invoke ctx File opt conda envs venv app lib python3 8 site packages click core py line 1657 in invoke return process result sub ctx command invoke sub ctx File opt conda envs venv app lib python3 8 site packages click core py line 1657 in invoke return process result sub ctx command invoke sub ctx File opt conda envs venv app lib python3 8 site packages click core py line 1651 in invoke cmd name cmd args = self resolve command ctx args File opt conda envs venv app lib python3 8 site packages click core py line 1698 in resolve command cmd = self get command ctx cmd name File opt conda envs venv app lib python3 8 site packages flytekit clis sdk in container run py line 607 in get command entity = load naive entity module exe entity project root File opt conda envs venv app lib python3 8 site packages flytekit clis sdk in container run py line 447 in load naive entity importlib import module module name File opt conda envs venv app lib python3 8 importlib init py line 127 in import module return bootstrap gcd import name level package level File lt frozen importlib bootstrap gt line 1014 in gcd import File lt frozen importlib bootstrap gt line 991 in find and load File lt frozen importlib bootstrap gt line 975 in find and load unlocked File lt frozen importlib bootstrap gt line 671 in load unlocked File lt frozen importlib bootstrap external gt line 843 in exec module File lt frozen importlib bootstrap gt line 219 in call with frames removed File usr src app sports py line 57 in lt module gt def wf data str uri str job id str callback url str result path str gt None File opt conda envs venv app lib python3 8 site packages flytekit core workflow py line 739 in workflow return wrapper workflow function File opt conda envs venv app lib python3 8 site packages flytekit core workflow py line 734 in wrapper workflow instance compile File opt conda envs venv app lib python3 8 site packages flytekit core workflow py line 614 in compile workflow outputs = exception scopes user entry point self workflow function input kwargs File opt conda envs venv app lib python3 8 site packages flytekit exceptions scopes py line 198 in user entry point return wrapped args kwargs File usr src app sports py line 58 in wf write video data=conditional input if data == 1 then download uri=uri else then data File opt conda envs venv app lib python3 8 site packages flytekit core base task py line 283 in call return flyte entity call handler self args kwargs File opt conda envs venv app lib python3 8 site packages flytekit core promise py line 1029 in flyte entity call handler return create and link node ctx entity=entity kwargs File opt conda envs venv app lib python3 8 site packages flytekit core promise py line 950 in create and link node raise AssertionError f Failed to Bind variable k for function entity name from e AssertionError Failed to Bind variable data for function sports write video ```

Samhita Alla · Answer

You ll need to include a task in the `else then` construct Could you return `data` from a task if that value needs to be included in `else then`

Pulkit Mishra · Answer

So I need to create a dummy task that takes in data and simply returns it Is there a better way of doing this maybe

Samhita Alla · Answer

Why do you want to return `data` in `else` You re sending `conditional` output to `write video` task so you need to return something to it When you include `data` in `else` it doesn t return any output Got it

Pulkit Mishra · Answer

This is how I am handling it for now Other than changing bytes handling to TypeTransformer is there something else you would suggest changing < Samhita Alla> ``` task def decode data bytes gt bytes try if base64 b64encode base64 b64decode data == data return base64 b64decode data else return data except Exception return data task def download uri str gt bytes data err = code to download video from uri return data task def return data data bytes gt bytes return data task def write video data bytes gt None with open input mp4 wb as out file out file write data write video data=decode data=conditional input if data == 1 then download uri=uri else then return data data=data ```

Samhita Alla · Answer

I m wondering why `return data` accepts `data` of type `bytes` when `data` is of type `str` as per the workflow input

Pulkit Mishra · Answer

Yeah my bad I only checked when the input is uri