I am running into an issue when using credentials from AWSSe

Question

I am running into an issue when using credentials from AWSSecretsManager to fetch data from PostgresFetch as below ```from prefect import Flow task from prefect tasks aws secrets manager import AWSSecretsManager from prefect tasks postgres postgres import PostgresFetch task def print me to print print to print with Flow fetching data as flow credentials = AWSSecretsManager secret= pg prefectsql prefect print me credentials pg user = credentials username pg pass = credentials password pg host = localhost pg port = credentials port pg query = SELECT FROM hdb catalog hdb table LIMIT 5 runsql = PostgresFetch db name= prefect user=pg user host= localhost result = runsql password=pg pass query=pg query fetch= all print me result flow run ``` the PostgresFetch initialization doesn t work when I try to use `user=credentials username ` but it does when I hardcode the username or even if I set `pg user` to the a string containing the username The error flow run produces the below output

nicholas · Answer

Hi < Hamza Ahmed> would you mind moving your flow run output here and removing it from the main channel Typically we like to avoid big walls of code in the main channel

Hamza Ahmed · Answer

``` 2020 09 18 17 29 36 INFO prefect FlowRunner | Beginning Flow run for fetching data 2020 09 18 17 29 36 INFO prefect TaskRunner | Task AWSSecretsManager Starting task run 2020 09 18 17 29 36 INFO prefect TaskRunner | Task AWSSecretsManager finished task run for task with final state Success 2020 09 18 17 29 36 INFO prefect TaskRunner | Task print me Starting task run username prefect password prefectpass engine postgres host <http dbidentifier randstring aws region rds amazonaws com|dbidentifier randstring aws region rds amazonaws com> port 5432 dbInstanceIdentifier instancename 2020 09 18 17 29 36 INFO prefect TaskRunner | Task print me finished task run for task with final state Success 2020 09 18 17 29 36 INFO prefect TaskRunner | Task GetItem Starting task run 2020 09 18 17 29 36 INFO prefect TaskRunner | Task GetItem finished task run for task with final state Success 2020 09 18 17 29 36 INFO prefect TaskRunner | Task PostgresFetch Starting task run 2020 09 18 17 29 37 ERROR prefect TaskRunner | Unexpected error OperationalError FATAL password authentication failed for user lt Task GetItem gt nFATAL password authentication failed for user lt Task GetItem gt n Traceback most recent call last File C Users hahmed repos prefectsecrets venv lib site packages prefect engine runner py line 48 in inner new state = method self state args kwargs File C Users hahmed repos prefectsecrets venv lib site packages prefect engine task runner py line 822 in get task run state value = timeout handler File C Users hahmed repos prefectsecrets venv lib site packages prefect utilities executors py line 188 in timeout handler return fn args kwargs File C Users hahmed repos prefectsecrets venv lib site packages prefect utilities tasks py line 427 in method return run method self args kwargs File C Users hahmed repos prefectsecrets venv lib site packages prefect tasks postgres postgres py line 190 in run conn = pg connect File C Users hahmed repos prefectsecrets venv lib site packages psycopg2 init py line 127 in connect conn = connect dsn connection factory=connection factory kwasync psycopg2 OperationalError FATAL password authentication failed for user lt Task GetItem gt FATAL password authentication failed for user lt Task GetItem gt 2020 09 18 17 29 37 INFO prefect TaskRunner | Task PostgresFetch finished task run for task with final state Failed 2020 09 18 17 29 37 INFO prefect TaskRunner | Task GetItem Starting task run 2020 09 18 17 29 37 INFO prefect TaskRunner | Task GetItem finished task run for task with final state Success 2020 09 18 17 29 37 INFO prefect TaskRunner | Task GetItem Starting task run 2020 09 18 17 29 37 INFO prefect TaskRunner | Task GetItem finished task run for task with final state Success 2020 09 18 17 29 37 INFO prefect TaskRunner | Task print me Starting task run 2020 09 18 17 29 37 INFO prefect TaskRunner | Task print me finished task run for task with final state TriggerFailed 2020 09 18 17 29 37 INFO prefect FlowRunner | Flow run FAILED some reference tasks failed Process finished with exit code 0```

Hamza Ahmed · Answer

Instead of passing the value for `credentials password ` it is passing the task

Hamza Ahmed · Answer

< nicholas> Sorry about that I thought it was one big message

Skip Breidbach · Answer

< nicholas> is there something about class init parameters that doesn t convert Tasks to their actual values i e the results of run That s what it looks like is happening here

nicholas · Answer

No worries < Hamza Ahmed> The answer is yes < Skip Breidbach> when you init a Class you can only pass in build time parameters as opposed to run time parameters When you need values that are only available as a result of another task s run method you need to pass them to downstream tasks as run time parameters Example ```Class FullName Task def init self first name last name self first name = first name self last name = last name def run self first name last name self first name = first name if first name is not None self last name = first name if last name is not None return first name self first name last name self last name with Flow some flow as flow first name = Parameter first name last name = Boseman full name = FullName last name=last name first name=first name At build time the FullName class is initialized only with the last name param which we pass as a static string However it can t initialize first name until runtime because first name is the result of another task s the Parameter task runtime if you tried to pass first name in this way full name = FullName first name=first name last name=last name you would be passing the a reference to the Parameter instantiation since instantiation of classes in the flow context happens at build time instead of passing a reference to its runtime result```

nicholas · Answer

So < Hamza Ahmed> this means that if you want to pass args to a class instantiation like `PostgresFetch` that are determined by the result of another task s run you ll need to pass them to the task s runtime context instead In this particular case you ll probably need to extend the `PostgresFetch` class since it doesn t take user and host args at runtime OR create a wrapper task that you can pass runtime args to which can return the result of `PostgresFetch run` ```Class RuntimePostgresFetch PostgresFetch def run self user db name password self user = user self password = password self db name = db name super RuntimePostgresFetch self run ``` ^ class extension method OR ``` task def postgres fetch wrapper user db name password pg query return PostgresFetch db name=db name user=user host= localhost run password=pg pass query=pg query fetch= all ```

Skip Breidbach · Answer

Got it < nicholas> < Hamza Ahmed> and I had tried something similar to the 2nd option but without the run we tried instantiating a `PostgreFetch` task and returning it essentially a class factory method but that gave us an error like you are trying to modify a task outside of the flow or something like that It s a good lesson to us as our takeaway in creating our own Task objects to be sure that only guaranteed static items are included in the constructor Thanks for the flavor and the workarounds

nicholas · Answer

Happy to help < Skip Breidbach> let me know if you hit any other snags

Skip Breidbach · Answer

Thanks that s appreciated We re really liking the platform just hit an occasional subtlety like this slightly smiling face

Hamza Ahmed · Answer

Thank you < nicholas>