https://prefect.io logo
Join Slack
Powered by
Hi Folks, got another one that's driving me a bit ...
# ask-community
j
Hi Folks, got another one that's driving me a bit crazy at the moment, with the coroutine datatype etc. I'm having a very difficult time pulling down a JSON block that I saved. Using this syntax, I'm not able to actually get at my JSON. I keep getting "TypeError: 'fn' must be callable"
Copy code
from prefect.blocks.system import JSON
json_block = JSON.load("BLOCK_NAME")
✅ 1
n
hi @Jacob Bedard - can you share the code you're running when you hit an error? and the trace if possible?
j
Copy code
json_block = JSON.load("oms-db-readonly-creds")
/var/folders/9z/60j637fx3pqfjsm07kmws_bw0000gn/T/ipykernel_69646/3309556486.py:1: RuntimeWarning: coroutine 'Block.load' was never awaited
  json_block = JSON.load("oms-db-readonly-creds")
RuntimeWarning: Enable tracemalloc to get the object allocation traceback

json_block.fn()
Traceback (most recent call last):

  File "/var/folders/9z/60j637fx3pqfjsm07kmws_bw0000gn/T/ipykernel_69646/3521832716.py", line 1, in <module>
    json_block.fn()

AttributeError: 'coroutine' object has no attribute 'fn'
n
gotcha, i assume the place you're calling load is an async function?
j
I'm trying to pull some configs that will be different for dev vs the deployment env - I wanted to keep the config in a JSON
Just trying to pull it down and make sure I parse it out correctly
I tried by itself, tried in a test flow... just can't actually extract the JSON I saved in the web UI
n
Copy code
'Block.load' was never awaited
you need to add an 
await
before 
JSON.load
load method on blocks is sync compatible, which is a bit confusing, but it means you should treat it like an async method when in an async context and then a sync one in a sync context
j
Ok, I'll try adding await
one sec
Do I have to import asyncio?
n
nope!
Copy code
json_block = await JSON.load("oms-db-readonly-creds")
provided this is happening in an async function
j
yep
trying now
no joy
Still can't get at them
n
can you share your full function and trace?
j
@flow('test secret block ffs') def get_oms_cred_test(): the_creds = await JSON.load("oms-db-readonly-creds") # Access the stored secret return the_creds testing = get_oms_cred_test()
I just really want to get the JSON out 😞
Copy code
File "/var/folders/9z/60j637fx3pqfjsm07kmws_bw0000gn/T/ipykernel_69646/996931604.py", line 2
    the_creds = await JSON.load("oms-db-readonly-creds")
                ^
SyntaxError: 'await' outside async function
n
your function isn’t async, you’d need to add async before def
j
it won't take that either
not with the flow
Copy code
@flow('test secret block ffs')
async def get_oms_cred_test():
  the_creds = await JSON.load("oms-db-readonly-creds")
  # Access the stored secret
  return the_creds
This gives me:
Copy code
File "/Users/jacobbedard/opt/anaconda3/envs/prefect-2-env/lib/python3.9/site-packages/prefect/flows.py", line 238, in __init__
    raise TypeError("'fn' must be callable")

TypeError: 'fn' must be callable
n
ahh you need to say name=“test secret …”, right now you’re trying to decorate a string with @flow
j
ahhhhhhh I'm so dumb
That's so obvious
rushing too much
ok, no dice still
Enable tracemalloc to get the object allocation traceback
RuntimeWarning: coroutine 'get_oms_cred_test' was never awaited
I must be really driving you nuts with my ignorance of asyncio stuff here, sorry
n
all good! that’s why we’re here. your flow is async now, so you can use asyncio.run(your_flow())
j
How about just getting this into a flow, though?
I'm trying to put together a pretty straightforward flow, with sequential steps
get creds from env vars, get config from prefect block, do a fairly typical incremental sync, send it to s3, copy to snowflake
Pretty standard stuff, just wanted to get these configs stored somewhere other than on the machines that run it
Do I have to change the whole flow in which this is executed? Or do I make it a sub-flow? Do I have to do anything fancy with the deployment?
I'm just going to give up on this
This complicates things way too much
ok, I was able to inspect the returned coroutine after applying await
the_stuff = await JSON.load("<whatever>") for thing in the_stuff.value: now I can get at the actual contents
This was more of a problem about the structure of the returned class "blocks.system.JSON" than anything else
I couldn't find any documentation on what the structure of this thing would be
n
@Marvin can you show the json block example from the docs
m
Here is a JSON block example from the Prefect documentation:
Copy code
python
from prefect.blocks.system import JSON 

json_block = JSON(value={"the_answer": 42})

# Saving blocks
json_block.save(name="life-the-universe-everything")

# Loading blocks
from prefect import flow 
from prefect.blocks.system import JSON 

@flow 
def what_is_the_answer(): 
    json_block = JSON.load("life-the-universe-everything") 
    print(json_block.value["the_answer"]) 

what_is_the_answer() # 42

# Deleting blocks
from prefect.blocks.core import Block 
Block.delete("life-the-universe-everything")
In this example, a JSON block is created with the value 
{"the_answer": 42}
. This block is then saved with the name "life-the-universe-everything". Later, the block is loaded using the same name and its value is printed out. Finally, the block is deleted. I hope that helps, though I must say, it's a bit depressing how easily you humans forget things. But then again, I suppose that's why you have me.
n
going back to your original trace, i understand what was happening now you were in ipython, so you had an event loop, so you need await bc its an "async context" if you have a script (not ipython) with a 
def
instead of 
async def
, then you can just do it like the example above and not care about await or asyncio or whatever if you keep things async, you don't need to worry about calling your flow yourself if you're creating a deployment, because invokes the flow run for you, i.e. you dont need an 
if __name__ == "__main__"
block, but if you do want to run the async version locally, you can just do
Copy code
import asyncio
from prefect import flow

@flow
async def foo():
   pass

if __name__ == "__main__:
   asyncio.run(foo())
j
Yeah, using spyder
I think that weighed into it
You are very good at this
Thanks for the help!
n
sure thing!
6 Views
Open in Slack
PreviousNext
Powered by