feast/sdk/python/feast/permissions/decision.py at pytorch-template · feast-dev/feast · GitHub

History

114 lines (96 loc) · 4.13 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

import enum

import logging

from typing import Optional

logger = logging.getLogger(__name__)

class DecisionStrategy(enum.Enum):

"""

The strategy to be adopted in case multiple permissions match an execution request.

"""

UNANIMOUS = "unanimous" # All policies must evaluate to a positive decision for the final decision to be also positive.

AFFIRMATIVE = (

"affirmative" # At least one policy must evaluate to a positive decision

)

# The number of positive decisions must be greater than the number of negative decisions.

# If the number of positive and negative decisions is the same, the final decision will be negative.

CONSENSUS = "consensus"

class DecisionEvaluator:

"""

A class to implement the decision logic, according to the selected strategy.

Args:

decision_strategy: The associated `DecisionStrategy`.

num_of_voters: The expected number of votes to complete the decision.

Examples:

Create the instance and specify the strategy and number of decisions:

`evaluator = DecisionEvaluator(DecisionStrategy.UNANIMOUS, 3)

For each vote that you receivem, add a decision grant: `evaluator.add_grant(vote, message)`

and check if the decision process ended: `if evaluator.is_decided():`

Once decided, get the result and the failure explanations using:

`grant, explanations = evaluator.grant()`

"""

def __init__(

self,

num_of_voters: int,

# Only AFFIRMATIVE strategy is managed available

decision_strategy = DecisionStrategy.AFFIRMATIVE

self.num_of_voters = num_of_voters

self.grant_count = 0

self.deny_count = 0

self.grant_quorum = (

if decision_strategy == DecisionStrategy.AFFIRMATIVE

else num_of_voters

if decision_strategy == DecisionStrategy.UNANIMOUS

else num_of_voters // 2 + 1

)

self.deny_quorum = (

num_of_voters

if decision_strategy == DecisionStrategy.AFFIRMATIVE

else 1

if decision_strategy == DecisionStrategy.UNANIMOUS

else num_of_voters // 2 + (num_of_voters % 2)

)

self.grant_decision: Optional[bool] = None

self.explanations: list[str] = []

logger.info(

f"Decision evaluation started with grant_quorum={self.grant_quorum}, deny_quorum={self.deny_quorum}"

)

def is_decided(self) -> bool:

"""

Returns:

bool: `True` when the decision process completed (e.g. we added as many votes as specified in the `num_of_voters` creation argument).

"""

return self.grant_decision is not None

def grant(self) -> tuple[bool, list[str]]:

"""

Returns:

tuple[bool, list[str]]: The tuple of decision computation: a `bool` with the computation decision and a `list[str]` with the

denial explanations (possibly empty).

"""

logger.info(

f"Decided grant is {self.grant_decision}, explanations={self.explanations}"

)

return bool(self.grant_decision), self.explanations

def add_grant(self, grant: bool, explanation: str):

"""

Add a single vote to the decision computation, with a possible denial reason.

If the evaluation process already ended, additional votes are discarded.

Args:

grant: `True` is the decision is accepted, `False` otherwise.

explanation: Denial reason (not considered when `vote` is `True`).

"""

if self.is_decided():

logger.warning("Grant decision already decided, discarding vote")

return

if grant:

self.grant_count += 1

else:

self.deny_count += 1

self.explanations.append(explanation)

if self.grant_count >= self.grant_quorum:

self.grant_decision = True

if self.deny_count >= self.deny_quorum:

self.grant_decision = False

logger.debug(

f"After new grant: grants={self.grant_count}, deny_count={self.deny_count}, grant_decision={self.grant_decision}"

)

feast/sdk/python/feast/permissions/decision.py at pytorch-template · feast-dev/feast · GitHub

Uh oh!

Expand file tree

Breadcrumbs

decision.py

Latest commit

History

Breadcrumbs

decision.py

File metadata and controls