urls

importing.urls

Attributes

app_name = 'importing' module-attribute

urlpatterns = [path('', DataImportListView.as_view(), name='dataimport_list'), path('<int:pk>/', DataImportDetailView.as_view(), name='dataimport_detail'), path('edit/<int:pk>', DataImportEditView.as_view(), name='dataimport_edit'), path('create/', DataImportCreateView.as_view(), name='dataimport_create'), path('delete/<int:pk>', DataImportDeleteView.as_view(), name='dataimport_delete'), path('api/upload/', DataImportUploadAPIView.as_view(), name='api_upload'), path('api/dataingestion/', DataIngestionQueryView.as_view(), name='api_data_ingestion'), path('thingsboard-import-maps/', ThingsBoardImportMapListView.as_view(), name='thingsboardimportmap_list'), path('thingsboard-import-maps/create/', ThingsboardImportMapCreateView.as_view(), name='thingsboardimportmap_create'), path('thingsboard-import-maps/<int:pk>/', ThingsboardImportMapDetailView.as_view(), name='thingsboardimportmap_detail'), path('thingsboard-import-maps/edit/<int:pk>/', ThingsboardImportMapEditView.as_view(), name='thingsboardimportmap_edit'), path('thingsboard-import-maps/delete/<int:pk>/', ThingsboardImportMapEditView.as_view(), name='thingsboardimportmap_delete'), path('thingsboard-data-retrieval/', ThingsboardDataRetrievalView.as_view(), name='thingsboard_data_retrieval'), path('map-layers/', MapLayerListView.as_view(), name='maplayerimport_list'), path('map-layers/create/', MapLayerCreateView.as_view(), name='maplayerimport_create'), path('map-layers/<int:pk>/', MapLayerDetailView.as_view(), name='maplayerimport_detail'), path('map-layers/edit/<int:pk>/', MapLayerEditView.as_view(), name='maplayerimport_edit'), path('map-layers/delete/<int:pk>/', MapLayerDeleteView.as_view(), name='maplayerimport_delete')] module-attribute

Classes

DataImportCreateView

Bases: CustomCreateView

View to create a data import.

DataImportDeleteView

Bases: CustomDeleteView

View to delete a data import.

DataImportDetailView

Bases: CustomDetailView

View to view a data import.

DataImportEditView

Bases: CustomEditView

View to edit a data import.

Functions

form_valid(form)

Reprocess if a new file is uploaded or the reprocess button is selected.

Source code in importing/views.py

def form_valid(self, form):
    """Reprocess if a new file is uploaded or the reprocess button is selected."""
    reprocess = self.request.POST.get("action") == "reprocess"
    new_file = self.object.pk and "rawfile" in form.changed_data
    if reprocess or new_file:
        self.object.status = "N"
        self.object.save()

    return super().form_valid(form)

DataImportListView

Bases: CustomTableView

View to list all data imports.

DataImportUploadAPIView

Bases: APIView

API endpoint for uploading data files for import.

Access is controlled via object-level permissions using django-guardian. Users can only upload data for Station objects for which they have the change_station permission assigned.

Functions

get_permitted_stations(user)

Get stations the user has permission to import data for.

Source code in importing/views.py

def get_permitted_stations(self, user):
    """Get stations the user has permission to import data for."""
    return get_objects_for_user(user, "change_station", klass=Station)

post(request)

Upload a data file and create a data import.

Source code in importing/views.py

@extend_schema(
    summary="Upload data file for import",
    description="""
    Upload a data file to create a new data import.

    **Permissions**: Users can only upload data for stations they have
    `change_station` permission for.

    **File Upload**: Send as multipart/form-data with all parameters in the body.
    """,
    request={
        "multipart/form-data": {
            "type": "object",
            "properties": {
                "station": {
                    "type": "string",
                    "description": "Station code",
                },
                "format": {
                    "type": "integer",
                    "description": "Format ID",
                },
                "rawfile": {
                    "type": "string",
                    "format": "binary",
                    "description": "Data file to upload",
                },
                "visibility": {
                    "type": "string",
                    "enum": ["public", "private"],
                    "default": "private",
                    "description": "Visibility level",
                },
                "observations": {
                    "type": "string",
                    "description": "Additional observations",
                },
            },
            "required": ["station", "format", "rawfile"],
        }
    },
    responses={
        201: OpenApiResponse(
            response=DataImportUploadResponseSerializer,
            description="Data import created successfully",
            examples=[
                OpenApiExample(
                    "Successful upload",
                    value={
                        "data_import_id": 123,
                        "station": "CAR_02_HC_01",
                        "format": 47,
                        "rawfile": "/media/imports/data_2024.csv",
                        "visibility": "private",
                        "date": "2024-12-09T10:30:00Z",
                        "status": "N",
                        "status_display": "Not queued",
                    },
                )
            ],
        ),
        400: OpenApiResponse(description="Invalid request parameters or file"),
        403: OpenApiResponse(
            description="User does not have permission to upload data."
        ),
        404: OpenApiResponse(description="Station or format not found"),
    },
    tags=["importing"],
)
def post(self, request):
    """Upload a data file and create a data import."""
    # Get the uploaded file
    rawfile = request.FILES.get("rawfile")
    if not rawfile:
        return Response(
            {"rawfile": ["No file was submitted."]},
            status=status.HTTP_400_BAD_REQUEST,
        )

    # Build data dict for serializer
    data = {
        "station": request.data.get("station"),
        "format": request.data.get("format"),
        "visibility": request.data.get("visibility", "private"),
        "observations": request.data.get("observations", ""),
        "rawfile": rawfile,
    }

    # Validate request parameters
    serializer = DataImportUploadRequestSerializer(data=data)
    if not serializer.is_valid():
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

    validated_data = serializer.validated_data
    station = validated_data["station"]

    # Check user permissions for this station
    permitted_stations = self.get_permitted_stations(request.user)
    if station not in permitted_stations:
        return Response(
            {
                "detail": "You do not have permission to upload data "
                f"for station '{station.station_code}'."
            },
            status=status.HTTP_403_FORBIDDEN,
        )

    # Create the data import
    try:
        data_import = DataImport.objects.create(
            station=station,
            format=validated_data["format"],
            origin=ImportOrigin.objects.get_or_create(origin="api")[0],
            visibility=validated_data["visibility"],
            observations=validated_data.get("observations", ""),
            rawfile=validated_data["rawfile"],
            owner=request.user,
        )
    except Exception as e:
        logging.exception(f"Error creating data import: {e}")
        return Response(
            {"detail": "An internal error occurred while creating data import."},
            status=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )

    # Return the created import
    response_serializer = DataImportUploadResponseSerializer(
        data_import, context={"request": request}
    )
    return Response(response_serializer.data, status=status.HTTP_201_CREATED)

DataIngestionQueryView

Bases: APIView

API endpoint for querying data ingestion status/list.

• No data_import_id query param: return list of imports the user may view.
• With data_import_id: only the owner may request detailed info (including log). Non-owners receive 403 for direct PK detail requests; they should use the list endpoint.

Functions

get(request)

Return list or owner-only detail for data import ingestion status.

Source code in importing/views.py

@extend_schema(
    summary="Query data ingestion status",
    description="""
    - If `data_import_id` query parameter is provided, the endpoint returns the
      details for that DataImport **only if the requesting user is the owner**.
      Owners receive the full detail including the ingestion `log`.
    - If no `data_import_id` is provided, the endpoint returns a list of
      DataImport objects the user can view, ordered by submission date.
    """,
    parameters=[
        OpenApiParameter(
            name="data_import_id",
            type=OpenApiTypes.INT,
            location=OpenApiParameter.QUERY,
            required=False,
            description="Primary key of a DataImport to return detailed info.",
        ),
    ],
    responses={
        200: OpenApiResponse(
            description="Data ingestion status retrieved successfully"
        ),
        400: OpenApiResponse(description="Invalid request parameters"),
        403: OpenApiResponse(
            description="Only the owner may request a specific data import by PK"
        ),
        404: OpenApiResponse(description="Data import not found"),
    },
    tags=["importing"],
)
def get(self, request):
    """Return list or owner-only detail for data import ingestion status."""
    data_import_id = request.query_params.get("data_import_id")

    if not data_import_id:
        # Return list of DataImport objects the user can view
        permitted_qs = get_objects_for_user(
            request.user, "view_dataimport", klass=DataImport
        ).order_by("-date")
        serializer = DataImportUploadResponseSerializer(
            permitted_qs, many=True, context={"request": request}
        )
        return Response(serializer.data, status=status.HTTP_200_OK)

    # If a PK is provided, fetch the object and enforce owner-only access.
    try:
        data_import = DataImport.objects.get(pk=data_import_id)
    except DataImport.DoesNotExist:
        return Response(
            {"detail": "Data import not found."}, status=status.HTTP_404_NOT_FOUND
        )

    # Only the owner may request detailed info by PK
    if data_import.owner != request.user:
        return Response(
            {
                "detail": "Only the owner may request this data import by primary key."  # noqa: E501
            },
            status=status.HTTP_403_FORBIDDEN,
        )

    # Owner may view full details (including log)
    serializer = DataImportDetailSerializer(
        data_import, context={"request": request}
    )
    return Response(serializer.data, status=status.HTTP_200_OK)

MapLayerCreateView

Bases: CustomCreateView

View to create a map layer import.

MapLayerDeleteView

Bases: CustomDeleteView

View to delete a map layer import.

MapLayerDetailView

Bases: CustomDetailView

View to view a data import.

MapLayerEditView

Bases: CustomEditView

View to edit a map layer imports.

MapLayerListView

Bases: CustomTableView

View to list all map layer imports.

ThingsBoardImportMapListView

Bases: CustomTableView

ThingsboardDataRetrievalView

Bases: LoginRequiredMixin, FormView

ThingsboardImportMapCreateView

Bases: CustomCreateView

ThingsboardImportMapDetailView

Bases: CustomDetailView

ThingsboardImportMapEditView

Bases: CustomEditView