Skip to content

prefect.variables

Variable

Bases: VariableCreate

Variables are named, mutable string values, much like environment variables. Variables are scoped to a Prefect server instance or a single workspace in Prefect Cloud. https://docs.prefect.io/latest/concepts/variables/

Parameters:

Name Type Description Default
name

A string identifying the variable.

 required
value

A string that is the value of the variable.

 required
tags

An optional list of strings to associate with the variable.

 required
Source code in prefect/variables.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
class Variable(VariableRequest):
    """
    Variables are named, mutable string values, much like environment variables. Variables are scoped to a Prefect server instance or a single workspace in Prefect Cloud.
    https://docs.prefect.io/latest/concepts/variables/

    Arguments:
        name: A string identifying the variable.
        value: A string that is the value of the variable.
        tags: An optional list of strings to associate with the variable.
    """

    @classmethod
    @sync_compatible
    async def set(
        cls,
        name: str,
        value: str,
        tags: Optional[List[str]] = None,
        overwrite: bool = False,
    ) -> Optional["Variable"]:
        """
        Sets a new variable. If one exists with the same name, user must pass `overwrite=True`
        ```
            from prefect.variables import Variable

            @flow
            def my_flow():
                var = Variable.set(name="my_var",value="test_value", tags=["hi", "there"], overwrite=True)
        ```
        or
        ```
            from prefect.variables import Variable

            @flow
            async def my_flow():
                var = await Variable.set(name="my_var",value="test_value", tags=["hi", "there"], overwrite=True)
        ```
        """
        client, _ = get_or_create_client()
        variable = await client.read_variable_by_name(name)
        var_dict = {"name": name, "value": value, "tags": tags or []}
        if variable:
            if not overwrite:
                raise ValueError(
                    "You are attempting to save a variable with a name that is already in use. If you would like to overwrite the values that are saved, then call .set with `overwrite=True`."
                )
            var = VariableUpdateRequest(**var_dict)
            await client.update_variable(variable=var)
            variable = await client.read_variable_by_name(name)
        else:
            var = VariableRequest(**var_dict)
            variable = await client.create_variable(variable=var)

        return variable if variable else None

    @classmethod
    @sync_compatible
    async def get(
        cls, name: str, default: Optional[str] = None
    ) -> Optional[Union["Variable", str]]:
        """
        Get a variable by name. If doesn't exist return the default.
        ```
            from prefect.variables import Variable

            @flow
            def my_flow():
                var = Variable.get("my_var")
        ```
        or
        ```
            from prefect.variables import Variable

            @flow
            async def my_flow():
                var = await Variable.get("my_var")
        ```
        """
        client, _ = get_or_create_client()
        variable = await client.read_variable_by_name(name)

        return (
            Variable(name=variable.name, value=variable.value, tags=variable.tags)
            if variable
            else default
        )

get async classmethod

Get a variable by name. If doesn't exist return the default. 

    from prefect.variables import Variable

    @flow
    def my_flow():
        var = Variable.get("my_var")
or 
    from prefect.variables import Variable

    @flow
    async def my_flow():
        var = await Variable.get("my_var")

Source code in prefect/variables.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
@classmethod
@sync_compatible
async def get(
    cls, name: str, default: Optional[str] = None
) -> Optional[Union["Variable", str]]:
    """
    Get a variable by name. If doesn't exist return the default.
    ```
        from prefect.variables import Variable

        @flow
        def my_flow():
            var = Variable.get("my_var")
    ```
    or
    ```
        from prefect.variables import Variable

        @flow
        async def my_flow():
            var = await Variable.get("my_var")
    ```
    """
    client, _ = get_or_create_client()
    variable = await client.read_variable_by_name(name)

    return (
        Variable(name=variable.name, value=variable.value, tags=variable.tags)
        if variable
        else default
    )

set async classmethod

Sets a new variable. If one exists with the same name, user must pass overwrite=True 

    from prefect.variables import Variable

    @flow
    def my_flow():
        var = Variable.set(name="my_var",value="test_value", tags=["hi", "there"], overwrite=True)
or 
    from prefect.variables import Variable

    @flow
    async def my_flow():
        var = await Variable.set(name="my_var",value="test_value", tags=["hi", "there"], overwrite=True)

Source code in prefect/variables.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
@classmethod
@sync_compatible
async def set(
    cls,
    name: str,
    value: str,
    tags: Optional[List[str]] = None,
    overwrite: bool = False,
) -> Optional["Variable"]:
    """
    Sets a new variable. If one exists with the same name, user must pass `overwrite=True`
    ```
        from prefect.variables import Variable

        @flow
        def my_flow():
            var = Variable.set(name="my_var",value="test_value", tags=["hi", "there"], overwrite=True)
    ```
    or
    ```
        from prefect.variables import Variable

        @flow
        async def my_flow():
            var = await Variable.set(name="my_var",value="test_value", tags=["hi", "there"], overwrite=True)
    ```
    """
    client, _ = get_or_create_client()
    variable = await client.read_variable_by_name(name)
    var_dict = {"name": name, "value": value, "tags": tags or []}
    if variable:
        if not overwrite:
            raise ValueError(
                "You are attempting to save a variable with a name that is already in use. If you would like to overwrite the values that are saved, then call .set with `overwrite=True`."
            )
        var = VariableUpdateRequest(**var_dict)
        await client.update_variable(variable=var)
        variable = await client.read_variable_by_name(name)
    else:
        var = VariableRequest(**var_dict)
        variable = await client.create_variable(variable=var)

    return variable if variable else None

get async

Get a variable by name. If doesn't exist return the default. 

    from prefect import variables

    @flow
    def my_flow():
        var = variables.get("my_var")
or 
    from prefect import variables

    @flow
    async def my_flow():
        var = await variables.get("my_var")

Source code in prefect/variables.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
@deprecated_callable(start_date="Apr 2024")
@sync_compatible
async def get(name: str, default: Optional[str] = None) -> Optional[str]:
    """
    Get a variable by name. If doesn't exist return the default.
    ```
        from prefect import variables

        @flow
        def my_flow():
            var = variables.get("my_var")
    ```
    or
    ```
        from prefect import variables

        @flow
        async def my_flow():
            var = await variables.get("my_var")
    ```
    """
    variable = await Variable.get(name, default=default)
    if isinstance(variable, Variable):
        variable = variable.value
    return variable