https://prefect.io logo
#prefect-community
Title
# prefect-community
c

Chern Hong Poh

10/12/2022, 9:55 AM
Hello guys, currently I am using prefect version 0.13.17 and working environment is Amazon Linux. Then I face one problem when using
ShellTask
that returns
Command failed with exit code 2
when I registered and quick run the prefect flow. I registered the flow using this command
prefect register flow --file testing.py --project staging
. Appreciated if someone can help. This has been bugging me since morning.
Copy code
## print2.py

print("hello")
Copy code
## testing.py

import os
import datetime
from datetime import timedelta
import pendulum

import prefect
from prefect import case
from prefect import Flow
from prefect import Parameter
from prefect import task
from prefect.environments.storage import S3
from prefect.schedules import filters
from prefect.schedules.clocks import IntervalClock
from prefect.schedules.schedules import Schedule
from prefect.tasks.control_flow import merge
from prefect.tasks.dbt import DbtShellTask
from prefect.tasks.shell import ShellTask

import subprocess

@task(name="Logging")
def logging_result(stuff):
    logger = prefect.context.get("logger")
    return <http://logger.info|logger.info>(stuff)

@task(name="Run Python Script", log_stdout=True)
def run_script():
    return ShellTask(command=f"python3 print2.py").run()

with Flow(name="DBT Python daily run") as flow:
    python_run = run_script()
    final = logging_result(python_run)

#flow_state = flow.run()
#shell_output = flow_state.result[python_run].result
#print(shell_output)
1
m

Mason Menges

10/12/2022, 4:08 PM
Hey @Chern Hong Poh It looks like you're calling the shelltask from within another task, tasks can only be called within flows can you try running your flow like this
Copy code
import os
import datetime
from datetime import timedelta
import pendulum

import prefect
from prefect import case
from prefect import Flow
from prefect import Parameter
from prefect import task
from prefect.environments.storage import S3
from prefect.schedules import filters
from prefect.schedules.clocks import IntervalClock
from prefect.schedules.schedules import Schedule
from prefect.tasks.control_flow import merge
from prefect.tasks.dbt import DbtShellTask
from prefect.tasks.shell import ShellTask

import subprocess

@task(name="Logging")
def logging_result(stuff):
    logger = prefect.context.get("logger")
    return <http://logger.info|logger.info>(stuff)

with Flow(name="DBT Python daily run") as flow:
    python_run = ShellTask(command=f"python3 print2.py")
    final = logging_result(python_run)
3 Views