## Manage workflow schedules

URL: https://docs.atlan.com/product/capabilities/build-apps/sdks/python/workflows/how-tos/manage-schedules

> Manage and schedule recurring workflow runs programmatically using the Atlan Python SDK (pyatlan) and WorkflowSchedule API.

# WorkflowSchedule: schedule and manage workflow runs

Use `WorkflowSchedule` in the Atlan Python SDK to programmatically schedule and manage recurring workflow runs.

## Schedule workflow run

### Directly on run:

You can directly add a schedule to a workflow run.
For example, with [Snowflake Miner](https://docs.atlan.com/llms/connectors/snowflake/miner-reference/llms.txt):

### Java

:::warning[Coming soon]
:::

### Python

```python showLineNumbers title="Add schedule directly on run"
from pyatlan.client.atlan import AtlanClient
from pyatlan.model.packages import SnowflakeMiner
from pyatlan.model.workflow import WorkflowSchedule

client = AtlanClient()

miner = ( # (1)
 SnowflakeMiner(
 connection_qualified_name="default/snowflake/1234567890"
 )
 .s3(
 s3_bucket="test-s3-bucket",
 s3_prefix="test-s3-prefix",
 s3_bucket_region="test-s3-bucket-region",
 sql_query_key="TEST_QUERY",
 default_database_key="TEST_SNOWFLAKE",
 default_schema_key="TEST_SCHEMA",
 session_id_key="TEST_SESSION_ID",
 )
 .popularity_window(days=15)
 .native_lineage(enabled=True)
 .custom_config(config={"test": True, "feature": 1234})
 .to_workflow()
)

schedule = WorkflowSchedule(
 cron_schedule="45 5 * * *",
 timezone="Europe/Paris",
) # (2)

response = client.workflow.run(
 workflow=miner, workflow_schedule=schedule
) # (3)
```

1. Begin by constructing the Snowflake miner workflow.
2. To create a new schedule for the workflow, specify:

 - [`cron schedule expression`](https://crontab.guru),
 for example: `45 5 * * *` (scheduled for tomorrow at `04:05:00`).
 - time zone for the cron schedule, such as `Europe/Paris`.

3. Finally, use the `client.workflow.run()` method to add this new schedule.
It will both add the schedule and execute the workflow in Atlan.

### Kotlin

:::warning[Coming soon]
:::

### Go

```go showLineNumbers title="Add schedule directly on run"
miner := assets.NewSnowflakeMiner("default/snowflake/1234567890"). // (1)
 S3(
 "test-s3-bucket",
 "test-s3-prefix",
 "TEST_QUERY",
 "TEST_SNOWFLAKE",
 "TEST_SCHEMA",
 "TEST_SESSION_ID",
 structs.StringPtr("test-s3-bucket-region"),
 ).
 PopularityWindow(30).
 NativeLineage(true).
 CustomConfig(map[string]interface{}{
 "test": true,
 "feature": 1234,
 }).
 ToWorkflow()

Schedule := structs.WorkflowSchedule{CronSchedule: "45 5 * * *", Timezone: "Europe/Paris"} // (2)

response, atlanErr := ctx.WorkflowClient.Run(miner, &Schedule) // (3)
if atlanErr != nil {
 logger.Log.Errorf("Error : %v", atlanErr)
}
```

1. Begin by constructing the Snowflake miner workflow.
2. To create a new schedule for the workflow, specify:

 - [`cron schedule expression`](https://crontab.guru),
 for example: `45 5 * * *` (scheduled for tomorrow at `04:05:00`).
 - time zone for the cron schedule, such as `Europe/Paris`.

3. Finally, use the `ctx.WorkflowClient.Run()` method to add this new schedule.
It will both add the schedule and execute the workflow in Atlan.

### Existing workflow:

You can also add a schedule to an existing workflow run:

### Java

:::warning[Coming soon]
:::

### Python

```python showLineNumbers title="Schedule an existing workflow run"
from pyatlan.client.atlan import AtlanClient
from pyatlan.model.workflow import WorkflowSchedule

client = AtlanClient()

existing_workflow = client.workflow.find_by_type(
 prefix=WorkflowPackage.SNOWFLAKE_MINER
)[0] # (1)

schedule = WorkflowSchedule(
 cron_schedule="45 5 * * *",
 timezone="Europe/Paris",
) # (2)

response = client.workflow.add_schedule(
 workflow=existing_workflow, workflow_schedule=schedule
) # (3)
```

1. You can retrieve workflows based on their type (prefix).
`Note:` Only workflows that have been run will be found.
2. To create a new schedule for an existing workflow, provide:

 - [`cron schedule expression`](https://crontab.guru), e.g: `45 5 * * *` (tomorrow at `04:05:00`).
 - time zone for the cron schedule, e.g: `Europe/Paris`.

3. Finally, to apply this new schedule to the existing workflow,
use `client.workflow.add_schedule()` method.

### Kotlin

:::warning[Coming soon]
:::

### Go

```go showLineNumbers title="Schedule an existing workflow run"
existingWorkflow, _ := ctx.WorkflowClient.FindByType(atlan.WorkflowPackageSnowflakeMiner, 1) // (1)
Schedule := structs.WorkflowSchedule{CronSchedule: "45 5 * * *", Timezone: "Europe/Paris"} // (2)
response, atlanErr := ctx.WorkflowClient.AddSchedule(existingWorkflow[0], &Schedule)
```

1. You can retrieve workflows based on their type (prefix). (You can also specify the **maximum number of resulting
workflows** you want to retrieve as results.)
`Note:` Only workflows that have been run will be found.

2. To create a new schedule for an existing workflow, provide:

 - [`cron schedule expression`](https://crontab.guru), e.g: `45 5 * * *` (tomorrow at `04:05:00`).
 - time zone for the cron schedule, e.g: `Europe/Paris`.

3. Finally, to apply this new schedule to the existing workflow,
use `ctx.WorkflowClient.AddSchedule()` method.
`Note:` The AddSchedule() method returns an error in case of failure, manage it accordingly by returning it.

## Retrieve scheduled workflow run

### Retrieve by name:

To retrieve an existing scheduled workflow run by its name:

### Java

:::warning[Coming soon]
:::

### Python

```python showLineNumbers title="Retrieve scheduled workflow run by its name"
from pyatlan.client.atlan import AtlanClient

client = AtlanClient()

response = client.workflow.get_scheduled_run(
 workflow_name="atlan-snowflake-miner-1714638976"
) # (1)
```

1. To retrieve an existing scheduled workflow runs, specify:

 - name of the workflow as displayed in the
 UI (e.g: `atlan-snowflake-miner-1714638976`).

### Go

```go showLineNumbers title="Retrieve scheduled workflow run by its name"
response, atlanErr := ctx.WorkflowClient.GetScheduledRun("atlan-snowflake-miner-1714638976") // (1)
if atlanErr != nil {
 logger.Log.Errorf("Error : %v", atlanErr)
}
```

1. To retrieve an existing scheduled workflow runs, specify:

 - name of the workflow as displayed in the
 UI (e.g: `atlan-snowflake-miner-1714638976`).

### Retrieve all scheduled runs:

To retrieve all existing scheduled workflow runs:

### Java

:::warning[Coming soon]
:::

### Python

```python showLineNumbers title="Retrieve all scheduled workflow runs"
from pyatlan.client.atlan import AtlanClient

client = AtlanClient()

response = client.workflow.get_all_scheduled_runs() # (1)
```

1. To retrieve all scheduled runs for workflows,
use `client.workflow.get_all_scheduled_runs()` method.

### Kotlin

:::warning[Coming soon]
:::

### Go

```go showLineNumbers title="Retrieve all scheduled workflow runs"
response, atlanErr := ctx.WorkflowClient.GetAllScheduledRuns() // (1)
```

1. To retrieve all scheduled runs for workflows,
use `ctx.WorkflowClient.GetAllScheduledRuns()` method.

## Remove schedule from workflow run

To remove a schedule from an existing workflow run:

### Java

:::warning[Coming soon]
:::

### Python

```python showLineNumbers title="Remove a schedule from an existing workflow run"
from pyatlan.client.atlan import AtlanClient

client = AtlanClient()

existing_workflow = client.workflow.find_by_type(
 prefix=WorkflowPackage.SNOWFLAKE_MINER
)[0] # (1)

response = client.workflow.remove_schedule(
 workflow=existing_workflow
) # (2)
```

1. You can retrieve workflows based on their type (prefix).
`Note:` Only workflows that have been run will be found.

2. Finally, to remove a schedule from this workflow,
use the `client.workflow.remove_schedule()` method.

### Kotlin

:::warning[Coming soon]
:::

### Go

```go showLineNumbers title="Remove a schedule from an existing workflow run"
existingWorkflow, _ := ctx.WorkflowClient.FindByType(atlan.WorkflowPackageSnowflakeMiner, 1) // (1)
response, atlanErr := ctx.WorkflowClient.RemoveSchedule(existingWorkflow[0]) // (2)
```

1. You can retrieve workflows based on their type (prefix). (You can also specify the **maximum number of resulting
workflows** you want to retrieve as results.)
`Note:` Only workflows that have been run will be found.

2. Finally, to remove a schedule from this workflow,
use the `ctx.WorkflowClient.RemoveSchedule()` method.

---