This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 23e3b14074 [python] Add REST policy management (#10384)
23e3b14074 is described below
commit 23e3b140748c70d13fa0f2f8c831900bb5d17fdc
Author: Jiajia Li <[email protected]>
AuthorDate: Thu Oct 8 16:16:14 2026 +0800
[python] Add REST policy management (#10384)
---
paimon-python/pypaimon/api/api_request.py | 94 ++++++++++
paimon-python/pypaimon/api/api_response.py | 34 ++++
paimon-python/pypaimon/api/resource_paths.py | 8 +
paimon-python/pypaimon/api/rest_api.py | 48 ++++-
.../pypaimon/api/rest_policy_management.py | 48 +++++
.../pypaimon/catalog/rest/rest_catalog.py | 5 +
paimon-python/pypaimon/management/column_mask.py | 49 ++++++
paimon-python/pypaimon/management/data_policy.py | 89 ++++++++++
paimon-python/pypaimon/management/java_string.py | 5 +
.../pypaimon/management/list_policies_request.py | 75 ++++++++
.../pypaimon/management/permission_resource.py | 4 +
.../pypaimon/management/policy_management.py | 55 ++++++
.../management/{java_string.py => policy_type.py} | 19 +-
paimon-python/pypaimon/management/row_filter.py | 42 +++++
.../pypaimon/tests/policy_management_json_test.py | 151 ++++++++++++++++
.../tests/rest/rest_policy_management_test.py | 193 +++++++++++++++++++++
16 files changed, 910 insertions(+), 9 deletions(-)
diff --git a/paimon-python/pypaimon/api/api_request.py
b/paimon-python/pypaimon/api/api_request.py
index c5ccf5399d..04482e9247 100644
--- a/paimon-python/pypaimon/api/api_request.py
+++ b/paimon-python/pypaimon/api/api_request.py
@@ -23,10 +23,15 @@ from pypaimon.common.identifier import Identifier
from pypaimon.common.json_util import json_field
from pypaimon.function.function_change import FunctionChange
from pypaimon.function.function_definition import FunctionDefinition
+from pypaimon.management.column_mask import ColumnMask
+from pypaimon.management.data_policy import DataPolicy
+from pypaimon.management.java_string import is_blank
from pypaimon.management.permission_access import PermissionAccess
from pypaimon.management.permission_assignment import PermissionAssignment
from pypaimon.management.permission_columns import PermissionColumns
from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.policy_type import PolicyType
+from pypaimon.management.row_filter import RowFilter
from pypaimon.schema.data_types import DataField
from pypaimon.schema.schema import Schema
from pypaimon.schema.schema_change import SchemaChange
@@ -288,3 +293,92 @@ class RevokePermissionRequest(RESTRequest):
resource = data.get(cls.FIELD_RESOURCE)
return cls(None if resource is None else
PermissionResource.from_dict(resource),
data.get(cls.FIELD_ACCESS), data.get(cls.FIELD_PRINCIPAL))
+
+
+class PolicyRequest(RESTRequest):
+ FIELD_ROW_FILTER = "rowFilter"
+ FIELD_COLUMN_MASK = "columnMask"
+ FIELD_PRINCIPAL = "principal"
+
+ def __init__(self, row_filter: Optional[RowFilter], column_mask:
Optional[ColumnMask],
+ principal: str):
+ self._row_filter = row_filter
+ self._column_mask = column_mask
+ self._principal = principal
+
+ @classmethod
+ def from_policy(cls, policy: DataPolicy) -> "PolicyRequest":
+ return cls(policy.get_row_filter(), policy.get_column_mask(),
policy.get_principal())
+
+ def policy(self, resource: PermissionResource) -> DataPolicy:
+ return DataPolicy(resource, self._row_filter, self._column_mask,
self._principal)
+
+ def get_row_filter(self) -> Optional[RowFilter]:
+ return self._row_filter
+
+ def get_column_mask(self) -> Optional[ColumnMask]:
+ return self._column_mask
+
+ def get_principal(self) -> str:
+ return self._principal
+
+ def to_dict(self) -> dict:
+ result = {}
+ if self._row_filter is not None:
+ result[self.FIELD_ROW_FILTER] = self._row_filter.to_dict()
+ if self._column_mask is not None:
+ result[self.FIELD_COLUMN_MASK] = self._column_mask.to_dict()
+ result[self.FIELD_PRINCIPAL] = self._principal
+ return result
+
+ @classmethod
+ def from_dict(cls, data: dict) -> "PolicyRequest":
+ row_filter = data.get(cls.FIELD_ROW_FILTER)
+ column_mask = data.get(cls.FIELD_COLUMN_MASK)
+ return cls(None if row_filter is None else
RowFilter.from_dict(row_filter),
+ None if column_mask is None else
ColumnMask.from_dict(column_mask),
+ data.get(cls.FIELD_PRINCIPAL))
+
+
+class DropPolicyRequest(RESTRequest):
+ FIELD_TYPE = "type"
+ FIELD_PRINCIPAL = "principal"
+ FIELD_COLUMN = "column"
+
+ def __init__(self, policy_type: PolicyType, principal: str, column:
Optional[str] = None):
+ if policy_type is None:
+ raise ValueError("policy type cannot be null")
+ self._type = policy_type
+ self._principal = PermissionAssignment.validate_principal(principal)
+ if policy_type == PolicyType.ROW_FILTER:
+ if not is_blank(column):
+ raise ValueError("ROW_FILTER identity cannot contain a
column.")
+ self._column = None
+ else:
+ if is_blank(column):
+ raise ValueError("column is required for COLUMN_MASKING
identity.")
+ self._column = column
+
+ def get_type(self) -> PolicyType:
+ return self._type
+
+ def get_principal(self) -> str:
+ return self._principal
+
+ def get_column(self) -> Optional[str]:
+ return self._column
+
+ def to_dict(self) -> dict:
+ result = {self.FIELD_TYPE: self._type.name, self.FIELD_PRINCIPAL:
self._principal}
+ if self._column is not None:
+ result[self.FIELD_COLUMN] = self._column
+ return result
+
+ @classmethod
+ def from_dict(cls, data: dict) -> "DropPolicyRequest":
+ policy_type = data.get(cls.FIELD_TYPE)
+ # Jackson matches enum names exactly.
+ if policy_type is not None and policy_type not in
PolicyType.__members__:
+ raise ValueError("Unknown policy type '{}'.".format(policy_type))
+ return cls(None if policy_type is None else PolicyType[policy_type],
+ data.get(cls.FIELD_PRINCIPAL), data.get(cls.FIELD_COLUMN))
diff --git a/paimon-python/pypaimon/api/api_response.py
b/paimon-python/pypaimon/api/api_response.py
index 081ec4945f..837e2ba7ff 100644
--- a/paimon-python/pypaimon/api/api_response.py
+++ b/paimon-python/pypaimon/api/api_response.py
@@ -23,6 +23,7 @@ from pypaimon.api.api_request import RESTRequest
from pypaimon.common.identifier import Identifier
from pypaimon.common.json_util import T, json_field, optional_json_field
from pypaimon.common.options import Options
+from pypaimon.management.data_policy import DataPolicy
from pypaimon.management.permission_assignment import PermissionAssignment
from pypaimon.schema.data_types import DataField
from pypaimon.schema.schema import Schema
@@ -54,6 +55,7 @@ class ErrorResponse(RESTResponse):
RESOURCE_TYPE_BRANCH = "BRANCH"
RESOURCE_TYPE_DEFINITION = "DEFINITION"
RESOURCE_TYPE_DIALECT = "DIALECT"
+ RESOURCE_TYPE_POLICY = "POLICY"
resource_type: Optional[str] = json_field("resourceType", default=None)
resource_name: Optional[str] = json_field("resourceName", default=None)
@@ -665,3 +667,35 @@ class
ListPermissionsResponse(PagedResponse[PermissionAssignment]):
None if permission is None else
PermissionAssignment.from_dict(permission)
for permission in permissions],
data.get(PagedResponse.FIELD_NEXT_PAGE_TOKEN))
+
+
+class ListPoliciesResponse(PagedResponse[DataPolicy]):
+ FIELD_POLICIES = "policies"
+
+ def __init__(self, policies: Optional[List[DataPolicy]], next_page_token:
Optional[str] = None):
+ self._policies = policies
+ self._next_page_token = next_page_token
+
+ def get_policies(self) -> Optional[List[DataPolicy]]:
+ return self._policies
+
+ def data(self) -> Optional[List[DataPolicy]]:
+ return self._policies
+
+ def get_next_page_token(self) -> Optional[str]:
+ return self._next_page_token
+
+ def to_dict(self) -> dict:
+ result = {self.FIELD_POLICIES: None if self._policies is None else [
+ None if policy is None else policy.to_dict() for policy in
self._policies]}
+ if self._next_page_token is not None:
+ result[PagedResponse.FIELD_NEXT_PAGE_TOKEN] = self._next_page_token
+ return result
+
+ @classmethod
+ def from_dict(cls, data: dict) -> "ListPoliciesResponse":
+ policies = data.get(cls.FIELD_POLICIES)
+ return cls(
+ None if policies is None else [
+ None if policy is None else DataPolicy.from_dict(policy) for
policy in policies],
+ data.get(PagedResponse.FIELD_NEXT_PAGE_TOKEN))
diff --git a/paimon-python/pypaimon/api/resource_paths.py
b/paimon-python/pypaimon/api/resource_paths.py
index 94f61948a3..45f82364f6 100644
--- a/paimon-python/pypaimon/api/resource_paths.py
+++ b/paimon-python/pypaimon/api/resource_paths.py
@@ -36,6 +36,7 @@ class ResourcePaths:
RENAME = "rename"
FORWARD = "forward"
PERMISSIONS = "permissions"
+ POLICIES = "policies"
def __init__(self, prefix: str):
# Java encodes the prefix once with URLEncoder, so it stays one path
segment. '~' stays raw:
@@ -154,3 +155,10 @@ class ResourcePaths:
def revoke_permission(self) -> str:
return "{}/revoke".format(self.permissions())
+
+ def policies(self, resource) -> str:
+ resource.validate_policy_attachment()
+ return "{}/{}".format(self.table(resource.get_database(),
resource.get_table()), self.POLICIES)
+
+ def drop_policy(self, resource) -> str:
+ return "{}/drop".format(self.policies(resource))
diff --git a/paimon-python/pypaimon/api/rest_api.py
b/paimon-python/pypaimon/api/rest_api.py
index b6ee8f4144..543c0858b3 100755
--- a/paimon-python/pypaimon/api/rest_api.py
+++ b/paimon-python/pypaimon/api/rest_api.py
@@ -25,12 +25,13 @@ from pypaimon.api.api_request import (AlterDatabaseRequest,
AlterFunctionRequest
CreateBranchRequest,
CreateDatabaseRequest,
CreateFunctionRequest,
CreatePartitionsRequest,
CreateTableRequest, CreateTagRequest,
- ForwardBranchRequest,
- GrantPermissionRequest,
+ DropPolicyRequest, ForwardBranchRequest,
+ GrantPermissionRequest, PolicyRequest,
RenameBranchRequest, RenameTableRequest,
RevokePermissionRequest,
RollbackTableRequest)
from pypaimon.api.api_response import (CommitTableResponse, ConfigResponse,
+ ErrorResponse,
GetDatabaseResponse,
GetFunctionResponse,
GetTableResponse,
GetTableTokenResponse, GetTagResponse,
@@ -40,6 +41,7 @@ from pypaimon.api.api_response import (CommitTableResponse,
ConfigResponse,
ListFunctionsGloballyResponse,
ListFunctionsResponse,
ListPermissionsResponse,
+ ListPoliciesResponse,
CreatePartitionsResponse,
ListPartitionsResponse,
ListTablesResponse, ListTagsResponse,
@@ -51,16 +53,20 @@ from pypaimon.api.api_response import (CommitTableResponse,
ConfigResponse,
from pypaimon.api.auth import AuthProviderFactory, RESTAuthFunction
from pypaimon.api.client import HttpClient
from pypaimon.api.resource_paths import ResourcePaths
+from pypaimon.api.rest_exception import NoSuchResourceException
from pypaimon.api.rest_util import RESTUtil
from pypaimon.api.typedef import T
from pypaimon.common import user_agent
from pypaimon.common.options import Options
from pypaimon.common.options.config import CatalogOptions
from pypaimon.common.identifier import Identifier
+from pypaimon.management.data_policy import DataPolicy
from pypaimon.management.list_permissions_request import \
ListPermissionsRequest
+from pypaimon.management.list_policies_request import ListPoliciesRequest
from pypaimon.management.permission_assignment import PermissionAssignment
from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.policy_type import PolicyType
from pypaimon.schema.schema import Schema
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.snapshot.snapshot_commit import PartitionStatistics
@@ -770,6 +776,44 @@ class RESTApi:
self.rest_auth_function,
)
+ def list_policies(self, request: ListPoliciesRequest) ->
ListPoliciesResponse:
+ query_params = {}
+ if request.get_type() is not None:
+ RESTApi._put_query_parameter(query_params, "type",
request.get_type().name)
+ RESTApi._put_query_parameter(query_params, "principal",
request.get_principal())
+ RESTApi._put_query_parameter(query_params, "column",
request.get_column())
+ if request.get_max_results() is not None:
+ query_params[RESTApi.MAX_RESULTS] = str(request.get_max_results())
+ RESTApi._put_query_parameter(query_params, RESTApi.PAGE_TOKEN,
request.get_page_token())
+ return self.client.get_with_params(
+ self.resource_paths.policies(request.get_resource()),
+ query_params,
+ ListPoliciesResponse,
+ self.rest_auth_function,
+ )
+
+ def create_policy(self, policy: DataPolicy) -> None:
+ self.client.post(
+ self.resource_paths.policies(policy.get_resource()),
+ PolicyRequest.from_policy(policy),
+ self.rest_auth_function,
+ )
+
+ def drop_policy(self, resource: PermissionResource, policy_type:
PolicyType, principal: str,
+ column: Optional[str], ignore_if_not_exists: bool) -> None:
+ if resource is None:
+ raise ValueError("resource cannot be null")
+ resource.validate_policy_attachment()
+ try:
+ self.client.post(
+ self.resource_paths.drop_policy(resource),
+ DropPolicyRequest(policy_type, principal, column),
+ self.rest_auth_function,
+ )
+ except NoSuchResourceException as e:
+ if not ignore_if_not_exists or e.resource_type !=
ErrorResponse.RESOURCE_TYPE_POLICY:
+ raise
+
@staticmethod
def _put_query_parameter(query_params: Dict[str, str], name: str, value:
Optional[str]):
# Java isNotEmpty: unlike __build_paged_query_params, a whitespace
value is sent.
diff --git a/paimon-python/pypaimon/api/rest_policy_management.py
b/paimon-python/pypaimon/api/rest_policy_management.py
new file mode 100644
index 0000000000..8d55d2ce43
--- /dev/null
+++ b/paimon-python/pypaimon/api/rest_policy_management.py
@@ -0,0 +1,48 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import Optional
+
+from pypaimon.api.api_response import ErrorResponse, PagedList
+from pypaimon.api.rest_exception import AlreadyExistsException
+from pypaimon.management.data_policy import DataPolicy
+from pypaimon.management.list_policies_request import ListPoliciesRequest
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.policy_management import (PolicyAlreadyExistException,
+ PolicyManagement)
+from pypaimon.management.policy_type import PolicyType
+
+
+class RESTPolicyManagement(PolicyManagement):
+
+ def __init__(self, api):
+ self.api = api
+
+ def list_policies(self, request: ListPoliciesRequest) ->
PagedList[DataPolicy]:
+ response = self.api.list_policies(request)
+ return PagedList(response.get_policies(),
response.get_next_page_token())
+
+ def create_policy(self, policy: DataPolicy) -> None:
+ try:
+ self.api.create_policy(policy)
+ except AlreadyExistsException as e:
+ if e.resource_type == ErrorResponse.RESOURCE_TYPE_POLICY:
+ raise PolicyAlreadyExistException(policy) from e
+ raise
+
+ def drop_policy(self, resource: PermissionResource, policy_type:
PolicyType, principal: str,
+ column: Optional[str], ignore_if_not_exists: bool) -> None:
+ self.api.drop_policy(resource, policy_type, principal, column,
ignore_if_not_exists)
diff --git a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
index 12fc512c29..ac33e55c30 100644
--- a/paimon-python/pypaimon/catalog/rest/rest_catalog.py
+++ b/paimon-python/pypaimon/catalog/rest/rest_catalog.py
@@ -20,6 +20,7 @@ from typing import Any, Callable, Dict, List, Optional, Union
from pypaimon.api.api_response import ErrorResponse, GetTableResponse,
GetTagResponse, PagedList, Partition
from pypaimon.api.rest_api import RESTApi
from pypaimon.api.rest_permission_management import RESTPermissionManagement
+from pypaimon.api.rest_policy_management import RESTPolicyManagement
from pypaimon.catalog.catalog_exception import IllegalArgumentError,
IllegalStateError
from pypaimon.api.rest_exception import (NoSuchResourceException,
AlreadyExistsException,
ForbiddenException,
BadRequestException,
@@ -48,6 +49,7 @@ from pypaimon.filesystem.caching_file_io import CachingFileIO
from pypaimon.common.identifier import Identifier
from pypaimon.common.json_util import JSON
from pypaimon.management.permission_management import PermissionManagement
+from pypaimon.management.policy_management import PolicyManagement
from pypaimon.schema.schema import Schema
from pypaimon.schema.schema_change import SchemaChange
from pypaimon.schema.table_schema import TableSchema
@@ -97,6 +99,9 @@ class RESTCatalog(Catalog):
def permission_management(self) -> PermissionManagement:
return RESTPermissionManagement(self.rest_api)
+ def policy_management(self) -> PolicyManagement:
+ return RESTPolicyManagement(self.rest_api)
+
def supports_version_management(self) -> bool:
"""
Return whether this catalog supports version management for tables.
diff --git a/paimon-python/pypaimon/management/column_mask.py
b/paimon-python/pypaimon/management/column_mask.py
new file mode 100644
index 0000000000..30eb2e48b8
--- /dev/null
+++ b/paimon-python/pypaimon/management/column_mask.py
@@ -0,0 +1,49 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import Any, Dict
+
+from pypaimon.management.java_string import is_blank, utf8_length
+
+
+class ColumnMask:
+ MAX_TRANSFORM_BYTES = 60 * 1024
+
+ FIELD_ON_COLUMN = "onColumn"
+ FIELD_TRANSFORM = "transform"
+
+ def __init__(self, on_column: str, transform: str):
+ if is_blank(on_column):
+ raise ValueError("onColumn cannot be empty.")
+ if is_blank(transform):
+ raise ValueError("transform cannot be empty.")
+ if utf8_length(transform) > self.MAX_TRANSFORM_BYTES:
+ raise ValueError("transform must not exceed {} UTF-8
bytes.".format(self.MAX_TRANSFORM_BYTES))
+ self._on_column = on_column
+ self._transform = transform
+
+ def get_on_column(self) -> str:
+ return self._on_column
+
+ def get_transform(self) -> str:
+ return self._transform
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {self.FIELD_ON_COLUMN: self._on_column, self.FIELD_TRANSFORM:
self._transform}
+
+ @classmethod
+ def from_dict(cls, data: Dict[str, Any]) -> "ColumnMask":
+ return cls(data.get(cls.FIELD_ON_COLUMN),
data.get(cls.FIELD_TRANSFORM))
diff --git a/paimon-python/pypaimon/management/data_policy.py
b/paimon-python/pypaimon/management/data_policy.py
new file mode 100644
index 0000000000..a98459bfc2
--- /dev/null
+++ b/paimon-python/pypaimon/management/data_policy.py
@@ -0,0 +1,89 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import Any, Dict, Optional
+
+from pypaimon.management.column_mask import ColumnMask
+from pypaimon.management.permission_assignment import PermissionAssignment
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.policy_type import PolicyType
+from pypaimon.management.row_filter import RowFilter
+
+
+class DataPolicy:
+ FIELD_RESOURCE = "resource"
+ FIELD_ROW_FILTER = "rowFilter"
+ FIELD_COLUMN_MASK = "columnMask"
+ FIELD_PRINCIPAL = "principal"
+
+ def __init__(self,
+ resource: PermissionResource,
+ row_filter: Optional[RowFilter],
+ column_mask: Optional[ColumnMask],
+ principal: str):
+ if resource is None:
+ raise ValueError("resource cannot be null")
+ resource.validate_policy_attachment()
+ if (row_filter is None) == (column_mask is None):
+ raise ValueError("A policy must contain exactly one of rowFilter
and columnMask.")
+ self._resource = resource
+ self._row_filter = row_filter
+ self._column_mask = column_mask
+ self._principal = PermissionAssignment.validate_principal(principal)
+
+ @classmethod
+ def row_filter(cls, resource: PermissionResource, row_filter: RowFilter,
+ principal: str) -> "DataPolicy":
+ return cls(resource, row_filter, None, principal)
+
+ @classmethod
+ def column_mask(cls, resource: PermissionResource, column_mask: ColumnMask,
+ principal: str) -> "DataPolicy":
+ return cls(resource, None, column_mask, principal)
+
+ def get_resource(self) -> PermissionResource:
+ return self._resource
+
+ def get_row_filter(self) -> Optional[RowFilter]:
+ return self._row_filter
+
+ def get_column_mask(self) -> Optional[ColumnMask]:
+ return self._column_mask
+
+ def type(self) -> PolicyType:
+ return PolicyType.COLUMN_MASKING if self._row_filter is None else
PolicyType.ROW_FILTER
+
+ def get_principal(self) -> str:
+ return self._principal
+
+ def to_dict(self) -> Dict[str, Any]:
+ result = {self.FIELD_RESOURCE: self._resource.to_dict()}
+ if self._row_filter is not None:
+ result[self.FIELD_ROW_FILTER] = self._row_filter.to_dict()
+ if self._column_mask is not None:
+ result[self.FIELD_COLUMN_MASK] = self._column_mask.to_dict()
+ result[self.FIELD_PRINCIPAL] = self._principal
+ return result
+
+ @classmethod
+ def from_dict(cls, data: Dict[str, Any]) -> "DataPolicy":
+ resource = data.get(cls.FIELD_RESOURCE)
+ row_filter = data.get(cls.FIELD_ROW_FILTER)
+ column_mask = data.get(cls.FIELD_COLUMN_MASK)
+ return cls(None if resource is None else
PermissionResource.from_dict(resource),
+ None if row_filter is None else
RowFilter.from_dict(row_filter),
+ None if column_mask is None else
ColumnMask.from_dict(column_mask),
+ data.get(cls.FIELD_PRINCIPAL))
diff --git a/paimon-python/pypaimon/management/java_string.py
b/paimon-python/pypaimon/management/java_string.py
index ba92ab453c..f53b3981d7 100644
--- a/paimon-python/pypaimon/management/java_string.py
+++ b/paimon-python/pypaimon/management/java_string.py
@@ -25,3 +25,8 @@ def is_blank(value: Optional[str]) -> bool:
def java_length(value: str) -> int:
"""Java ``String.length()``: UTF-16 code units, not code points."""
return len(value.encode('utf-16-le', 'surrogatepass')) // 2
+
+
+def utf8_length(value: str) -> int:
+ """Java ``getBytes(UTF_8).length``: an unpaired surrogate counts as one
byte."""
+ return len(value.encode('utf-8', 'replace'))
diff --git a/paimon-python/pypaimon/management/list_policies_request.py
b/paimon-python/pypaimon/management/list_policies_request.py
new file mode 100644
index 0000000000..a9adb2a8d0
--- /dev/null
+++ b/paimon-python/pypaimon/management/list_policies_request.py
@@ -0,0 +1,75 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import Optional
+
+from pypaimon.management.java_string import is_blank
+from pypaimon.management.list_permissions_request import \
+ ListPermissionsRequest
+from pypaimon.management.permission_assignment import PermissionAssignment
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.policy_type import PolicyType
+
+
+class ListPoliciesRequest:
+
+ def __init__(self,
+ resource: PermissionResource,
+ policy_type: Optional[PolicyType] = None,
+ principal: Optional[str] = None,
+ column: Optional[str] = None,
+ page_token: Optional[str] = None,
+ max_results: Optional[int] = None):
+ if resource is None:
+ raise ValueError("resource cannot be null")
+ resource.validate_policy_attachment()
+ if not is_blank(principal):
+ PermissionAssignment.validate_principal(principal)
+ if max_results is not None and max_results <= 0:
+ raise ValueError("maxResults must be greater than 0.")
+ if max_results is not None and max_results >
ListPermissionsRequest.MAX_PAGE_SIZE:
+ raise ValueError(
+ "maxResults must be at most
{}.".format(ListPermissionsRequest.MAX_PAGE_SIZE))
+ if not is_blank(column) and policy_type != PolicyType.COLUMN_MASKING:
+ raise ValueError("column filter requires type COLUMN_MASKING.")
+ self._resource = resource
+ self._type = policy_type
+ self._principal = None if is_blank(principal) else principal
+ self._column = None if is_blank(column) else column
+ self._page_token = page_token
+ self._max_results = max_results
+
+ def get_resource(self) -> PermissionResource:
+ return self._resource
+
+ def get_type(self) -> Optional[PolicyType]:
+ return self._type
+
+ def get_principal(self) -> Optional[str]:
+ return self._principal
+
+ def get_column(self) -> Optional[str]:
+ return self._column
+
+ def get_page_token(self) -> Optional[str]:
+ return self._page_token
+
+ def get_max_results(self) -> Optional[int]:
+ return self._max_results
+
+ def with_page_token(self, new_page_token: Optional[str]) ->
"ListPoliciesRequest":
+ return ListPoliciesRequest(self._resource, self._type,
self._principal, self._column,
+ new_page_token, self._max_results)
diff --git a/paimon-python/pypaimon/management/permission_resource.py
b/paimon-python/pypaimon/management/permission_resource.py
index 57ad0d8cac..032c4fe947 100644
--- a/paimon-python/pypaimon/management/permission_resource.py
+++ b/paimon-python/pypaimon/management/permission_resource.py
@@ -59,6 +59,10 @@ class PermissionResource:
def get_view(self) -> Optional[str]:
return self._view
+ def validate_policy_attachment(self):
+ if self._type != ResourceType.TABLE:
+ raise ValueError("Policies can currently be attached only to TABLE
resources.")
+
def to_dict(self) -> Dict[str, Any]:
result = {self.FIELD_TYPE: self._type.name}
for name, value in ((self.FIELD_DATABASE, self._database),
diff --git a/paimon-python/pypaimon/management/policy_management.py
b/paimon-python/pypaimon/management/policy_management.py
new file mode 100644
index 0000000000..377593de9e
--- /dev/null
+++ b/paimon-python/pypaimon/management/policy_management.py
@@ -0,0 +1,55 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from abc import ABC, abstractmethod
+from typing import Optional
+
+from pypaimon.api.api_response import PagedList
+from pypaimon.management.data_policy import DataPolicy
+from pypaimon.management.list_policies_request import ListPoliciesRequest
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.policy_type import PolicyType
+
+
+class PolicyManagement(ABC):
+
+ @abstractmethod
+ def list_policies(self, request: ListPoliciesRequest) ->
PagedList[DataPolicy]:
+ pass
+
+ @abstractmethod
+ def create_policy(self, policy: DataPolicy) -> None:
+ pass
+
+ @abstractmethod
+ def drop_policy(self, resource: PermissionResource, policy_type:
PolicyType, principal: str,
+ column: Optional[str], ignore_if_not_exists: bool) -> None:
+ pass
+
+
+class PolicyAlreadyExistException(Exception):
+
+ def __init__(self, policy: DataPolicy):
+ target = policy.type().name
+ if policy.get_column_mask() is not None:
+ target += "(" + policy.get_column_mask().get_on_column() + ")"
+ resource = policy.get_resource()
+ super().__init__("{} policy for principal '{}' already exists on table
'{}.{}'.".format(
+ target, policy.get_principal(), resource.get_database(),
resource.get_table()))
+ self._policy = policy
+
+ def policy(self) -> DataPolicy:
+ return self._policy
diff --git a/paimon-python/pypaimon/management/java_string.py
b/paimon-python/pypaimon/management/policy_type.py
similarity index 65%
copy from paimon-python/pypaimon/management/java_string.py
copy to paimon-python/pypaimon/management/policy_type.py
index ba92ab453c..cb3155d84b 100644
--- a/paimon-python/pypaimon/management/java_string.py
+++ b/paimon-python/pypaimon/management/policy_type.py
@@ -14,14 +14,19 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from enum import Enum
from typing import Optional
-def is_blank(value: Optional[str]) -> bool:
- """Java ``trim().isEmpty()``: only characters <= U+0020 are blank, unlike
``str.strip()``."""
- return value is None or all(ch <= ' ' for ch in value)
+class PolicyType(Enum):
+ ROW_FILTER = "ROW_FILTER"
+ COLUMN_MASKING = "COLUMN_MASKING"
-
-def java_length(value: str) -> int:
- """Java ``String.length()``: UTF-16 code units, not code points."""
- return len(value.encode('utf-16-le', 'surrogatepass')) // 2
+ @staticmethod
+ def from_string(value: Optional[str]) -> Optional["PolicyType"]:
+ if value is None:
+ return None
+ name = value.upper()
+ if name not in PolicyType.__members__:
+ raise ValueError("No enum constant PolicyType.{}".format(name))
+ return PolicyType[name]
diff --git a/paimon-python/pypaimon/management/row_filter.py
b/paimon-python/pypaimon/management/row_filter.py
new file mode 100644
index 0000000000..ec43d74811
--- /dev/null
+++ b/paimon-python/pypaimon/management/row_filter.py
@@ -0,0 +1,42 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from typing import Any, Dict
+
+from pypaimon.management.java_string import is_blank, utf8_length
+
+
+class RowFilter:
+ MAX_PREDICATE_BYTES = 60 * 1024
+
+ FIELD_PREDICATE = "predicate"
+
+ def __init__(self, predicate: str):
+ if is_blank(predicate):
+ raise ValueError("predicate cannot be empty.")
+ if utf8_length(predicate) > self.MAX_PREDICATE_BYTES:
+ raise ValueError("predicate must not exceed {} UTF-8
bytes.".format(self.MAX_PREDICATE_BYTES))
+ self._predicate = predicate
+
+ def get_predicate(self) -> str:
+ return self._predicate
+
+ def to_dict(self) -> Dict[str, Any]:
+ return {self.FIELD_PREDICATE: self._predicate}
+
+ @classmethod
+ def from_dict(cls, data: Dict[str, Any]) -> "RowFilter":
+ return cls(data.get(cls.FIELD_PREDICATE))
diff --git a/paimon-python/pypaimon/tests/policy_management_json_test.py
b/paimon-python/pypaimon/tests/policy_management_json_test.py
new file mode 100644
index 0000000000..af30d30b36
--- /dev/null
+++ b/paimon-python/pypaimon/tests/policy_management_json_test.py
@@ -0,0 +1,151 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+import json
+import unittest
+
+from pypaimon.api.api_request import DropPolicyRequest, PolicyRequest
+from pypaimon.api.api_response import ListPoliciesResponse
+from pypaimon.common.json_util import JSON
+from pypaimon.management.column_mask import ColumnMask
+from pypaimon.management.data_policy import DataPolicy
+from pypaimon.management.list_policies_request import ListPoliciesRequest
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.policy_type import PolicyType
+from pypaimon.management.resource_type import ResourceType
+from pypaimon.management.row_filter import RowFilter
+
+PREDICATE_JSON = (
+ '{"kind":"LEAF","transform":{"name":"FIELD_REF",'
+ '"fieldRef":{"index":0,"name":"region","type":"STRING"}},'
+ '"function":"EQUAL","literals":["APAC"]}')
+TRANSFORM_JSON = (
+ '{"name":"CONCAT","inputs":[{"index":0,"name":"region",'
+ '"type":"STRING"},"****"]}')
+
+
+def catalog_resource():
+ return PermissionResource(ResourceType.CATALOG, None, None, None, None)
+
+
+def table_resource():
+ return PermissionResource(ResourceType.TABLE, "sales", "orders", None,
None)
+
+
+class PolicyManagementJsonTest(unittest.TestCase):
+
+ def test_policy_definitions_round_trip(self):
+ policy = DataPolicy.column_mask(
+ table_resource(), ColumnMask("email", TRANSFORM_JSON), "analyst")
+
+ round_trip = JSON.from_json(JSON.to_json(policy), DataPolicy)
+ self.assertEqual(PolicyType.COLUMN_MASKING, round_trip.type())
+ self.assertEqual(table_resource(), round_trip.get_resource())
+ self.assertEqual("email", round_trip.get_column_mask().get_on_column())
+ self.assertEqual(TRANSFORM_JSON,
round_trip.get_column_mask().get_transform())
+ self.assertEqual("analyst", round_trip.get_principal())
+
+ request = PolicyRequest.from_policy(policy)
+ wire = json.loads(JSON.to_json(request))
+ self.assertEqual(2, len(wire))
+ self.assertIsNotNone(wire["columnMask"])
+ self.assertEqual("analyst", wire["principal"])
+ self.assertEqual(table_resource(),
request.policy(table_resource()).get_resource())
+
+ def test_row_filter_round_trip(self):
+ policy = DataPolicy.row_filter(table_resource(),
RowFilter(PREDICATE_JSON), "analyst")
+
+ round_trip = JSON.from_json(JSON.to_json(policy), DataPolicy)
+ self.assertEqual(PolicyType.ROW_FILTER, round_trip.type())
+ self.assertEqual(PREDICATE_JSON,
round_trip.get_row_filter().get_predicate())
+ self.assertIsNone(round_trip.get_column_mask())
+
+ def test_policy_validation_and_payload_bounds(self):
+ with self.assertRaisesRegex(ValueError, "predicate"):
+ RowFilter(" ")
+ with self.assertRaisesRegex(ValueError, "transform"):
+ ColumnMask("email", " ")
+ with self.assertRaisesRegex(ValueError, "exactly one"):
+ DataPolicy(table_resource(), RowFilter(PREDICATE_JSON),
+ ColumnMask("email", TRANSFORM_JSON), "analyst")
+ with self.assertRaisesRegex(ValueError, "TABLE"):
+ DataPolicy.column_mask(catalog_resource(), ColumnMask("email",
TRANSFORM_JSON), "analyst")
+ with self.assertRaisesRegex(ValueError, "UTF-8 bytes"):
+ RowFilter("p" * (RowFilter.MAX_PREDICATE_BYTES + 1))
+ with self.assertRaisesRegex(ValueError, "UTF-8 bytes"):
+ ColumnMask("email", "t" * (ColumnMask.MAX_TRANSFORM_BYTES + 1))
+
+ def test_drop_policy_request_round_trip_and_identity_validation(self):
+ request = DropPolicyRequest(PolicyType.COLUMN_MASKING, "analyst",
"email")
+ round_trip = JSON.from_json(JSON.to_json(request), DropPolicyRequest)
+
+ self.assertEqual(PolicyType.COLUMN_MASKING, round_trip.get_type())
+ self.assertEqual("analyst", round_trip.get_principal())
+ self.assertEqual("email", round_trip.get_column())
+ with self.assertRaisesRegex(ValueError, "column is required"):
+ DropPolicyRequest(PolicyType.COLUMN_MASKING, "analyst", None)
+ with self.assertRaisesRegex(ValueError, "cannot contain a column"):
+ DropPolicyRequest(PolicyType.ROW_FILTER, "analyst", "email")
+
+ def test_list_policies_validation_and_opaque_page_token(self):
+ request = ListPoliciesRequest(
+ table_resource(), PolicyType.COLUMN_MASKING, "analyst", "email",
None, 25)
+
+ self.assertEqual(" \t", request.with_page_token("
\t").get_page_token())
+ with self.assertRaisesRegex(ValueError, "COLUMN_MASKING"):
+ ListPoliciesRequest(table_resource(), None, None, "email", None,
25)
+ with self.assertRaisesRegex(ValueError, "at most 1000"):
+ ListPoliciesRequest(table_resource(), None, None, None, None, 1001)
+ with self.assertRaisesRegex(ValueError, "TABLE"):
+ ListPoliciesRequest(catalog_resource(), None, None, None, None, 25)
+
+ def test_payload_bounds_count_utf8_bytes(self):
+ RowFilter("\u20ac" * (RowFilter.MAX_PREDICATE_BYTES // 3))
+ with self.assertRaisesRegex(ValueError, "UTF-8 bytes"):
+ RowFilter("\u20ac" * (RowFilter.MAX_PREDICATE_BYTES // 3 + 1))
+ with self.assertRaisesRegex(ValueError, "UTF-8 bytes"):
+ ColumnMask("email", "\U0001F600" * (ColumnMask.MAX_TRANSFORM_BYTES
// 4 + 1))
+ RowFilter("\ud800" * RowFilter.MAX_PREDICATE_BYTES)
+
+ def test_drop_identity_column_follows_java_string_trim(self):
+ self.assertEqual(
+ "\u00a0", DropPolicyRequest(PolicyType.COLUMN_MASKING, "analyst",
"\u00a0").get_column())
+ with self.assertRaisesRegex(ValueError, "cannot contain a column"):
+ DropPolicyRequest(PolicyType.ROW_FILTER, "analyst", "\u00a0")
+ with self.assertRaisesRegex(ValueError, "column is required"):
+ DropPolicyRequest(PolicyType.COLUMN_MASKING, "analyst", "\x00")
+ self.assertIsNone(DropPolicyRequest(PolicyType.ROW_FILTER, "analyst",
"\x00").get_column())
+
+ def test_drop_policy_request_type_is_case_sensitive(self):
+ with self.assertRaisesRegex(ValueError, "column_masking"):
+ JSON.from_json(
+
'{"type":"column_masking","principal":"analyst","column":"email"}',
+ DropPolicyRequest)
+ with self.assertRaisesRegex(ValueError, "policy type cannot be null"):
+ JSON.from_json('{"principal":"analyst"}', DropPolicyRequest)
+
+ def test_listed_policies_are_validated(self):
+ both = json.dumps({"policies": [{
+ "resource": {"type": "TABLE", "database": "sales", "table":
"orders"},
+ "rowFilter": {"predicate": PREDICATE_JSON},
+ "columnMask": {"onColumn": "email", "transform": TRANSFORM_JSON},
+ "principal": "analyst"}]})
+ with self.assertRaisesRegex(ValueError, "exactly one"):
+ JSON.from_json(both, ListPoliciesResponse)
+
+
+if __name__ == '__main__':
+ unittest.main()
diff --git a/paimon-python/pypaimon/tests/rest/rest_policy_management_test.py
b/paimon-python/pypaimon/tests/rest/rest_policy_management_test.py
new file mode 100644
index 0000000000..6c6059e8d5
--- /dev/null
+++ b/paimon-python/pypaimon/tests/rest/rest_policy_management_test.py
@@ -0,0 +1,193 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+import json
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from urllib.parse import parse_qsl, urlsplit
+
+from pypaimon.api.resource_paths import ResourcePaths
+from pypaimon.api.rest_api import RESTApi
+from pypaimon.api.rest_exception import (AlreadyExistsException,
+ NoSuchResourceException)
+from pypaimon.api.rest_policy_management import RESTPolicyManagement
+from pypaimon.common.json_util import JSON
+from pypaimon.management.column_mask import ColumnMask
+from pypaimon.management.data_policy import DataPolicy
+from pypaimon.management.list_policies_request import ListPoliciesRequest
+from pypaimon.management.permission_resource import PermissionResource
+from pypaimon.management.policy_management import PolicyAlreadyExistException
+from pypaimon.management.policy_type import PolicyType
+from pypaimon.management.resource_type import ResourceType
+
+COLLECTION_PATH = "/v1/catalog+id/databases/sales/tables/orders/policies"
+DROP_PATH = COLLECTION_PATH + "/drop"
+
+
+def table_resource():
+ return PermissionResource(ResourceType.TABLE, "sales", "orders", None,
None)
+
+
+def policy():
+ return DataPolicy.column_mask(
+ table_resource(),
+ ColumnMask("email", '{"name":"FIELD_REF","fieldRef":{"index":0,'
+ '"name":"region","type":"STRING"}}'),
+ "analyst")
+
+
+class RESTPolicyManagementTest(unittest.TestCase):
+
+ def setUp(self):
+ recorded = self.recorded = {"drop_calls": 0, "create_error": None,
"drop_error": None}
+ list_response = '{"policies":[' + JSON.to_json(policy()) +
'],"nextPageToken":"next"}'
+
+ class Handler(BaseHTTPRequestHandler):
+
+ def log_message(self, *args):
+ pass
+
+ def respond(self, code, body=None):
+ data = body.encode("utf-8") if body else b""
+ self.send_response(code)
+ self.send_header("Content-Type", "application/json")
+ self.send_header("Content-Length", str(len(data)))
+ self.end_headers()
+ self.wfile.write(data)
+
+ def do_GET(self):
+ url = urlsplit(self.path)
+ if url.path == COLLECTION_PATH:
+ recorded["list_query"] = dict(parse_qsl(url.query,
keep_blank_values=True))
+ self.respond(200, list_response)
+ else:
+ self.respond(404, '{"message":"missing","code":404}')
+
+ def do_POST(self):
+ body = self.rfile.read(int(self.headers.get("Content-Length",
0))).decode("utf-8")
+ path = urlsplit(self.path).path
+ if path == COLLECTION_PATH:
+ recorded["create_body"] = body
+ error = recorded["create_error"]
+ self.respond(409, error) if error else self.respond(200)
+ elif path == DROP_PATH:
+ recorded["drop_body"] = body
+ recorded["drop_calls"] += 1
+ error = recorded["drop_error"]
+ self.respond(404, error) if error else self.respond(200)
+ else:
+ self.respond(404, '{"message":"missing","code":404}')
+
+ self.server = HTTPServer(("127.0.0.1", 0), Handler)
+ threading.Thread(target=self.server.serve_forever, daemon=True).start()
+ options = {
+ "uri": "http://127.0.0.1:{}".format(self.server.server_port),
+ "token.provider": "bear",
+ "token": "secret",
+ "prefix": "catalog id",
+ }
+ self.api = RESTApi(options, False)
+ self.management = RESTPolicyManagement(self.api)
+
+ def tearDown(self):
+ self.server.shutdown()
+ self.server.server_close()
+
+ def drop(self, ignore_if_not_exists):
+ dropped = policy()
+ self.management.drop_policy(
+ dropped.get_resource(), dropped.type(), dropped.get_principal(),
+ dropped.get_column_mask().get_on_column(), ignore_if_not_exists)
+
+ def test_policies_are_nested_under_attachment_resource(self):
+ paths = ResourcePaths("catalog/id")
+ catalog = PermissionResource(ResourceType.CATALOG, None, None, None,
None)
+ database = PermissionResource(ResourceType.DATABASE, "sales db", None,
None, None)
+ table = PermissionResource(ResourceType.TABLE, "sales db",
"orders/all", None, None)
+
+ for resource in (catalog, database):
+ with self.assertRaisesRegex(ValueError, "TABLE"):
+ paths.policies(resource)
+ self.assertEqual(
+ "/v1/catalog%2Fid/databases/sales+db/tables/orders%2Fall/policies",
+ paths.policies(table))
+ self.assertEqual(
+
"/v1/catalog%2Fid/databases/sales+db/tables/orders%2Fall/policies/drop",
+ paths.drop_policy(table))
+
+ def test_list_uses_resource_nested_path_and_identity_filters(self):
+ policies = self.management.list_policies(ListPoliciesRequest(
+ table_resource(), PolicyType.COLUMN_MASKING, "analyst", "email",
"start", 25))
+
+ self.assertEqual(1, len(policies.elements))
+ self.assertEqual("next", policies.next_page_token)
+ self.assertEqual(
+ {"type": "COLUMN_MASKING", "principal": "analyst", "column":
"email",
+ "maxResults": "25", "pageToken": "start"},
+ self.recorded["list_query"])
+
+ def test_list_sends_a_whitespace_page_token_verbatim(self):
+ request = ListPoliciesRequest(table_resource(), page_token=" \t")
+ self.management.list_policies(request)
+ self.assertEqual({"pageToken": " \t"}, self.recorded["list_query"])
+
+ self.management.list_policies(request.with_page_token(""))
+ self.assertEqual({}, self.recorded["list_query"])
+
+ def test_create_and_drop_use_post_endpoints(self):
+ self.management.create_policy(policy())
+ self.drop(False)
+
+ create = json.loads(self.recorded["create_body"])
+ self.assertEqual("analyst", create["principal"])
+ self.assertNotIn("resource", create)
+ drop = json.loads(self.recorded["drop_body"])
+ self.assertEqual("COLUMN_MASKING", drop["type"])
+ self.assertEqual("email", drop["column"])
+ self.assertEqual(1, self.recorded["drop_calls"])
+
+ def test_create_maps_only_policy_conflict(self):
+ self.recorded["create_error"] = (
+
'{"resourceType":"POLICY","resourceName":"COLUMN_MASKING:analyst:email",'
+ '"message":"already exists","code":409}')
+ with self.assertRaisesRegex(PolicyAlreadyExistException,
r"COLUMN_MASKING\(email\)") as ctx:
+ self.management.create_policy(policy())
+ self.assertIsInstance(ctx.exception.__cause__, AlreadyExistsException)
+
+ self.recorded["create_error"] = (
+ '{"resourceType":"TABLE","resourceName":"orders",'
+ '"message":"table conflict","code":409}')
+ with self.assertRaisesRegex(AlreadyExistsException, "table conflict"):
+ self.management.create_policy(policy())
+
+ def test_drop_if_exists_only_ignores_missing_policy(self):
+ self.recorded["drop_error"] = (
+
'{"resourceType":"POLICY","resourceName":"COLUMN_MASKING:analyst:email",'
+ '"message":"missing","code":404}')
+ self.drop(True)
+ with self.assertRaises(NoSuchResourceException):
+ self.drop(False)
+
+ self.recorded["drop_error"] = (
+ '{"resourceType":"TABLE","resourceName":"orders",'
+ '"message":"missing table","code":404}')
+ with self.assertRaisesRegex(NoSuchResourceException, "missing table"):
+ self.drop(True)
+
+
+if __name__ == '__main__':
+ unittest.main()
