from typing import List, Optional from frostfs_sdk.models.dto.selector import Selector from frostfs_sdk.models.enums.selector_clause import SelectorClause from frostfs_sdk.protos.models.netmap import types_pb2 as types_pb2_netmap class SelectorMapper: @staticmethod def to_grpc_messages(selectors: List[Selector]) -> Optional[List[types_pb2_netmap.Selector]]: if not selectors: return None return [SelectorMapper.to_grpc_message(selector) for selector in selectors] @staticmethod def to_grpc_message(selector: Selector) -> Optional[types_pb2_netmap.Selector]: """ Converts Selector DTO to gRPC message Args: selector: Selector DTO object Returns: gRPC Selector message """ if not selector: return None clause_grpc = types_pb2_netmap.Clause(selector.clause.value) if clause_grpc is None: raise ValueError(f"Unknown enum value: {selector.clause.name} for {types_pb2_netmap.Clause.__name__}") return types_pb2_netmap.Selector( name=selector.name, count=selector.count, clause=clause_grpc, attribute=selector.attribute, filter=selector.filter ) @staticmethod def to_models(grpc_selectors: List[types_pb2_netmap.Selector]) -> Optional[List[Selector]]: if not grpc_selectors: return None return [SelectorMapper.to_model(grpc_selector) for grpc_selector in grpc_selectors] @staticmethod def to_model(selector_grpc: types_pb2_netmap.Selector) -> Optional[Selector]: """ Converts gRPC message to Selector DTO Args: selector_grpc: gRPC Selector message Returns: Selector DTO object """ if not selector_grpc or selector_grpc.ByteSize() == 0: return None clause = SelectorClause.get(selector_grpc.clause) if clause is None: raise ValueError(f"Unknown enum value: {selector_grpc.clause} for {SelectorClause.__name__}") return Selector( name=selector_grpc.name, count=selector_grpc.count, clause=clause, attribute=selector_grpc.attribute, filter=selector_grpc.filter )