Environment Operations

def get_environment(
    self,
    request: Optional[Union[cloudshell.GetEnvironmentRequest, dict]] = None,
    *,
    name: Optional[str] = None,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> cloudshell.Environment:
    """
    Get environment details. Returns NOT_FOUND if environment does not exist.
    
    Args:
        request: GetEnvironmentRequest object or dict
        name: Environment resource name (users/{user}/environments/{environment})
        retry: Retry configuration for the request
        timeout: Timeout for the request
        metadata: Additional metadata to send with the request
        
    Returns:
        Environment object with current state and connection details
        
    Raises:
        google.api_core.exceptions.NotFound: Environment does not exist
    """

def start_environment(
    self,
    request: Optional[Union[cloudshell.StartEnvironmentRequest, dict]] = None,
    *,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> operation.Operation:
    """
    Start an existing environment. This is a long-running operation.
    
    Args:
        request: StartEnvironmentRequest object or dict
        retry: Retry configuration for the request
        timeout: Timeout for the request
        metadata: Additional metadata to send with the request
        
    Returns:
        Long-running operation that resolves to StartEnvironmentResponse
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: API call failed
    """

def authorize_environment(
    self,
    request: Optional[Union[cloudshell.AuthorizeEnvironmentRequest, dict]] = None,
    *,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> operation.Operation:
    """
    Send OAuth credentials to running environment. This is a long-running operation.
    
    Args:
        request: AuthorizeEnvironmentRequest object or dict
        retry: Retry configuration for the request
        timeout: Timeout for the request
        metadata: Additional metadata to send with the request
        
    Returns:
        Long-running operation that resolves to AuthorizeEnvironmentResponse
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: API call failed
    """

class GetEnvironmentRequest(proto.Message):
    """Request to get environment details."""
    
    name: str  # Required. Environment resource name (e.g., users/me/environments/default)

class StartEnvironmentRequest(proto.Message):
    """Request to start environment."""
    
    name: str                          # Environment name to start
    access_token: str                  # Optional initial access token
    public_keys: MutableSequence[str]  # Public keys to add before starting

class AuthorizeEnvironmentRequest(proto.Message):
    """Request to send OAuth credentials to environment."""
    
    name: str                                        # Environment name
    access_token: str                               # OAuth access token
    id_token: str                                   # OAuth ID token
    expire_time: google.protobuf.timestamp_pb2.Timestamp  # Token expiration time

class Environment(proto.Message):
    """A Cloud Shell environment combining Docker image and persistent home directory."""
    
    name: str                           # Immutable resource name: users/{owner}/environments/{id}
    id: str                            # Output only environment identifier
    docker_image: str                  # Required immutable Docker image path
    state: Environment.State           # Output only current execution state
    web_host: str                      # Output only HTTPS/WSS connection host
    ssh_username: str                  # Output only SSH username
    ssh_host: str                      # Output only SSH connection host
    ssh_port: int                      # Output only SSH connection port
    public_keys: MutableSequence[str]  # Output only associated public keys

class State(proto.Enum):
    """Possible execution states for an environment."""
    STATE_UNSPECIFIED = 0  # Unknown state
    SUSPENDED = 1          # Not running, can be started
    PENDING = 2           # Starting but not ready
    RUNNING = 3           # Running and ready for connections
    DELETING = 4          # Being deleted

class StartEnvironmentResponse(proto.Message):
    """Response from start environment operation."""
    
    environment: Environment  # Started environment

class AuthorizeEnvironmentResponse(proto.Message):
    """Response from authorize environment operation."""
    # Empty message

class StartEnvironmentMetadata(proto.Message):
    """Metadata for start environment operation."""
    
    state: StartEnvironmentMetadata.State  # Current startup state

class State(proto.Enum):
    """Current startup state."""
    STATE_UNSPECIFIED = 0           # Unknown state
    STARTING = 1                    # Starting without details
    UNARCHIVING_DISK = 2           # Unarchiving user disk
    AWAITING_COMPUTE_RESOURCES = 4  # Waiting for compute resources
    FINISHED = 3                    # Startup complete

class CreateEnvironmentMetadata(proto.Message):
    """Operation metadata for create operations."""
    # Empty message

class DeleteEnvironmentMetadata(proto.Message):
    """Operation metadata for delete operations."""
    # Empty message

class AuthorizeEnvironmentMetadata(proto.Message):
    """Operation metadata for authorize operations."""
    # Empty message

from google.cloud.shell import CloudShellServiceClient

client = CloudShellServiceClient()

# Get default environment
environment = client.get_environment(
    name="users/me/environments/default"
)

print(f"Environment ID: {environment.id}")
print(f"State: {environment.state}")
print(f"Docker Image: {environment.docker_image}")

if environment.state == environment.State.RUNNING:
    print(f"SSH: {environment.ssh_username}@{environment.ssh_host}:{environment.ssh_port}")
    print(f"Web Host: {environment.web_host}")
    print(f"Public Keys: {len(environment.public_keys)} keys")

from google.cloud.shell import CloudShellServiceClient, StartEnvironmentRequest

client = CloudShellServiceClient()

# Start environment with SSH keys
request = StartEnvironmentRequest(
    name="users/me/environments/default",
    public_keys=[
        "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABgQC... user@example.com"
    ]
)

operation = client.start_environment(request=request)
print(f"Operation name: {operation.name}")

# Wait for completion
response = operation.result(timeout=300)  # 5 minute timeout
print(f"Started environment: {response.environment.name}")
print(f"Environment state: {response.environment.state}")

# Monitor operation metadata
metadata = operation.metadata
if hasattr(metadata, 'state'):
    print(f"Startup state: {metadata.state}")

from google.cloud.shell import CloudShellServiceClient, AuthorizeEnvironmentRequest
from google.protobuf import timestamp_pb2
import datetime

client = CloudShellServiceClient()

# Create expiration timestamp (1 hour from now)
expire_time = timestamp_pb2.Timestamp()
expire_time.FromDatetime(datetime.datetime.utcnow() + datetime.timedelta(hours=1))

request = AuthorizeEnvironmentRequest(
    name="users/me/environments/default",
    access_token="ya29.c.KqKB...",
    id_token="eyJhbGciOiJSUzI1NiI...",
    expire_time=expire_time
)

operation = client.authorize_environment(request=request)
response = operation.result()
print("OAuth credentials sent to environment")

import asyncio
from google.cloud.shell import CloudShellServiceAsyncClient

async def start_environment_async():
    async with CloudShellServiceAsyncClient() as client:
        # Get environment status
        environment = await client.get_environment(
            name="users/me/environments/default"
        )
        
        if environment.state != environment.State.RUNNING:
            # Start if not running
            operation = await client.start_environment(
                name="users/me/environments/default"
            )
            response = await operation.result()
            print(f"Started: {response.environment.name}")
        else:
            print("Environment already running")

asyncio.run(start_environment_async())

from google.cloud.shell import CloudShellServiceClient
from google.api_core import exceptions

client = CloudShellServiceClient()

try:
    environment = client.get_environment(
        name="users/me/environments/nonexistent"
    )
except exceptions.NotFound:
    print("Environment not found")
except exceptions.GoogleAPICallError as e:
    print(f"API call failed: {e}")