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()

get_view_schema(self, request) async

Get the schema for a ResourceView type, including all fields and their types. This is similar to get_config_for_component but for ResourceViews.

Source code in ofrak/gui/server.py 
@exceptions_to_http(SerializedError)
async def get_view_schema(self, request: Request) -> Response:
    """
    Get the schema for a ResourceView type, including all fields and their types.
    This is similar to get_config_for_component but for ResourceViews.
    """
    view_type: Type[ViewableResourceTag] = self._serializer.from_pjson(
        await request.json(), Type[ViewableResourceTag]
    )

    # Build schema similar to get_config_for_component
    fields_info = []
    if dataclasses.is_dataclass(view_type):
        for field in dataclasses.fields(view_type):
            if field.name.startswith("_"):  # Skip private fields like _resource, _deleted
                continue

            field_info = {
                "name": field.name,
                "type": self._convert_to_class_name_str(field.type),
                "args": self._construct_arg_response(field.type),
                "fields": self._construct_field_response(field.type),
                "enum": self._construct_enum_response(field.type),
                "default": _format_default(field.default)
                if not isinstance(field.default, dataclasses._MISSING_TYPE)
                else None,
                "required": field.default == dataclasses.MISSING,
            }
            fields_info.append(field_info)

    # Return schema in same format as get_config_for_component
    schema = {
        "name": view_type.__name__,
        "type": self._convert_to_class_name_str(view_type),
        "fields": fields_info,
        # Optionally include the composed attribute types for reference
        "composed_attributes": self._serializer.to_pjson(
            list(getattr(view_type, "composed_attributes_types", [])),
            List[Type[ResourceAttributes]],
        ),
    }

    return json_response(schema)

add_view_to_resource(self, request) async

Add a view to a resource using the view type and field values.

Source code in ofrak/gui/server.py 
@exceptions_to_http(SerializedError)
async def add_view_to_resource(self, request: Request) -> Response:
    """
    Add a view to a resource using the view type and field values.
    """
    resource = await self._get_resource_for_request(request)
    body = await request.json()

    view_type: Type[ResourceViewInterface] = self._serializer.from_pjson(
        body["view_type"], Type[ResourceViewInterface]
    )
    field_values = body["fields"]
    view_instance = view_type(**field_values)
    script_str = (
        """
    {resource}"""
        f""".add_view({view_instance.__repr__()})
    """
        """await {resource}.save()
    """
    )
    await self.script_builder.add_action(resource, script_str, ActionType.MOD)

    resource.add_view(view_instance)
    await resource.save()
    return json_response(self._serialize_resource(resource))

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()}