def _get_path(self, context: Union[InputContext, OutputContext]) -> UPath:
        """
        Look for a custom destination path in the metadata, otherwise use the default
        """
        # Get the actual output object from the output context
        event_data = None
        if isinstance(context, OutputContext):
            events = context.step_context.instance.all_logs(
                run_id=context.run_id, of_type=DagsterEventType.STEP_OUTPUT
            )
            for e in events:
                event_specific_data = e.dagster_event.event_specific_data
                if (
                    event_specific_data.step_output_handle.step_key == context.step_key
                    and event_specific_data.step_output_handle.output_name
                    == context.name
                ):
                    event_data = event_specific_data
                    break

        if event_data is not None and "path" in event_data.metadata:
            return UPath(event_data.metadata["path"].value)
        return super()._get_path(context)