Merge fa465c8 into 9291830

flexmeasures/api/v3_0/sensors.py

@@ -46,6 +46,7 @@
 get_sensor_schema = GetSensorDataSchema()
 post_sensor_schema = PostSensorDataSchema()
 sensors_schema = SensorSchema(many=True)
+sensor_schema = SensorSchema()
 
 
 class SensorAPI(FlaskView):
@@ -494,3 +495,39 @@ def get_schedule(self, sensor: Sensor, job_id: str, duration: timedelta, **kwarg
 
         d, s = request_processed()
         return dict(**response, **d), s
+
+    @route("/<id>", methods=["GET"])
+    @use_kwargs({"sensor": SensorIdField(data_key="id")}, location="path")
+    @permission_required_for_context("read", arg_name="sensor")
+    @as_json
+    def fetch_one(self, id, sensor):
+        """Fetch a given sensor.
+
+        .. :quickref: Sensor; Get a sensor
+
+        This endpoint gets a sensor.
+
+        **Example response**
+
+        .. sourcecode:: json
+
+            {
+                "name": "some gas sensor",
+                "unit": "m³/h",
+                "entity_address": "ea1.2023-08.localhost:fm1.1",
+                "event_resolution": 10,
+                "generic_asset_id": 4,
+                "timezone": "UTC",
+            }
+
+        :reqheader Authorization: The authentication token
+        :reqheader Content-Type: application/json
+        :resheader Content-Type: application/json
+        :status 200: PROCESSED
+        :status 400: INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS
+        :status 401: UNAUTHORIZED
+        :status 403: INVALID_SENDER
+        :status 422: UNPROCESSABLE_ENTITY
+        """
+        sensor = Sensor.query.filter(Sensor.id == 1).one_or_none()
+        return sensor_schema.dump(sensor), 200

flexmeasures/api/v3_0/tests/test_sensors_api.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+
+from flask import url_for
+
+
+from flexmeasures import Sensor
+from flexmeasures.api.tests.utils import get_auth_token
+
+
+def test_fetch_one_sensor(
+    client,
+    setup_api_test_data: dict[str, Sensor],
+):
+    sensor_id = 1
+    assert_response = {
+        "name": "some gas sensor",
+        "unit": "m³/h",
+        "entity_address": "ea1.2023-08.localhost:fm1.1",
+        "event_resolution": 10,
+        "generic_asset_id": 4,
+        "timezone": "UTC",
+        "status": 200,
+    }
+    headers = make_headers_for("test_supplier_user_4@seita.nl", client)
+    response = client.get(
+        url_for("SensorAPI:fetch_one", id=sensor_id),
+        headers=headers,
+    )
+    print("Server responded with:\n%s" % response.json)
+    assert response.status_code == 200
+    assert response.json == assert_response
+
+
+def make_headers_for(user_email: str | None, client) -> dict:
+    headers = {"content-type": "application/json"}
+    if user_email:
+        headers["Authorization"] = get_auth_token(client, user_email, "testtest")
+    return headers