Is it possible to separate tasks in different scripts from the flows that call them in Prefect 2.0? I'm getting this error currently. @task(name='Get Email Credentials', NameError: name 'task' is not defined

@task(name='Get Email Credentials',
      retries = 2,
      retry_delay_seconds = 10)
async def GetEmailCredentials(aws_credentials: AwsCredentials) -> EmailCredentials:
    session = aws_credentials.get_boto3_session()
    secrets_client = session.client(service_name='secretsmanager')
    aws_secret = secrets_client.get_secret_value(SecretId='DEV_EMAIL')
    secret_json = aws_secret['SecretString']
    emailDict = json.loads(secret_json)
    return EmailCredentials(emailDict)

yes, this is possible do you have

from prefect import task

in the file where you're defining

GetEmailCredentials

?

Yes

import prefect
import json
from prefect_aws import AwsCredentials
from prefect import tasks

@task(name='Get Email Credentials',
      retries = 2,
      retry_delay_seconds = 10)
async def GetEmailCredentials(aws_credentials: AwsCredentials) -> EmailCredentials:
    session = aws_credentials.get_boto3_session()
    secrets_client = session.client(service_name='secretsmanager')
    aws_secret = secrets_client.get_secret_value(SecretId='DEV_EMAIL')
    secret_json = aws_secret['SecretString']
    emailDict = json.loads(secret_json)
    return EmailCredentials(emailDict)

I think we've got an extra "s" on

task

😄

from prefect import tasks

🤦‍♂️

earlier today I accidentally swapped my aws key id and secret key in my

~/.aws/credentials

profile and for like a half hour I was like "what is going on?!?!" haha it happens

Thanks again, Nate. Looks like I'm able to hydrate that EmailCredentials class now

glad to have helped!