<s> import time <EOL> from . . import jwt <EOL> import warnings <EOL> warnings . simplefilter ( '<STR_LIT>' , DeprecationWarning ) <EOL> TASK_ROUTER_BASE_URL = '<STR_LIT>' <EOL> TASK_ROUTER_BASE_EVENTS_URL = '<STR_LIT>' <EOL> TASK_ROUTER_VERSION = "<STR_LIT>" <EOL> REQUIRED = { '<STR_LIT>' : True } <EOL> OPTIONAL = { '<STR_LIT>' : False } <EOL> def deprecated ( func ) : <EOL> def log_warning ( * args , ** kwargs ) : <EOL> warnings . warn ( "<STR_LIT>" . <EOL> format ( func . __name__ ) , <EOL> stacklevel = <NUM_LIT:2> , <EOL> category = DeprecationWarning ) <EOL> return func ( * args , ** kwargs ) <EOL> return log_warning <EOL> class TaskRouterCapability ( object ) : <EOL> def __init__ ( self , account_sid , auth_token , workspace_sid , channel_id ) : <EOL> self . account_sid = account_sid <EOL> self . auth_token = auth_token <EOL> self . policies = [ ] <EOL> self . workspace_sid = workspace_sid <EOL> self . channel_id = channel_id <EOL> self . base_url = "<STR_LIT>" . format ( TASK_ROUTER_BASE_URL , <EOL> TASK_ROUTER_VERSION , <EOL> workspace_sid ) <EOL> self . validate_jwt ( ) <EOL> self . setup_resource ( ) <EOL> self . allow_web_sockets ( channel_id ) <EOL> self . add_policy ( self . resource_url , "<STR_LIT:GET>" , True ) <EOL> @ property <EOL> def channel_prefix ( self ) : <EOL> return self . channel_id [ <NUM_LIT:0> : <NUM_LIT:2> ] <EOL> def setup_resource ( self ) : <EOL> if self . channel_prefix == "<STR_LIT>" : <EOL> self . resource_url = self . base_url <EOL> elif self . channel_prefix == "<STR_LIT>" : <EOL> self . resource_url = self . base_url + "<STR_LIT>" + self . channel_id <EOL> activity_url = self . base_url + "<STR_LIT>" <EOL> self . allow ( activity_url , "<STR_LIT:GET>" ) <EOL> reservations_url = self . base_url + "<STR_LIT>" <EOL> self . allow ( reservations_url , "<STR_LIT:GET>" ) <EOL> elif self . channel_prefix == "<STR_LIT>" : <EOL> self . resource_url = "<STR_LIT>" . format ( <EOL> self . base_url , self . channel_id ) <EOL> def allow_web_sockets ( self , channel_id ) : <EOL> web_socket_url = "<STR_LIT>" . format ( TASK_ROUTER_BASE_EVENTS_URL , <EOL> self . account_sid , <EOL> self . channel_id ) <EOL> self . policies . append ( self . make_policy ( web_socket_url , "<STR_LIT:GET>" , True ) ) <EOL> self . policies . append ( self . make_policy ( web_socket_url , "<STR_LIT:POST>" , True ) ) <EOL> def validate_jwt ( self ) : <EOL> if self . account_sid is None or self . account_sid [ <NUM_LIT:0> : <NUM_LIT:2> ] != "<STR_LIT>" : <EOL> raise ValueError ( '<STR_LIT>' + <EOL> self . account_sid ) <EOL> if self . workspace_sid is None or self . workspace_sid [ <NUM_LIT:0> : <NUM_LIT:2> ] != "<STR_LIT>" : <EOL> raise ValueError ( '<STR_LIT>' + <EOL> self . workspace_sid ) <EOL> if self . channel_id is None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if self . channel_prefix != "<STR_LIT>" and self . channel_prefix != "<STR_LIT>" and self . channel_prefix != "<STR_LIT>" : <EOL> raise ValueError ( '<STR_LIT>' + self . channel_id ) <EOL> def allow_fetch_subresources ( self ) : <EOL> self . allow ( self . resource_url + "<STR_LIT>" , "<STR_LIT:GET>" ) <EOL> def allow_updates ( self ) : <EOL> self . allow ( self . resource_url , "<STR_LIT:POST>" ) <EOL> def allow_updates_subresources ( self ) : <EOL> self . allow ( self . resource_url + "<STR_LIT>" , "<STR_LIT:POST>" ) <EOL> def allow_delete ( self ) : <EOL> self . allow ( self . resource_url , "<STR_LIT>" ) <EOL> def allow_delete_subresources ( self ) : <EOL> self . allow ( self . resource_url + "<STR_LIT>" , "<STR_LIT>" ) <EOL> @ deprecated <EOL> def allow_worker_fetch_attributes ( self ) : <EOL> if self . channel_prefix != "<STR_LIT>" : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> else : <EOL> self . policies . append ( self . make_policy ( <EOL> self . resource_url , <EOL> '<STR_LIT:GET>' ) ) <EOL> @ deprecated <EOL> def allow_worker_activity_updates ( self ) : <EOL> if self . channel_prefix == "<STR_LIT>" : <EOL> self . policies . append ( self . make_policy ( <EOL> self . resource_url , <EOL> '<STR_LIT:POST>' , <EOL> True , <EOL> post_filter = { '<STR_LIT>' : REQUIRED } ) ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> @ deprecated <EOL> def allow_task_reservation_updates ( self ) : <EOL> if self . channel_prefix == "<STR_LIT>" : <EOL> tasks_url = self . base_url + "<STR_LIT>" <EOL> self . policies . append ( self . make_policy ( <EOL> tasks_url , <EOL> '<STR_LIT:POST>' , <EOL> True ) ) <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def add_policy ( self , url , method , <EOL> allowed , query_filter = None , post_filter = None ) : <EOL> policy = self . make_policy ( url , method , <EOL> allowed , query_filter , post_filter ) <EOL> self . policies . append ( policy ) <EOL> def allow ( self , url , method , query_filter = None , post_filter = None ) : <EOL> self . add_policy ( url , method , True , query_filter , post_filter ) <EOL> def deny ( self , url , method , query_filter = None , post_filter = None ) : <EOL> self . add_policy ( url , method , False , query_filter , post_filter ) <EOL> def make_policy ( self , url , method , <EOL> allowed = True , query_filter = None , post_filter = None ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> '<STR_LIT:url>' : url , <EOL> '<STR_LIT>' : method , <EOL> '<STR_LIT>' : allowed , <EOL> '<STR_LIT>' : query_filter or { } , <EOL> '<STR_LIT>' : post_filter or { } <EOL> } <EOL> def get_resource_url ( self ) : <EOL> return self . resource_url <EOL> def generate_token ( self , ttl = <NUM_LIT> ) : <EOL> task_router_attributes = { <EOL> '<STR_LIT>' : self . account_sid , <EOL> '<STR_LIT>' : self . workspace_sid , <EOL> '<STR_LIT>' : self . channel_id <EOL> } <EOL> if self . channel_prefix == "<STR_LIT>" : <EOL> task_router_attributes [ "<STR_LIT>" ] = self . channel_id <EOL> elif self . channel_prefix == "<STR_LIT>" : <EOL> task_router_attributes [ "<STR_LIT>" ] = self . channel_id <EOL> return self . _generate_token ( ttl , task_router_attributes ) <EOL> def _generate_token ( self , ttl , attributes = None ) : <EOL> payload = { <EOL> '<STR_LIT>' : self . account_sid , <EOL> '<STR_LIT>' : int ( time . time ( ) ) + ttl , <EOL> '<STR_LIT:version>' : TASK_ROUTER_VERSION , <EOL> '<STR_LIT>' : self . channel_id , <EOL> '<STR_LIT>' : self . policies , <EOL> } <EOL> if attributes is not None : <EOL> payload . update ( attributes ) <EOL> return jwt . encode ( payload , self . auth_token , '<STR_LIT>' ) <EOL> class TaskRouterWorkerCapability ( TaskRouterCapability ) : <EOL> def __init__ ( self , account_sid , auth_token , workspace_sid , worker_sid ) : <EOL> super ( TaskRouterWorkerCapability , self ) . __init__ ( account_sid , <EOL> auth_token , <EOL> workspace_sid , <EOL> worker_sid ) <EOL> self . reservations_url = self . base_url + "<STR_LIT>" <EOL> self . activity_url = self . base_url + "<STR_LIT>" <EOL> self . allow ( self . reservations_url , "<STR_LIT:GET>" ) <EOL> self . allow ( self . activity_url , "<STR_LIT:GET>" ) <EOL> def setup_resource ( self ) : <EOL> self . resource_url = self . base_url + "<STR_LIT>" + self . channel_id <EOL> def allow_activity_updates ( self ) : <EOL> self . policies . append ( self . make_policy ( <EOL> self . resource_url , <EOL> '<STR_LIT:POST>' , <EOL> True , <EOL> post_filter = { '<STR_LIT>' : REQUIRED } ) ) <EOL> def allow_reservation_updates ( self ) : <EOL> self . policies . append ( self . make_policy ( <EOL> self . reservations_url , <EOL> '<STR_LIT:POST>' , <EOL> True ) ) <EOL> class TaskRouterTaskQueueCapability ( TaskRouterCapability ) : <EOL> def setup_resource ( self ) : <EOL> self . resource_url = self . base_url + "<STR_LIT>" + self . channel_id <EOL> class TaskRouterWorkspaceCapability ( TaskRouterCapability ) : <EOL> def __init__ ( self , account_sid , auth_token , workspace_sid ) : <EOL> super ( TaskRouterWorkspaceCapability , self ) . __init__ ( account_sid , <EOL> auth_token , <EOL> workspace_sid , <EOL> workspace_sid ) <EOL> def setup_resource ( self ) : <EOL> self . resource_url = self . base_url <EOL> from . taskrouter_config import ( <EOL> TaskRouterConfig <EOL> ) <EOL> from . workflow_config import ( <EOL> WorkflowConfig <EOL> ) <EOL> from . workflow_ruletarget import ( <EOL> WorkflowRuleTarget <EOL> ) <EOL> from . workflow_rule import ( <EOL> WorkflowRule <EOL> ) </s>
