https://prefect.io
Join Slack
Hello, I have using Git storage for my flow and I have successfully set up the Git repository. Now I...
# ask-community
t
Hello, I have using Git storage for my flow and I have successfully set up the Git repository. Now I am running my local agent. When I run the flow using my Local agent I get the following error: "Failed to load and execute Flow's environment: NotADirectoryError(20, 'The directory name is invalid')". But if I run a local agent through a docker container using docker-compose file. It works fine. Can someone please help me what can be possible problem. I am lost here. I am trying to find anything in documentation that can be helpful, but haven't had much of success
k
Hey @Talha, can you show you do configure the Git storage? Just remove any private info
t
# Configure remote flow storage. storage = Git(     repo_host="dev.azure.com/Myhostname",     repo="prefect-flows",     flow_path="flows/flow.py",     git_token_secret_name=GIT_ACCESS_TOKEN_NAME, )
with Flow("hello-docker-flow", storage=storage, run_config=LocalRun(labels=["label"])) as flow:     say_hello()
If I use the Github as storage option then the local agent is working fine
k
I think the repo host here is the one that has to be changed, but I’m not sure to what since I haven’t used Azure DevOps
Might have to be 
<http://dev.azure.com/username/username/_git/|dev.azure.com/username/username/_git/>
t
I don't think this is the problem : f.storage = Git(     repo="talhacheema77/prefect",      #path="flows/myflow.py",     flow_path="flows/myflow.py",     repo_host="github.com",     git_token_secret_name =GITHUB_ACCESS_TOKEN,     add_default_labels=False) f.run_config=LocalRun(labels =["talha"])
See at this example here I am using github.com
It is also not working. But If I change the f.storage = GitHub(     repo="talhacheema77/prefect",      #path="flows/myflow.py",     flow_path="flows/myflow.py",     repo_host="github.com",     git_token_secret_name =GITHUB_ACCESS_TOKEN,     add_default_labels=False) f.run_config=LocalRun(labels =["talha"])
It works
k
Ah ok I see what you’re saying. One sec.
I can’t replicate. I added a flow here . I was able to register and run with a local agent. What Prefect version are you on? This fails for you?
Copy code
f.storage = Git(
    repo="talhacheema77/prefect", 
    #path="flows/myflow.py",
    flow_path="flows/myflow.py",
    repo_host="<http://github.com|github.com>",
    git_token_secret_name =GITHUB_ACCESS_TOKEN,
    add_default_labels=False)
Could you see if you can run my flow?
t
I am on version 0.14.17.
I will try to run your flow
k
There seems to have been no changes since 0.14.17 so i think that’s fine
t
Hi, sorry for such late reply. I have tried your flow as well. But it is still not working with my local agent, the same error which is Failed to load and execute Flow's environment: NotADirectoryError(20, 'The directory name is invalid')
I am so confused how to debug it, as the current error log is not providing any information. Also I don't know what is (20, directory name). I am not providing any. Do I need to make any change in my config file
k
Are you on server or cloud?
t
server
k
What is the version of your server and agent?
t
How to check that ?
According to docker image core-0.14.17
k
Use this query
Copy code
query Api {
  api {
    backend
    mode
    version
    core_version
    release_timestamp
  }
}
t
"version": "2021.05.25",
"core_version": "0.14.19+21.g174f4ee60",
k
Seems like your server is ahead of the Flow (which was 0.14.17). You might be able to see the agent version in the UI when you look at agents
Actually, did you my demo flow as-is or did you do anything like add a run config?
t
i added run config
k
What RunConfig did you add?
t
flow.run_config=LocalRun(labels =["talha_local"])
k
That should be fine. I went down to 0.14.17 and it worked for me. Did you try turning on debug logs?
t
yes, i think that maybe the issue. Agent has version 0.14.17, server has version 0.14.21. Don't know the version of flow. how I can check that
k
flow.run_config=LocalRun(labels =["talha_local"], env = {'PREFECT__LOGGING__LEVEL': "DEBUG"})
You can see the info in the UI for what version it was registered with
t
flow has version 0.14.17
k
I think that should all work. Maybe try debug logs and see? Did you use my Github repo under the Git storage like this? This script is public so it should work for you
Copy code
flow.storage = Git(
repo="kvnkho/demos", 
flow_path="prefect/git_storage.py",
repo_host="<http://github.com|github.com>")
t
k
This looks good…I can’t think of any reason why it would fail
t
same error.
k
Did the DEBUG logs appear?
t
In LOGs of UI
k
Is there any clues there?
t
k
Let me ask the team and get back to you
t
How can I update the agent version
??
Do I have to install the whole prefect again
m
Hello Talha! Do you use any virtual env? Could you upgrade prefect version (
pip install --upgrade prefect==0.14.22
) and test it using that version?
t
virtual env, I just use skype or Microsoft teams
upgrade the prefect doesnot help
k
Are you on Windows?
t
yes
k
You use VSCode on Windows?
t
yes
k
Ok I’ll try it on my Windows computer
t
I have no idea why it is not working for me. My colleague is using the same system and he has prefect version 0.14.21. For him the the error is different
k
What is their error?
t
Copy code
Failed to load and execute Flow's environment: PermissionError (13, 'The process cannot access the file because it is being used by another process')
k
@Talha I was able to replicate your error when using VSCode on Windows to register my flow. Agent also running on Windows.
t
Okay. What do you think can be the reason ]
How can it can be resolved ?
k
Windows messes up file paths a lot of the time because it has a different convention. I'll figure out how we can get it working with Git storage but that will likely made the code not work in Linux environments. Just meant to be a temporary fix while we file the issue on our end.
t
what can be the temporary fix?
k
Working on it
t
yeah, it make sense now, when I use the docker container Linux environment the flow runs. which means this is the windows issue
k
Ok I spoke to an engineer on the core team. This fix will go out today for the Windows working Git storage. When it goes out, you can install the latest master version of Prefect and this Git storage will work. The Permission denied is looking related also.
t
Thanksa
for the fix, the git storage is working fine now
👍 1
I am trying to figure out where my results will be if my flow storage is Git Storage. Will the result be stored at Git or I have to define the storage and location of results separatly?
k
Result is separate from storage. Those have to be defined separate and I think we don’t have Git based results.
t
Results will have the information about my tasks as well as the return value. So if I store my results in the S3. It will have the information/return values, which I can access afterwards. Right?
k
Yes that’s right. Just give the S3 file path so you know where to retrieve it
t
Hi, a confusion: I am writing a flow for a office desk. The desk has some installed python packages which I don't have but when I replicate their code and try to run them through my system to register the flow. It gives me error. what is the way around that. How can I register a code with out installing any of the additioanal packages
k
What storage are you using? Still Git storage?
t
yes, the git
k
Yeah I don't think there is a way around this then unfortunately. You need the dependencies locally to register.
t
for any other storage it is possible ???
k
Let me check with the team
Ok so the answer is no because dependencies are needed to register a flow. I thought Git storage might work, which is why I asked, but I was wrong. You need dependencies to register
You can move your imports to a task to defer their execution. This way, the imports won’t run immediately. This is not good practice though
t
Hi, is it possible to call python modules with in the flow.py script. As I am importing some python files as Import module.py in the flow code. but the flow is giving me error that the module is not found. If I put that py file in the site packages where my prefect is. It works.
Can you tell me how to solve this issue. Can I use my python files as modules without placing them in the site-packages of python. As I am using git storage so I want all my files to be on git and then flow files should import the other py file as module and run the flow.
k
If you need you package dependent modules like this, you need to use Docker storage or have them installed where the code will be running (where the agent is). You can do this by packaging your project into a Python module and then installing it in Docker.
You won’t be able to do this with Git storage (though this is on the roadmap), because the agent will only pull the Flow file from Git storage
This part of the docs may help you
t
Hi, thanks for the help earlier. Now, I have confusion regarding the prefect agent install. I have gone through the documentation. I want to run the local agent in the background. I read about #supervisord but it is not compatiable with windows. What is the alternative . curently i am using a python script and running it as background script
k
Anything that lets you run it as a background script. Maybe you can try this? https://stackoverflow.com/a/59125542
I haven’t had to do this. You may want to post again in the community so other community members cna see that
t
Thanks. That is exactly what i am doing right now. Sure
Also I am running the email notification task , It is flow is successfully executing but i am receiving no emails. Cam you see the code below and let me know
k
Do you a task inside another task, use 
.run()
like 
EmailTask.run()
, but you should have to do this. you can just do something like:
Copy code
email_task = EmailTask(..., name= ..., trigger=...)

with Flow("test") as flow:
    email_task()
t
Email USer name and Password of the Email to should be added as Secret or Email from Username or Password
I added the Prefect Secret as set PREFECT__CONTEXT__SECRETS__EMAIL_USERNAME=tzchee77@gmail.com
And so as password
Getting this error "TimeoutError: [WinError 10060] A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond "
k
Run this to make sure they are right:
Copy code
from prefect.client import Secret
print(Secret("EMAIL_USERNAME").get())
print(Secret("EMAIL_PASSWORD").get())
If they are right, this is some sort of networking issue. Do you have a firewall?
t
yes
k
So the EmailTask uses smtplib. Can you try:
Copy code
import smtplib
smtp=smtplib.SMTP('<http://smtp.gmail.com|smtp.gmail.com>')
smtp.starttls()
print(smtp.ehlo())
Without Prefect and see if this works
👍 1
t
Hello, Is it possible to label flows such that they it can decide on which agent to run based on the state of the task? For example : I have a flow A which has two tasks (task A and task B). My flow is labelled as Run A. Now I have two Agents running on two different systems, and both have flow A as a label.
I want my flow A to run and on the runtime after checking the state of the task select or switch to the other agent?
k
You can’t do this mid flow, but ou can use a parent Flow that triggers child flow runs
And those child flow runs have different Exectuors and Labels. You can do this with the 
StartFlowRun
task
t
Hi, Is there a way to check the success or failure of a flow with in another flow. For example I have registered Flow A and in flow B I want to check if Flow-A was successful then run Flow -B
Can you guide to a proper documentation for such case
k
Check this new interface in 0.15.0
t
Hi, I have trying to figure out the #flowrunview thing recently added to the prefect core. I am running the following command but it is giving me error that the No Flow runs found while querrying for the flow runs ?
flow_state = FlowRunView.from_flow_run_id(flow_run_id="0163108d-e0bd-4663-bc14-2c1130a8c375")
why is such behaviour happening, Because I do have this flow id, at the UI
flow_run = FlowView.from_flow_name(flow_name="parent", project_name="Test",last_updated=True)
Also when I run Flow.view, it is working perfectly fine
k
I think you might be confusing FlowView from FlowRunView? The FlowRun id different from the Flow id. Each run gets a new FlowRun id, while the Flow id will not change from run to run.
👍 1
t
yeah, I got it and changed the code accordingly, but the confusion still lie that the FlowRunview gives me states. How can I check that the last flow run was successful. FlowView has not attribute as State in it
k
I think it would be the last item of FlowRunView.states. It should return something like 
[Scheduled(), Submitted(), Running(), Success()]
, and the last element will the most recent one.
t
I was trying to build my code such that it looks at a flow and depending on the the flow success or failure decide to run the current flow. I have started the flow with create flow run got the flow_run_id from there, Then using the Flowrunviewer I have got the states. But the states are continously changing.
In order to make a state based dependency, I think I have to introduce the flow state handler. Am I right?
k
Actually, the KV Store might be a good approach for this if you haven’t seen it yet. You can persist small pieces of information. The upstream Flow can post a flag whether or not the downstream flow can run. You mean you have Flow B that looks at Flow A and checks the state to see whether or not to run Flow B?
The flow state handler is to perform some action if a certain state is reached. You could use a flow state handler from A to trigger B with the 
client.create_flow_run
or 
StartFlowRun
task.
t
Hi, I am having a problem with prefect UI. Sometime the I change flow and register it or change flow label from UI. But the Metadata is not modified. I have to delete the flow and then register with a different name to make it work. I want to mention again, it is happening with some flows. Not a regular occurring. Also, if i delete the flow, it is kept in the archieve, if I register flow with same name the re-register does not work. I think metadata from archieved flows is messing up with the new flow Meta-data.
Is there a way to delete flow from archieve. (permantly delete it from the database)
k
How are you registering it? With the 
prefect register
CLI?
Could you post this as a new thread so I can direct it to someone on the team who knows more than me?
t
Not with CLI with flow.register(projectname) in python code
Posted it in the new Thread. Please let me know if I can delete an archieved flow or not?
Hi, @Kevin Kho I just want to ask is it possible to show graph result of a python script on Prefect as artifact or if there is any other options
k
Do you mean the flow visualization? Would’t it be under the schematic? Or is that graph an output of your Flow?
t
Graph is the output of the flow
k
I see. So your images in Markdown artifacts, you need to store it somewhere like S3, and then add it in the markdown. If you use S3, your bucket needs to have public access.
t
Hi, Thanks for the reply, I need help on a separate topic. I have created multiple tenants as i was testing something. Now I want to delete them but couldn't find any documentation on how to delete it. Can you help please. Thanks
k
You can use the 
delete-tenant
in the GraphQL API. You just need to pass the tenant ID
t
Hi, need some help.
I am using the new features, FlowRunView. I am confused about how to provide the flow_run_id
k
when you do 
create_flow_run
, the new task. this replaces the old 
StartFlowRun
. this returns a 
flow_run_id
for you that you can feed in FlowRunView
t
I am registering the flow on the UI and running it with UI, But I want to but a condition in my code with respect to flow state. As if my flow state is successful or failed then what to do,
But create flow run is code specific, I am starting my code from UI.
k
I think that’s the case for state handlers rather than flow run view. Have you seen the docs around them ?
t
But can the state handler help with flows state as well? I was of the opinion that the state handlers are task specific.
If state handlers can help me with flow states, then can I use create flow run with state handler ? My Use case: If the flow fails of Agent A then run the flow on Agent B. want to change the labels in the run time
k
state handlers can both be used by flow-level and task-level. you can use 
create_flow_run
in two ways. you can use 
client.create_flow_run
and you can use the 
create_flow_run
task inside the state handler with 
create_flow_run().run()
t
Thanks a lot. Big help
👍 1
Hi,
Is it possible i can set the state of my flow. for example based on an if condition i can change my flow state to success even if it is failed.
I see such thing in UI, but need to do this via code
k
That should be in the flow state handler
Copy code
def mystatehandler(flow, old_state, new_state):
    if new_state.is_failed():
       return Success(...)
where Success is the state here
👍 1
2 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by