SaltStack custom state: How to download file from "salt:" URI

I'm trying to write a custom SaltStack state. It will be used like this:

Example of my.state usage:
  my.state:
    - name: /some/path
    - source: salt://path/to/dsl/file

This state is planned to grab a file from the Salt Master file server, transfer it to the minion, then execute the file using a program in the minion, analyze the output, delete the downloaded file, and return a proper state dict based on success/failure of the program that executes the DSL file.

My question:

How to write the custom state so it's able to download from salt:// URI ?

Note: I do not envision the need to download the file from anywhere else but the Salt Master.

Edit: In the interest of full disclosure of facts, "success/failure" is gleaned by analyzing the output of the program, not a simple exitcode analysis. Just FYI.

Solution

save the following as /srv/salt/_states/my_state.py

import os

def execute_dsl(name, source):
    """
    Custom Salt state to download a file from Salt Master, execute it, and analyze the result.

    :param name: Path where the file will be stored temporarily on the minion.
    :param source: salt:// path to the DSL file.
    :return: Salt state dictionary with success/failure status.
    """
    ret = {
        "name": name,
        "changes": {},
        "result": False,
        "comment": ""
    }
# here you are fetching the file from the Salt Master
    local_path = __salt__["cp.get_file"](source, name)
    
    if not local_path:
        ret["comment"] = f"Failed to fetch {source} from Salt Master."
        return ret

    ret["changes"]["downloaded"] = local_path

              # her we are executing the DSL file using the appropriate interpreter (modify if needed)
    cmd_result = __salt__["cmd.run_all"](f"/usr/bin/dsl_interpreter {local_path}", python_shell=True)

  # you should check theexccutiong status here
    if cmd_result["retcode"] == 0:
        ret["result"] = True
        ret["changes"]["output"] = cmd_result["stdout"]
        ret["comment"] = "DSL file executed successfully."
    else:
        ret["comment"] = f"Execution failed: {cmd_result['stderr']}"
    
    # cleanup and download the file.
    try:
        os.remove(local_path)
        ret["changes"]["cleaned_up"] = f"Deleted {local_path}"
    except Exception as e:
        ret["comment"] += f" However, cleanup failed: {str(e)}"

    return ret