daily pastebin goal
41%
SHARE
TWEET

Untitled

a guest Jan 18th, 2019 58 Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
  1. from sys import Type
  2. from typing import List
  3.  
  4. class Policy:
  5.     ...
  6.  
  7.  
  8. class RetryPolicy(Policy):
  9.  
  10.     def __init__(self, max_retries=4, timeout=None):
  11.         self.max_retries = max_retries
  12.         self.timeout = timeout
  13.  
  14. class LoggingPolicy(Policy):
  15.     ...
  16.  
  17. class TracingPolicy(Policy):
  18.     ...
  19.  
  20.  
  21. class Pipeline:
  22.  
  23.     def __init__(self, policies: "List[Policy]"):
  24.         self.policies = policies
  25.         self.prepare_request_handlers = []
  26.         self.sending_request_handlers = []
  27.  
  28.     def send(self, request):
  29.         """ Send the given request through the pipeline"""
  30.  
  31.         # Each prepare_request_handler is called in the order they were registered
  32.         for handler in self.prepare_request_handlers:
  33.             request = handler(request)
  34.  
  35.         request = self.policies[0].send(request)
  36.  
  37.         # Each sending_request_handler is called in the order they were registered
  38.         for handler in self.prepare_request_handlers:
  39.             request = handler(request)
  40.  
  41.         # The last policy is the "transport" policy
  42.         response = self.policies[-1].send(request)
  43.  
  44.         return response
  45.  
  46.     def add_prepare_request_handler(self, handler):
  47.         self.prepare_request_handlers.add(handler)
  48.        
  49.     def get_policy(self, policy_type: "Type") -> "Policy":
  50.         """ Get the policy of the given type
  51.  
  52.         Will return the policy if there is one and only one policy of the given type in the
  53.         pipeline. Will throw otherwise.
  54.         """
  55.         matches = list([policy for policy in self.policies if isinstance(policy, policy_type)])
  56.         if len(matches) != 1:
  57.             raise ValueError('Found {} matches for policy type {}'.format(len(matches), policy_type.__name__))
  58.         return matches[0]
  59.  
  60.     def set_policy(self, policy: "Policy", *, policy_type: "Type"=None):
  61.         """ Set the policy of the given type to a new instance.
  62.         """
  63.         # Ensure that there is ona and only one policy of the type to replace/set
  64.         policy_type_to_replace = policy_type or type(policy)
  65.         self.get_policy(policy_type_to_replace)
  66.         self.policies = list([policy if isinstance(existing_policy, policy_type_to_replace)
  67.                               else existing_policy
  68.                               for existing_policy in self.policies])
  69.  
  70.  
  71. # Usage example
  72.  
  73. # Somehow the default pipeline is constructed for a given client
  74. pipeline = Pipeline([
  75.     LoggingPolicy(),
  76.     RetryPolicy(),
  77.     TracingPolicy()
  78. ])
  79.  
  80. retry_policy = pipeline.get_policy(RetryPolicy)
  81. retry_policy.max_retries = 1
RAW Paste Data
We use cookies for various purposes including analytics. By continuing to use Pastebin, you agree to our use of cookies as described in the Cookies Policy. OK, I Understand
 
Top