BrickFlow

https://engineering.nike.com/brickflow/latest/
Admin
BrickFlow is a CLI tool for developing and deploying Python-based Databricks workflows in a...
Tokens:126,698
Snippets:641
Trust Score:10
Update:7 hours ago
Show doc for...
Tokens:
Raw
### Initialize a new BrickFlow project

Source: https://engineering.nike.com/brickflow/latest/bundles-quickstart/

Run `bf projects add` to start the interactive project setup. Follow the prompts to configure project name, paths, Git URL, and versions.

```bash
bf projects add
```

--------------------------------

### Entrypoint example with brickflow import

Source: https://engineering.nike.com/brickflow/latest/upgrades/upgrade-pre-0-10-0-to-0-10-0/

Example entrypoint.py showing the required import order: brickflow at the top, then your modules. Includes a Project context with plugins enabled.

```python
# Databricks notebook source

# COMMAND ----------

from brickflow import Project # (1)!

import workflows # (2)!

def main() -> None:
    """Project entrypoint"""
    with Project(
        "product_abc_workflows_2",
        git_repo="https://github.com/stikkireddy/mono-repo-test",
        provider="github",
        libraries=[  # (3)!
            # PypiTaskLibrary(package="spark-expectations==0.5.0"), # Uncomment if spark-expectations is needed
        ],
        enable_plugins=True, # (4)!
    ) as f:
        f.add_pkg(workflows)


if __name__ == "__main__":
    main()
```

--------------------------------

### Install and configure Databricks CLI

Source: https://engineering.nike.com/brickflow/latest/bundles-quickstart/

Install the Databricks CLI and configure it with a token. This is a prerequisite for using BrickFlow.

```bash
pip install databricks-cli
databricks configure -t
```

--------------------------------

### Install brickflows and check version

Source: https://engineering.nike.com/brickflow/latest/upgrades/upgrade-pre-0-10-0-to-0-10-0/

Install the new brickflows package (version 0.10.0 or later) and verify the installation with bf --version.

```bash
pip install brickflows>=0.10.0
bf --version
```

--------------------------------

### Install BrickFlow CLI

Source: https://engineering.nike.com/brickflow/latest/bundles-quickstart/

Install the BrickFlow CLI using pip.

```bash
pip install brickflows
```

--------------------------------

### Install task-specific libraries

Source: https://engineering.nike.com/brickflow/latest/tasks/

Shows how to install libraries specific to a task using the libraries parameter of the @wf.task decorator. Refer to the workflows section for implementation details.

```python
from brickflow import Workflow
wf = Workflow(...)

@wf.task(libraries=[...])  # (1)!
def custom_libraries():
    pass

```

--------------------------------

### Install optional plugin extras with pip

Source: https://engineering.nike.com/brickflow/latest/upgrades/upgrade-airflow-free/

Use these pip commands to install optional plugin backends (Snowflake, Tableau, Box, Cerberus) locally. The 'all-plugins' extra installs all of them. At runtime on a Databricks cluster, attach the libraries via 'enable_plugins=True' on your Project or manually via 'PypiTaskLibrary'.

```bash
pip install "brickflows[snowflake]"
pip install "brickflows[tableau]"
pip install "brickflows[box]"
pip install "brickflows[cerberus]"
# or, all of them:
pip install "brickflows[all-plugins]"
```

--------------------------------

### JAR Task examples

Source: https://engineering.nike.com/brickflow/latest/tasks/

Shows three examples of defining a JAR Task: using @wf.spark_jar_task with a dbfs jar, with an s3 jar, and using @wf.task with TaskType.SPARK_JAR_TASK.

```python
# Example:1
@wf.spark_jar_task(
    libraries=[
        JarTaskLibrary(
            jar="dbfs:/Volumes/development/global_sustainability_dev/raju_spark_jar_test/PrintArgs.jar"
        )
    ]
)
def spark_jar_task_a():
    return SparkJarTask(
        main_class_name="PrintArgs",
        parameters=["Hello", "World!"],
    ) 
# Example: 2
@wf.spark_jar_task(
    libraries=[
        JarTaskLibrary(
            jar="s3:/Volumes/development/global_sustainability_dev/raju_spark_jar_test/PrintArgs.jar"
        )
    ]
)
def spark_jar_task_test():
    return SparkJarTask(
        main_class_name="PrintArgs",
    )
# Example: 3
# we can also pass task type as parameter
@wf.task(task_type=TaskType.SPARK_JAR_TASK, libraries=[
        JarTaskLibrary(
            jar="dbfs:/Volumes/development/global_sustainability_dev/raju_spark_jar_test/PrintArgs.jar"
        )])
def run_job_task_a():
    return SparkJarTask(main_class_name="MainClass")
```

--------------------------------

### Polyrepo project structure

Source: https://engineering.nike.com/brickflow/latest/bundles-quickstart/

Example directory layout for a polyrepo-style BrickFlow project.

```text
repo-root/
├── .git
├── src/
│   ├── lib/
│   │   ├── __init__.py
│   │   └── shared_functions.py
│   ├── workflows/
│   │   ├── __init__.py
│   │   ├── entrypoint.py
│   │   └── workflow.py
├── .brickflow-project-root.yml
```

--------------------------------

### Monorepo project structure

Source: https://engineering.nike.com/brickflow/latest/bundles-quickstart/

Example directory layout for a monorepo-style BrickFlow project.

```text
repo-root/
├── .git
├── projects/
│   ├── project_abc/
│   │   ├── lib/
│   │   │   ├── __init__.py
│   │   │   └── shared_functions.py
│   │   ├── workflows/
│   │   │   ├── __init__.py
│   │   │   ├── entrypoint.py
│   │   │   └── workflow_abc.py
│   │   └── .brickflow-project-root.yml
```

--------------------------------

### Conditional Run Example with TaskRunCondition.AT_LEAST_ONE_FAILED

Source: https://engineering.nike.com/brickflow/latest/tasks/

Shows how to set a run condition on a task using TaskSettings with run_if. The example uses AT_LEAST_ONE_FAILED, which runs the task if at least one dependency failed. Other options include ALL_SUCCESS, AT_LEAST_ONE_SUCCESS, NONE_FAILED, ALL_DONE, and ALL_FAILED.

```python
from brickflow import Workflow, TaskRunCondition, TaskSettings
wf = Workflow(...)

@wf.task(
   task_settings=TaskSettings(run_if=TaskRunCondition.AT_LEAST_ONE_FAILED)
)
def none_failed_task():
   pass
```

--------------------------------

### Run Job Task examples

Source: https://engineering.nike.com/brickflow/latest/tasks/

Shows three ways to define a Run Job Task: using @wf.run_job_task, using @wf.task with TaskType.RUN_JOB_TASK, and referencing another workflow in the same project.

```python
from brickflow import RunJobTask

@wf.run_job_task
def run_job_task_a():
    return RunJobTask(job_name="run_job_task")

# we can also pass task type as parameter
@wf.task(task_type=TaskType.RUN_JOB_TASK)
def run_job_task_a():
    return RunJobTask(job_name="run_job_task")

# reference the job in the same Brickflow project
@wf.run_job_task
def run_job_task_a():
    from workflows.workflow_a import wf
    return wf
```

--------------------------------

### Notebook Task example

Source: https://engineering.nike.com/brickflow/latest/tasks/

Shows how to define a Notebook Task using the @wf.notebook_task decorator. The task returns a NotebookTask instance with a notebook path and base parameters accessible via dbutils widgets.

```python
@wf.task
def notebook_task():
   pass

@wf.notebook_task
# this task runs a databricks notebook
def example_notebook():
    return NotebookTask(
        notebook_path="notebooks/example_notebook.py",
        base_parameters={
            "some_parameter": "some_value",  # in the notebook access these via dbutils.widgets.get("some_parameter")
        },
    )
```

--------------------------------

### Use different task types (notebook and DLT)

Source: https://engineering.nike.com/brickflow/latest/tasks/

Shows how to specify a task type using the task_type parameter. The default task type is NOTEBOOK. The example also demonstrates using TaskType.DLT for a DLT task.

```python
from brickflow import Workflow, TaskType, BrickflowTriggerRule, TaskResponse
wf = Workflow(...)

@wf.task
def notebook_task():
   pass

@wf.task(task_type=TaskType.DLT)
def dlt_task():
    pass

```

--------------------------------

### SQL Task Examples with @wf.sql_task

Source: https://engineering.nike.com/brickflow/latest/tasks/

Shows multiple SQL task definitions using the @wf.sql_task decorator: query ID, file path, alert, dashboard, and using task_type parameter. Each returns a SqlTask instance with required warehouse_id and optional parameters.

```python
@wf.sql_task
def sample_sql_task_query():
    return SqlTask(
        query_id="your_sql_query_id", warehouse_id="your_warehouse_id"
    )


@wf.sql_task
def sample_sql_task_file() -> any:
    return SqlTask(file_path="products/brickflow_test/src/sql/sql_task_file_test.sql", warehouse_id="your_warehouse_id")


@wf.sql_task
def sample_sql_alert() -> any:
    return SqlTask(
        alert_id="Your_Alert_ID",
        pause_subscriptions=False,
        subscriptions={
            "usernames": ["YOUR_USERNAME", "YOUR_USERNAME"]
        },
        warehouse_id="your_warehouse_id",
    )


@wf.sql_task
def sample_sql_dashboard() -> any:
    return SqlTask(
        dashboard_id="Your_Dashboard_ID",
        dashboard_custom_subject="Raju Legacy Dashboard Test",
        pause_subscriptions=True,
        subscriptions={
            "usernames": ["YOUR_USERNAME", "YOUR_USERNAME"],
            "destination_id": ["your_destination_id"],
        },
        warehouse_id="your_warehouse_id",
    )

# we can also pass task type as parameter
@wf.task(task_type=TaskType.SQL)
def sample_sql_dashboard_task() -> any:
    return SqlTask(
        dashboard_id="Your_Dashboard_ID",
        dashboard_custom_subject="Raju Legacy Dashboard Test",
        pause_subscriptions=True,
        subscriptions={
            "usernames": ["YOUR_USERNAME", "YOUR_USERNAME"],
            "destination_id": ["your_destination_id"],
        },
        warehouse_id="your_warehouse_id",
    )
```

--------------------------------

### Configure Workflow Notification Settings with Email Notifications

Source: https://engineering.nike.com/brickflow/latest/faq/faq/

Use this snippet to set up a Brickflow Workflow with email notifications for start, success, failure, and duration warning threshold exceeded events. It also configures workflow-level notification settings to suppress alerts for canceled and skipped runs. The example includes optional parameters like schedule, timezone, and pause status.

```python
from datetime import timedelta
from brickflow import (Workflow, Cluster, TaskSettings, EmailNotifications)
from brickflow.engine.workflow import WorkflowNotificationSettings

wf = Workflow(
    "wf_test",
    default_cluster=Cluster.from_existing_cluster("your_existing_cluster_id"),

    # Optional parameters below
    schedule_quartz_expression="0 0/20 0 ? * * *",
    timezone="UTC",
    schedule_pause_status="PAUSED",
    default_task_settings=TaskSettings(
        email_notifications=EmailNotifications(
            on_start=["email@nike.com"],
            on_success=["email@nike.com"],
            on_failure=["email@nike.com"],
            on_duration_warning_threshold_exceeded=["email@nike.com"]
        ),
        timeout_seconds=timedelta(hours=2).seconds
    ),
   notification_settings=WorkflowNotificationSettings(
      no_alert_for_canceled_runs=True,
      no_alert_for_skipped_runs=True
   )
)
```

--------------------------------

### Project entrypoint example (entrypoint.py)

Source: https://engineering.nike.com/brickflow/latest/projects/

Defines the main entrypoint for a Brickflow project. The first line comment treats the file as a Databricks notebook. The main() function creates a Project with name 'brickflow-demo', git repo, provider, and libraries, then adds a package. The if __name__ == '__main__' guard ensures it runs only when executed directly.

```python
# Databricks notebook source  (1)

import examples.brickflow_examples.workflows

from brickflow import Project, PypiTaskLibrary, MavenTaskLibrary


def main() -> None:
    """Project entrypoint"""
    with Project(
            "brickflow-demo",  # (3)!
            git_repo="https://github.com/nike-inc/brickflow",  # (4)!
            provider="github",  # (5)!
            libraries=[  # (6)!
                PypiTaskLibrary(package="networkx"),
            ],
    ) as f:
        f.add_pkg(examples.brickflow_examples.workflows)  # (7)!


if __name__ == "__main__":  # (2)!
    main()
```

--------------------------------

### __post_init__ method for Brickflow Project initialization

Source: https://engineering.nike.com/brickflow/latest/api/project/

Initializes the Project instance: sets logging, mode, entry point, libraries (auto-adding Brickflow libraries if configured), validates project name consistency, sets current project context, and populates bundle and git settings from environment variables. Raises ValueError if project names mismatch or if git_reference is missing when deploying to non-local environments.

```python
def __post_init__(self) -> None:
    # during entry of project enable logging
    _ilog.setLevel(logging.INFO)

    self._mode = Stage[
        config(BrickflowEnvVars.BRICKFLOW_MODE.value, default=Stage.execute.value)
    ]
    self.entry_point_path = self.entry_point_path or get_caller_info()
    deploy_settings = BrickflowProjectDeploymentSettings()
    # setup current_project
    env_project_name = deploy_settings.brickflow_project_name

    self.libraries = self.libraries or []
    if deploy_settings.brickflow_auto_add_libraries is True:
        _ilog.info("Auto adding brickflow libraries...")
        self.libraries = filter_bf_related_libraries(self.libraries)
        # here libraries should not be null anymore if this branch is invoked
        self.libraries += get_brickflow_libraries()

    if (
        env_project_name is not None
        and self.name is not None
        and env_project_name != self.name
    ):
        raise ValueError(
            "Project name in config files and entrypoint must be the same"
        )

    ctx.set_current_project(self.name or env_project_name)  # always setup first

    # populate bundle info via env vars
    self.bundle_obj_name = config(
        BrickflowEnvVars.BRICKFLOW_BUNDLE_OBJ_NAME.value,
        default=".brickflow_bundles",
    )
    self.bundle_base_path = config(
        BrickflowEnvVars.BRICKFLOW_BUNDLE_BASE_PATH.value,
        default="/Users/${workspace.current_user.userName}",
    )

    self.git_reference = config(
        BrickflowEnvVars.BRICKFLOW_GIT_REF.value, default=self.get_git_ref()
    )

    if (
        self._mode == Stage.deploy
        and ctx.is_local() is False
        and self.git_reference is None
    ):
        raise ValueError(
            "git_reference must be set when deploying to non-local envs"
        )

    self.provider = config(
        BrickflowEnvVars.BRICKFLOW_GIT_PROVIDER.value, default=self.provider
    )
    self.git_repo = config(
        BrickflowEnvVars.BRICKFLOW_GIT_REPO.value, default=self.git_repo
    )

    deployment_mode = config(
        BrickflowEnvVars.BRICKFLOW_DEPLOYMENT_MODE.value, default="bundle"
    )
    if deployment_mode == BrickflowDeployMode.BUNDLE.value:
        self.codegen_mechanism = DatabricksBundleCodegen
    if self.codegen_kwargs is None:
        self.codegen_kwargs = {}
```

--------------------------------

### bf CLI options

Source: https://engineering.nike.com/brickflow/latest/cli/reference/

Lists the global options for the bf command: --version and --help.

```bash
  --version  Show the version and exit.
  --help     Show this message and exit.
```

--------------------------------

### Creating if/else tasks with IfElseConditionTask

Source: https://engineering.nike.com/brickflow/latest/tasks/

Shows how to create if/else condition tasks using the @wf.if_else_condition_task decorator. Example 1 creates a simple condition task, Example 2 creates a task that triggers when the parent returns true, Example 3 triggers when the parent returns false, and Example 4 creates a SQL alert task that depends on Example 3. The if_else_outcome parameter controls which parent outcome triggers the task.

```python
# Example 1: creating a if/else task with some params
@wf.if_else_condition_task
def sample_if_else_condition_task():
    return IfElseConditionTask(
        left="value1", right="value2", op="=="
    )
# Let me walk you through how we can make use of if/else condition task. we created a task with name `sample_if_else_condition_task` and it will return either true ot false. Now based on the returned bool, now we're going to decide which task to run. check the below examples.

# Example 2: creating a if/else task that depends on example 1 and this task only triggers if example 1 returns true.
@wf.if_else_condition_task(depends_on="sample_if_else_condition_task", name="new_conditon_task", if_else_outcome={"sample_if_else_condition_task":"true"})
def sample_condition_true():
    return IfElseConditionTask(
        left='{{job.id}}',
        op="==",
        right='{{job.id}}')

'''
 Now i created on more condition task (you can create any task type), since my new task named `new_conditon_task` is dependent on `sample_if_else_condition_task` (if/else task). Now, If my parent tasks runs sucessfully (returns true) then only this task will trigger, cz i mentioned 
if_else_outcome={"sample_if_else_condition_task":"true"}, to seee the false case see the example below.
'''
#Example 3: this task will trigger only when example1 task fails (returns false).
@wf.if_else_condition_task(depends_on="sample_if_else_condition_task", name="example_task_3", if_else_outcome={"sample_if_else_condition_task":"false"})
def sample_condition_false():
    return IfElseConditionTask(
        left='{{job.trigger.type}}',
        op="==",
        right='{{job.trigger.type}}')

# Note: As we can have multiple deps same way we can keep multiple deps for if_else_outcome:
# Ex: if_else_outcome={"task1":"false", "task2":"true"}

# Example 4: creating a SQL Alert Task that depends on example_task_3, now the below sql task will trigger only if above tasks returns true.
@wf.sql_task(depends_on="example_task_3", if_else_outcome={"example_task_3":"true"})
def sample_sql_alert() ->any:
    # it automatically validates user emails
    return SqlTask(alert_id="ALERT_ID", pause_subscriptions=False, subscriptions={"usernames":["YOUR_EMAIL", 'YOUR_EMAIL']} ,warehouse_id="WAREHOUSE_ID")
# Note: Since SQL task doesn't return any bool, we can't make use of if_else_outcome params for the tasks that depends on sql Task
```

--------------------------------

### Verify BrickFlow installation

Source: https://engineering.nike.com/brickflow/latest/bundles-quickstart/

Check that the BrickFlow CLI is installed and the Databricks workspace is accessible. Add `--profile <profile>` if using a specific profile.

```bash
bf --help
databricks workspace list /  # Add --profile <profile> if using specific profile
```

--------------------------------

### Context.__init__ method

Source: https://engineering.nike.com/brickflow/latest/api/context/

Initializes the Context object, setting up internal attributes and calling _configure(). The order of initialization is noted as important.

```python
def __init__(self) -> None:
    # Order of init matters todo: fix this

    self._dbutils: Optional[Any] = None
    self._spark: Optional[Any] = None
    self._task_coms: BrickflowTaskComs
    self._current_task: Optional[str] = None
    self._configure()
    self._current_project: Optional[str] = None
```

--------------------------------

### bf projects add options

Source: https://engineering.nike.com/brickflow/latest/cli/reference/

Options for the bf projects add subcommand, including --name, --path-from-repo-root-to-project-root, --path-project-root-to-workflows-dir, -g/--git-https-url, -bfv/--brickflow-version, -sev/--spark-expectations-version, --skip-entrypoint, and --help.

```bash
  --name TEXT                     Name of the project
  --path-from-repo-root-to-project-root DIRECTORY
                                  Path from repo root to project root
  --path-project-root-to-workflows-dir TEXT
                                  Path from project root to workflows dir
  -g, --git-https-url TEXT        Provide the github URL for your project,
                                  example: https://github.com/nike-eda-
                                  apla/brickflow
  -bfv, --brickflow-version TEXT
  -sev, --spark-expectations-version TEXT
  --skip-entrypoint               Skip creating entrypoint.py file
  --help                          Show this message and exit.
```

--------------------------------

### TableauRefreshWorkBookOperator usage example

Source: https://engineering.nike.com/brickflow/latest/api/airflow_tableau_operators/

Example of instantiating and executing TableauRefreshWorkBookOperator to refresh a list of Tableau workbooks. Requires server, username, password, site, project, and workbooks parameters.

```python
TableauRefreshWorkBookOperator(
    server="https://tableau.example.com",
    username="me",
    password="pw",
    site="my_site",
    project="my_project",
    workbooks=["wb_a", "wb_b"],
).execute()
```

--------------------------------

### Project.__enter__ method

Source: https://engineering.nike.com/brickflow/latest/api/project/

Creates and returns an internal _Project instance using the Project's configuration. Used when entering a context manager.

```python
def __enter__(self) -> "_Project":
    self._project = _Project(
        self.name,
        self.git_repo,
        self.provider,
        self.git_reference,
        self.entry_point_path,
        libraries=self.libraries,
        batch=self.batch,
        bundle_obj_name=self.bundle_obj_name,
        bundle_base_path=self.bundle_base_path,
    )
    return self._project
```

--------------------------------

### Deploy to dev, test, and production environments

Source: https://engineering.nike.com/brickflow/latest/bundles-quickstart/

Deploy the project to different environments using the `-e` flag. Each environment is a separate deployment.

```bash
# Dev environment
bf projects deploy --project <project> -p <profile> -e dev --force-acquire-lock

# Test environment
bf projects deploy --project <project> -p <profile> -e test --force-acquire-lock

# Production environment
bf projects deploy --project <project> -p <profile> -e prod --force-acquire-lock
```

--------------------------------

### AutosysSensor usage example

Source: https://engineering.nike.com/brickflow/latest/api/airflow_external_task_dependency/

Example of creating an AutosysSensor instance and calling poke(). The sensor polls an Autosys REST endpoint for a job and waits until it reports a successful status recent enough to satisfy time_delta.

```python
sensor = AutosysSensor(
    url="https://autosys.example.com/api/jobs",
    job_name="my_upstream_job",
    poke_interval=60,
    time_delta=timedelta(hours=1),
)
sensor.poke()
```

--------------------------------

### Synthesize project configuration

Source: https://engineering.nike.com/brickflow/latest/bundles-quickstart/

Generate the bundle.yml for your project. Replace `<project_name>` and `<profile>` with your values.

```bash
bf projects synth --project <project_name> --profile <profile>
```

--------------------------------

### Tableau Refresh Operators Usage Example

Source: https://engineering.nike.com/brickflow/latest/tasks/

Example of using TableauRefreshDataSourceOperator and TableauRefreshWorkBookOperator within a Brickflow workflow. Requires a Tableau server URL, credentials, site, and project. The operators trigger refreshes for the specified data sources or workbooks and poll for status.

```python
from brickflow.context import ctx
from brickflow_plugins import TableauRefreshDataSourceOperator, TableauRefreshWorkBookOperator

wf = Workflow(...)


@wf.task
def tableau_refresh_datasource():
    return TableauRefreshDataSourceOperator(
        server="https://my-tableau.com",
        username="foo",
        password="bar",
        site="site",
        project="project",
        data_sources=["datasource1", "datasource2"],
    )


@wf.task
def tableau_refresh_workbook():
    return TableauRefreshWorkBookOperator(
        server="https://my-tableau.com",
        username="foo",
        password="bar",
        site="site",
        project="project",
        workbooks=["workbook1", "workbook2"],
    )
```

--------------------------------

### brickflow.cli.docs() - Open docs in browser

Source: https://engineering.nike.com/brickflow/latest/api/cli/

Defines the `docs` CLI command that opens the Brickflow documentation site in the user's default browser and prints a confirmation message.

```python
@cli.command
def docs() -> None:
    """Use to open docs in your browser"""
    docs_site = "https://engineering.nike.com/brickflow/"
    webbrowser.open(docs_site, new=2)
    click.echo(f"Opening browser for docs... site: {docs_site}")
```

--------------------------------

### Configure serverless compute for Brickflow entrypoint task using MAGIC commands

Source: https://engineering.nike.com/brickflow/latest/faq/faq/

Use this snippet to configure serverless compute for a Brickflow entrypoint task. It requires adding MAGIC commands to the top of the notebook, including a mandatory `brickflows` dependency that should point to a version with serverless support or a wheel file.

```python
# Databricks notebook source
# `brickflows` dependency is mandatory!
# It should always point to the `brickflows` version with serverless support or the wheel file with the same
# MAGIC %pip install brickflows==x.x.x
# MAGIC %pip install my-dependency==x.x.x
# MAGIC %restart_python

# COMMAND ----------
```

--------------------------------

### bf projects sync options

Source: https://engineering.nike.com/brickflow/latest/cli/reference/

Options for bf projects sync, including watch, full, interval-duration, tags, key-value parameters, workflow names, debug, lock, fail-on-active-runs, skip-libraries, auto-approve, profile, project, env, and help.

```bash
  --watch                   Enable filewatcher to sync files over.
  --full                    Run a full sync.
  --interval-duration TEXT  File system polling interval (for --watch).
  -t, --tag TEXT            Provide the runtime key-value tags, each key-value
                            separated by space! Example: bf projects deploy -p
                            DEFAULT -e local -t t_key1=value1 -kv
                            t_key2=value2
  -kv, --key-value TEXT     Provide the runtime key-value parameters, each
                            key-value separated by space! Example: bf projects
                            deploy -p DEFAULT -e local -kv key1=value1 -kv
                            key2=value2
  -w, --workflow TEXT       Provide the workflow names (local mode only) to
                            deploy, each workflow separated by space! Note:
                            provide the workflow names without the env prefix
                            or the file extension. Example: bf projects deploy
                            -p DEFAULT -e local -w wf1 -w wf2
  --debug                   Enable debug logs.
  --force-acquire-lock      Force acquire lock for databricks bundles destroy.
  --fail-on-active-runs     Fail if there are running jobs or pipelines in the
                            deployment.
  --skip-libraries          Skip automatically adding brickflow libraries.
  --auto-approve            Auto approve brickflow pipeline without being
                            prompted to approve.
  -p, --profile TEXT        The databricks profile to use for authenticating
                            to databricks during deployment.
  --project []              Select the project of workflows you would like to
                            deploy.
  -e, --env TEXT            Set the environment value, certain tags [TBD] get
                            added to the workflows based on this value.
  --help                    Show this message and exit.
```

--------------------------------

### Before: TaskDependencySensor with AirflowProxyOktaClusterAuth

Source: https://engineering.nike.com/brickflow/latest/upgrades/upgrade-airflow-free/

Original usage of TaskDependencySensor with AirflowProxyOktaClusterAuth, which is removed. Use the 'After' example instead.

```python
from brickflow_plugins import TaskDependencySensor, AirflowProxyOktaClusterAuth

sensor = TaskDependencySensor(
    task_id="sensor",
    timeout=180,
    airflow_cluster_auth=AirflowProxyOktaClusterAuth(
        oauth2_conn_id=f"b64://{data}",
        airflow_cluster_url="https://proxy.../.../cluster_id/",
        airflow_version="2.0.2",
    ),
    external_dag_id="external_airflow_dag",
    external_task_id="hello",
    allowed_states=["success"],
    execution_delta=timedelta(hours=-2),
    poke_interval=60,
)
```

--------------------------------

### Project.__enter__()

Source: https://engineering.nike.com/brickflow/latest/api/project/

Creates and returns an internal _Project instance using the Project's configuration. This method is called when entering a 'with' block.

```APIDOC
## Project.__enter__()

### Description
Creates and returns an internal _Project instance using the Project's configuration. This method is called when entering a 'with' block.

### Method
__enter__

### Signature
`__enter__() -> _Project`

### Parameters
None

### Returns
- **_Project** - The internal project instance.

### Source Code
```python
def __enter__(self) -> "_Project":
    self._project = _Project(
        self.name,
        self.git_repo,
        self.provider,
        self.git_reference,
        self.entry_point_path,
        libraries=self.libraries,
        batch=self.batch,
        bundle_obj_name=self.bundle_obj_name,
        bundle_base_path=self.bundle_base_path,
    )
    return self._project
```
```

--------------------------------

### Set runtime project parameters via CLI

Source: https://engineering.nike.com/brickflow/latest/environment-variables/

Provides runtime key-value parameters to the deploy command using the -kv flag, each key-value separated by space.

```bash
bf projects deploy -p DEFAULT -e local -kv key1=value1 -kv key2=value2
```

--------------------------------

### Uninstall brickflow package

Source: https://engineering.nike.com/brickflow/latest/upgrades/upgrade-pre-0-10-0-to-0-10-0/

Run this command to uninstall the old brickflow package before installing the new brickflows package.

```bash
pip uninstall brickflow
```

--------------------------------

### CranTaskLibrary

Source: https://engineering.nike.com/brickflow/latest/api/task/

Dataclass representing a CRAN package task library. Use it to specify a CRAN package to install, optionally from a custom repository.

```APIDOC
## CranTaskLibrary

### Description
Dataclass representing a CRAN package task library. Use it to specify a CRAN package to install, optionally from a custom repository.

### Constructor
`CranTaskLibrary(package: str, repo: Optional[str] = None)`

### Parameters
- **package** (str) - Required - The name of the CRAN package to install.
- **repo** (Optional[str]) - Optional - The repository where the package can be found. If not specified, the default CRAN repo is used.

### Attributes
- **dict** (Dict[str, Union[str, Dict[str, str]]]) - property - Dictionary representation of the task library.
- **package** (str) - instance attribute - The name of the CRAN package.
- **repo** (Optional[str]) - class/instance attribute - The repository URL or None.

```

--------------------------------

### bf projects synth options

Source: https://engineering.nike.com/brickflow/latest/cli/reference/

Lists all available options for the 'bf projects synth' command, including tags, key-value parameters, workflow names, fail-on-active-runs, skip-libraries, profile, project, and environment.

```bash
  -t, --tag TEXT         Provide the runtime key-value tags, each key-value
                         separated by space! Example: bf projects deploy -p
                         DEFAULT -e local -t t_key1=value1 -kv t_key2=value2
  -kv, --key-value TEXT  Provide the runtime key-value parameters, each key-
                         value separated by space! Example: bf projects deploy
                         -p DEFAULT -e local -kv key1=value1 -kv key2=value2
  -w, --workflow TEXT    Provide the workflow names (local mode only) to
                         deploy, each workflow separated by space! Note:
                         provide the workflow names without the env prefix or
                         the file extension. Example: bf projects deploy -p
                         DEFAULT -e local -w wf1 -w wf2
  --fail-on-active-runs  Fail if there are running jobs or pipelines in the
                         deployment.
  --skip-libraries       Skip automatically adding brickflow libraries.
  -p, --profile TEXT     The databricks profile to use for authenticating to
                         databricks during deployment.
  --project []           Select the project of workflows you would like to
                         deploy.
  -e, --env TEXT         Set the environment value, certain tags [TBD] get
                         added to the workflows based on this value.
  --help                 Show this message and exit.
```

--------------------------------

### get_existing_folder_id(parent_folder_id, folder_name)

Source: https://engineering.nike.com/brickflow/latest/api/box_operator/

Get the ID of an existing folder in a Box folder. Returns the folder ID if found, otherwise None.

```APIDOC
## get_existing_folder_id(parent_folder_id, folder_name)

### Description
Get the ID of an existing folder in a Box folder.

### Method
Function call

### Parameters
- `parent_folder_id` (str) - Required - The ID of the parent Box folder.
- `folder_name` (str) - Required - The name of the folder.

### Returns
- `str` - The ID of the existing folder, or None if not found.

```

--------------------------------

### __post_init__() method

Source: https://engineering.nike.com/brickflow/latest/api/compute/

Calls validate() after initialization to ensure the cluster configuration is valid.

```python
def __post_init__(self) -> None:
    self.validate()
```

--------------------------------

### AirflowTaskDependencySensor usage for Airflow 2.x

Source: https://engineering.nike.com/brickflow/latest/api/airflow_external_task_dependency/

Example of using AirflowTaskDependencySensor with an AirflowCluster for Airflow 2.x. Requires only requests, no apache-airflow. The token is provided by the user.

```python
from datetime import timedelta

from brickflow_plugins import AirflowCluster, AirflowTaskDependencySensor

sensor = AirflowTaskDependencySensor(
    dag_id="my_upstream_dag",
    task_id="final_task",
    cluster=AirflowCluster(
        url="https://airflow.example.com",
        version="2.0.2",
        token=my_bearer_token,  # compute this from Okta/etc. yourself
    ),
    allowed_states=["success"],
    execution_delta=timedelta(hours=0),
    timeout_seconds=3600,
    poke_interval=60,
)
sensor.execute()
```

--------------------------------

### brickflow.cli.docs() -> None

Source: https://engineering.nike.com/brickflow/latest/api/cli/

Opens the Brickflow documentation website in the user's default browser. This command is part of the CLI and provides quick access to the official documentation.

```APIDOC
## brickflow.cli.docs() -> None

### Description
Opens the Brickflow documentation website in the user's default browser. This command is part of the CLI and provides quick access to the official documentation.

### Method
CLI Command

### Endpoint
brickflow.cli.docs

### Parameters
None

### Request Example
```
brickflow docs
```

### Response
#### Success Response
- Opens the browser to https://engineering.nike.com/brickflow/ and prints a confirmation message.

#### Response Example
```
Opening browser for docs... site: https://engineering.nike.com/brickflow/
```
```

--------------------------------

### TableauWrapper.__init__ source code

Source: https://engineering.nike.com/brickflow/latest/api/airflow_tableau_operators/

Constructor for TableauWrapper. Initializes connection parameters and polling settings. Calls _ensure_tableau() to verify the Tableau client library is installed.

```python
def __init__(
    self,
    server: str,
    username: str,
    password: str,
    site: str,
    project: Optional[str] = None,
    parent_project: Optional[str] = None,
    version: str = "3.14",
    max_async_workers: int = 5,
    polling_required: bool = True,
    polling_interval: int = 30,
    polling_timeout: int = 600,
) -> None:
    """
    Parameters
    ----------
    server : str
        Tableau server address, e.g. ``https://tableau-server.com``.
    username : str
        Log in username.
    password : str
        Log in password.
    site : str
        Tableau site.
    project : str
        Tableau project.
    parent_project : str
        Name of the parent Tableau project. Use ``"/"`` for the site root.
    version : str
        Tableau server API version.
    max_async_workers : int
        Maximum number of asynchronous tasks that will trigger jobs and
        wait for completion.
    polling_required : bool
        Wait for job completion to proceed, otherwise trigger the job and
        proceed without waiting.
    polling_interval : int
        Polling interval for the job status updates (seconds).
    polling_timeout : int
        Stop polling if the job was not completed within the specified
        interval (seconds).
    """
    _ensure_tableau()
    self.server = server
    self.version = version
    self.username = username
    self.password = password
    self.site = site
    self.project = project
    self.parent_project = parent_project

    self.max_async_workers = max_async_workers
    self.polling_required = polling_required
    self.polling_interval = polling_interval
    self.polling_timeout = polling_timeout

    self._logger = log
    self._ip = None
```

--------------------------------

### Example: TableauRefreshDataSourceOperator usage

Source: https://engineering.nike.com/brickflow/latest/api/airflow_tableau_operators/

Shows how to instantiate and execute the operator to refresh data sources. Requires server, username, password, site, project, and data_sources parameters.

```python
TableauRefreshDataSourceOperator(
    server="https://tableau.example.com",
    username="me",
    password="pw",
    site="my_site",
    project="my_project",
    data_sources=["ds_a", "ds_b"],
).execute()
```

--------------------------------

### TableauRefreshDataSourceOperator and TableauRefreshWorkBookOperator usage

Source: https://engineering.nike.com/brickflow/latest/api/airflow_tableau_operators/

Example of instantiating and executing the Tableau refresh operators. The operators are plain Python classes with an execute() method, requiring the tableauserverclient library.

```APIDOC
## TableauRefreshDataSourceOperator and TableauRefreshWorkBookOperator

### Description
Refresh Tableau data sources or workbooks by triggering async refresh jobs and polling the Tableau server until they complete. These are plain Python classes with an `execute()` method, no Airflow dependency.

### Usage Example
```python
from brickflow_plugins import (
    TableauRefreshDataSourceOperator,
    TableauRefreshWorkBookOperator,
)

TableauRefreshDataSourceOperator(
    server="https://tableau.example.com",
    username="me",
    password="pw",
    site="my_site",
    project="my_project",
    data_sources=["ds_a", "ds_b"],
).execute()
```

### Requirements
Requires the `tableauserverclient` library to be installed on the cluster (via `PypiTaskLibrary("tableauserverclient==0.25")` or the `brickflow[tableau]` extra during local development).
```
Tokens:
Raw