Skip to content

server.py

ofrak.gui.server

AiohttpOFRAKServer

start(self) async

Start the server then return.

Source code in ofrak/gui/server.py 
async def start(self):  # pragma: no cover
    """
    Start the server then return.
    """
    self.runner = web.AppRunner(self._app)
    await self.runner.setup()
    server = web.TCPSite(self.runner, host=self._host, port=self._port)
    await server.start()

stop(self) async

Stop the server.

Source code in ofrak/gui/server.py 
async def stop(self):  # pragma: no cover
    """
    Stop the server.
    """
    await self.runner.server.shutdown()
    await self.runner.cleanup()

run_until_cancelled(self) async

To be run after start_server, within an asyncio Task. cancel() that task to shutdown the server.

Source code in ofrak/gui/server.py 
async def run_until_cancelled(self):  # pragma: no cover
    """
    To be run after `start_server`, within an asyncio Task.
    cancel() that task to shutdown the server.
    """
    try:
        while True:
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        pass
    finally:
        await self.runner.cleanup()

add_comment(self, request) async

Expected POST body is a comment in the form Tuple[Optional[Range], str] (serialized to JSON).

Source code in ofrak/gui/server.py 
async def add_comment(self, request: Request) -> Response:
    """
    Expected POST body is a comment in the form Tuple[Optional[Range], str] (serialized to JSON).
    """
    resource = await self._get_resource_for_request(request)
    comment = self._serializer.from_pjson(await request.json(), Tuple[Optional[Range], str])
    script_str = (
        """
    await {resource}.run"""
        f"""(AddCommentModifier, AddCommentModifierConfig({comment}))
    """
    )
    await self.script_builder.add_action(resource, script_str, ActionType.MOD)
    try:
        result = await resource.run(AddCommentModifier, AddCommentModifierConfig(comment))
        await self.script_builder.commit_to_script(resource)
    except Exception as e:
        await self.script_builder.clear_script_queue(resource)
        raise e
    return json_response(await self._serialize_component_result(result))

_serialize_resource_model(self, resource_model) private

Serialize the resource model, stripped of information irrelevant to the frontend.

Source code in ofrak/gui/server.py 
def _serialize_resource_model(self, resource_model: ResourceModel) -> PJSONType:
    """
    Serialize the resource model, stripped of information irrelevant to the frontend.
    """
    result = {
        "id": resource_model.id.hex(),
        "data_id": resource_model.data_id.hex() if resource_model.data_id else None,
        "parent_id": resource_model.parent_id.hex() if resource_model.parent_id else None,
        "tags": [tag.__module__ + "." + tag.__qualname__ for tag in resource_model.tags],
        "attributes": self._serializer.to_pjson(
            resource_model.attributes, Dict[Type[ResourceAttributes], ResourceAttributes]
        ),
        "caption": resource_model.caption,
    }
    return result

_serialize_resource(self, resource) private

Serialize the resource as a serialized model, stripped of information irrelevant to the frontend.

Source code in ofrak/gui/server.py 
def _serialize_resource(self, resource: Resource) -> PJSONType:
    """
    Serialize the resource as a serialized model, stripped of information irrelevant to the
    frontend.
    """
    return self._serialize_resource_model(resource.get_model())

_serialize_multi_resource(self, resources) private

Serialize the resources as serialized models, stripped of information irrelevant to the frontend.

Source code in ofrak/gui/server.py 
def _serialize_multi_resource(self, resources: Iterable[Resource]) -> PJSONType:
    """
    Serialize the resources as serialized models, stripped of information irrelevant to the
    frontend.
    """
    return list(map(self._serialize_resource, resources))

exceptions_to_http(error_class)

Decorator for a server function that attempts to do some work, and forwards the exception, if any, to the client over HTTP.

Usage:

@exceptions_to_http(MyErrorClass) async def handle_some_request(self, request...): ...

Source code in ofrak/gui/server.py 
def exceptions_to_http(error_class: Type[SerializedError]):
    """
    Decorator for a server function that attempts to do some work, and
    forwards the exception, if any, to the client over HTTP.

    Usage:

    @exceptions_to_http(MyErrorClass)
    async def handle_some_request(self, request...):
        ...
    """

    def exceptions_to_http_decorator(func: Callable):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            try:
                return await func(*args, **kwargs)
            except Exception as error:
                LOGGER.exception("Exception raised in aiohttp endpoint")
                return respond_with_error(error, error_class)

        return wrapper

    return exceptions_to_http_decorator

get_query_string_as_pjson(request)

URL-encoded GET parameters are all strings. For example, None is encoded as 'None', or 1 as '1', which isn't valid PJSON. We fix this by applying json.loads on each parameter.

Source code in ofrak/gui/server.py 
def get_query_string_as_pjson(request: Request) -> Dict[str, PJSONType]:
    """
    URL-encoded GET parameters are all strings. For example, None is encoded as 'None',
    or 1 as '1', which isn't valid PJSON. We fix this by applying `json.loads` on each parameter.
    """
    return {key: json.loads(value) for key, value in request.query.items()}