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

Emerson Franks

07/12/2022, 5:51 PM
Hey there Prefect folks! I'm getting a Prefect -> FiveTran POC up and encountering an issue where the flow runs fine locally from terminal but doesn't appear in the UI, and therefore can't be run in the UI, once I launch prefect agent local start. I'm using 1.0 as I didn't see a FiveTran module for 2.0. If anyone can point me at what I'm missing I would greatly appreciate it!
1
a

Anna Geller

07/12/2022, 5:53 PM
please could you move the code block into the thread to keep the main channel clean? Thanks in advance!
e

Emerson Franks

07/12/2022, 5:59 PM
I assume this is what you mean? 🙂 Here is my code:
import prefect
from prefect import task, Flow
from prefect.tasks.fivetran import FivetranSyncTask
logger = prefect.context.get("logger")
def log_message(message):
<http://logger.info|logger.info>(message)
def start_job():
api_key = 'PUT THESE BACK' #These need to come from Key Vault eventually.
api_secret = 'PUT THESE BACK'
connector_id = 'CONNECTOR ID GOES HERE'
poll_status_every_n_seconds = 60
fivetran_task = FivetranSyncTask(connector_id)
status = fivetran_task.run(api_key, api_secret, connector_id=connector_id, poll_status_every_n_seconds=poll_status_every_n_seconds)
return status
@task
def fivetran_flow():
log_message('Kicking off FiveTran task.')
status = start_job()
log_message(f'FiveTran run completed with status of: {status}')
flow = Flow("five-tran-flow", tasks=[fivetran_flow])
flow.register(project_name="five-tran-poc")
a

Anna Geller

07/12/2022, 6:02 PM
thank you! Can you try this syntax?
Copy code
from prefect import task, Flow, Parameter
from prefect.tasks.fivetran import FivetranSyncTask
from prefect.tasks.secrets import PrefectSecret


fivetran_task = FivetranSyncTask(connector_id="CONNECTOR ID GOES HERE")

with Flow("five-tran-flow") as flow:
    api_key = PrefectSecret("FIVETRAN_API_KEY")
    api_secret = PrefectSecret("FIVETRAN_API_SECRET")
    poll_seconds = Parameter(name="poll_seconds", default=60)
    fivetran_task(
        api_key=api_key, api_secret=api_secret, poll_status_every_n_seconds=poll_seconds
    )

if __name__ == "__main__":
    flow.register(project_name="five-tran-poc")
🙌 1
this assumes you use Prefect Cloud and store Secrets there - you can do that from the UI
e

Emerson Franks

07/12/2022, 8:27 PM
That did it, thanks!
🙌 1
6 Views