Jon
10/10/2022, 2:51 PMAttributeError: 'Parameter' object ...
. Why is the flow trying to run when I register it?
the more unclear thing is that when i declare a parameter, it returns a parameter object when i really just want the value that's passed to the flowKalise Richmond
10/10/2022, 3:12 PMJon
10/10/2022, 3:32 PMprefect register --project platform_medallion-etl--dev -p flows/cbe/electricity_ami/extract.py
i get this exception
Collecting flows...
[2022-10-10 11:30:40-0400] INFO - prefect | Getting credentials for provider_code: cbe client_code: <Parameter: clientCode>
Error loading 'flows/cbe/electricity_ami/extract.py':
Traceback (most recent call last):
File "/Users/jonyoung/Dev/workflows/.venv/lib/python3.9/site-packages/prefect/cli/build_register.py", line 133, in load_flows_from_script
namespace = runpy.run_path(abs_path, run_name="<flow>")
File "/Users/jonyoung/.pyenv/versions/3.9.13/lib/python3.9/runpy.py", line 288, in run_path
return _run_module_code(code, init_globals, run_name,
File "/Users/jonyoung/.pyenv/versions/3.9.13/lib/python3.9/runpy.py", line 97, in _run_module_code
_run_code(code, mod_globals, init_globals,
File "/Users/jonyoung/.pyenv/versions/3.9.13/lib/python3.9/runpy.py", line 87, in _run_code
exec(code, run_globals)
File "/Users/jonyoung/Dev/workflows/workflows/flows/cbe/electricity_ami/extract.py", line 34, in <module>
auth_payload = provider.get_login_credentials(
File "/Users/jonyoung/Dev/workflows/workflows/models/DataProvider.py", line 58, in get_login_credentials
credentials_info = RuntimeParameters().get_ingest_secret(
File "/Users/jonyoung/Dev/workflows/workflows/models/RuntimeParameters.py", line 100, in get_ingest_secret
customer = customer_code.lower()
AttributeError: 'Parameter' object has no attribute 'lower'
and the code looks something like this
import prefect
from common import workflow_utils, logging
from models import TempFileModel
from flows import constants
from tasks import register_workflow
from tasks.cbe import extract
logger = logging.get_logger()
# --------------------------------------------------
# Main Flow Configuration
# --------------------------------------------------
FLOW_NAME = "TEST"
with prefect.Flow(
FLOW_NAME,
run_config=prefect.run_configs.UniversalRun(),
terminal_state_handler=workflow_utils.workflow_terminal_state_handler,
) as flow:
client_code = prefect.Parameter("clientCode", required=True)
with extract.ExtractCBE() as provider:
auth_payload = provider.get_login_credentials(
extract.ExtractCBE.PROVIDER_CODE,
client_code,
)
flow.run(
parameters=dict(
clientCode="test",
)
)
# --------------------------------------------------
# Main
# --------------------------------------------------
if __name__ == "__main__":
pass
clientCode
?Kalise Richmond
10/10/2022, 3:53 PMJon
10/10/2022, 3:54 PMBianca Hoch
10/10/2022, 4:24 PM