Michael Michael
11/12/2024, 5:59 PM@task
def db_connection() -> connection:
try:
connection = db.connect(
user=Secret.load("user").get(),
password=Secret.load("password").get(),
# additional connection params...
)
return connection
except Exception as e:
raise PrefectException(f"DB connection error: {e}")
Next step: we want to spin up a test work pool/worker to trigger these flows in a test environment before they go live in production.
These test flows would also need to connect to separate test databases.
Question: Do we need separate secrets for the test environment in Prefect Cloud, and conditionally load them based on the environment (test vs. production)?
Or is there a more streamlined way to handle this setup?
Anyone tackled this before? Thanks in advance!Nate
11/12/2024, 6:04 PMSecret
block
so for example
{
"test": "sk-XXX",
"prod": "sk-XXX"
}
could be a secret value, and you'd get the actual dict
instance when you call Secret.load('your-secret')