tessl/pypi-grpc-google-iam-v1
gRPC client library for Google Cloud Identity and Access Management (IAM) services with protocol buffer definitions.
—
Pending
grpc-services.mddocs/
gRPC Services
Client and server classes for IAM policy service operations. This module provides both client stubs for making remote calls to IAM services and server-side servicer base classes for implementing IAM services.
Capabilities
Client Service Stub
Client stub for making gRPC calls to IAM policy services. Provides methods for setting, getting, and testing permissions on Google Cloud resources.
class IAMPolicyStub:
"""
Client stub for IAM Policy service.
Args:
channel: gRPC channel to the service endpoint
"""
def __init__(self, channel): ...
def SetIamPolicy(self, request: SetIamPolicyRequest, timeout=None, metadata=None, credentials=None) -> Policy:
"""
Sets the access control policy on a resource.
Args:
request (SetIamPolicyRequest): The request containing resource and policy
timeout (float, optional): Request timeout in seconds
metadata (List[Tuple[str, str]], optional): gRPC metadata
credentials (grpc.CallCredentials, optional): Call credentials
Returns:
Policy: The updated policy with etag
Raises:
grpc.RpcError: If the RPC fails
"""
def GetIamPolicy(self, request: GetIamPolicyRequest, timeout=None, metadata=None, credentials=None) -> Policy:
"""
Gets the access control policy for a resource.
Args:
request (GetIamPolicyRequest): The request containing resource name
timeout (float, optional): Request timeout in seconds
metadata (List[Tuple[str, str]], optional): gRPC metadata
credentials (grpc.CallCredentials, optional): Call credentials
Returns:
Policy: The current IAM policy for the resource
Raises:
grpc.RpcError: If the RPC fails
"""
def TestIamPermissions(self, request: TestIamPermissionsRequest, timeout=None, metadata=None, credentials=None) -> TestIamPermissionsResponse:
"""
Tests the specified permissions against the IAM access control policy.
Args:
request (TestIamPermissionsRequest): The request containing resource and permissions
timeout (float, optional): Request timeout in seconds
metadata (List[Tuple[str, str]], optional): gRPC metadata
credentials (grpc.CallCredentials, optional): Call credentials
Returns:
TestIamPermissionsResponse: Response containing permissions the caller has
Raises:
grpc.RpcError: If the RPC fails
"""Server Service Implementation
Base class for implementing IAM policy services on the server side.
class IAMPolicyServicer:
"""
Base class for implementing IAM Policy service.
Subclass this to provide actual implementations of the service methods.
"""
def SetIamPolicy(self, request: SetIamPolicyRequest, context: grpc.ServicerContext) -> Policy:
"""
Sets the access control policy on a resource.
Args:
request (SetIamPolicyRequest): The request containing resource and policy
context (grpc.ServicerContext): gRPC service context
Returns:
Policy: The updated policy
Raises:
NotImplementedError: Must be implemented by subclass
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def GetIamPolicy(self, request: GetIamPolicyRequest, context: grpc.ServicerContext) -> Policy:
"""
Gets the access control policy for a resource.
Args:
request (GetIamPolicyRequest): The request containing resource name
context (grpc.ServicerContext): gRPC service context
Returns:
Policy: The current IAM policy
Raises:
NotImplementedError: Must be implemented by subclass
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def TestIamPermissions(self, request: TestIamPermissionsRequest, context: grpc.ServicerContext) -> TestIamPermissionsResponse:
"""
Tests permissions against the IAM access control policy.
Args:
request (TestIamPermissionsRequest): The request containing resource and permissions
context (grpc.ServicerContext): gRPC service context
Returns:
TestIamPermissionsResponse: Response with permissions the caller has
Raises:
NotImplementedError: Must be implemented by subclass
"""
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')Service Registration
Utility function for registering an IAM policy servicer with a gRPC server.
def add_IAMPolicyServicer_to_server(servicer: IAMPolicyServicer, server: grpc.Server) -> None:
"""
Registers an IAM Policy servicer with a gRPC server.
Args:
servicer (IAMPolicyServicer): The servicer implementation
server (grpc.Server): The gRPC server to register with
"""Request/Response Messages
Core message types used in gRPC service operations.
class SetIamPolicyRequest:
"""
Request to set IAM policy on a resource.
Attributes:
resource (str): Resource name (e.g., "projects/my-project/topics/my-topic")
policy (Policy): The IAM policy to set
update_mask (google.protobuf.FieldMask): Optional field mask for partial updates
"""
resource: str
policy: Policy
update_mask: google.protobuf.FieldMask
class GetIamPolicyRequest:
"""
Request to get IAM policy from a resource.
Attributes:
resource (str): Resource name
options (GetPolicyOptions): Optional policy retrieval options
"""
resource: str
options: GetPolicyOptions
class TestIamPermissionsRequest:
"""
Request to test permissions on a resource.
Attributes:
resource (str): Resource name
permissions (List[str]): Permissions to test (e.g., ["storage.objects.get", "storage.objects.create"])
"""
resource: str
permissions: List[str]
class TestIamPermissionsResponse:
"""
Response containing permissions the caller has.
Attributes:
permissions (List[str]): Subset of requested permissions that the caller has
"""
permissions: List[str]Usage Examples
Client Usage
import grpc
from google.iam.v1 import iam_policy_pb2_grpc
from google.iam.v1 import iam_policy_pb2
from google.iam.v1 import policy_pb2
# Create secure channel to IAM service
credentials = grpc.ssl_channel_credentials()
channel = grpc.secure_channel('iam.googleapis.com:443', credentials)
# Create client stub
client = iam_policy_pb2_grpc.IAMPolicyStub(channel)
# Set IAM policy
policy = policy_pb2.Policy()
binding = policy_pb2.Binding()
binding.role = "roles/viewer"
binding.members.append("user:alice@example.com")
policy.bindings.append(binding)
set_request = iam_policy_pb2.SetIamPolicyRequest()
set_request.resource = "projects/my-project/topics/my-topic"
set_request.policy.CopyFrom(policy)
try:
response = client.SetIamPolicy(set_request, timeout=30.0)
print(f"Policy set successfully, etag: {response.etag}")
except grpc.RpcError as e:
print(f"Failed to set policy: {e.code()}: {e.details()}")
# Get IAM policy
get_request = iam_policy_pb2.GetIamPolicyRequest()
get_request.resource = "projects/my-project/topics/my-topic"
try:
policy_response = client.GetIamPolicy(get_request)
print(f"Retrieved policy with {len(policy_response.bindings)} bindings")
for binding in policy_response.bindings:
print(f" Role: {binding.role}, Members: {list(binding.members)}")
except grpc.RpcError as e:
print(f"Failed to get policy: {e.code()}: {e.details()}")
# Test permissions
test_request = iam_policy_pb2.TestIamPermissionsRequest()
test_request.resource = "projects/my-project/topics/my-topic"
test_request.permissions.extend([
"pubsub.topics.get",
"pubsub.topics.publish",
"pubsub.topics.delete"
])
try:
permissions_response = client.TestIamPermissions(test_request)
print(f"User has permissions: {list(permissions_response.permissions)}")
except grpc.RpcError as e:
print(f"Failed to test permissions: {e.code()}: {e.details()}")
# Clean up
channel.close()Server Implementation
import grpc
from concurrent import futures
from google.iam.v1 import iam_policy_pb2_grpc
from google.iam.v1 import iam_policy_pb2
from google.iam.v1 import policy_pb2
class MyIAMPolicyService(iam_policy_pb2_grpc.IAMPolicyServicer):
"""Custom IAM policy service implementation."""
def __init__(self):
# In practice, you'd have a database or other storage
self.policies = {}
def SetIamPolicy(self, request, context):
"""Implement policy setting logic."""
try:
# Validate resource name
if not request.resource:
context.abort(grpc.StatusCode.INVALID_ARGUMENT, "Resource name is required")
# Store policy (in practice, save to database)
self.policies[request.resource] = request.policy
# Return the policy with etag for optimistic concurrency
response_policy = policy_pb2.Policy()
response_policy.CopyFrom(request.policy)
response_policy.etag = b"etag-" + request.resource.encode()
return response_policy
except Exception as e:
context.abort(grpc.StatusCode.INTERNAL, f"Internal error: {str(e)}")
def GetIamPolicy(self, request, context):
"""Implement policy retrieval logic."""
try:
if request.resource not in self.policies:
context.abort(grpc.StatusCode.NOT_FOUND, "Policy not found")
return self.policies[request.resource]
except Exception as e:
context.abort(grpc.StatusCode.INTERNAL, f"Internal error: {str(e)}")
def TestIamPermissions(self, request, context):
"""Implement permission testing logic."""
try:
# In practice, you'd evaluate permissions against stored policies
# This is a simplified example
granted_permissions = []
if request.resource in self.policies:
policy = self.policies[request.resource]
# Simple check - in practice would be more complex
for binding in policy.bindings:
if "user:current-user@example.com" in binding.members:
# Grant all requested permissions for demo
granted_permissions = list(request.permissions)
break
response = iam_policy_pb2.TestIamPermissionsResponse()
response.permissions.extend(granted_permissions)
return response
except Exception as e:
context.abort(grpc.StatusCode.INTERNAL, f"Internal error: {str(e)}")
# Create and start server
def serve():
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
# Register service
iam_policy_pb2_grpc.add_IAMPolicyServicer_to_server(
MyIAMPolicyService(), server
)
# Listen on port
listen_addr = '[::]:50051'
server.add_insecure_port(listen_addr)
print(f"Starting server on {listen_addr}")
server.start()
server.wait_for_termination()
if __name__ == '__main__':
serve()Install with Tessl CLI
npx tessl i tessl/pypi-grpc-google-iam-v1