We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
specifying ssh port for RemoteTarget
It would be nice to be able to specify a port number when establishing a RemoteTarget.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `luigi/contrib/ssh.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 #
3 # Copyright 2012-2015 Spotify AB
4 #
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
8 #
9 # http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #
17 """
18 Light-weight remote execution library and utilities.
19 
20 There are some examples in the unittest, but
21 I added another more luigi-specific in the examples directory (examples/ssh_remote_execution.py
22 
23 :class:`RemoteContext` is meant to provide functionality similar to that of the standard library subprocess module,
24 but where the commands executed are run on a remote machine instead,
25 without the user having to think about prefixing everything with "ssh" and credentials etc.
26 
27 Using this mini library (which is just a convenience wrapper for subprocess),
28 :class:`RemoteTarget` is created to let you stream data from a remotely stored file using
29 the luigi :class:`~luigi.target.FileSystemTarget` semantics.
30 
31 As a bonus, :class:`RemoteContext` also provides a really cool feature that let's you
32 set up ssh tunnels super easily using a python context manager (there is an example
33 in the integration part of unittests).
34 
35 This can be super convenient when you want secure communication using a non-secure
36 protocol or circumvent firewalls (as long as they are open for ssh traffic).
37 """
38 
39 import contextlib
40 import os
41 import random
42 import subprocess
43 
44 import luigi
45 import luigi.format
46 import luigi.target
47 
48 
49 class RemoteContext(object):
50 
51     def __init__(self, host, username=None, key_file=None, connect_timeout=None):
52         self.host = host
53         self.username = username
54         self.key_file = key_file
55         self.connect_timeout = connect_timeout
56 
57     def __repr__(self):
58         return '%s(%r, %r, %r, %r)' % (
59             type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout)
60 
61     def __eq__(self, other):
62         return repr(self) == repr(other)
63 
64     def __hash__(self):
65         return hash(repr(self))
66 
67     def _host_ref(self):
68         if self.username:
69             return "{0}@{1}".format(self.username, self.host)
70         else:
71             return self.host
72 
73     def _prepare_cmd(self, cmd):
74         connection_cmd = ["ssh", self._host_ref(),
75                           "-S", "none",  # disable ControlMaster since it causes all sorts of weird behaviour with subprocesses...
76                           "-o", "BatchMode=yes",  # no password prompts etc
77                           ]
78 
79         if self.connect_timeout is not None:
80             connection_cmd += ['-o', 'ConnectTimeout=%d' % self.connect_timeout]
81 
82         if self.key_file:
83             connection_cmd.extend(["-i", self.key_file])
84         return connection_cmd + cmd
85 
86     def Popen(self, cmd, **kwargs):
87         """
88         Remote Popen.
89         """
90         prefixed_cmd = self._prepare_cmd(cmd)
91         return subprocess.Popen(prefixed_cmd, **kwargs)
92 
93     def check_output(self, cmd):
94         """
95         Execute a shell command remotely and return the output.
96 
97         Simplified version of Popen when you only want the output as a string and detect any errors.
98         """
99         p = self.Popen(cmd, stdout=subprocess.PIPE)
100         output, _ = p.communicate()
101         if p.returncode != 0:
102             raise subprocess.CalledProcessError(p.returncode, cmd)
103         return output
104 
105     @contextlib.contextmanager
106     def tunnel(self, local_port, remote_port=None, remote_host="localhost"):
107         """
108         Open a tunnel between localhost:local_port and remote_host:remote_port via the host specified by this context.
109 
110         Remember to close() the returned "tunnel" object in order to clean up
111         after yourself when you are done with the tunnel.
112         """
113         tunnel_host = "{0}:{1}:{2}".format(local_port, remote_host, remote_port)
114         proc = self.Popen(
115             # cat so we can shut down gracefully by closing stdin
116             ["-L", tunnel_host, "echo -n ready && cat"],
117             stdin=subprocess.PIPE,
118             stdout=subprocess.PIPE,
119         )
120         # make sure to get the data so we know the connection is established
121         ready = proc.stdout.read(5)
122         assert ready == b"ready", "Didn't get ready from remote echo"
123         yield  # user code executed here
124         proc.communicate()
125         assert proc.returncode == 0, "Tunnel process did an unclean exit (returncode %s)" % (proc.returncode,)
126 
127 
128 class RemoteFileSystem(luigi.target.FileSystem):
129 
130     def __init__(self, host, username=None, key_file=None):
131         self.remote_context = RemoteContext(host, username, key_file)
132 
133     def exists(self, path):
134         """
135         Return `True` if file or directory at `path` exist, False otherwise.
136         """
137         try:
138             self.remote_context.check_output(["test", "-e", path])
139         except subprocess.CalledProcessError as e:
140             if e.returncode == 1:
141                 return False
142             else:
143                 raise
144         return True
145 
146     def remove(self, path, recursive=True):
147         """
148         Remove file or directory at location `path`.
149         """
150         if recursive:
151             cmd = ["rm", "-r", path]
152         else:
153             cmd = ["rm", path]
154 
155         self.remote_context.check_output(cmd)
156 
157     def _scp(self, src, dest):
158         cmd = ["scp", "-q", "-B", "-C", "-o", "ControlMaster=no"]
159         if self.remote_context.key_file:
160             cmd.extend(["-i", self.remote_context.key_file])
161         cmd.extend([src, dest])
162         p = subprocess.Popen(cmd)
163         output, _ = p.communicate()
164         if p.returncode != 0:
165             raise subprocess.CalledProcessError(p.returncode, cmd)
166 
167     def put(self, local_path, path):
168         # create parent folder if not exists
169         normpath = os.path.normpath(path)
170         folder = os.path.dirname(normpath)
171         if folder and not self.exists(folder):
172             self.remote_context.check_output(['mkdir', '-p', folder])
173 
174         tmp_path = path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)
175         self._scp(local_path, "%s:%s" % (self.remote_context._host_ref(), tmp_path))
176         self.remote_context.check_output(['mv', tmp_path, path])
177 
178     def get(self, path, local_path):
179         # Create folder if it does not exist
180         normpath = os.path.normpath(local_path)
181         folder = os.path.dirname(normpath)
182         if folder and not os.path.exists(folder):
183             os.makedirs(folder)
184 
185         tmp_local_path = local_path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)
186         self._scp("%s:%s" % (self.remote_context._host_ref(), path), tmp_local_path)
187         os.rename(tmp_local_path, local_path)
188 
189 
190 class AtomicRemoteFileWriter(luigi.format.OutputPipeProcessWrapper):
191 
192     def __init__(self, fs, path):
193         self._fs = fs
194         self.path = path
195 
196         # create parent folder if not exists
197         normpath = os.path.normpath(self.path)
198         folder = os.path.dirname(normpath)
199         if folder and not self.fs.exists(folder):
200             self.fs.remote_context.check_output(['mkdir', '-p', folder])
201 
202         self.__tmp_path = self.path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)
203         super(AtomicRemoteFileWriter, self).__init__(
204             self.fs.remote_context._prepare_cmd(['cat', '>', self.__tmp_path]))
205 
206     def __del__(self):
207         super(AtomicRemoteFileWriter, self).__del__()
208         if self.fs.exists(self.__tmp_path):
209             self.fs.remote_context.check_output(['rm', self.__tmp_path])
210 
211     def close(self):
212         super(AtomicRemoteFileWriter, self).close()
213         self.fs.remote_context.check_output(['mv', self.__tmp_path, self.path])
214 
215     @property
216     def tmp_path(self):
217         return self.__tmp_path
218 
219     @property
220     def fs(self):
221         return self._fs
222 
223 
224 class RemoteTarget(luigi.target.FileSystemTarget):
225     """
226     Target used for reading from remote files.
227 
228     The target is implemented using ssh commands streaming data over the network.
229     """
230 
231     def __init__(self, path, host, format=None, username=None, key_file=None):
232         super(RemoteTarget, self).__init__(path)
233         if format is None:
234             format = luigi.format.get_default_format()
235         self.format = format
236         self._fs = RemoteFileSystem(host, username, key_file)
237 
238     @property
239     def fs(self):
240         return self._fs
241 
242     def open(self, mode='r'):
243         if mode == 'w':
244             file_writer = AtomicRemoteFileWriter(self.fs, self.path)
245             if self.format:
246                 return self.format.pipe_writer(file_writer)
247             else:
248                 return file_writer
249         elif mode == 'r':
250             file_reader = luigi.format.InputPipeProcessWrapper(
251                 self.fs.remote_context._prepare_cmd(["cat", self.path]))
252             if self.format:
253                 return self.format.pipe_reader(file_reader)
254             else:
255                 return file_reader
256         else:
257             raise Exception("mode must be r/w")
258 
259     def put(self, local_path):
260         self.fs.put(local_path, self.path)
261 
262     def get(self, local_path):
263         self.fs.get(self.path, local_path)
264
```
--- 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/luigi/contrib/ssh.py b/luigi/contrib/ssh.py
--- a/luigi/contrib/ssh.py
+++ b/luigi/contrib/ssh.py
@@ -48,15 +48,16 @@
 
 class RemoteContext(object):
 
-    def __init__(self, host, username=None, key_file=None, connect_timeout=None):
+    def __init__(self, host, username=None, key_file=None, connect_timeout=None, port=None):
         self.host = host
         self.username = username
         self.key_file = key_file
         self.connect_timeout = connect_timeout
+        self.port = port
 
     def __repr__(self):
-        return '%s(%r, %r, %r, %r)' % (
-            type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout)
+        return '%s(%r, %r, %r, %r, %r)' % (
+            type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout, self.port)
 
     def __eq__(self, other):
         return repr(self) == repr(other)
@@ -75,6 +76,8 @@
                           "-S", "none",  # disable ControlMaster since it causes all sorts of weird behaviour with subprocesses...
                           "-o", "BatchMode=yes",  # no password prompts etc
                           ]
+        if self.port:
+            connection_cmd.extend(["-p", self.port])
 
         if self.connect_timeout is not None:
             connection_cmd += ['-o', 'ConnectTimeout=%d' % self.connect_timeout]
@@ -127,8 +130,8 @@
 
 class RemoteFileSystem(luigi.target.FileSystem):
 
-    def __init__(self, host, username=None, key_file=None):
-        self.remote_context = RemoteContext(host, username, key_file)
+    def __init__(self, host, username=None, key_file=None, port=None):
+        self.remote_context = RemoteContext(host, username, key_file, port)
 
     def exists(self, path):
         """
@@ -158,6 +161,8 @@
         cmd = ["scp", "-q", "-B", "-C", "-o", "ControlMaster=no"]
         if self.remote_context.key_file:
             cmd.extend(["-i", self.remote_context.key_file])
+        if self.remote_context.port:
+            cmd.extend(["-P", self.remote_context.port])
         cmd.extend([src, dest])
         p = subprocess.Popen(cmd)
         output, _ = p.communicate()
@@ -228,12 +233,12 @@
     The target is implemented using ssh commands streaming data over the network.
     """
 
-    def __init__(self, path, host, format=None, username=None, key_file=None):
+    def __init__(self, path, host, format=None, username=None, key_file=None, port=None):
         super(RemoteTarget, self).__init__(path)
         if format is None:
             format = luigi.format.get_default_format()
         self.format = format
-        self._fs = RemoteFileSystem(host, username, key_file)
+        self._fs = RemoteFileSystem(host, username, key_file, port)
 
     @property
     def fs(self):

{"golden_diff": "diff --git a/luigi/contrib/ssh.py b/luigi/contrib/ssh.py\n--- a/luigi/contrib/ssh.py\n+++ b/luigi/contrib/ssh.py\n@@ -48,15 +48,16 @@\n \n class RemoteContext(object):\n \n-    def __init__(self, host, username=None, key_file=None, connect_timeout=None):\n+    def __init__(self, host, username=None, key_file=None, connect_timeout=None, port=None):\n         self.host = host\n         self.username = username\n         self.key_file = key_file\n         self.connect_timeout = connect_timeout\n+        self.port = port\n \n     def __repr__(self):\n-        return '%s(%r, %r, %r, %r)' % (\n-            type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout)\n+        return '%s(%r, %r, %r, %r, %r)' % (\n+            type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout, self.port)\n \n     def __eq__(self, other):\n         return repr(self) == repr(other)\n@@ -75,6 +76,8 @@\n                           \"-S\", \"none\",  # disable ControlMaster since it causes all sorts of weird behaviour with subprocesses...\n                           \"-o\", \"BatchMode=yes\",  # no password prompts etc\n                           ]\n+        if self.port:\n+            connection_cmd.extend([\"-p\", self.port])\n \n         if self.connect_timeout is not None:\n             connection_cmd += ['-o', 'ConnectTimeout=%d' % self.connect_timeout]\n@@ -127,8 +130,8 @@\n \n class RemoteFileSystem(luigi.target.FileSystem):\n \n-    def __init__(self, host, username=None, key_file=None):\n-        self.remote_context = RemoteContext(host, username, key_file)\n+    def __init__(self, host, username=None, key_file=None, port=None):\n+        self.remote_context = RemoteContext(host, username, key_file, port)\n \n     def exists(self, path):\n         \"\"\"\n@@ -158,6 +161,8 @@\n         cmd = [\"scp\", \"-q\", \"-B\", \"-C\", \"-o\", \"ControlMaster=no\"]\n         if self.remote_context.key_file:\n             cmd.extend([\"-i\", self.remote_context.key_file])\n+        if self.remote_context.port:\n+            cmd.extend([\"-P\", self.remote_context.port])\n         cmd.extend([src, dest])\n         p = subprocess.Popen(cmd)\n         output, _ = p.communicate()\n@@ -228,12 +233,12 @@\n     The target is implemented using ssh commands streaming data over the network.\n     \"\"\"\n \n-    def __init__(self, path, host, format=None, username=None, key_file=None):\n+    def __init__(self, path, host, format=None, username=None, key_file=None, port=None):\n         super(RemoteTarget, self).__init__(path)\n         if format is None:\n             format = luigi.format.get_default_format()\n         self.format = format\n-        self._fs = RemoteFileSystem(host, username, key_file)\n+        self._fs = RemoteFileSystem(host, username, key_file, port)\n \n     @property\n     def fs(self):\n", "issue": "specifying ssh port for RemoteTarget\nIt would be nice to be able to specify a port number when establishing a RemoteTarget.\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\"\"\"\nLight-weight remote execution library and utilities.\n\nThere are some examples in the unittest, but\nI added another more luigi-specific in the examples directory (examples/ssh_remote_execution.py\n\n:class:`RemoteContext` is meant to provide functionality similar to that of the standard library subprocess module,\nbut where the commands executed are run on a remote machine instead,\nwithout the user having to think about prefixing everything with \"ssh\" and credentials etc.\n\nUsing this mini library (which is just a convenience wrapper for subprocess),\n:class:`RemoteTarget` is created to let you stream data from a remotely stored file using\nthe luigi :class:`~luigi.target.FileSystemTarget` semantics.\n\nAs a bonus, :class:`RemoteContext` also provides a really cool feature that let's you\nset up ssh tunnels super easily using a python context manager (there is an example\nin the integration part of unittests).\n\nThis can be super convenient when you want secure communication using a non-secure\nprotocol or circumvent firewalls (as long as they are open for ssh traffic).\n\"\"\"\n\nimport contextlib\nimport os\nimport random\nimport subprocess\n\nimport luigi\nimport luigi.format\nimport luigi.target\n\n\nclass RemoteContext(object):\n\n    def __init__(self, host, username=None, key_file=None, connect_timeout=None):\n        self.host = host\n        self.username = username\n        self.key_file = key_file\n        self.connect_timeout = connect_timeout\n\n    def __repr__(self):\n        return '%s(%r, %r, %r, %r)' % (\n            type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout)\n\n    def __eq__(self, other):\n        return repr(self) == repr(other)\n\n    def __hash__(self):\n        return hash(repr(self))\n\n    def _host_ref(self):\n        if self.username:\n            return \"{0}@{1}\".format(self.username, self.host)\n        else:\n            return self.host\n\n    def _prepare_cmd(self, cmd):\n        connection_cmd = [\"ssh\", self._host_ref(),\n                          \"-S\", \"none\",  # disable ControlMaster since it causes all sorts of weird behaviour with subprocesses...\n                          \"-o\", \"BatchMode=yes\",  # no password prompts etc\n                          ]\n\n        if self.connect_timeout is not None:\n            connection_cmd += ['-o', 'ConnectTimeout=%d' % self.connect_timeout]\n\n        if self.key_file:\n            connection_cmd.extend([\"-i\", self.key_file])\n        return connection_cmd + cmd\n\n    def Popen(self, cmd, **kwargs):\n        \"\"\"\n        Remote Popen.\n        \"\"\"\n        prefixed_cmd = self._prepare_cmd(cmd)\n        return subprocess.Popen(prefixed_cmd, **kwargs)\n\n    def check_output(self, cmd):\n        \"\"\"\n        Execute a shell command remotely and return the output.\n\n        Simplified version of Popen when you only want the output as a string and detect any errors.\n        \"\"\"\n        p = self.Popen(cmd, stdout=subprocess.PIPE)\n        output, _ = p.communicate()\n        if p.returncode != 0:\n            raise subprocess.CalledProcessError(p.returncode, cmd)\n        return output\n\n    @contextlib.contextmanager\n    def tunnel(self, local_port, remote_port=None, remote_host=\"localhost\"):\n        \"\"\"\n        Open a tunnel between localhost:local_port and remote_host:remote_port via the host specified by this context.\n\n        Remember to close() the returned \"tunnel\" object in order to clean up\n        after yourself when you are done with the tunnel.\n        \"\"\"\n        tunnel_host = \"{0}:{1}:{2}\".format(local_port, remote_host, remote_port)\n        proc = self.Popen(\n            # cat so we can shut down gracefully by closing stdin\n            [\"-L\", tunnel_host, \"echo -n ready && cat\"],\n            stdin=subprocess.PIPE,\n            stdout=subprocess.PIPE,\n        )\n        # make sure to get the data so we know the connection is established\n        ready = proc.stdout.read(5)\n        assert ready == b\"ready\", \"Didn't get ready from remote echo\"\n        yield  # user code executed here\n        proc.communicate()\n        assert proc.returncode == 0, \"Tunnel process did an unclean exit (returncode %s)\" % (proc.returncode,)\n\n\nclass RemoteFileSystem(luigi.target.FileSystem):\n\n    def __init__(self, host, username=None, key_file=None):\n        self.remote_context = RemoteContext(host, username, key_file)\n\n    def exists(self, path):\n        \"\"\"\n        Return `True` if file or directory at `path` exist, False otherwise.\n        \"\"\"\n        try:\n            self.remote_context.check_output([\"test\", \"-e\", path])\n        except subprocess.CalledProcessError as e:\n            if e.returncode == 1:\n                return False\n            else:\n                raise\n        return True\n\n    def remove(self, path, recursive=True):\n        \"\"\"\n        Remove file or directory at location `path`.\n        \"\"\"\n        if recursive:\n            cmd = [\"rm\", \"-r\", path]\n        else:\n            cmd = [\"rm\", path]\n\n        self.remote_context.check_output(cmd)\n\n    def _scp(self, src, dest):\n        cmd = [\"scp\", \"-q\", \"-B\", \"-C\", \"-o\", \"ControlMaster=no\"]\n        if self.remote_context.key_file:\n            cmd.extend([\"-i\", self.remote_context.key_file])\n        cmd.extend([src, dest])\n        p = subprocess.Popen(cmd)\n        output, _ = p.communicate()\n        if p.returncode != 0:\n            raise subprocess.CalledProcessError(p.returncode, cmd)\n\n    def put(self, local_path, path):\n        # create parent folder if not exists\n        normpath = os.path.normpath(path)\n        folder = os.path.dirname(normpath)\n        if folder and not self.exists(folder):\n            self.remote_context.check_output(['mkdir', '-p', folder])\n\n        tmp_path = path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        self._scp(local_path, \"%s:%s\" % (self.remote_context._host_ref(), tmp_path))\n        self.remote_context.check_output(['mv', tmp_path, path])\n\n    def get(self, path, local_path):\n        # Create folder if it does not exist\n        normpath = os.path.normpath(local_path)\n        folder = os.path.dirname(normpath)\n        if folder and not os.path.exists(folder):\n            os.makedirs(folder)\n\n        tmp_local_path = local_path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        self._scp(\"%s:%s\" % (self.remote_context._host_ref(), path), tmp_local_path)\n        os.rename(tmp_local_path, local_path)\n\n\nclass AtomicRemoteFileWriter(luigi.format.OutputPipeProcessWrapper):\n\n    def __init__(self, fs, path):\n        self._fs = fs\n        self.path = path\n\n        # create parent folder if not exists\n        normpath = os.path.normpath(self.path)\n        folder = os.path.dirname(normpath)\n        if folder and not self.fs.exists(folder):\n            self.fs.remote_context.check_output(['mkdir', '-p', folder])\n\n        self.__tmp_path = self.path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        super(AtomicRemoteFileWriter, self).__init__(\n            self.fs.remote_context._prepare_cmd(['cat', '>', self.__tmp_path]))\n\n    def __del__(self):\n        super(AtomicRemoteFileWriter, self).__del__()\n        if self.fs.exists(self.__tmp_path):\n            self.fs.remote_context.check_output(['rm', self.__tmp_path])\n\n    def close(self):\n        super(AtomicRemoteFileWriter, self).close()\n        self.fs.remote_context.check_output(['mv', self.__tmp_path, self.path])\n\n    @property\n    def tmp_path(self):\n        return self.__tmp_path\n\n    @property\n    def fs(self):\n        return self._fs\n\n\nclass RemoteTarget(luigi.target.FileSystemTarget):\n    \"\"\"\n    Target used for reading from remote files.\n\n    The target is implemented using ssh commands streaming data over the network.\n    \"\"\"\n\n    def __init__(self, path, host, format=None, username=None, key_file=None):\n        super(RemoteTarget, self).__init__(path)\n        if format is None:\n            format = luigi.format.get_default_format()\n        self.format = format\n        self._fs = RemoteFileSystem(host, username, key_file)\n\n    @property\n    def fs(self):\n        return self._fs\n\n    def open(self, mode='r'):\n        if mode == 'w':\n            file_writer = AtomicRemoteFileWriter(self.fs, self.path)\n            if self.format:\n                return self.format.pipe_writer(file_writer)\n            else:\n                return file_writer\n        elif mode == 'r':\n            file_reader = luigi.format.InputPipeProcessWrapper(\n                self.fs.remote_context._prepare_cmd([\"cat\", self.path]))\n            if self.format:\n                return self.format.pipe_reader(file_reader)\n            else:\n                return file_reader\n        else:\n            raise Exception(\"mode must be r/w\")\n\n    def put(self, local_path):\n        self.fs.put(local_path, self.path)\n\n    def get(self, local_path):\n        self.fs.get(self.path, local_path)\n", "path": "luigi/contrib/ssh.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\"\"\"\nLight-weight remote execution library and utilities.\n\nThere are some examples in the unittest, but\nI added another more luigi-specific in the examples directory (examples/ssh_remote_execution.py\n\n:class:`RemoteContext` is meant to provide functionality similar to that of the standard library subprocess module,\nbut where the commands executed are run on a remote machine instead,\nwithout the user having to think about prefixing everything with \"ssh\" and credentials etc.\n\nUsing this mini library (which is just a convenience wrapper for subprocess),\n:class:`RemoteTarget` is created to let you stream data from a remotely stored file using\nthe luigi :class:`~luigi.target.FileSystemTarget` semantics.\n\nAs a bonus, :class:`RemoteContext` also provides a really cool feature that let's you\nset up ssh tunnels super easily using a python context manager (there is an example\nin the integration part of unittests).\n\nThis can be super convenient when you want secure communication using a non-secure\nprotocol or circumvent firewalls (as long as they are open for ssh traffic).\n\"\"\"\n\nimport contextlib\nimport os\nimport random\nimport subprocess\n\nimport luigi\nimport luigi.format\nimport luigi.target\n\n\nclass RemoteContext(object):\n\n    def __init__(self, host, username=None, key_file=None, connect_timeout=None, port=None):\n        self.host = host\n        self.username = username\n        self.key_file = key_file\n        self.connect_timeout = connect_timeout\n        self.port = port\n\n    def __repr__(self):\n        return '%s(%r, %r, %r, %r, %r)' % (\n            type(self).__name__, self.host, self.username, self.key_file, self.connect_timeout, self.port)\n\n    def __eq__(self, other):\n        return repr(self) == repr(other)\n\n    def __hash__(self):\n        return hash(repr(self))\n\n    def _host_ref(self):\n        if self.username:\n            return \"{0}@{1}\".format(self.username, self.host)\n        else:\n            return self.host\n\n    def _prepare_cmd(self, cmd):\n        connection_cmd = [\"ssh\", self._host_ref(),\n                          \"-S\", \"none\",  # disable ControlMaster since it causes all sorts of weird behaviour with subprocesses...\n                          \"-o\", \"BatchMode=yes\",  # no password prompts etc\n                          ]\n        if self.port:\n            connection_cmd.extend([\"-p\", self.port])\n\n        if self.connect_timeout is not None:\n            connection_cmd += ['-o', 'ConnectTimeout=%d' % self.connect_timeout]\n\n        if self.key_file:\n            connection_cmd.extend([\"-i\", self.key_file])\n        return connection_cmd + cmd\n\n    def Popen(self, cmd, **kwargs):\n        \"\"\"\n        Remote Popen.\n        \"\"\"\n        prefixed_cmd = self._prepare_cmd(cmd)\n        return subprocess.Popen(prefixed_cmd, **kwargs)\n\n    def check_output(self, cmd):\n        \"\"\"\n        Execute a shell command remotely and return the output.\n\n        Simplified version of Popen when you only want the output as a string and detect any errors.\n        \"\"\"\n        p = self.Popen(cmd, stdout=subprocess.PIPE)\n        output, _ = p.communicate()\n        if p.returncode != 0:\n            raise subprocess.CalledProcessError(p.returncode, cmd)\n        return output\n\n    @contextlib.contextmanager\n    def tunnel(self, local_port, remote_port=None, remote_host=\"localhost\"):\n        \"\"\"\n        Open a tunnel between localhost:local_port and remote_host:remote_port via the host specified by this context.\n\n        Remember to close() the returned \"tunnel\" object in order to clean up\n        after yourself when you are done with the tunnel.\n        \"\"\"\n        tunnel_host = \"{0}:{1}:{2}\".format(local_port, remote_host, remote_port)\n        proc = self.Popen(\n            # cat so we can shut down gracefully by closing stdin\n            [\"-L\", tunnel_host, \"echo -n ready && cat\"],\n            stdin=subprocess.PIPE,\n            stdout=subprocess.PIPE,\n        )\n        # make sure to get the data so we know the connection is established\n        ready = proc.stdout.read(5)\n        assert ready == b\"ready\", \"Didn't get ready from remote echo\"\n        yield  # user code executed here\n        proc.communicate()\n        assert proc.returncode == 0, \"Tunnel process did an unclean exit (returncode %s)\" % (proc.returncode,)\n\n\nclass RemoteFileSystem(luigi.target.FileSystem):\n\n    def __init__(self, host, username=None, key_file=None, port=None):\n        self.remote_context = RemoteContext(host, username, key_file, port)\n\n    def exists(self, path):\n        \"\"\"\n        Return `True` if file or directory at `path` exist, False otherwise.\n        \"\"\"\n        try:\n            self.remote_context.check_output([\"test\", \"-e\", path])\n        except subprocess.CalledProcessError as e:\n            if e.returncode == 1:\n                return False\n            else:\n                raise\n        return True\n\n    def remove(self, path, recursive=True):\n        \"\"\"\n        Remove file or directory at location `path`.\n        \"\"\"\n        if recursive:\n            cmd = [\"rm\", \"-r\", path]\n        else:\n            cmd = [\"rm\", path]\n\n        self.remote_context.check_output(cmd)\n\n    def _scp(self, src, dest):\n        cmd = [\"scp\", \"-q\", \"-B\", \"-C\", \"-o\", \"ControlMaster=no\"]\n        if self.remote_context.key_file:\n            cmd.extend([\"-i\", self.remote_context.key_file])\n        if self.remote_context.port:\n            cmd.extend([\"-P\", self.remote_context.port])\n        cmd.extend([src, dest])\n        p = subprocess.Popen(cmd)\n        output, _ = p.communicate()\n        if p.returncode != 0:\n            raise subprocess.CalledProcessError(p.returncode, cmd)\n\n    def put(self, local_path, path):\n        # create parent folder if not exists\n        normpath = os.path.normpath(path)\n        folder = os.path.dirname(normpath)\n        if folder and not self.exists(folder):\n            self.remote_context.check_output(['mkdir', '-p', folder])\n\n        tmp_path = path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        self._scp(local_path, \"%s:%s\" % (self.remote_context._host_ref(), tmp_path))\n        self.remote_context.check_output(['mv', tmp_path, path])\n\n    def get(self, path, local_path):\n        # Create folder if it does not exist\n        normpath = os.path.normpath(local_path)\n        folder = os.path.dirname(normpath)\n        if folder and not os.path.exists(folder):\n            os.makedirs(folder)\n\n        tmp_local_path = local_path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        self._scp(\"%s:%s\" % (self.remote_context._host_ref(), path), tmp_local_path)\n        os.rename(tmp_local_path, local_path)\n\n\nclass AtomicRemoteFileWriter(luigi.format.OutputPipeProcessWrapper):\n\n    def __init__(self, fs, path):\n        self._fs = fs\n        self.path = path\n\n        # create parent folder if not exists\n        normpath = os.path.normpath(self.path)\n        folder = os.path.dirname(normpath)\n        if folder and not self.fs.exists(folder):\n            self.fs.remote_context.check_output(['mkdir', '-p', folder])\n\n        self.__tmp_path = self.path + '-luigi-tmp-%09d' % random.randrange(0, 1e10)\n        super(AtomicRemoteFileWriter, self).__init__(\n            self.fs.remote_context._prepare_cmd(['cat', '>', self.__tmp_path]))\n\n    def __del__(self):\n        super(AtomicRemoteFileWriter, self).__del__()\n        if self.fs.exists(self.__tmp_path):\n            self.fs.remote_context.check_output(['rm', self.__tmp_path])\n\n    def close(self):\n        super(AtomicRemoteFileWriter, self).close()\n        self.fs.remote_context.check_output(['mv', self.__tmp_path, self.path])\n\n    @property\n    def tmp_path(self):\n        return self.__tmp_path\n\n    @property\n    def fs(self):\n        return self._fs\n\n\nclass RemoteTarget(luigi.target.FileSystemTarget):\n    \"\"\"\n    Target used for reading from remote files.\n\n    The target is implemented using ssh commands streaming data over the network.\n    \"\"\"\n\n    def __init__(self, path, host, format=None, username=None, key_file=None, port=None):\n        super(RemoteTarget, self).__init__(path)\n        if format is None:\n            format = luigi.format.get_default_format()\n        self.format = format\n        self._fs = RemoteFileSystem(host, username, key_file, port)\n\n    @property\n    def fs(self):\n        return self._fs\n\n    def open(self, mode='r'):\n        if mode == 'w':\n            file_writer = AtomicRemoteFileWriter(self.fs, self.path)\n            if self.format:\n                return self.format.pipe_writer(file_writer)\n            else:\n                return file_writer\n        elif mode == 'r':\n            file_reader = luigi.format.InputPipeProcessWrapper(\n                self.fs.remote_context._prepare_cmd([\"cat\", self.path]))\n            if self.format:\n                return self.format.pipe_reader(file_reader)\n            else:\n                return file_reader\n        else:\n            raise Exception(\"mode must be r/w\")\n\n    def put(self, local_path):\n        self.fs.put(local_path, self.path)\n\n    def get(self, local_path):\n        self.fs.get(self.path, local_path)\n", "path": "luigi/contrib/ssh.py"}]}