<s> import time <EOL> from . . import jwt <EOL> TASK_ROUTER_BASE_URL = '<STR_LIT>' <EOL> TASK_ROUTER_BASE_WS_URL = '<STR_LIT>' <EOL> REQUIRED = { '<STR_LIT>' : True } <EOL> OPTIONAL = { '<STR_LIT>' : False } <EOL> class TaskRouterCapability ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , account_sid , auth_token , workspace_sid , worker_sid , <EOL> base_url = TASK_ROUTER_BASE_URL , <EOL> version = '<STR_LIT>' , <EOL> base_ws_url = TASK_ROUTER_BASE_WS_URL ) : <EOL> self . account_sid = account_sid <EOL> self . auth_token = auth_token <EOL> self . workspace_sid = workspace_sid <EOL> self . worker_sid = worker_sid <EOL> self . version = version <EOL> self . base_url = '<STR_LIT>' % ( base_url , self . version ) <EOL> self . base_ws_url = base_ws_url <EOL> self . policies = [ ] <EOL> self . _allow_worker_websocket_urls ( ) <EOL> self . _allow_activity_list_fetch ( ) <EOL> @ property <EOL> def workspace_url ( self ) : <EOL> return '<STR_LIT>' % ( self . base_url , self . workspace_sid ) <EOL> @ property <EOL> def worker_url ( self ) : <EOL> return '<STR_LIT>' % ( self . workspace_url , self . worker_sid ) <EOL> def _allow_worker_websocket_urls ( self ) : <EOL> worker_event_url = '<STR_LIT>' % ( <EOL> self . base_ws_url , <EOL> self . account_sid , <EOL> self . worker_sid , <EOL> ) <EOL> self . policies . append ( make_policy ( <EOL> worker_event_url , <EOL> '<STR_LIT:GET>' , <EOL> ) ) <EOL> self . policies . append ( make_policy ( <EOL> worker_event_url , <EOL> '<STR_LIT:POST>' , <EOL> ) ) <EOL> def _allow_activity_list_fetch ( self ) : <EOL> self . policies . append ( make_policy ( <EOL> '<STR_LIT>' % self . workspace_url , <EOL> '<STR_LIT:GET>' , <EOL> ) ) <EOL> def allow_worker_activity_updates ( self ) : <EOL> self . policies . append ( make_policy ( <EOL> self . worker_url , <EOL> '<STR_LIT:POST>' , <EOL> post_filter = { '<STR_LIT>' : REQUIRED } , <EOL> ) ) <EOL> def allow_worker_fetch_attributes ( self ) : <EOL> self . policies . append ( make_policy ( <EOL> self . worker_url , <EOL> '<STR_LIT:GET>' , <EOL> ) ) <EOL> def allow_task_reservation_updates ( self ) : <EOL> tasks_url = '<STR_LIT>' % self . workspace_url <EOL> self . policies . append ( make_policy ( <EOL> tasks_url , <EOL> '<STR_LIT:POST>' , <EOL> post_filter = { '<STR_LIT>' : REQUIRED } , <EOL> ) ) <EOL> def generate_token ( self , ttl = <NUM_LIT> , attributes = None ) : <EOL> """<STR_LIT>""" <EOL> return self . _generate_token ( <EOL> ttl , <EOL> { <EOL> '<STR_LIT>' : self . account_sid , <EOL> '<STR_LIT>' : self . worker_sid , <EOL> '<STR_LIT>' : self . worker_sid , <EOL> '<STR_LIT>' : self . workspace_sid , <EOL> } <EOL> ) <EOL> def _generate_token ( self , ttl , attributes = None ) : <EOL> payload = { <EOL> '<STR_LIT:version>' : self . version , <EOL> '<STR_LIT>' : self . worker_sid , <EOL> '<STR_LIT>' : self . policies , <EOL> '<STR_LIT>' : self . account_sid , <EOL> '<STR_LIT>' : int ( time . time ( ) ) + ttl , <EOL> } <EOL> if attributes is not None : <EOL> payload . update ( attributes ) <EOL> return jwt . encode ( payload , self . auth_token , '<STR_LIT>' ) <EOL> def make_policy ( url , method , query_filter = None , post_filter = None , <EOL> allowed = True ) : <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> } </s>
