Hi Everyone, new to prefect, looking for some poin...
# prefect-community
r
Hi Everyone, new to prefect, looking for some pointers/guidance how to run PowerShell with prefect. I started looking at using the shell task but, I am not having much luck. My agent is running a Windows Server 2019. Thanks in advance.
👀 1
n
Hi @Richard Hughes, sorry for the delay! I don't think the current
ShellTask
support PowerShell, unfortunately. This is a place we would really welcome contributors though!
r
@nicholas I looked at the code for the
ShellTask
and it looks similar to a `PowerShell`example I found on the web using the `POPEN`function. I think maybe some subtle difference in the code might make this available for different shells. Python is new for me so it will take some time if I have to write it.
n
No worries @Richard Hughes - can you describe what you're seeing? I'd like to open a ticket for this.
r
@nicholas let me get you some details
👍 2
@nicholas I think its best if I try to work on this some more and get back to you, it looks like when passing the CMD or PowreShell in the shell parameter is doesn't this input on the popen sub_process . My server is acting up so I am going to have to prob catch you up next week.
n
No worries at all @Richard Hughes - we'll be here 🙂
r
@nicholas I was able to write a flow with using PowerShell. I don't know if you want to create a PowerShell Task based upon this code or modify the shell task to accept PowerShell and handle it accordingly. Probably a separate task would be easier.
Copy code
import os
import shlex
import sys
from prefect import task, Flow
from subprocess import Popen, PIPE
from prefect.triggers import all_successful, any_failed, some_successful

@task(name="PowerShell")
def sub_process_task(command):
    command=shlex.split(command)
    commandArray = ["powershell.exe"] + command
    p = Popen(commandArray, stdout=PIPE, stderr=PIPE)
    output, error = p.communicate()
    if p.returncode != 0: 
        raise Exception("powershell failed %d %s %s" % (p.returncode, output.decode("utf-8"), error.decode("utf-8")))
    else:
        #print("powershell succeeded %d %s" % (p.returncode, output.decode("utf-8")))
        return output.decode("utf-8")

@task(name="Print", trigger=all_successful)
def output_handler(output):
    print('Finished Running the flow')
    print(output)
    print('Finished Printing the flow')

with Flow('PowerShell') as flow:
    command="Write-Host 'Hello World'; Write-Host 'Test12345'; Get-Location; Get-ChildItem -Dir;"
    output=sub_process_task(command=command)
    output_handler(output)

flow.run()
n
Hi @Richard Hughes, this looks great! Could I encourage you to open a ticket or even a PR with this code? The Core folks will have stronger opinions on how this could/should be structured
👍 2