1+ #!/usr/bin/env python
2+
3+ from io import StringIO
4+ import sys
5+ import asyncio
6+ import paramiko
7+ from gitpod import Gitpod
8+ import gitpod .lib as util
9+ from gitpod .types .environment_spec_param import EnvironmentSpecParam
10+
11+ # Examples:
12+ # - ./examples/fs_access.py
13+ # - ./examples/fs_access.py https://github.com/gitpod-io/empty
14+ async def main (cleanup : util .Disposables ) -> None :
15+ client = Gitpod ()
16+
17+ context_url = sys .argv [1 ] if len (sys .argv ) > 1 else None
18+
19+ env_class = util .find_most_used_environment_class (client )
20+ if not env_class :
21+ print ("Error: No environment class found. Please create one first." )
22+ sys .exit (1 )
23+ print (f"Found environment class: { env_class .display_name } { env_class .description } )
24+
25+ print ("Generating SSH key pair" )
26+ key = paramiko .RSAKey .generate (2048 )
27+ private_key_file = StringIO ()
28+ key .write_private_key (private_key_file )
29+ private_key_file .seek (0 ) # Reset position to start
30+ public_key = f"{ key .get_name ()} { key .get_base64 ()}
31+
32+ print ("Creating environment with SSH access" )
33+ key_id = "fs-access-example"
34+ spec : EnvironmentSpecParam = {
35+ "desired_phase" : "ENVIRONMENT_PHASE_RUNNING" ,
36+ "machine" : {"class" : env_class .id },
37+ "ssh_public_keys" : [{
38+ "id" : key_id ,
39+ "value" : public_key
40+ }]
41+ }
42+ if context_url :
43+ spec ["content" ] = {
44+ "initializer" : {"specs" : [{
45+ "contextUrl" : {
46+ "url" : context_url
47+ }
48+ }]}
49+ }
50+
51+ environment_id = client .environments .create (spec = spec ).environment .id
52+ cleanup .add (lambda : client .environments .delete (environment_id = environment_id ))
53+
54+ env = util .EnvironmentState (client , environment_id )
55+ cleanup .add (lambda : asyncio .run (env .close ()))
56+
57+ print ("Waiting for environment to be running" )
58+ await env .wait_until_running ()
59+
60+ print ("Waiting for SSH key to be applied" )
61+ await env .wait_for_ssh_key_applied (key_id = key_id , key_value = public_key )
62+
63+ print ("Waiting for SSH URL" )
64+ ssh_url = await env .wait_for_ssh_url ()
65+
66+ print (f"Setting up SSH connection to { ssh_url } )
67+ # Parse ssh://username@host:port format
68+ url_parts = ssh_url .split ('://' )[- 1 ]
69+ username , rest = url_parts .split ('@' )
70+ host , port = rest .split (':' )
71+ port = int (port )
72+
73+ ssh = paramiko .SSHClient ()
74+ ssh .set_missing_host_key_policy (paramiko .AutoAddPolicy ())
75+ ssh .connect (
76+ hostname = host ,
77+ port = port ,
78+ username = username ,
79+ pkey = key
80+ )
81+ cleanup .add (lambda : ssh .close ())
82+
83+ print ("Creating SFTP client" )
84+ sftp = ssh .open_sftp ()
85+ cleanup .add (lambda : sftp .close ())
86+
87+ print ("Writing test file" )
88+ test_content = "Hello from Gitpod Python SDK!"
89+ with sftp .file ('test.txt' , 'w' ) as f :
90+ f .write (test_content )
91+
92+ with sftp .file ('test.txt' , 'r' ) as f :
93+ content = f .read ()
94+ print (f"File content: { content } )
95+
96+ if __name__ == "__main__" :
97+ disposables = util .Disposables ()
98+ with disposables :
99+ asyncio .run (main (disposables ))
0 commit comments