50 lines
1.7 KiB
Python
50 lines
1.7 KiB
Python
from typing import Optional
|
|
from frostfs_sdk.models.mappers.filter_mapper import FilterMapper
|
|
import protos.models.netmap.types_pb2 as types_pb2_netmap
|
|
from frostfs_sdk.models.dto.placement_policy import PlacementPolicy
|
|
|
|
|
|
class PlacementPolicyMapper:
|
|
@staticmethod
|
|
def to_grpc_message(policy: PlacementPolicy) -> Optional[types_pb2_netmap.PlacementPolicy]:
|
|
"""
|
|
Converts PlacementPolicy DTO to gRPC message
|
|
|
|
Args:
|
|
policy: PlacementPolicy DTO object
|
|
|
|
Returns:
|
|
gRPC PlacementPolicy message
|
|
"""
|
|
if not policy:
|
|
return None
|
|
|
|
return types_pb2_netmap.PlacementPolicy(
|
|
unique=policy.unique,
|
|
container_backup_factor=policy.backup_factor,
|
|
filters=FilterMapper.to_grpc_messages(policy.filters),
|
|
# selectors=SelectorMapper.to_grpc_messages(policy.selectors),
|
|
# replicas=ReplicaMapper.to_grpc_messages(policy.replicas)
|
|
)
|
|
|
|
@staticmethod
|
|
def to_model(policy_grpc: types_pb2_netmap.PlacementPolicy) -> Optional[PlacementPolicy]:
|
|
"""
|
|
Converts gRPC message to PlacementPolicy DTO
|
|
|
|
Args:
|
|
policy_grpc: gRPC PlacementPolicy message
|
|
|
|
Returns:
|
|
PlacementPolicy DTO object
|
|
"""
|
|
if not policy_grpc or policy_grpc.ByteSize() == 0:
|
|
return None
|
|
|
|
return PlacementPolicy(
|
|
# replicas=ReplicaMapper.to_models(policy_grpc.replicas),
|
|
unique=policy_grpc.unique,
|
|
backup_factor=policy_grpc.container_backup_factor,
|
|
filters=FilterMapper.to_models(policy_grpc.filters),
|
|
# selectors=SelectorMapper.to_models(policy_grpc.selectors)
|
|
)
|