Hi Team - For Prefect 2.0, we would like to use our S3 bucket to store some allocations files. These will be emailed/sftp to clients. I'm not sure what I'm doing wrong with the state and await from the write_path function.

aws_credentials_block = AwsCredentials.load("****-user")
s3_block = S3.load("***-s3")

s3_bucket_block = S3Bucket(
    bucket_name=s3_block.bucket_path,
    aws_credentials=aws_credentials_block,
    basepath=f"Allocations/{date.today().year}"
)
output_file = f'''SR Allocations {datetime.now().strftime('%Y%m%d-%H%M%S')}.csv'''
bytes_to_write = df_alloc.to_csv(None).encode()
csv_file = s3_bucket_block.write_path(path=output_file, content=bytes_to_write)

<http://logging.info|logging.info>('Filtering for **** Trades: Rows = %s, Accounts = %s' % (len(df_alloc), df_alloc.custodianAccnt.nunique()))

return csv_file

@flow(name = "***** Allocations")
def ****AllocationsFlow():
    try:
        slack_webhook_block = SlackWebhook.load("****-webhook")
        state = allocations_process(return_state = True)
        excel_file = state.result()
        if 'Completed'== state.name:
            slack_webhook_block.notify("**** Allocations was successful")
        else:
            slack_webhook_block.notify("***** Allocations failed")

Error: cannot pickle 'coroutine' object 11:23:31.736 | INFO | Task run 'Fidelity Allocations-e728df66-0' - Crash detected! Execution was interrupted by an unexpected exception. 11:23:32.443 | ERROR | Flow run 'visionary-cat' - Finished in state Failed('1/1 states failed.') Traceback (most recent call last): File "/opt/pysetup/.venv/lib/python3.9/site-packages/anyio/_core/_eventloop.py", line 70, in run return asynclib.run(func, *args, **backend_options) File "/opt/pysetup/.venv/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 292, in run return native_run(wrapper(), debug=debug) File "/usr/local/lib/python3.9/asyncio/runners.py", line 44, in run return loop.run_until_complete(main) File "/usr/local/lib/python3.9/asyncio/base_events.py", line 647, in run_until_complete return future.result() File "/opt/pysetup/.venv/lib/python3.9/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper return await func(*args) File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/client/orion.py", line 82, in with_injected_client return await fn(*args, **kwargs) File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/engine.py", line 239, in create_then_begin_flow_run return state.result() File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/orion/schemas/states.py", line 159, in result state.result() File "/opt/pysetup/.venv/lib/python3.9/site-packages/prefect/orion/schemas/states.py", line 145, in result raise data TypeError: cannot pickle 'coroutine' object

what does your

allocations_process

look like?

I am querying the database and creating a data frame. The data frame is converted to a csv file and the content is the data of the csv file.

what is the return type of

allocations_process

(if any)?

🤦‍♂️ I didn't have one. It looks like bytes is appropriate from return s3_bucket_block.read_path(csv_file)

I was able to reproduce that error, but I was able to get it to work by making my task async (and then awaiting my block load and block method calls) - will look into why this error is occurring in a sync context

Like this:self.__secret_name = await String.load("***-database")

is that line of code in a class's

__init__

function?

Yes, so the ___init_ cannot be async

I believe that issue has been resolved and merged but is not yet released - not exactly sure when that will be included as for the issue you raised with

S3Bucket

above (thanks for raising that by the way), I've gone ahead and opened an issue (my PR to resolve it is being reviewed now) which will allow you to call

S3Bucket

methods in a sync context without running into that

TypeError: cannot pickle 'coroutine' object

error - you can track the progress here

Thanks Nate!