refactor: dataclass and file naming for queue info (#301)

* refactor: dataclass and file naming for queue info

* apply suggestions

src/braket/aws/aws_quantum_task.py

@@ -24,6 +24,7 @@
 from braket.ahs.analog_hamiltonian_simulation import AnalogHamiltonianSimulation
 from braket.annealing.problem import Problem
 from braket.aws.aws_session import AwsSession
+from braket.aws.queue_information import QuantumTaskQueueInfo, QueuePriority
 from braket.circuits import Instruction
 from braket.circuits.circuit import Circuit, Gate, QubitSet
 from braket.circuits.circuit_helpers import validate_circuit_and_shots
@@ -54,7 +55,6 @@
 from braket.ir.blackbird import Program as BlackbirdProgram
 from braket.ir.openqasm import Program as OpenQASMProgram
 from braket.pulse.pulse_sequence import PulseSequence
-from braket.queue_information.queue_position import QueuePosition, QueuePriority
 from braket.schema_common import BraketSchemaBase
 from braket.task_result import (
     AnalogHamiltonianSimulationTaskResult,
@@ -315,12 +315,12 @@ def state(self, use_cached_value: bool = False) -> str:
         """
         return self._status(use_cached_value)
 
-    def queue_position(self) -> QueuePosition:
+    def queue_position(self) -> QuantumTaskQueueInfo:
         """
         The queue position details for the quantum task.
 
         Returns:
-            QueuePosition: Instance containing the queue position information
+            QuantumTaskQueueInfo: Instance containing the queue position information
             for the task.
 
             Note: The queue_position is only returned when task is not in
@@ -329,12 +329,12 @@ def queue_position(self) -> QueuePosition:
         Examples:
             task status = QUEUED
             >>> task.queue_position()
-            QueuePosition(queue_position='2', queue_priority=<QueuePriority.NORMAL: 'Normal'>,
-            message=None)
+            QuantumTaskQueueInfo(queue_position='2',
+            queue_priority=<QueuePriority.NORMAL: 'Normal'>, message=None)
 
             task status = COMPLETED
             >>> task.queue_position()
-            QueuePosition(queue_position='2',
+            QuantumTaskQueueInfo(queue_position='2',
             queue_priority=<QueuePriority.NORMAL: 'Normal'>,
             message='Task is in COMPLETED status. AmazonBraket does
                         not show queue position for this status.')
@@ -345,9 +345,9 @@ def queue_position(self) -> QueuePosition:
 
         if queue_position == "None":
             message = response["message"]
-            return QueuePosition(queue_position, queue_priority, message)
+            return QuantumTaskQueueInfo(queue_position, queue_priority, message)
 
-        return QueuePosition(queue_position, queue_priority)
+        return QuantumTaskQueueInfo(queue_position, queue_priority)
 
     def _status(self, use_cached_value: bool = False) -> str:
         metadata = self.metadata(use_cached_value)

src/braket/queue_information/queue_position.py → src/braket/aws/queue_information.py

@@ -30,9 +30,9 @@ class QueuePriority(str, Enum):
 
 
 @dataclass
-class QueuePosition:
+class QuantumTaskQueueInfo:
     """
-    Represents information related to data fetched from metadata.
+    Represents quantum tasks queue information.
 
     Attributes:
         queue_position (str): current position of your quantum task within a respective

src/braket/tasks/local_quantum_task.py

@@ -12,7 +12,7 @@
 # language governing permissions and limitations under the License.
 
 import asyncio
-from typing import Dict, Union
+from typing import Union
 
 from braket.tasks import (
     AnnealingQuantumTaskResult,
@@ -48,9 +48,6 @@ def cancel(self) -> None:
     def state(self) -> str:
         return "COMPLETED"
 
-    def queue_position(self) -> Dict[str, str]:
-        raise NotImplementedError("Cannot show queue position for completed task")
-
     def result(
         self,
     ) -> Union[

src/braket/tasks/quantum_task.py

@@ -15,7 +15,6 @@
 from abc import ABC, abstractmethod
 from typing import Any, Dict, Union
 
-from braket.queue_information.queue_position import QueuePosition
 from braket.tasks.annealing_quantum_task_result import AnnealingQuantumTaskResult
 from braket.tasks.gate_model_quantum_task_result import GateModelQuantumTaskResult
 from braket.tasks.photonic_model_quantum_task_result import PhotonicModelQuantumTaskResult
@@ -43,13 +42,6 @@ def state(self) -> str:
             str: State of the quantum task.
         """
 
-    @abstractmethod
-    def queue_position(self) -> QueuePosition:
-        """Get the queue position of the quantum task.
-        Returns:
-            QueuePosition: queue position of the quantum task.
-        """
-
     @abstractmethod
     def result(
         self,

test/integ_tests/test_queue_position.py

@@ -12,9 +12,9 @@
 # language governing permissions and limitations under the License.
 
 from braket.aws import AwsDevice
+from braket.aws.queue_information import QuantumTaskQueueInfo, QueuePriority
 from braket.circuits import Circuit
 from braket.devices import Devices
-from braket.queue_information import QueuePosition, QueuePriority
 
 
 def test_task_queue_position():
@@ -27,7 +27,7 @@ def test_task_queue_position():
     queue_information = task.queue_position()
 
     # data type validations
-    assert isinstance(queue_information, QueuePosition)
+    assert isinstance(queue_information, QuantumTaskQueueInfo)
     assert isinstance(queue_information.queue_priority, QueuePriority)
     assert isinstance(queue_information.queue_position, str)
 

test/unit_tests/braket/aws/test_aws_quantum_task.py

@@ -26,6 +26,7 @@
 from braket.aws import AwsQuantumTask
 from braket.aws.aws_quantum_task import _create_annealing_device_params
 from braket.aws.aws_session import AwsSession
+from braket.aws.queue_information import QuantumTaskQueueInfo, QueuePriority
 from braket.circuits import Circuit
 from braket.circuits.gates import PulseGate
 from braket.circuits.serialization import (
@@ -47,7 +48,6 @@
 from braket.ir.blackbird import Program as BlackbirdProgram
 from braket.ir.openqasm import Program as OpenQASMProgram
 from braket.pulse import Frame, Port, PulseSequence
-from braket.queue_information import QueuePosition, QueuePriority
 from braket.tasks import (
     AnalogHamiltonianSimulationQuantumTaskResult,
     AnnealingQuantumTaskResult,
@@ -206,7 +206,7 @@ def test_metadata_call_if_none(quantum_task):
 def test_queue_position(quantum_task):
     state_1 = "QUEUED"
     _mock_metadata(quantum_task._aws_session, state_1)
-    assert quantum_task.queue_position() == QueuePosition(
+    assert quantum_task.queue_position() == QuantumTaskQueueInfo(
         queue_position="2", queue_priority=QueuePriority.NORMAL, message=None
     )
 
@@ -215,7 +215,7 @@ def test_queue_position(quantum_task):
         f"'Task is in {state_2} status. AmazonBraket does not show queue position for this status.'"
     )
     _mock_metadata(quantum_task._aws_session, state_2)
-    assert quantum_task.queue_position() == QueuePosition(
+    assert quantum_task.queue_position() == QuantumTaskQueueInfo(
         queue_position="None", queue_priority=QueuePriority.NORMAL, message=message
     )
 

test/unit_tests/braket/tasks/test_local_quantum_task.py

@@ -56,12 +56,6 @@ def test_async():
     TASK.async_result()
 
 
-def test_queue_position():
-    error_message = "Cannot show queue position for completed task"
-    with pytest.raises(NotImplementedError, match=error_message):
-        TASK.queue_position()
-
-
 def test_str():
     expected = "LocalQuantumTask('id':{})".format(TASK.id)
     assert str(TASK) == expected