0
# Operations Management
1

2
Operations for managing and monitoring long-running service management tasks. Many service operations (enable, disable, batch operations) are asynchronous and return Operation objects that can be monitored for completion.
3

4
## Capabilities
5

6
### List Operations
7

8
Lists long-running operations associated with the Service Usage API, allowing you to monitor and track operation status.
9

10
```python { .api }
11
def list_operations(
12
    request: ListOperationsRequest,
13
    *,
14
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
15
    timeout: Optional[float] = None,
16
    metadata: Sequence[Tuple[str, str]] = ()
17
) -> ListOperationsPager:
18
    """
19
    List long-running operations.
20
    
21
    Args:
22
        request: The request object for listing operations
23
        retry: Retry configuration for the request
24
        timeout: Request timeout in seconds
25
        metadata: Additional metadata for the request
26
        
27
    Returns:
28
        Paginated iterator over Operation objects
29
        
30
    Raises:
31
        google.api_core.exceptions.GoogleAPICallError: If the request fails
32
    """
33
```
34

35
Usage example:
36

37
```python
38
from google.cloud import service_usage
39
from google.longrunning import operations_pb2
40

41
client = service_usage.ServiceUsageClient()
42

43
# List all operations
44
request = operations_pb2.ListOperationsRequest(
45
    name="projects/my-project-id/operations"
46
)
47

48
operations = client.list_operations(request=request)
49
for operation in operations:
50
    print(f"Operation: {operation.name}")
51
    print(f"Done: {operation.done}")
52
    if operation.done:
53
        print(f"Result: {operation.response}")
54
    else:
55
        print(f"Metadata: {operation.metadata}")
56

57
# List operations with filtering
58
filtered_request = operations_pb2.ListOperationsRequest(
59
    name="projects/my-project-id/operations",
60
    filter="done=false"  # Only show running operations
61
)
62

63
running_ops = client.list_operations(request=filtered_request)
64
print("Running operations:")
65
for op in running_ops:
66
    print(f"  {op.name}")
67
```
68

69
### Get Operation
70

71
Retrieves detailed information about a specific long-running operation, including its current status and results.
72

73
```python { .api }
74
def get_operation(
75
    request: GetOperationRequest,
76
    *,
77
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
78
    timeout: Optional[float] = None,
79
    metadata: Sequence[Tuple[str, str]] = ()
80
) -> Operation:
81
    """
82
    Get details of a long-running operation.
83
    
84
    Args:
85
        request: The request object containing operation name
86
        retry: Retry configuration for the request
87
        timeout: Request timeout in seconds
88
        metadata: Additional metadata for the request
89
        
90
    Returns:
91
        Operation object with current status and result
92
        
93
    Raises:
94
        google.api_core.exceptions.NotFound: If the operation doesn't exist
95
        google.api_core.exceptions.GoogleAPICallError: If the request fails
96
    """
97
```
98

99
Usage example:
100

101
```python
102
from google.cloud import service_usage
103
from google.longrunning import operations_pb2
104

105
client = service_usage.ServiceUsageClient()
106

107
# Get a specific operation by name
108
operation_name = "projects/my-project-id/operations/operation-123456789"
109
request = operations_pb2.GetOperationRequest(name=operation_name)
110

111
operation = client.get_operation(request=request)
112
print(f"Operation: {operation.name}")
113
print(f"Done: {operation.done}")
114

115
if operation.done:
116
    if operation.HasField('response'):
117
        print("Operation completed successfully")
118
        print(f"Response: {operation.response}")
119
    elif operation.HasField('error'):
120
        print("Operation failed")
121
        print(f"Error: {operation.error.message}")
122
else:
123
    print("Operation still running")
124
    print(f"Metadata: {operation.metadata}")
125
```
126

127
## Working with Long-Running Operations
128

129
### Operation Monitoring
130

131
```python
132
import time
133
from google.cloud import service_usage
134
from google.longrunning import operations_pb2
135

136
def monitor_operation(client, operation_name, timeout=300):
137
    """Monitor an operation until completion or timeout."""
138
    start_time = time.time()
139
    
140
    while time.time() - start_time < timeout:
141
        request = operations_pb2.GetOperationRequest(name=operation_name)
142
        operation = client.get_operation(request=request)
143
        
144
        if operation.done:
145
            if operation.HasField('response'):
146
                print("Operation completed successfully")
147
                return operation.response
148
            elif operation.HasField('error'):
149
                print(f"Operation failed: {operation.error.message}")
150
                return None
151
        
152
        print("Operation still running, waiting...")
153
        time.sleep(10)  # Wait 10 seconds before checking again
154
    
155
    print("Operation monitoring timed out")
156
    return None
157

158
# Usage
159
client = service_usage.ServiceUsageClient()
160

161
# Start a service enable operation
162
service_name = "projects/my-project-id/services/storage.googleapis.com"
163
enable_request = service_usage.EnableServiceRequest(name=service_name)
164
operation = client.enable_service(request=enable_request)
165

166
# Monitor the operation
167
result = monitor_operation(client, operation.name)
168
if result:
169
    print(f"Service enabled: {result}")
170
```
171

172
### Using Operation.result()
173

174
The recommended way to wait for operation completion is using the `result()` method:
175

176
```python
177
from google.cloud import service_usage
178

179
client = service_usage.ServiceUsageClient()
180

181
# Enable a service and wait for completion
182
service_name = "projects/my-project-id/services/storage.googleapis.com"
183
request = service_usage.EnableServiceRequest(name=service_name)
184

185
operation = client.enable_service(request=request)
186
print(f"Started operation: {operation.name}")
187

188
try:
189
    # Wait up to 5 minutes for completion
190
    result = operation.result(timeout=300)
191
    print(f"Service enabled successfully: {result.service.name}")
192
    print(f"Service state: {result.service.state}")
193
except Exception as e:
194
    print(f"Operation failed or timed out: {e}")
195
```
196

197
### Handling Operation Errors
198

199
```python
200
from google.cloud import service_usage
201
from google.api_core import exceptions
202

203
client = service_usage.ServiceUsageClient()
204

205
try:
206
    # Attempt to enable a service
207
    service_name = "projects/my-project-id/services/invalid-service.googleapis.com"
208
    request = service_usage.EnableServiceRequest(name=service_name)
209
    
210
    operation = client.enable_service(request=request)
211
    result = operation.result(timeout=60)
212
    
213
except exceptions.NotFound:
214
    print("Service not found")
215
except exceptions.PermissionDenied:
216
    print("Permission denied - check project permissions")
217
except exceptions.DeadlineExceeded:
218
    print("Operation timed out")
219
except Exception as e:
220
    print(f"Unexpected error: {e}")
221
```
222

223
## Request Types
224

225
```python { .api }
226
class ListOperationsRequest:
227
    """Request to list operations."""
228
    name: str  # Format: "projects/{project}/operations"
229
    filter: str  # Filter expression (e.g., "done=false")
230
    page_size: int  # Maximum number of operations per page
231
    page_token: str  # Token for pagination
232

233
class GetOperationRequest:
234
    """Request to get a specific operation."""
235
    name: str  # Full operation resource name
236
```
237

238
## Response Types
239

240
```python { .api }
241
from google.longrunning import operations_pb2
242
from google.rpc import status_pb2
243

244
class Operation:
245
    """A long-running operation."""
246
    name: str  # Operation resource name
247
    metadata: OperationMetadata  # Operation metadata
248
    done: bool  # Whether the operation is complete
249
    
250
    # One of the following will be set when done=True:
251
    error: status_pb2.Status  # Error information if operation failed
252
    response: Any  # Result if operation succeeded
253

254
class ListOperationsResponse:
255
    """Response from listing operations."""
256
    operations: List[Operation]  # List of operations
257
    next_page_token: str  # Token for next page
258

259
# Standard operations types
260
ListOperationsRequest = operations_pb2.ListOperationsRequest
261
GetOperationRequest = operations_pb2.GetOperationRequest
262
ListOperationsResponse = operations_pb2.ListOperationsResponse
263
Operation = operations_pb2.Operation
264
Status = status_pb2.Status
265
```
266

267
## Operation Metadata
268

269
```python { .api }
270
class OperationMetadata:
271
    """Metadata for Service Usage operations."""
272
    resource_names: List[str]  # Names of resources being operated on
273
```
274

275
The metadata provides information about which resources are being affected by the operation, helping you understand what services or configurations are being modified.
276

277
## Best Practices
278

279
1. **Always check operation completion**: Use `operation.result()` or poll with `get_operation()`
280
2. **Set appropriate timeouts**: Service operations can take several minutes
281
3. **Handle errors gracefully**: Check for both request errors and operation failures
282
4. **Monitor long-running operations**: Use `list_operations()` to track multiple operations
283
5. **Use batch operations**: More efficient than individual operations when possible