https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> i am trying to create the deployment ...
# marvin-ai
k
@Marvin i am trying to create the deployment using the prefect.yaml which looks like this - name: '[Tarinika-Inc(US Store)]: Amazon Transactions Breakdown Transformations' version: null tags: ['marketplaces', 'tarinika-inc', 'dev', 'amazon'] concurrency_limit: null description: |- This runs the python transformation for the amazon transactions Args: database: Database name store_timezone_offset_minutes: Timezone offset (default: "330") entrypoint: flows/amazon_transactions_breakdown_transformation.py:transform_amazon_transactions parameters: database: "tarinika-inc" store_timezone_offset_minutes: "330" work_pool: name: local-work-pool work_queue_name: null job_variables: {} schedules: - interval: 1800.0 anchor_date: '2025-09-09T083020.631532+00:00' timezone: UTC active: false the entry point file which is supposed to create the deployment of the flow is ths from prefect import flow, get_run_logger from prefect_dbt import PrefectDbtRunner import os from prefect.blocks.system import Secret from dotenv import load_dotenv from helpers.env_loader import load_env_vars_dynamic from post_dbt_transformations.marketplaces.amazon.process_transactions import transform_items @flow def transform_amazon_transactions( database: str, store_timezone_offset_minutes: str = "330" ): """ This runs the python transformation for the amazon transactions Args: database: Database name _store_timezone_offset_minutes: Timezone offset (default: "330")_ """ logger = get_run_logger() # Load the environment variables for the deployment -- THE ENVIRONMENT HAS BEEN HARDCODED FOR NOW JUST TO TEST THIS WORKFLOW _# load_env_vars_dynamic(_ # database=database, _# store_timezone_offset_minutes=store_timezone_offset_minutes_ # ) logger.info(f"Prefect-Dbt Environment is Ready For Database: {database}") logger.info(f"Running For Database: {database}") try: logger.info("Starting 'process_transaction.py' function...") result = transform_items() logger.info("run completed successfully") return result except Exception as e: logger.error(f"run failed: {str(e)}") raise # GOOD: This block will NOT run during deployment if name == "__main__": transform_amazon_transactions( database="tarinika-inc", store_timezone_offset_minutes="330" ) now when i am deploying this prefect deploy -n '[Tarinika-Inc(US Store)]: Amazon Transactions Breakdown Transformations' instead of creating the deployment in the UI it is runnning the flow instantly why and how can i correct this,
m
✅ thought for 296.4 seconds
Short answer: your flow is being executed because your entrypoint file is being executed as a script during deployment. When 
prefect deploy
loads an entrypoint given as a file path, it executes that file in a way that sets 
__name__ == "__main__"
, which triggers your 
if __name__ == "__main__":
block and runs the flow. Why this happens - 
prefect deploy
must import your entrypoint to build the deployment. - With a file path entrypoint like 
flows/amazon_transactions_breakdown_transformation.py:transform_amazon_transactions
, Prefect executes the file in a mode where 
__name__ == "__main__"
. - That means your “GOOD: This block will NOT run during deployment” block actually does run, calling 
transform_amazon_transactions(...)
immediately. How to fix it You have a few good options. Pick one: 1) Use a module path entrypoint instead of a file path - Make sure your project is importable (add 
__init__.py
to the 
flows/
folder and ensure your project root is on 
PYTHONPATH
). - Change the entrypoint in your prefect.yaml to a module path:
Copy code
entrypoint: flows.amazon_transactions_breakdown_transformation:transform_amazon_transactions
- Then redeploy:
Copy code
prefect deploy -n '[Tarinika-Inc(US Store)]: Amazon Transactions Breakdown Transformations'
- This way, Prefect will import the module normally and 
__name__
will not be 
"__main__"
. 2) Remove the main block from the flow module - Delete the block that calls the flow at the bottom of `amazon_transactions_breakdown_transformation.py`:
Copy code
if __name__ == "__main__":
      transform_amazon_transactions(
          database="tarinika-inc",
          store_timezone_offset_minutes="330"
      )
- If you want a quick local runner, put that into a separate script (e.g., 
scripts/dev_run_amazon.py
) that imports and invokes the flow instead. 3) Ensure nothing runs on import in any imported modules - Because 
prefect deploy
imports your entrypoint, any side-effects in imported modules will also run. Double-check that 
post_dbt_transformations/marketplaces/amazon/process_transactions.py
doesn’t do any work at import time (e.g., running transforms) and that the flow is only executed when explicitly called. A couple of additional notes - Your schedule is set to 
active: false
, so even after deployment it won’t schedule runs until you set it to 
true
or create a schedule via the UI/CLI. - Make sure a worker is running for 
local-work-pool
to actually pick up scheduled or manual runs. Docs for reference - Prefect YAML deployments: prefect.yaml deployments - Creating deployments: Create deployments If you switch to the module path entrypoint and still see it running during 
prefect deploy
, please share: - The updated 
prefect.yaml
entrypoint value - Whether there’s any top-level code in 
process_transactions.py
or other imports that could be doing work on import - Your Prefect version (
prefect version
)
Open in Slack
PreviousNext
Powered by