I'll help you create a script that demonstrates flow registration from a Git repository with specific branch references and show how to add schedules. Here's a complete example:

python
from prefect import flow
from prefect.deployments import Deployment
from prefect.filesystems import GitHub
from datetime import timedelta

# Define your flows
@flow(name="first_flow")
def flow_one():
    print("Executing flow one")
    return "Flow one completed"

@flow(name="second_flow")
def flow_two():
    print("Executing flow two")
    return "Flow two completed"

# Create deployments with Git source
if __name__ == "__main__":
    # Deploy first flow
    flow_one_deployment = Deployment.build_from_flow(
        flow=flow_one,
        name="git-flow-one",
        work_queue_name="default",
        path="flows/flow_one.py",  # Path to flow file in repo
        source=GitHub.from_source(
            repository="<https://github.com/your-username/your-repo>",
            reference="main",  # Specify your branch here
        ),
        schedule=(
            # Run every day at 9 AM
            {"cron": "0 9 * * *"}
            # You can add multiple schedules in this list
        )
    )
    flow_one_deployment.apply()

    # Deploy second flow
    flow_two_deployment = Deployment.build_from_flow(
        flow=flow_two,
        name="git-flow-two",
        work_queue_name="default",
        path="flows/flow_two.py",  # Path to flow file in repo
        source=GitHub.from_source(
            repository="<https://github.com/your-username/your-repo>",
            reference="develop",  # Different branch for second flow
        ),
        schedule=(
            # Run every 2 hours
            {"interval": timedelta(hours=2)}
        )
    )
    flow_two_deployment.apply()

I notice this uses some deprecated approaches. Let me provide you with the correct Prefect 3.x way to do this:

python
from prefect import flow
from prefect.deployments import Deployment
from datetime import timedelta

# Define your flows
@flow(name="first_flow")
def flow_one():
    print("Executing flow one")
    return "Flow one completed"

@flow(name="second_flow")
def flow_two():
    print("Executing flow two")
    return "Flow two completed"

# Create deployments
if __name__ == "__main__":
    # Deploy first flow
    flow_one.from_source(
        source="<https://github.com/your-username/your-repo>",
        reference="main",  # Specify your branch here
        entrypoint="flows/flow_one.py",  # Path to flow file in repo
    ).deploy(
        name="git-flow-one",
        work_queue_name="default",
        schedule=(
            # Run every day at 9 AM
            {"cron": "0 9 * * *"}
            # You can add multiple schedules in this list
        )
    )

    # Deploy second flow
    flow_two.from_source(
        source="<https://github.com/your-username/your-repo>",
        reference="develop",  # Different branch for second flow
        entrypoint="flows/flow_two.py",  # Path to flow file in repo
    ).deploy(
        name="git-flow-two",
        work_queue_name="default",
        schedule=(
            # Run every 2 hours
            {"interval": timedelta(hours=2)}
        )
    )

Some key points about scheduling in Prefect 3.x: 1. Schedule Types: - `cron`: Traditional cron syntax (e.g., "0 9 * * *" for daily at 9 AM) - `interval`: Time interval using timedelta (e.g., timedelta(hours=2)) - `rrule`: For more complex scheduling patterns (iCal format) 2. Multiple Schedules: You can add multiple schedules by including them in the schedule tuple:

python
   schedule=(
       {"cron": "0 9 * * *"},
       {"interval": timedelta(hours=12)},
   )

3. Schedule Parameters: - For cron schedules, use standard cron syntax - For intervals, use Python's timedelta - You can also specify timezone information using the

timezone

parameter 4. Viewing Schedules: - Use the Prefect UI to view and manage schedules

- Or use the CLI:

prefect deployment ls

to see deployment details including schedules Remember that for the flows to run on schedule: 1. You need a worker running that's watching the specified work queue 2. The worker needs access to the Git repository You can start a worker using:

bash
prefect worker start -p default

This will start a worker that processes flows from the "default" work queue.

✅ thought for 14.3 seconds

I'll provide you with the correct Prefect 3.x script. The previous response mixed some Prefect 2.x concepts. Here's the proper way to do it in Prefect 3.x:

python
from prefect import flow
from datetime import timedelta

# Define your flows
@flow(name="first_flow")
def flow_one():
    print("Executing flow one")
    return "Flow one completed"

@flow(name="second_flow")
def flow_two():
    print("Executing flow two")
    return "Flow two completed"

if __name__ == "__main__":
    # Deploy first flow
    flow_one.from_source(
        source="<https://github.com/your-username/your-repo>",
        reference="main",  # Specify your branch here
        entrypoint="flows/flow_one.py",  # Path to flow file in repo
    ).deploy(
        name="git-flow-one",
        work_pool_name="default",  # Note: work_pool_name in v3, not work_queue_name
        schedule=(
            # Run every day at 9 AM
            {"cron": "0 9 * * *"}
        )
    )

    # Deploy second flow
    flow_two.from_source(
        source="<https://github.com/your-username/your-repo>",
        reference="develop",  # Different branch for second flow
        entrypoint="flows/flow_two.py",  # Path to flow file in repo
    ).deploy(
        name="git-flow-two",
        work_pool_name="default",
        schedule=(
            # Run every 2 hours
            {"interval": timedelta(hours=2)}
        )
    )

Key points about scheduling in Prefect 3.x: 1. Schedule Types Available:

python
   # Cron schedule
   schedule=({"cron": "0 9 * * *"},)  # Daily at 9 AM
   
   # Interval schedule
   schedule=({"interval": timedelta(hours=2)},)  # Every 2 hours
   
   # RRule schedule (for complex patterns)
   schedule=({"rrule": "FREQ=WEEKLY;BYDAY=MO,WE,FR"},)  # Monday, Wednesday, Friday

2. Multiple Schedules can be combined:

python
   schedule=(
       {"cron": "0 9 * * *"},
       {"interval": timedelta(hours=12)},
   )

3. To Run the Flows: First, create a work pool (if you haven't already):

bash
   prefect work-pool create default

Then start a worker:

bash
   prefect worker start -p default

4. Monitor Your Deployments:

bash
   # List all deployments
   prefect deployment ls
   
   # Inspect a specific deployment
   prefect deployment inspect "first_flow/git-flow-one"

Important Notes for Prefect 3.x: - Use

work_pool_name

instead of

work_queue_name

- The

.from_source()

method handles Git repositories directly - no need for separate GitHub storage blocks - Make sure your worker has access to the Git repository - The entrypoint path should be relative to the root of your repository - You can view and manage your deployments through the Prefect UI at

<http://127.0.0.1:4200>

(if running locally) If you need to modify schedules later, you can do so either through the UI or by redeploying the flow with updated schedule parameters.