We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Profiling bug when creating a dataframe from a dictionary using dask
**Describe the bug**
calling `df.profile("*")` for a dataframe created using `op.create.dataframe` causes an error `KeyError: 'id'`.

**To Reproduce**
```
df = op.create.dataframe({"id":[5,6,10,11,79,100]})
df.profile("*")
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `optimus/engines/base/dask/dataframe.py`
Content:
```
1 from abc import abstractmethod
2 
3 import dask
4 import humanize
5 from dask.distributed import Variable
6 from dask.utils import parse_bytes
7 
8 from optimus.engines.base.basedataframe import BaseDataFrame
9 from optimus.engines.pandas.dataframe import PandasDataFrame
10 from optimus.helpers.functions import random_int
11 from optimus.helpers.raiseit import RaiseIt
12 from optimus.infer import is_one_element
13 
14 
15 class DaskBaseDataFrame(BaseDataFrame):
16 
17     def __init__(self, root, data):
18         super().__init__(root, data)
19 
20     def _assign(self, kw_columns):
21 
22         dfd = self.root.data
23 
24         if dfd.known_divisions:
25             for key in kw_columns:
26                 kw_column = kw_columns[key]
27                 if not is_one_element(kw_column) and not callable(kw_column) and not kw_column.known_divisions:
28                     kw_columns[key] = kw_column.reset_index().set_index('index')[key]
29         return dfd.assign(**kw_columns)
30 
31     @staticmethod
32     @abstractmethod
33     def _base_to_dfd(df, n_partitions):
34         """
35         Convert a dataframe from the dataframe base (pandas, cudf) in a distributed engine dataframe (dask, dask_cudf)
36         :param pdf:
37         :param n_partitions:
38         :return:
39         """
40         pass
41 
42     def execute(self):
43         self.data = self.data.persist()
44         return self
45 
46     def compute(self):
47         df = self.data
48         return df.compute()
49 
50     def visualize(self):
51         return display(self.data.visualize())
52 
53     def export(self):
54         """
55         Helper function to export all the dataframe in text format. Aimed to be used in test functions
56         :return:
57         """
58         df = self.root
59         df_data = df.to_json()
60         df_schema = df.cols.dtypes()
61 
62         return f"{df_schema}, {df_data}"
63 
64     def _reset_buffer(self):
65         if self.buffer:
66             Variable(self.buffer).delete()
67             self.buffer = None
68 
69     def get_buffer(self):
70         return Variable(self.buffer).get() if self.buffer else None
71 
72     def _buffer_window(self, input_cols, lower_bound, upper_bound):
73         def func(value):
74             return value[lower_bound:upper_bound]
75 
76         return PandasDataFrame(self.data[input_cols].partitions[0].map_partitions(func).compute())
77 
78     def sample(self, n=10, random=False):
79         """
80         Return a n number of sample from a dataFrame
81         :param n: Number of samples
82         :param random: if true get a semi random sample
83         :return:
84         """
85         df = self.root
86         if random is True:
87             seed = random_int()
88         elif random is False:
89             seed = 0
90         else:
91             RaiseIt.value_error(random, ["True", "False"])
92 
93         rows_count = df.rows.count()
94         if n < rows_count:
95             # n/rows_count can return a number that represent less the total number we expect. multiply by 1.1
96             fraction = (n / rows_count) * 1.1
97         else:
98             fraction = 1.0
99         return self.root.new(df.data.sample(frac=fraction, random_state=seed))
100 
101     def stratified_sample(self, col_name, seed: int = 1):
102         """
103         Stratified Sampling
104         :param col_name:
105         :param seed:
106         :return:
107         """
108         df = self.data
109         n = min(5, df[col_name].value_counts().min())
110         df = df.groupby(col_name).apply(lambda x: x.sample(2))
111         # df_.index = df_.index.droplevel(0)
112         return self.root.new(df)
113 
114     @staticmethod
115     def pivot(index, column, values):
116         """
117         Return reshaped DataFrame organized by given index / column values.
118         :param index: Column to use to make new frame's index.
119         :param column: Column to use to make new frame's columns.
120         :param values: Column(s) to use for populating new frame's values.
121         :return:
122         """
123         raise NotImplementedError
124 
125     @staticmethod
126     def melt(id_vars, value_vars, var_name="variable", value_name="value", data_type="str"):
127         """
128         Convert DataFrame from wide to long format.
129         :param id_vars: column with unique values
130         :param value_vars: Column names that are going to be converted to columns values
131         :param var_name: Column name for vars
132         :param value_name: Column name for values
133         :param data_type: All columns must have the same type. It will transform all columns to this data type.
134         :return:
135         """
136 
137         raise NotImplementedError
138 
139     def size(self, deep=False, format=None):
140         """
141         Get the size of a dask in bytes
142         :return:
143         """
144         df = self.data
145         result = df.memory_usage(index=True, deep=deep).sum().compute()
146         if format == "human":
147             result = humanize.naturalsize(result)
148 
149         return result
150 
151     def run(self):
152         """
153         This method is a very useful function to break lineage of transformations. By default Spark uses the lazy
154         evaluation approach in processing data: transformation functions are not computed into an action is called.
155         Sometimes when transformations are numerous, the computations are very extensive because the high number of
156         operations that spark needs to run in order to get the results.
157 
158         Other important thing is that Apache Spark save task but not result of dataFrame, so tasks are
159         accumulated and the same situation happens.
160 
161         :return:
162         """
163         df = self.data
164         df.cache().count()
165         return df
166 
167     @staticmethod
168     def query(sql_expression):
169         raise NotImplementedError
170 
171     def partitions(self):
172         return self.data.npartitions
173 
174     @staticmethod
175     def partitioner():
176         print("Dask not support custom partitioner")
177         raise NotImplementedError
178 
179     def repartition(self, n=None, *args, **kwargs):
180         dfd = self.data
181         df = self
182         if n == "auto":
183             # Follow a heuristic for partitioning a mentioned
184             # https://docs.dask.org/en/latest/best-practices.html#avoid-very-large-partitions
185             client = dask.distributed.get_client()
186             worker_memory = parse_bytes(client.cluster.worker_spec[0]["options"]["memory_limit"])
187             nthreads = client.cluster.worker_spec[0]["options"]["nthreads"]
188 
189             part_recommended_size = worker_memory / nthreads / 10
190             n = int(df.size() / part_recommended_size)
191 
192             # Partition can not be lower than 1
193             n = n if n < 0 else 1
194             # TODO .repartition(partition_size="100MB"). https://stackoverflow.com/questions/44657631/strategy-for-partitioning-dask-dataframes-efficiently
195         dfd = dfd.repartition(npartitions=n, *args, **kwargs)
196 
197         return self.new(dfd, meta=self.meta)
198 
199     @staticmethod
200     def debug():
201         """
202 
203         :return:
204         """
205         raise NotImplementedError
206 
207     @staticmethod
208     def create_id(column="id"):
209         """
210         Create a unique id for every row.
211         :param column: Columns to be processed
212         :return:
213         """
214 
215         raise NotImplementedError
216 
217     def to_dict(self, orient="records", index=True):
218         """
219         Create a dict
220         :param orient:
221         :param index: Return the series index
222         :return:
223         """
224 
225         series = self.data
226         if index is True:
227             return series.compute().to_dict(orient)
228         else:
229             return series.compute().to_list()
230 
231     def to_pandas(self):
232         return self.data.compute()
233 
234     @property
235     def constants(self):
236         from optimus.engines.base.dask.constants import constants
237         return constants(self)
238
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/optimus/engines/base/dask/dataframe.py b/optimus/engines/base/dask/dataframe.py
--- a/optimus/engines/base/dask/dataframe.py
+++ b/optimus/engines/base/dask/dataframe.py
@@ -25,7 +25,13 @@
             for key in kw_columns:
                 kw_column = kw_columns[key]
                 if not is_one_element(kw_column) and not callable(kw_column) and not kw_column.known_divisions:
-                    kw_columns[key] = kw_column.reset_index().set_index('index')[key]
+                    _dfd = kw_column.reset_index().set_index('index')
+                    if key in _dfd:
+                        # the incoming series has the same column key
+                        kw_columns[key] = _dfd[key]
+                    else:
+                        # the incoming series has no column key
+                        kw_columns[key] = _dfd[0]
         return dfd.assign(**kw_columns)
 
     @staticmethod

{"golden_diff": "diff --git a/optimus/engines/base/dask/dataframe.py b/optimus/engines/base/dask/dataframe.py\n--- a/optimus/engines/base/dask/dataframe.py\n+++ b/optimus/engines/base/dask/dataframe.py\n@@ -25,7 +25,13 @@\n             for key in kw_columns:\n                 kw_column = kw_columns[key]\n                 if not is_one_element(kw_column) and not callable(kw_column) and not kw_column.known_divisions:\n-                    kw_columns[key] = kw_column.reset_index().set_index('index')[key]\n+                    _dfd = kw_column.reset_index().set_index('index')\n+                    if key in _dfd:\n+                        # the incoming series has the same column key\n+                        kw_columns[key] = _dfd[key]\n+                    else:\n+                        # the incoming series has no column key\n+                        kw_columns[key] = _dfd[0]\n         return dfd.assign(**kw_columns)\n \n     @staticmethod\n", "issue": "Profiling bug when creating a dataframe from a dictionary using dask\n**Describe the bug**\r\ncalling `df.profile(\"*\")` for a dataframe created using `op.create.dataframe` causes an error `KeyError: 'id'`.\r\n\r\n**To Reproduce**\r\n```\r\ndf = op.create.dataframe({\"id\":[5,6,10,11,79,100]})\r\ndf.profile(\"*\")\r\n```\r\n\n", "before_files": [{"content": "from abc import abstractmethod\n\nimport dask\nimport humanize\nfrom dask.distributed import Variable\nfrom dask.utils import parse_bytes\n\nfrom optimus.engines.base.basedataframe import BaseDataFrame\nfrom optimus.engines.pandas.dataframe import PandasDataFrame\nfrom optimus.helpers.functions import random_int\nfrom optimus.helpers.raiseit import RaiseIt\nfrom optimus.infer import is_one_element\n\n\nclass DaskBaseDataFrame(BaseDataFrame):\n\n    def __init__(self, root, data):\n        super().__init__(root, data)\n\n    def _assign(self, kw_columns):\n\n        dfd = self.root.data\n\n        if dfd.known_divisions:\n            for key in kw_columns:\n                kw_column = kw_columns[key]\n                if not is_one_element(kw_column) and not callable(kw_column) and not kw_column.known_divisions:\n                    kw_columns[key] = kw_column.reset_index().set_index('index')[key]\n        return dfd.assign(**kw_columns)\n\n    @staticmethod\n    @abstractmethod\n    def _base_to_dfd(df, n_partitions):\n        \"\"\"\n        Convert a dataframe from the dataframe base (pandas, cudf) in a distributed engine dataframe (dask, dask_cudf)\n        :param pdf:\n        :param n_partitions:\n        :return:\n        \"\"\"\n        pass\n\n    def execute(self):\n        self.data = self.data.persist()\n        return self\n\n    def compute(self):\n        df = self.data\n        return df.compute()\n\n    def visualize(self):\n        return display(self.data.visualize())\n\n    def export(self):\n        \"\"\"\n        Helper function to export all the dataframe in text format. Aimed to be used in test functions\n        :return:\n        \"\"\"\n        df = self.root\n        df_data = df.to_json()\n        df_schema = df.cols.dtypes()\n\n        return f\"{df_schema}, {df_data}\"\n\n    def _reset_buffer(self):\n        if self.buffer:\n            Variable(self.buffer).delete()\n            self.buffer = None\n\n    def get_buffer(self):\n        return Variable(self.buffer).get() if self.buffer else None\n\n    def _buffer_window(self, input_cols, lower_bound, upper_bound):\n        def func(value):\n            return value[lower_bound:upper_bound]\n\n        return PandasDataFrame(self.data[input_cols].partitions[0].map_partitions(func).compute())\n\n    def sample(self, n=10, random=False):\n        \"\"\"\n        Return a n number of sample from a dataFrame\n        :param n: Number of samples\n        :param random: if true get a semi random sample\n        :return:\n        \"\"\"\n        df = self.root\n        if random is True:\n            seed = random_int()\n        elif random is False:\n            seed = 0\n        else:\n            RaiseIt.value_error(random, [\"True\", \"False\"])\n\n        rows_count = df.rows.count()\n        if n < rows_count:\n            # n/rows_count can return a number that represent less the total number we expect. multiply by 1.1\n            fraction = (n / rows_count) * 1.1\n        else:\n            fraction = 1.0\n        return self.root.new(df.data.sample(frac=fraction, random_state=seed))\n\n    def stratified_sample(self, col_name, seed: int = 1):\n        \"\"\"\n        Stratified Sampling\n        :param col_name:\n        :param seed:\n        :return:\n        \"\"\"\n        df = self.data\n        n = min(5, df[col_name].value_counts().min())\n        df = df.groupby(col_name).apply(lambda x: x.sample(2))\n        # df_.index = df_.index.droplevel(0)\n        return self.root.new(df)\n\n    @staticmethod\n    def pivot(index, column, values):\n        \"\"\"\n        Return reshaped DataFrame organized by given index / column values.\n        :param index: Column to use to make new frame's index.\n        :param column: Column to use to make new frame's columns.\n        :param values: Column(s) to use for populating new frame's values.\n        :return:\n        \"\"\"\n        raise NotImplementedError\n\n    @staticmethod\n    def melt(id_vars, value_vars, var_name=\"variable\", value_name=\"value\", data_type=\"str\"):\n        \"\"\"\n        Convert DataFrame from wide to long format.\n        :param id_vars: column with unique values\n        :param value_vars: Column names that are going to be converted to columns values\n        :param var_name: Column name for vars\n        :param value_name: Column name for values\n        :param data_type: All columns must have the same type. It will transform all columns to this data type.\n        :return:\n        \"\"\"\n\n        raise NotImplementedError\n\n    def size(self, deep=False, format=None):\n        \"\"\"\n        Get the size of a dask in bytes\n        :return:\n        \"\"\"\n        df = self.data\n        result = df.memory_usage(index=True, deep=deep).sum().compute()\n        if format == \"human\":\n            result = humanize.naturalsize(result)\n\n        return result\n\n    def run(self):\n        \"\"\"\n        This method is a very useful function to break lineage of transformations. By default Spark uses the lazy\n        evaluation approach in processing data: transformation functions are not computed into an action is called.\n        Sometimes when transformations are numerous, the computations are very extensive because the high number of\n        operations that spark needs to run in order to get the results.\n\n        Other important thing is that Apache Spark save task but not result of dataFrame, so tasks are\n        accumulated and the same situation happens.\n\n        :return:\n        \"\"\"\n        df = self.data\n        df.cache().count()\n        return df\n\n    @staticmethod\n    def query(sql_expression):\n        raise NotImplementedError\n\n    def partitions(self):\n        return self.data.npartitions\n\n    @staticmethod\n    def partitioner():\n        print(\"Dask not support custom partitioner\")\n        raise NotImplementedError\n\n    def repartition(self, n=None, *args, **kwargs):\n        dfd = self.data\n        df = self\n        if n == \"auto\":\n            # Follow a heuristic for partitioning a mentioned\n            # https://docs.dask.org/en/latest/best-practices.html#avoid-very-large-partitions\n            client = dask.distributed.get_client()\n            worker_memory = parse_bytes(client.cluster.worker_spec[0][\"options\"][\"memory_limit\"])\n            nthreads = client.cluster.worker_spec[0][\"options\"][\"nthreads\"]\n\n            part_recommended_size = worker_memory / nthreads / 10\n            n = int(df.size() / part_recommended_size)\n\n            # Partition can not be lower than 1\n            n = n if n < 0 else 1\n            # TODO .repartition(partition_size=\"100MB\"). https://stackoverflow.com/questions/44657631/strategy-for-partitioning-dask-dataframes-efficiently\n        dfd = dfd.repartition(npartitions=n, *args, **kwargs)\n\n        return self.new(dfd, meta=self.meta)\n\n    @staticmethod\n    def debug():\n        \"\"\"\n\n        :return:\n        \"\"\"\n        raise NotImplementedError\n\n    @staticmethod\n    def create_id(column=\"id\"):\n        \"\"\"\n        Create a unique id for every row.\n        :param column: Columns to be processed\n        :return:\n        \"\"\"\n\n        raise NotImplementedError\n\n    def to_dict(self, orient=\"records\", index=True):\n        \"\"\"\n        Create a dict\n        :param orient:\n        :param index: Return the series index\n        :return:\n        \"\"\"\n\n        series = self.data\n        if index is True:\n            return series.compute().to_dict(orient)\n        else:\n            return series.compute().to_list()\n\n    def to_pandas(self):\n        return self.data.compute()\n\n    @property\n    def constants(self):\n        from optimus.engines.base.dask.constants import constants\n        return constants(self)\n", "path": "optimus/engines/base/dask/dataframe.py"}], "after_files": [{"content": "from abc import abstractmethod\n\nimport dask\nimport humanize\nfrom dask.distributed import Variable\nfrom dask.utils import parse_bytes\n\nfrom optimus.engines.base.basedataframe import BaseDataFrame\nfrom optimus.engines.pandas.dataframe import PandasDataFrame\nfrom optimus.helpers.functions import random_int\nfrom optimus.helpers.raiseit import RaiseIt\nfrom optimus.infer import is_one_element\n\n\nclass DaskBaseDataFrame(BaseDataFrame):\n\n    def __init__(self, root, data):\n        super().__init__(root, data)\n\n    def _assign(self, kw_columns):\n\n        dfd = self.root.data\n\n        if dfd.known_divisions:\n            for key in kw_columns:\n                kw_column = kw_columns[key]\n                if not is_one_element(kw_column) and not callable(kw_column) and not kw_column.known_divisions:\n                    _dfd = kw_column.reset_index().set_index('index')\n                    if key in _dfd:\n                        # the incoming series has the same column key\n                        kw_columns[key] = _dfd[key]\n                    else:\n                        # the incoming series has no column key\n                        kw_columns[key] = _dfd[0]\n        return dfd.assign(**kw_columns)\n\n    @staticmethod\n    @abstractmethod\n    def _base_to_dfd(df, n_partitions):\n        \"\"\"\n        Convert a dataframe from the dataframe base (pandas, cudf) in a distributed engine dataframe (dask, dask_cudf)\n        :param pdf:\n        :param n_partitions:\n        :return:\n        \"\"\"\n        pass\n\n    def execute(self):\n        self.data = self.data.persist()\n        return self\n\n    def compute(self):\n        df = self.data\n        return df.compute()\n\n    def visualize(self):\n        return display(self.data.visualize())\n\n    def export(self):\n        \"\"\"\n        Helper function to export all the dataframe in text format. Aimed to be used in test functions\n        :return:\n        \"\"\"\n        df = self.root\n        df_data = df.to_json()\n        df_schema = df.cols.dtypes()\n\n        return f\"{df_schema}, {df_data}\"\n\n    def _reset_buffer(self):\n        if self.buffer:\n            Variable(self.buffer).delete()\n            self.buffer = None\n\n    def get_buffer(self):\n        return Variable(self.buffer).get() if self.buffer else None\n\n    def _buffer_window(self, input_cols, lower_bound, upper_bound):\n        def func(value):\n            return value[lower_bound:upper_bound]\n\n        return PandasDataFrame(self.data[input_cols].partitions[0].map_partitions(func).compute())\n\n    def sample(self, n=10, random=False):\n        \"\"\"\n        Return a n number of sample from a dataFrame\n        :param n: Number of samples\n        :param random: if true get a semi random sample\n        :return:\n        \"\"\"\n        df = self.root\n        if random is True:\n            seed = random_int()\n        elif random is False:\n            seed = 0\n        else:\n            RaiseIt.value_error(random, [\"True\", \"False\"])\n\n        rows_count = df.rows.count()\n        if n < rows_count:\n            # n/rows_count can return a number that represent less the total number we expect. multiply by 1.1\n            fraction = (n / rows_count) * 1.1\n        else:\n            fraction = 1.0\n        return self.root.new(df.data.sample(frac=fraction, random_state=seed))\n\n    def stratified_sample(self, col_name, seed: int = 1):\n        \"\"\"\n        Stratified Sampling\n        :param col_name:\n        :param seed:\n        :return:\n        \"\"\"\n        df = self.data\n        n = min(5, df[col_name].value_counts().min())\n        df = df.groupby(col_name).apply(lambda x: x.sample(2))\n        # df_.index = df_.index.droplevel(0)\n        return self.root.new(df)\n\n    @staticmethod\n    def pivot(index, column, values):\n        \"\"\"\n        Return reshaped DataFrame organized by given index / column values.\n        :param index: Column to use to make new frame's index.\n        :param column: Column to use to make new frame's columns.\n        :param values: Column(s) to use for populating new frame's values.\n        :return:\n        \"\"\"\n        raise NotImplementedError\n\n    @staticmethod\n    def melt(id_vars, value_vars, var_name=\"variable\", value_name=\"value\", data_type=\"str\"):\n        \"\"\"\n        Convert DataFrame from wide to long format.\n        :param id_vars: column with unique values\n        :param value_vars: Column names that are going to be converted to columns values\n        :param var_name: Column name for vars\n        :param value_name: Column name for values\n        :param data_type: All columns must have the same type. It will transform all columns to this data type.\n        :return:\n        \"\"\"\n\n        raise NotImplementedError\n\n    def size(self, deep=False, format=None):\n        \"\"\"\n        Get the size of a dask in bytes\n        :return:\n        \"\"\"\n        df = self.data\n        result = df.memory_usage(index=True, deep=deep).sum().compute()\n        if format == \"human\":\n            result = humanize.naturalsize(result)\n\n        return result\n\n    def run(self):\n        \"\"\"\n        This method is a very useful function to break lineage of transformations. By default Spark uses the lazy\n        evaluation approach in processing data: transformation functions are not computed into an action is called.\n        Sometimes when transformations are numerous, the computations are very extensive because the high number of\n        operations that spark needs to run in order to get the results.\n\n        Other important thing is that Apache Spark save task but not result of dataFrame, so tasks are\n        accumulated and the same situation happens.\n\n        :return:\n        \"\"\"\n        df = self.data\n        df.cache().count()\n        return df\n\n    @staticmethod\n    def query(sql_expression):\n        raise NotImplementedError\n\n    def partitions(self):\n        return self.data.npartitions\n\n    @staticmethod\n    def partitioner():\n        print(\"Dask not support custom partitioner\")\n        raise NotImplementedError\n\n    def repartition(self, n=None, *args, **kwargs):\n        dfd = self.data\n        df = self\n        if n == \"auto\":\n            # Follow a heuristic for partitioning a mentioned\n            # https://docs.dask.org/en/latest/best-practices.html#avoid-very-large-partitions\n            client = dask.distributed.get_client()\n            worker_memory = parse_bytes(client.cluster.worker_spec[0][\"options\"][\"memory_limit\"])\n            nthreads = client.cluster.worker_spec[0][\"options\"][\"nthreads\"]\n\n            part_recommended_size = worker_memory / nthreads / 10\n            n = int(df.size() / part_recommended_size)\n\n            # Partition can not be lower than 1\n            n = n if n < 0 else 1\n            # TODO .repartition(partition_size=\"100MB\"). https://stackoverflow.com/questions/44657631/strategy-for-partitioning-dask-dataframes-efficiently\n        dfd = dfd.repartition(npartitions=n, *args, **kwargs)\n\n        return self.new(dfd, meta=self.meta)\n\n    @staticmethod\n    def debug():\n        \"\"\"\n\n        :return:\n        \"\"\"\n        raise NotImplementedError\n\n    @staticmethod\n    def create_id(column=\"id\"):\n        \"\"\"\n        Create a unique id for every row.\n        :param column: Columns to be processed\n        :return:\n        \"\"\"\n\n        raise NotImplementedError\n\n    def to_dict(self, orient=\"records\", index=True):\n        \"\"\"\n        Create a dict\n        :param orient:\n        :param index: Return the series index\n        :return:\n        \"\"\"\n\n        series = self.data\n        if index is True:\n            return series.compute().to_dict(orient)\n        else:\n            return series.compute().to_list()\n\n    def to_pandas(self):\n        return self.data.compute()\n\n    @property\n    def constants(self):\n        from optimus.engines.base.dask.constants import constants\n        return constants(self)\n", "path": "optimus/engines/base/dask/dataframe.py"}]}