handlers

importing.signals.handlers

Attributes

User = get_user_model() module-attribute

Classes

DataImport

Bases: PermissionsBase

Model to store the data imports.

This model stores the data imports, which are, often, files with data that are uploaded to the system. The data is then processed asynchronously and stored in the database.

Attributes:

Name	Type	Description
station	ForeignKey	Station to which the data belongs.
format	ForeignKey	Format of the data.
rawfile	FileField	File with the data to be imported.
date	DateTimeField	Date of submission of the data.
start_date	DateTimeField	Start date of the data.
end_date	DateTimeField	End date of the data.
records	IntegerField	Number of records in the data.
observations	TextField	Notes or observations about the data.
status	TextField	Status of the import.
log	TextField	Log of the data ingestion, indicating any errors.

Functions

clean()

Validate information and uploads the measurement data.

Source code in importing/models.py

def clean(self) -> None:
    """Validate information and uploads the measurement data."""
    tz = self.station.timezone
    if not tz:
        raise ValidationError("Station must have a timezone set.")

    if self.origin.origin == "Thingsboard" and not self.format.thingsboard:
        raise ValidationError("Ensure a Thingsboard-specific format is specified.")

MapLayerImport

Bases: PermissionsBase

Functions

clean()

Validate the uploaded GeoTIFF and its transformed lon/lat bounds.

Source code in importing/models.py

def clean(self) -> None:
    """Validate the uploaded GeoTIFF and its transformed lon/lat bounds."""

    if not self.file:
        return

    import rasterio
    from rasterio.warp import transform_bounds

    file_obj = self.file.file
    try:
        with rasterio.MemoryFile(file_obj.read()) as memfile:
            with memfile.open() as dataset:
                if dataset.count == 0:
                    raise ValidationError(
                        {"file": "File contains no raster bands."}
                    )
                if dataset.crs is None:
                    raise ValidationError(
                        {"file": "File has no coordinate reference system (CRS)."}
                    )

                left, bottom, right, top = transform_bounds(
                    dataset.crs,
                    "EPSG:4326",
                    dataset.bounds.left,
                    dataset.bounds.bottom,
                    dataset.bounds.right,
                    dataset.bounds.top,
                )

                if not (-180 <= left <= 180 and -180 <= right <= 180):
                    raise ValidationError(
                        {
                            "file": (
                                "GeoTIFF longitude values are out of range "
                                "[-180, 180]."
                            )
                        }
                    )
                if not (-90 <= bottom <= 90 and -90 <= top <= 90):
                    raise ValidationError(
                        {
                            "file": (
                                "GeoTIFF latitude values are out of range "
                                "[-90, 90]."
                            )
                        }
                    )
    except ValidationError:
        raise
    except Exception as e:
        raise ValidationError(
            {"file": f"Could not open as a valid GeoTIFF or read coordinates: {e}"}
        )
    finally:
        file_obj.seek(0)

Functions

ingest_data(data_import_pk)

Initiate the ingestion of data into the DB.

If the status of the data import is "not queued", the request is processed. The data loaded and saved to the database. The status is updated to completed or failed depending on the outcome.

Source code in importing/tasks.py

@on_commit_task()
def ingest_data(data_import_pk: int) -> None:
    """Initiate the ingestion of data into the DB.

    If the status of the data import is "not queued", the request is processed. The data
    loaded and saved to the database. The status is updated to completed or failed
    depending on the outcome.
    """
    data_import = DataImport.objects.get(pk=data_import_pk)
    if data_import.status != "N":
        return

    data_import.status = "Q"
    data_import.save()

    try:
        getLogger("huey").info("Ingesting data for %s", data_import)
        data_import.start_date, data_import.end_date, data_import.records = (
            save_temp_data_to_permanent(data_import)
        )
        update_variables_for_station(data_import.station.station_code)
        data_import.status = "C"
        data_import.log = "Data ingestion completed successfully"
        getLogger("huey").info("Data ingestion for %s completed", data_import)
    except Exception as e:
        data_import.status = "F"
        data_import.log = str(e)
        getLogger("huey").exception("Error ingesting data for %s", data_import)
    finally:
        data_import.save()
        clear_cache()

process_data_ingestion(sender, instance, **kwargs)

Schedules the data ingestion task.

Source code in importing/signals/handlers.py

@receiver(post_save, sender=DataImport)
def process_data_ingestion(sender, instance: PermissionsBase, **kwargs):
    """Schedules the data ingestion task."""
    ingest_data(instance.pk)

set_model_permissions(sender, **kwargs)

Set model-level permissions.

Source code in importing/signals/handlers.py

@receiver(post_migrate)
def set_model_permissions(sender, **kwargs):
    """Set model-level permissions."""
    DataImport.set_model_permissions()

set_object_permissions(sender, instance, **kwargs)

Set object-level permissions.

Source code in importing/signals/handlers.py

@receiver(post_save, sender=DataImport)
@receiver(post_save, sender=MapLayerImport)
def set_object_permissions(sender, instance: PermissionsBase, **kwargs):
    """Set object-level permissions."""
    instance.set_object_permissions()