0
# Dagster SSH
1
2
SSH and SFTP integration for Dagster data orchestration workflows. This package provides secure remote connections to servers with support for multiple authentication methods, file transfer operations, and SSH tunneling capabilities designed to seamlessly integrate with Dagster jobs and assets.
3
4
## Package Information
5
6
- **Package Name**: dagster-ssh
7
- **Language**: Python
8
- **Installation**: `pip install dagster-ssh`
9
- **Dependencies**: dagster==1.11.9, sshtunnel, paramiko
10
11
## Core Imports
12
13
```python
14
from dagster_ssh import SSHResource, ssh_resource
15
```
16
17
Additional imports:
18
19
```python
20
from dagster_ssh import __version__
21
# or
22
from dagster_ssh.version import __version__
23
```
24
25
## Basic Usage
26
27
```python
28
from dagster_ssh import SSHResource
29
from dagster import Definitions, asset
30
31
# Create SSH resource with key-based authentication
32
ssh_resource = SSHResource(
33
remote_host="example.com",
34
username="myuser",
35
key_file="/path/to/private/key"
36
)
37
38
# Or with password authentication
39
ssh_resource_pwd = SSHResource(
40
remote_host="example.com",
41
username="myuser",
42
password="my_secure_password"
43
)
44
45
@asset
46
def download_remote_file(ssh: SSHResource):
47
# Download file from remote server
48
local_file = ssh.sftp_get("/remote/path/data.csv", "/local/path/data.csv")
49
return local_file
50
51
@asset
52
def upload_processed_file(ssh: SSHResource):
53
# Upload file to remote server
54
ssh.sftp_put("/remote/path/output.csv", "/local/path/processed.csv")
55
56
defs = Definitions(
57
assets=[download_remote_file, upload_processed_file],
58
resources={"ssh": ssh_resource}
59
)
60
```
61
62
## Capabilities
63
64
### SSH Resource Class
65
66
Modern Dagster resource using Pydantic configuration for SSH connections and file operations.
67
68
```python { .api }
69
class SSHResource(ConfigurableResource):
70
"""
71
A Dagster resource for establishing SSH connections and file operations.
72
73
Args:
74
remote_host (str): Hostname or IP address of remote server
75
remote_port (Optional[int]): SSH port (default: None, uses 22)
76
username (Optional[str]): Username for authentication (default: None, uses system user)
77
password (Optional[str]): Password for authentication (default: None)
78
key_file (Optional[str]): Path to SSH private key file (default: None)
79
key_string (Optional[str]): SSH private key as string (default: None)
80
timeout (int): Connection timeout in seconds (default: 10)
81
keepalive_interval (int): Keepalive packet interval (default: 30)
82
compress (bool): Enable transport compression (default: True)
83
no_host_key_check (bool): Disable host key verification (default: True)
84
allow_host_key_change (bool): Allow changed host keys (default: False)
85
"""
86
87
def setup_for_execution(self, context: InitResourceContext) -> None:
88
"""
89
Initialize resource for execution context.
90
91
Sets up the logger, creates RSA key from string if provided,
92
auto-detects username if not specified, and configures SSH proxy
93
from ~/.ssh/config if available.
94
95
Args:
96
context (InitResourceContext): Dagster resource initialization context
97
"""
98
99
def set_logger(self, logger: logging.Logger) -> None:
100
"""Set logger instance for the resource."""
101
102
def get_connection(self) -> SSHClient:
103
"""
104
Open SSH connection to remote host.
105
106
Returns:
107
paramiko.client.SSHClient: Connected SSH client
108
"""
109
110
def get_tunnel(
111
self,
112
remote_port,
113
remote_host="localhost",
114
local_port=None
115
) -> SSHTunnelForwarder:
116
"""
117
Create SSH tunnel forwarder.
118
119
Args:
120
remote_port: Remote port to forward
121
remote_host: Remote host to connect to (default: "localhost")
122
local_port: Local port to bind (default: None, auto-assign)
123
124
Returns:
125
SSHTunnelForwarder: Configured tunnel forwarder
126
"""
127
128
def sftp_get(self, remote_filepath: str, local_filepath: str) -> str:
129
"""
130
Download file from remote server via SFTP.
131
132
Args:
133
remote_filepath (str): Path to remote file
134
local_filepath (str): Local destination path
135
136
Returns:
137
str: Path to downloaded local file
138
"""
139
140
def sftp_put(
141
self,
142
remote_filepath: str,
143
local_filepath: str,
144
confirm: bool = True
145
) -> str:
146
"""
147
Upload file to remote server via SFTP.
148
149
Args:
150
remote_filepath (str): Remote destination path
151
local_filepath (str): Path to local file
152
confirm (bool): Confirm file transfer (default: True)
153
154
Returns:
155
str: Path to uploaded local file
156
"""
157
158
@property
159
def log(self) -> logging.Logger:
160
"""Logger instance for the resource."""
161
```
162
163
### SSH Resource Factory Function
164
165
Legacy Dagster resource function for creating SSH resources from configuration.
166
167
```python { .api }
168
@resource
169
def ssh_resource(init_context: InitResourceContext) -> SSHResource:
170
"""
171
Factory function for creating SSHResource from Dagster context.
172
173
Args:
174
init_context (InitResourceContext): Dagster resource initialization context
175
176
Returns:
177
SSHResource: Configured SSH resource instance
178
179
Configuration Schema:
180
remote_host (StringSource): Remote host to connect to (required)
181
remote_port (IntSource): SSH port (default: 22)
182
username (StringSource): Username for connection (optional)
183
password (StringSource): Password for authentication (optional)
184
key_file (StringSource): Key file path for authentication (optional)
185
key_string (StringSource): Key string for authentication (optional)
186
timeout (IntSource): Connection timeout (default: 10)
187
keepalive_interval (IntSource): Keepalive interval (default: 30)
188
compress (BoolSource): Enable compression (default: True)
189
no_host_key_check (BoolSource): Disable host key check (default: True)
190
allow_host_key_change (BoolSource): Allow host key changes (default: False)
191
"""
192
```
193
194
### Utility Functions
195
196
Helper functions for SSH key management.
197
198
```python { .api }
199
def key_from_str(key_str: str) -> RSAKey:
200
"""
201
Create paramiko SSH key from string representation.
202
203
Args:
204
key_str (str): String containing private key data
205
206
Returns:
207
RSAKey: RSA key object
208
209
Raises:
210
ValueError: If key string is invalid or cannot be parsed
211
"""
212
```
213
214
### Version Information
215
216
Access to package version information.
217
218
```python { .api }
219
__version__: str
220
# Version string for the dagster-ssh package (e.g., "0.27.9")
221
```
222
223
## Types
224
225
```python { .api }
226
from paramiko.client import SSHClient
227
from paramiko import RSAKey
228
from sshtunnel import SSHTunnelForwarder
229
from dagster._core.execution.context.init import InitResourceContext
230
import logging
231
```
232
233
## Advanced Usage Examples
234
235
### SSH Tunneling
236
237
```python
238
from dagster_ssh import SSHResource
239
from dagster import asset
240
241
@asset
242
def connect_through_tunnel(ssh: SSHResource):
243
# Create SSH tunnel to access remote database
244
tunnel = ssh.get_tunnel(remote_port=5432, local_port=5433)
245
tunnel.start()
246
247
try:
248
# Connect to database through tunnel using localhost:5433
249
# Database connection code here
250
pass
251
finally:
252
tunnel.stop()
253
```
254
255
### Multiple Authentication Methods
256
257
```python
258
# Key-based authentication with file
259
ssh_key_file = SSHResource(
260
remote_host="server.example.com",
261
username="deploy_user",
262
key_file="/home/user/.ssh/id_rsa"
263
)
264
265
# Key-based authentication with string
266
ssh_key_string = SSHResource(
267
remote_host="server.example.com",
268
username="deploy_user",
269
key_string="""-----BEGIN RSA PRIVATE KEY-----
270
MIIEpAIBAAKCAQEA...
271
-----END RSA PRIVATE KEY-----"""
272
)
273
274
# Password authentication
275
ssh_password = SSHResource(
276
remote_host="server.example.com",
277
username="deploy_user",
278
password="secure_password"
279
)
280
```
281
282
### File Transfer Operations
283
284
```python
285
from dagster import asset, Definitions
286
from dagster_ssh import SSHResource
287
288
@asset
289
def batch_file_transfer(ssh: SSHResource):
290
# Download multiple files
291
files = [
292
("/remote/data1.csv", "/local/data1.csv"),
293
("/remote/data2.csv", "/local/data2.csv"),
294
("/remote/config.json", "/local/config.json")
295
]
296
297
downloaded_files = []
298
for remote_path, local_path in files:
299
downloaded_file = ssh.sftp_get(remote_path, local_path)
300
downloaded_files.append(downloaded_file)
301
302
return downloaded_files
303
304
@asset
305
def upload_results(ssh: SSHResource):
306
# Upload processed results
307
ssh.sftp_put("/remote/results/output.csv", "/local/processed_data.csv")
308
ssh.sftp_put("/remote/logs/process.log", "/local/processing.log")
309
```
310
311
### Legacy Resource Configuration
312
313
```python
314
from dagster import Definitions, job, op
315
from dagster_ssh import ssh_resource
316
317
@op
318
def transfer_files(context, ssh):
319
# Use SSH resource in operation
320
ssh.sftp_get("/remote/file.txt", "/local/file.txt")
321
322
@job(resource_defs={"ssh": ssh_resource})
323
def ssh_transfer_job():
324
transfer_files()
325
326
defs = Definitions(
327
jobs=[ssh_transfer_job],
328
resources={
329
"ssh": ssh_resource.configured({
330
"remote_host": "example.com",
331
"username": "myuser",
332
"key_file": "/path/to/key"
333
})
334
}
335
)
336
```
337
338
## Authentication Methods
339
340
The SSH resource supports multiple authentication approaches:
341
342
1. **SSH Key File**: Most secure, specify `key_file` path
343
2. **SSH Key String**: Secure, embed key content in `key_string`
344
3. **Password**: Less secure, use `password` parameter
345
4. **SSH Config**: Automatic detection from `~/.ssh/config`
346
5. **System Keys**: Automatic key discovery via paramiko
347
348
## Security Considerations
349
350
- **Host Key Verification**: Default configuration disables host key checking (`no_host_key_check=True`)
351
- **Key-based Authentication**: Preferred over password authentication
352
- **Private Key Security**: Store keys securely, avoid hardcoding in source
353
- **Network Security**: Use appropriate network isolation and firewall rules
354
- **Connection Timeout**: Configure appropriate timeout values to prevent hanging connections
355
356
## Error Handling
357
358
Common exceptions and error patterns:
359
360
- **Connection Errors**: Network connectivity, authentication failures
361
- **File Transfer Errors**: Permission issues, disk space, invalid paths
362
- **Key Format Errors**: Invalid private key format in `key_from_str`
363
- **Timeout Errors**: Connection or operation timeouts