Adam
09/17/2020, 9:32 AMPostgresFetch
. It seems to require user
and host
arguments when declaring the task and then password
can be supplied when instantiating it. How can I send all when I instantiate it rather? I’m trying to achieve something like this:
get_customers = PostgresFetch(
name="Get customers ready for migration",
db_name="postgres",
user="CAN_I_USE_ENV_SECRET_HERE?",
host="CAN_I_USE_ENV_SECRET_HERE?",
query="""
SELECT DISTINCT customer_id
FROM customer_migration_request
""",
fetch="all",
)
with Flow("migrate customers") as flow:
postgres_user = EnvVarSecret("POSTGRES_USER", raise_if_missing=True)
postgres_password = EnvVarSecret("POSTGRES_PASSWORD", raise_if_missing=True)
postgres_host = EnvVarSecret("POSTGRES_HOST", raise_if_missing=True)
customers = get_customers(
password=postgres_password, user=postgres_user, host=postgres_host
)
get_customers
Dylan
09/17/2020, 1:52 PMget_customers
inside your flow context so your secrets are defined:
with Flow("migrate customers") as flow:
postgres_user = EnvVarSecret("POSTGRES_USER", raise_if_missing=True)
postgres_password = EnvVarSecret("POSTGRES_PASSWORD", raise_if_missing=True)
postgres_host = EnvVarSecret("POSTGRES_HOST", raise_if_missing=True)
customers = PostgresFetch(
name="Get customers ready for migration",
db_name="postgres",
user=postgres_user,
host=postgres_host,
password=postgres_password,
query="""
SELECT DISTINCT customer_id
FROM customer_migration_request
""",
fetch="all",
)()
postgres_user = EnvVarSecret("POSTGRES_USER", raise_if_missing=True)
postgres_password = EnvVarSecret("POSTGRES_PASSWORD", raise_if_missing=True)
postgres_host = EnvVarSecret("POSTGRES_HOST", raise_if_missing=True)
get_customers = PostgresFetch(
name="Get customers ready for migration",
db_name="postgres",
user=postgres_user,
host=postgres_host,
password=postgres_password,
query="""
SELECT DISTINCT customer_id
FROM customer_migration_request
""",
fetch="all",
)
with Flow("migrate customers") as flow:
customers = get_customers()
Adam
09/17/2020, 4:10 PMDylan
09/17/2020, 4:26 PMAdam
09/18/2020, 11:46 AMPostgresFetch
task is not handling the EnvVarSecret
task properly. It seems to stringify it:
[2020-09-18 11:45:38] ERROR - prefect.TaskRunner | Unexpected error: OperationalError('could not translate host name "<Task: POSTGRES_HOST>" to address: nodename nor servname provided, or not known\n')