async def METHOD_NAME(self, uuid):
    request_data: Dict = tornado.escape.json_decode(self.request.body)
    params = request_data['dataset']['params']
    params["type"] = ds_type = params["type"].upper()
    cls = get_dataset_cls(ds_type)
    ConverterCls = cls.get_msg_converter()
    converter = ConverterCls()
    try:
        dataset_params = converter.to_python(params)
        executor = self.state.executor_state.get_executor()
        ds = await load(filetype=cls, executor=executor, enable_async=True, **dataset_params)
        with tracer.start_as_current_span("prime_numba_caches"):
            await self.prime_numba_caches(ds)
        self.dataset_state.register(
            uuid=uuid,
            dataset=ds,
            params=request_data['dataset'],
            converted=dataset_params,
        )
        details = await self.dataset_state.serialize(dataset_id=uuid)
        msg = Message(self.state).create_dataset(dataset=uuid, details=details)
        log_message(msg)
        self.write(msg)
        self.event_registry.broadcast_event(msg)
    except Exception as e:
        if uuid in self.dataset_state:
            await self.dataset_state.remove(uuid)
        msg = Message(self.state).create_dataset_error(uuid, str(e))
        log_message(msg, exception=True)
        self.write(msg)
        return