def get(self, actor_id, execution_id):
     def get_hypermedia(actor, exc):
         return {'_links': {'self': '{}/actors/v2/{}/executions/{}/logs'.format(actor.api_server, actor.id, exc.id),
                            'owner': '{}/profiles/v2/{}'.format(actor.api_server, actor.owner),
                            'execution': '{}/actors/v2/{}/executions/{}'.format(actor.api_server, actor.id, exc.id)},
                 }
     dbid = Actor.get_dbid(g.tenant, actor_id)
     try:
         actor = Actor.from_db(actors_store[dbid])
     except KeyError:
         raise APIException(
             "actor not found: {}'".format(actor_id), 404)
     try:
         excs = executions_store[dbid]
     except KeyError:
         raise APIException("No executions found for actor {}.".format(actor_id))
     try:
         exc = Execution.from_db(excs[execution_id])
     except KeyError:
         raise APIException("Execution not found {}.".format(execution_id))
     try:
         logs = logs_store[execution_id]
     except KeyError:
         logs = ""
     result={'logs': logs}
     result.update(get_hypermedia(actor, exc))
     return ok(result, msg="Logs retrieved successfully.")

 def put(self, actor_id):
     dbid = Actor.get_dbid(g.tenant, actor_id)
     try:
         actor = Actor.from_db(actors_store[dbid])
     except KeyError:
         raise APIException(
             "actor not found: {}'".format(actor_id), 404)
     previous_image = actor.image
     args = self.validate_put(actor)
     args['tenant'] = g.tenant
     update_image = False
     if args['image'] == previous_image:
         args['status'] = actor.status
     else:
         update_image = True
         args['status'] = SUBMITTED
     args['api_server'] = g.api_server
     args['owner'] = g.user
     actor = Actor(**args)
     actors_store[actor.db_id] = actor.to_db()
     if update_image:
         ch = CommandChannel()
         ch.put_cmd(actor_id=actor.db_id, image=actor.image, tenant=args['tenant'])
     # return ok(result={'update_image': str(update_image)},
     #           msg="Actor updated successfully.")
     return ok(result=actor.display(),
               msg="Actor updated successfully.")

    def do_save(self):
        for hexsha in self.cached_data:

            val = self.cached_data[hexsha]
            try:
                actor = Actor.objects.get(full_name = val['name'])
            except Actor.DoesNotExist:
                actor = Actor(full_name = val['name'])
                actor.save()
                #Create the actor

            try:
                commit = Commit.objects.get(hexsha = hexsha)
            except Commit.DoesNotExist:
                commit = Commit(hexsha = hexsha, repo = self.repo_model, actor = actor)
                commit.save()

            for path, fun in val['funcs']:
                if not Function.objects.filter(name = fun, path = path).exists():
                    fmodel = Function(name = fun, commit = commit, path = path)
                    fmodel.save()
                    print "Saved  `%s` : `%s`" % (path[-16:], fun)

            for file_name in val['files_changed']:
                FileChange(path = file_name, actor = actor, commit = commit).save()


        self.cached_data.clear()

def subscribe(actor_id, worker_ch):
    """
    Main loop for the Actor executor worker. Subscribes to the actor's inbox and executes actor
    containers when message arrive. Also subscribes to the worker channel for future communications.
    :return:
    """
    actor_ch = ActorMsgChannel(actor_id)
    t = threading.Thread(target=process_worker_ch, args=(worker_ch, actor_id, actor_ch))
    t.start()
    print("Worker subscribing to actor channel...")
    while keep_running:
        update_worker_status(actor_id, worker_ch.name, READY)
        try:
            msg = actor_ch.get(timeout=2)
        except channelpy.ChannelTimeoutException:
            continue
        print("Received message {}. Starting actor container...".format(str(msg)))
        message = msg.pop("msg", "")
        try:
            stats, logs = execute_actor(actor_id, worker_ch, image, message, msg)
        except DockerStartContainerError as e:
            print("Got DockerStartContainerError: {}".format(str(e)))
            Actor.set_status(actor_id, ERROR)
            continue
        # add the execution to the actor store
        print("Actor container finished successfully. Got stats object:{}".format(str(stats)))
        exc_id = Execution.add_execution(actor_id, stats)
        Execution.set_logs(exc_id, logs)

 def post(self, actor_id):
     id = Actor.get_dbid(g.tenant, actor_id)
     try:
         actor = Actor.from_db(actors_store[id])
     except KeyError:
         raise APIException(
             "actor not found: {}'".format(actor_id), 404)
     args = self.validate_post()
     Execution.add_execution(id, args)
     return ok(result=actor.display(), msg="Actor execution added successfully.")

 def get(self, actor_id, ch_name):
     try:
         Actor.from_db(actors_store[actor_id])
     except KeyError:
         raise WorkerException("actor not found: {}'".format(actor_id))
     try:
         worker = get_worker(actor_id, ch_name)
     except WorkerException as e:
         raise APIException(e.message, 404)
     return ok(result=worker, msg="Worker retrieved successfully.")

 def post(self, actor_id):
     """Add new permissions for an actor"""
     try:
         Actor.from_db(actors_store[actor_id])
     except KeyError:
         raise APIException(
             "actor not found: {}'".format(actor_id), 404)
     args = self.validate_post()
     add_permission(args['user'], actor_id, args['level'])
     permissions = get_permissions(actor_id)
     return ok(result=permissions, msg="Permission added successfully.")

 def get(self, actor_id):
     try:
         Actor.from_db(actors_store[actor_id])
     except KeyError:
         raise APIException(
             "actor not found: {}'".format(actor_id), 404)
     try:
         permissions = get_permissions(actor_id)
     except PermissionsException as e:
         raise APIException(e.message, 404)
     return ok(result=permissions, msg="Permissions retrieved successfully.")

 def post(self):
     args = self.validate_post()
     args['executions'] = {}
     args['state'] = ''
     args['subscriptions'] = []
     args['status'] = SUBMITTED
     actor = Actor(args)
     actors_store[actor.id] = actor.to_db()
     ch = CommandChannel()
     ch.put_cmd(actor_id=actor.id, image=actor.image)
     return ok(result=actor, msg="Actor created successfully.")

 def delete(self, actor_id):
     id = Actor.get_dbid(g.tenant, actor_id)
     shutdown_workers(id)
     try:
         actor = Actor.from_db(actors_store[id])
         executions = actor.get('executions') or {}
         for ex_id, val in executions.items():
             del logs_store[ex_id]
     except KeyError:
         print("Did not find actor with id: {}".format(id))
     del actors_store[id]
     del permissions_store[id]
     return ok(result=None, msg='Actor deleted successfully.')

 def get(self, actor_id):
     try:
         actor = Actor.from_db(actors_store[actor_id])
     except KeyError:
         raise APIException(
             "actor not found: {}'".format(actor_id), 404)
     return ok(result=actor, msg="Actor retrieved successfully.")

    def test_serialize_unicode(self):
        """Tests that unicode makes the roundtrip intact"""
        actor_name = u"Za\u017c\u00f3\u0142\u0107"
        movie_title = u'G\u0119\u015bl\u0105 ja\u017a\u0144'
        ac = Actor(name=actor_name)
        mv = Movie(title=movie_title, actor=ac)
        ac.save()
        mv.save()

        serial_str = serializers.serialize(self.serializer_name, [mv])
        self.assertEqual(self._get_field_values(serial_str, "title")[0], movie_title)
        self.assertEqual(self._get_field_values(serial_str, "actor")[0], actor_name)

        obj_list = list(serializers.deserialize(self.serializer_name, serial_str))
        mv_obj = obj_list[0].object
        self.assertEqual(mv_obj.title, movie_title)

 def validate_put(self, actor):
     # inherit derived attributes from the original actor, including id and db_id:
     parser = Actor.request_parser()
     # remove since name is only required for POST, not PUT
     parser.remove_argument('name')
     # this update overrides all required and optional attributes
     actor.update(parser.parse_args())
     return actor

def main(worker_ch_name, image):
    worker_ch = WorkerChannel(name=worker_ch_name)
    # first, attempt to pull image from docker hub:
    try:
        print("Worker pulling image {}...".format(image))
        pull_image(image)
    except DockerError as e:
        # return a message to the spawner that there was an error pulling image and abort
        worker_ch.put({'status': 'error', 'msg': str(e)})
        raise e
    # inform spawner that image pulled successfully
    print("Image pulled successfully")

    # wait to receive message from spawner that it is time to subscribe to the actor channel
    print("Worker waiting on message from spawner...")
    result = worker_ch.put_sync({'status': 'ok'})

    if result['status'] == 'error':
        print("Worker received error message from spawner: {}. Quiting...".format(str(result)))
        raise WorkerException(str(result))
    actor_id = result.get('actor_id')
    tenant = result.get('tenant')
    print("Worker received ok from spawner. Message: {}, actor_id:{}".format(result, actor_id))
    api_server = None
    client_id = None
    client_secret = None
    access_token = None
    refresh_token = None
    if result.get('client') == 'yes':
        api_server = result.get('api_server')
        client_id = result.get('client_id')
        client_secret = result.get('client_secret')
        access_token = result.get('access_token')
        refresh_token = result.get('refresh_token')
    else:
        print("Did not get client:yes, got client:{}".format(result.get('client')))
    Actor.set_status(actor_id, READY)
    subscribe(tenant,
              actor_id,
              api_server,
              client_id,
              client_secret,
              access_token,
              refresh_token,
              worker_ch)

def manage_workers(actor_id):
    """Scale workers for an actor if based on message queue size and policy."""
    print("Entering manage_workers for {}".format(actor_id))
    try:
        actor = Actor.from_db(actors_store[actor_id])
    except KeyError:
        print("Did not find actor; returning.")
        return
    workers = Worker.get_workers(actor_id)

 def get(self, actor_id):
     # check that actor exists
     try:
         actor = Actor.from_db(actors_store[actor_id])
     except KeyError:
         raise APIException(
             "actor not found: {}'".format(actor_id), 404)
     # TODO
     # retrieve pending messages from the queue
     return ok(result={'messages': []})

def handle(socket, address):
    fileobj = socket.makefile('rw')
    while not Actor.by_socket(socket).disconnected:
        line = fileobj.readline()
        if not line:
            Actor.by_socket(socket).flush()
            Actor.by_socket(socket).disconnect()
            continue
        try:
            msg = Message.from_string(line)
            log.debug('<= %s %s' % (repr(msg.target), repr(msg)))
            resp = dispatcher.dispatch(socket, msg)
        except Exception as e:
            log.exception(e)
            actor = Actor.by_socket(socket)
            if actor.is_user() and actor.get_user().registered.nick and actor.get_user().registered.user:
                resp = [
                    Message(actor, 'NOTICE', 'The message your client has just sent could not be parsed or processed.'),
                    Message(actor, 'NOTICE', 'If this is a problem with the server, please open an issue at:'),
                    Message(actor, 'NOTICE', 'https://github.com/abesto/python-ircd'),
                    Message(actor, 'NOTICE', '---'),
                    Message(actor, 'NOTICE', 'The message sent by your client was:'),
                    Message(actor, 'NOTICE', line.strip("\n")),
                    Message(actor, 'NOTICE', 'The error was:'),
                    Message(actor, 'NOTICE', str(e)),
                    Message(actor, 'NOTICE', '---'),
                    Message(actor, 'NOTICE', 'Closing connection.')
                ]
                quit_resp = dispatcher.dispatch(socket, Message(None, 'QUIT', 'Protocol error'))
                if isinstance(quit_resp, list):
                    resp += quit_resp
                else:
                    resp.append(quit_resp)
            else:
                resp = Message(actor, 'ERROR')
            Actor.by_socket(socket).disconnect()

        try:
            router.send(resp)
        except Exception as e:
            log.exception(e)
            Actor.by_socket(socket).disconnect()

 def post(self, actor_id):
     args = self.validate_post()
     state = args['state']
     try:
         actor = Actor.from_db(actors_store[actor_id])
     except KeyError:
         raise APIException(
             "actor not found: {}'".format(actor_id), 404)
     actor.state = state
     actors_store[actor_id] = actor.to_db()
     return ok(result=actor, msg="State updated successfully.")

 def check_new_params(self, cmd):
     valid, msg = self.check_common(cmd)
     # validate the actor_id
     try:
         actor = Actor.from_db(actors_store[cmd.get('actor_id')])
     except KeyError:
         return False, "Unable to look up actor with id: {}".format(cmd.get('actor_id')), None
     # validate the worker id
     try:
         Worker.get_worker(actor_id=cmd.get('actor_id'), ch_name=cmd.get('worker_id'))
     except WorkerException as e:
         return False, "Unable to look up worker: {}".format(e.msg), None
     return valid, msg, actor.owner

 def put(self, actor_id):
     try:
         actor = Actor.from_db(actors_store[actor_id])
     except KeyError:
         raise APIException(
             "actor not found: {}'".format(actor_id), 404)
     args = self.validate_put()
     update_image = False
     args['name'] = actor['name']
     args['id'] = actor['id']
     args['executions'] = actor['executions']
     args['state'] = actor['state']
     if args['image'] == actor.image:
         args['status'] = actor.status
     else:
         update_image = True
         args['status'] = SUBMITTED
     actor = Actor(args)
     actors_store[actor.id] = actor.to_db()
     if update_image:
         ch = CommandChannel()
         ch.put_cmd(actor_id=actor.id, image=actor.image)
     return ok(result=actor, msg="Actor updated successfully.")