< Marvin> can you give me a sample code how can i use the pr Prefect Community #marvin-ai

<@ULVA73B9P> can you give me a sample code, how ca...

Kartik

10/23/2025, 5:21 PM

@Marvin can you give me a sample code, how can i use the prefect.yaml to create a deployment for a flow which runs function execute() , the schedule is attached to the prefect.yaml such that the sample prefect.yaml for the deployment would look 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: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 and on executing ; prefect deploy -n '[Tarinika-Inc(US Store)]: Amazon Transactions Breakdown Transformations' it should create the deployment in the prefect UI

Marvin

10/23/2025, 5:21 PM

✅ thought for 350.9 seconds

Marvin

10/23/2025, 5:27 PM

Great question! Here’s a minimal, working example for Prefect 3.x that: - defines a flow that runs your

execute()

function - configures a deployment in

prefect.yaml

with a schedule - lets you create the deployment with

prefect deploy -n '...'

- notes how to create/start a local work pool so the deployment can run Project layout:

Copy code

.
├── flows
│   └── amazon_transactions.py
└── prefect.yaml

flows/amazon_transactions.py

Copy code

from prefect import flow, task, get_run_logger

@task
def execute(database: str, store_timezone_offset_minutes: int = 330):
    logger = get_run_logger()
    <http://logger.info|logger.info>(
        f"Executing transformation for database={database}, "
        f"store_timezone_offset_minutes={store_timezone_offset_minutes}"
    )
    # TODO: your transformation logic here
    return "ok"

@flow
def transform_amazon_transactions(
    database: str = "tarinika-inc",
    store_timezone_offset_minutes: int = 330,
):
    return execute(database, store_timezone_offset_minutes)

prefect.yaml

Copy code

name: tarinika-amazon-transformations
prefect-version: ">=3.0.0"

deployments:
- name: "[Tarinika-Inc(US Store)]: Amazon Transactions Breakdown Transformations"
  description: |-
    This runs the python transformation for the amazon transactions
    Args:
      database: Database name
      store_timezone_offset_minutes: Timezone offset (default: 330)
  tags: ["marketplaces", "tarinika-inc", "dev", "amazon"]

  # IMPORTANT: Use a file path to your flow file, then colon, then the flow function name
  # e.g., ./flows/amazon_transactions.py:transform_amazon_transactions
  entrypoint: flows/amazon_transactions.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 is a LIST; each item can be an interval, cron, or rrule schedule
  schedules:
  - interval: 1800          # seconds (every 30 minutes)
    anchor_date: "2025-09-09T08:30:20Z"
    timezone: "UTC"
    active: false           # set true to start creating runs

Then, from the project root, create the deployment:

Copy code

prefect deploy -n "[Tarinika-Inc(US Store)]: Amazon Transactions Breakdown Transformations"

If you haven’t already, create and start a local work pool so the deployment has a place to run:

Copy code

prefect work-pool create "local-work-pool" --type process
prefect worker start -p local-work-pool

Notes and tips: - In Prefect 3.x, schedules go under

schedules

(plural) and each item may define exactly one of

interval

cron

, or

rrule

. -

interval

is in seconds (integer).

anchor_date

should be an ISO 8601 datetime.

active: true/false

controls whether new runs are created. - The

entrypoint

should be a file path followed by

:flow_function_name

. Your flow function must be decorated with

@flow

. - You can omit

anchor_date

if you don’t need to pin the schedule start. Docs for reference: - Schedules: Prefect 3.x schedules - prefect.yaml and deployments: prefect.yaml reference and create schedules in deployments If you want me to adapt this exactly to your module layout (e.g., using your

flows.amazon_transactions_breakdown_transformation:transform_amazon_transactions

naming), share your file path and I’ll tailor the

entrypoint

and code accordingly.

Open in Slack

Previous Next