FEAT-modin-project#1821: Add obtaining of df.dtypes.items()

Signed-off-by: Vasilij Litvinov <vasilij.n.litvinov@intel.com>

modin/experimental/cloud/rpyc_proxy.py

@@ -23,11 +23,17 @@
 def _batch_loads(items):
     return tuple(pickle.loads(item) for item in items)
 
+def _tuplize(arg):
+    '''turns any sequence or iterator into a flat tuple'''
+    return tuple(arg)
+
 class WrappingConnection(rpyc.Connection):
     def __init__(self, *a, **kw):
         super().__init__(*a, **kw)
         self._remote_batch_loads = None
         self._remote_cls_cache = {}
+        self._static_cache = {}
+
     def __wrap(self, local_obj):
         while True:
             # unwrap magic wrappers first
@@ -61,6 +67,21 @@ def deliver(self, args, kw):
 
         return tuple(delivered_args), delivered_kw
 
+    def obtain(self, remote):
+        while True:
+            try:
+                remote = object.__getattribute__(remote, '__remote_end__')
+            except AttributeError:
+                break
+        return pickle.loads(self._remote_dumps(remote))
+
+    def obtain_tuple(self, remote):
+        while True:
+            try:
+                remote = object.__getattribute__(remote, '__remote_end__')
+            except AttributeError:
+                break
+        return self._remote_tuplize(remote)
 
     def sync_request(self, handler, *args):
         if handler == consts.HANDLE_INSPECT:
@@ -159,6 +180,8 @@ def _box(self, obj):
 
     def _init_deliver(self):
         self._remote_batch_loads = self.modules["modin.experimental.cloud.rpyc_proxy"]._batch_loads
+        self._remote_dumps = self.modules["rpyc.lib.compat"].pickle.dumps
+        self._remote_tuplize = self.modules["modin.experimental.cloud.rpyc_proxy"]._tuplize
 
 
 class WrappingService(rpyc.ClassicService):
@@ -488,10 +511,28 @@ def make_dataframe_wrapper(DataFrame):
     It makes DF.loc, DF.groupby() and other methods listed below deliver their
     arguments to remote end by value.
     """
+
+    from modin.pandas.series import Series
+
+    conn = get_connection()
+    class ObtainingItems:
+        def items(self):
+            return conn.obtain_tuple(self.__remote_end__.items())
+    ObtainingItems = _deliveringWrapper(Series, mixin=ObtainingItems)
+
+
+    @property
+    def dtypes(self):
+        remote_dtypes = self.__remote_end__.dtypes
+        return ObtainingItems(__remote_end__=remote_dtypes)
+
+    DataFrameOverrides = _prepare_loc_mixin()
+    DataFrameOverrides.dtypes = dtypes
+
     DeliveringDataFrame = _deliveringWrapper(
         DataFrame,
         ["groupby", "agg", "aggregate", "__getitem__", "astype", "drop", "merge"],
-        _prepare_loc_mixin(),
+        DataFrameOverrides,
         "DataFrame",
     )
     return DeliveringDataFrame