Reference

Dataset

dms.datasets.models.Dataset

Bases: RulesModel

Source code in src/dms/datasets/models.py

class Dataset(RulesModel):
    id = models.CharField(primary_key=True, default=uuid.uuid4)
    version = models.CharField(null=True, blank=True)
    title = models.CharField()
    name = AutoSlugField(populate_from="title")
    project = models.ForeignKey(
        "projects.Project",
        on_delete=models.PROTECT,
        related_name="datasets",
        null=True,
        db_constraint=False,
    )
    created_at = models.DateTimeField(
        db_default=models.functions.Now(),
        verbose_name=_("Created at"),
        blank=True,
    )
    last_modified_at = models.DateTimeField(
        auto_now=True, verbose_name=_("Last modified at")
    )
    metadata = JSONBField(
        schema=dataset_metadata.schema, default=dict, encoder=DjangoJSONEncoder
    )
    embargo_end_date = models.DateField(null=True, blank=True)
    contributors = models.ManyToManyField(
        "users.User", through="DatasetContribution", blank=True
    )
    tags = TaggableManager(through=GenericStringTaggedItem, blank=True)

    extent = gis_models.GeometryField(
        null=True,
        blank=True,
        verbose_name="Spatial Extent",
    )

    def __str__(self):
        return self.title

    def get_absolute_url(self):
        return reverse("datasets:dataset_detail", kwargs={"pk": self.pk})

    def clone(self):
        """
        Create a clone of the current dataset.
        """
        old_instance = self

        # Create new instance with a new UUID
        new_instance = Dataset.objects.create(
            title=self.title,
            version=self.version,
            project=self.project,
            metadata=self.metadata,
            embargo_end_date=self.embargo_end_date,
        )

        # Copy many-to-many relationships
        # Tags
        for tag in old_instance.tags.all():
            new_instance.tags.add(tag)

        # Contributors (through DatasetContribution)
        for contribution in old_instance.contributor_roles.all():
            DatasetContribution.objects.create(
                dataset=new_instance, user=contribution.user, roles=contribution.roles
            )

        return new_instance

    def compute_extent(self):
        """
        Compute the union of the extents of all resources and update the dataset extent.
        """
        self.extent = None
        for r in self.resources.exclude(extent=None):
            if self.extent:
                self.extent = self.extent.union(r.extent)
            else:
                self.extent = r.extent

        if self.extent:
            self.extent = Polygon.from_bbox(self.extent.extent)

        self.save(update_fields=["extent"])

    @cached_property
    def under_embargo(self):
        return (
            self.embargo_end_date
            and self.embargo_end_date > tz.localtime(tz.now()).date()
        )

    class Meta:
        rules_permissions = {
            "add": rules.is_authenticated,
            "view": rules.always_allow,
            "change": dataset_in_user_projects,
            "delete": dataset_in_user_projects,
        }

clone()

Create a clone of the current dataset.

Source code in src/dms/datasets/models.py

def clone(self):
    """
    Create a clone of the current dataset.
    """
    old_instance = self

    # Create new instance with a new UUID
    new_instance = Dataset.objects.create(
        title=self.title,
        version=self.version,
        project=self.project,
        metadata=self.metadata,
        embargo_end_date=self.embargo_end_date,
    )

    # Copy many-to-many relationships
    # Tags
    for tag in old_instance.tags.all():
        new_instance.tags.add(tag)

    # Contributors (through DatasetContribution)
    for contribution in old_instance.contributor_roles.all():
        DatasetContribution.objects.create(
            dataset=new_instance, user=contribution.user, roles=contribution.roles
        )

    return new_instance

compute_extent()

Compute the union of the extents of all resources and update the dataset extent.

Source code in src/dms/datasets/models.py

def compute_extent(self):
    """
    Compute the union of the extents of all resources and update the dataset extent.
    """
    self.extent = None
    for r in self.resources.exclude(extent=None):
        if self.extent:
            self.extent = self.extent.union(r.extent)
        else:
            self.extent = r.extent

    if self.extent:
        self.extent = Polygon.from_bbox(self.extent.extent)

    self.save(update_fields=["extent"])

dms.datasets.models.Resource

Bases: LifecycleModelMixin, RulesModel

Source code in src/dms/datasets/models.py

class Resource(LifecycleModelMixin, RulesModel):
    class Role(models.TextChoices):
        DATA = "data", "Data"
        PRESENTATION = "presentation", "Presentation"
        DOCUMENTATION = "documentation", "Documentation"
        METADATA = "metadata", "Metadata"
        RAW_DATA = "raw_data", "Raw Data"

    class AccessType(models.TextChoices):
        PUBLIC = "public", "Public"
        INTERNAL = "internal", "Internal"
        PERMISSION_REQUIRED = "permission_required", "Permission required"

    id = models.CharField(
        primary_key=True,
    )
    title = models.CharField(
        help_text="A name that describes the resource", default="", blank=True
    )
    description = models.CharField(default="", blank=True)
    uri = models.CharField(
        verbose_name="URI of the resource",
        help_text="",
    )
    dataset = models.ForeignKey(
        "Dataset", on_delete=models.CASCADE, related_name="resources"
    )
    created_at = models.DateTimeField(
        db_default=models.functions.Now(), verbose_name=_("Created at"), blank=True
    )
    last_modified_at = models.DateTimeField(
        auto_now=True, verbose_name=_("Last modified at")
    )
    metadata = models.JSONField(
        blank=True, null=True, encoder=DjangoJSONEncoder, verbose_name="GDAL Metadata"
    )
    role = models.CharField(blank=True, null=True, choices=Role.choices)
    access_type = models.CharField(
        blank=True, null=True, default=AccessType.INTERNAL, choices=AccessType.choices
    )

    user_metadata = models.JSONField(
        blank=True,
        null=True,
        encoder=DjangoJSONEncoder,
        verbose_name="User provided metadata",
    )

    last_sync = models.JSONField(null=True, blank=True, encoder=DjangoJSONEncoder)
    is_metadata_manual = models.BooleanField(
        default=False,
    )

    objects = InheritanceManager()
    tags = TaggableManager(through=GenericStringTaggedItem, blank=True)

    extent = gis_models.GeometryField(null=True, blank=True)

    class Meta:
        rules_permissions = {
            "add": rules.is_authenticated,
            "view": rules.always_allow,
            "change": resource_in_user_projects,
            "delete": resource_in_user_projects,
        }

    def __str__(self):
        return self.title

    def get_absolute_url(self):
        return reverse(
            "datasets:resource_detail",
            kwargs={"pk": self.pk, "dataset_pk": self.dataset_id},
        )

    @property
    def detail_url(self):
        self.__class__.objects.get_subclass(id=self.pk).get_absolute_url()

    @property
    def type(self):
        return None

    def _get_http_headers(self):
        """Extract Last-Modified and ETag headers for HTTP resources."""
        if not self.uri.startswith("http"):
            return {}

        try:
            response = requests.head(self.uri, timeout=30)
            headers = {}

            if "Last-Modified" in response.headers:
                headers["last_modified"] = response.headers["Last-Modified"]

            return headers
        except requests.RequestException:
            return {}

    def infer_metadata(self, deferred=True):
        """
        Infer the metadata of the resource. For generic resources this
        only checks if the uri supports HTTP protocol and in case extracts
        the http headers using the HEAD method.
        """
        if self.is_metadata_manual:
            return

        if deferred:
            app.configure_task(name="dms.datasets.tasks.infer_metadata_task").defer(
                resource_id=self.pk
            )
            return

        http_headers = self._get_http_headers()
        if http_headers:
            if not self.metadata:
                self.metadata = {}
            self.metadata["http_headers"] = http_headers
            self.last_sync = {"timestamp": now(), "status": "ok"}
            self.save(update_fields=["metadata", "last_sync"])

    @hook(
        AFTER_SAVE,
        condition=WhenFieldHasChanged("uri", has_changed=True),
        on_commit=True,
    )
    def populate(self):
        """
        Automatically infer metadata when the URI changes

        **NOTE**: this is triggered by LifecycleModelMixin after saving the model
        """
        self.__class__.objects.get_subclass(id=self.pk).infer_metadata()

    def get_edit_url(self):
        return reverse(
            "datasets:resource_update",
            kwargs={"dataset_pk": self.dataset_id, "pk": self.pk},
        )

    @property
    def edit_url(self):
        return self.__class__.objects.get_subclass(id=self.pk).get_edit_url()

    @transaction.atomic()
    def to_class(self, cls):
        """
        Convert the current resource to another subclass of Resource.

        Args:
            cls (Type[Resource]): The desired subclass to convert to.
        Raises:
            TypeError: If the desired class is not a subclass of Resource.
        """
        cursor = connection.cursor()
        if cls != self.__class__:
            if issubclass(cls, Resource):
                # make sure that the desired class is an instance of resource
                if cls == Resource:
                    # If the desidered class is Resource
                    cursor.execute(
                        f"DELETE FROM {self.__class__._meta.db_table} WHERE resource_ptr_id = %s",  # noqa: E501, S608
                        [self.pk],
                    )
                    Resource.objects.get(pk=self.pk).infer_metadata()
                elif self.__class__ == Resource:
                    # make sure that the current class is Resource and not a subclass
                    child = cls()
                    child.resource_ptr = self
                    child.__dict__.update(self.__dict__)
                    child.save()
                    child.infer_metadata()
                else:
                    cursor.execute(
                        f"DELETE FROM {self.__class__._meta.db_table} WHERE resource_ptr_id = %s",  # noqa: E501, S608
                        [self.pk],
                    )
                    child = cls()
                    child.resource_ptr = self
                    child.__dict__.update(self.__dict__)
                    child.save()
                    child.infer_metadata()
            else:
                raise TypeError(f"{cls} is not a subclass of {self.__class__}")
        else:
            return

infer_metadata(deferred=True)

Infer the metadata of the resource. For generic resources this only checks if the uri supports HTTP protocol and in case extracts the http headers using the HEAD method.

Source code in src/dms/datasets/models.py

def infer_metadata(self, deferred=True):
    """
    Infer the metadata of the resource. For generic resources this
    only checks if the uri supports HTTP protocol and in case extracts
    the http headers using the HEAD method.
    """
    if self.is_metadata_manual:
        return

    if deferred:
        app.configure_task(name="dms.datasets.tasks.infer_metadata_task").defer(
            resource_id=self.pk
        )
        return

    http_headers = self._get_http_headers()
    if http_headers:
        if not self.metadata:
            self.metadata = {}
        self.metadata["http_headers"] = http_headers
        self.last_sync = {"timestamp": now(), "status": "ok"}
        self.save(update_fields=["metadata", "last_sync"])

populate()

Automatically infer metadata when the URI changes

NOTE: this is triggered by LifecycleModelMixin after saving the model

Source code in src/dms/datasets/models.py

@hook(
    AFTER_SAVE,
    condition=WhenFieldHasChanged("uri", has_changed=True),
    on_commit=True,
)
def populate(self):
    """
    Automatically infer metadata when the URI changes

    **NOTE**: this is triggered by LifecycleModelMixin after saving the model
    """
    self.__class__.objects.get_subclass(id=self.pk).infer_metadata()

to_class(cls)

Convert the current resource to another subclass of Resource.

Parameters:

Name	Type	Description	Default
cls	Type[Resource]	The desired subclass to convert to.	required

Raises: TypeError: If the desired class is not a subclass of Resource.

Source code in src/dms/datasets/models.py

@transaction.atomic()
def to_class(self, cls):
    """
    Convert the current resource to another subclass of Resource.

    Args:
        cls (Type[Resource]): The desired subclass to convert to.
    Raises:
        TypeError: If the desired class is not a subclass of Resource.
    """
    cursor = connection.cursor()
    if cls != self.__class__:
        if issubclass(cls, Resource):
            # make sure that the desired class is an instance of resource
            if cls == Resource:
                # If the desidered class is Resource
                cursor.execute(
                    f"DELETE FROM {self.__class__._meta.db_table} WHERE resource_ptr_id = %s",  # noqa: E501, S608
                    [self.pk],
                )
                Resource.objects.get(pk=self.pk).infer_metadata()
            elif self.__class__ == Resource:
                # make sure that the current class is Resource and not a subclass
                child = cls()
                child.resource_ptr = self
                child.__dict__.update(self.__dict__)
                child.save()
                child.infer_metadata()
            else:
                cursor.execute(
                    f"DELETE FROM {self.__class__._meta.db_table} WHERE resource_ptr_id = %s",  # noqa: E501, S608
                    [self.pk],
                )
                child = cls()
                child.resource_ptr = self
                child.__dict__.update(self.__dict__)
                child.save()
                child.infer_metadata()
        else:
            raise TypeError(f"{cls} is not a subclass of {self.__class__}")
    else:
        return

dms.datasets.models.MapResource

Bases: Resource

Source code in src/dms/datasets/models.py

class MapResource(Resource):
    class Type(models.TextChoices):
        NINA = "nina", "Nina Map configuration"
        URL = "url", "Link to a published map"

    map_type = models.CharField(default=Type.URL, choices=Type.choices)

    def get_edit_url(self):
        return reverse(
            "datasets:mapresource_update",
            kwargs={"dataset_pk": self.dataset_id, "pk": self.pk},
        )

    @property
    def type(self):
        return "map"

dms.datasets.models.RasterResource

Bases: Resource

Source code in src/dms/datasets/models.py

class RasterResource(Resource):
    @property
    def type(self):
        return "raster"

    titiler = models.JSONField(
        default=dict, verbose_name="Titiler configuration", blank=True
    )

    @property
    def preview_url(self):
        """
        Return a URL to preview the resource using titiler
        **NOTE** this only works for cloud optimized geotiffs

        Returns:
            url (str): the url of the preview as a TMS endpoint
        """
        params = {
            **self.titiler,
            "url": self.uri,
        }
        if len(self.metadata["bands"]) > 1:
            params = [(k, v) for k, v in params.items()] + [
                (
                    "bidx",
                    f"{band['band']}",
                )
                for band in self.metadata["bands"]
                if band.get("colorInterpretation") in ["Red", "Green", "Blue"]
            ]
        else:
            params["colormap_name"] = "viridis"

            params = (
                [(k, v) for k, v in params.items()]
                + [
                    (
                        "rescale",
                        f"{band.get('minimum', 0)},{band.get('maximum', 1)}",
                    )
                    for band in self.metadata["bands"]
                ]
                + [
                    (
                        "bidx",
                        f"{band['band']}",
                    )
                    for band in self.metadata["bands"]
                ]
            )
        return settings.DATASETS_TITILER_URL + "/cog/preview/?" + urlencode(params)

    def infer_metadata(self, deferred=True):
        """
        Infer the metadata of the resuurce using GDAL.

        Args:
            deferred (bool): should the inference be executed in a deferred task?
        """
        if self.is_metadata_manual:
            return

        if deferred:
            app.configure_task(name="dms.datasets.tasks.infer_metadata_task").defer(
                resource_id=self.pk
            )
            return

        if not re.search(r"^https?://", self.uri):
            self.last_sync = {
                "status": "not started",
                "timestamp": now(),
                "error": "resource is not reachable",
                "warnings": [],
            }
            self.metadata = {}
            self.save(update_fields=["last_sync", "metadata"])
            return

        try:
            # Disable permenent auxillary files to prevent GDAL
            # creating a stats file with a remote resource
            uri = f"/vsicurl/{self.uri}" if self.uri.startswith("http") else self.uri
            with gdal.config_options({"GDAL_PAM_ENABLED": "NO"}):
                with gdal.Run(
                    "raster",
                    "info",
                    input=uri,
                    approx_stats=True,
                ) as alg:
                    metadata = alg.Output()
                    # Add HTTP headers to metadata if present
                    http_headers = self._get_http_headers()
                    if http_headers:
                        metadata["http_headers"] = http_headers

                    self.metadata = metadata
                    self.last_sync = {"timestamp": now(), "status": "ok"}
                    self.save(update_fields=["metadata", "last_sync"])
        except Exception as e:
            self.last_sync = {
                "status": "fail",
                "timestamp": now(),
                "error": str(e),
                "warnings": [],
            }
            self.metadata = {}
            self.save(update_fields=["last_sync", "metadata"])

    @hook(
        AFTER_SAVE,
        condition=WhenFieldHasChanged("metadata", has_changed=True),
    )
    def _extract_from_metadata(self):
        if not self.last_sync:
            self.last_sync = {
                "status": "never",
                "warnings": [],
            }
        try:
            self.extent = GEOSGeometry(json.dumps(self.metadata.get("wgs84Extent")))
            self.save(update_fields=["extent"])
        except Exception:
            self.last_sync["warnings"].append(traceback.format_exc())
            self.save(update_fields=["last_sync"])

    def get_edit_url(self):
        return reverse(
            "datasets:rasterresource_update",
            kwargs={"dataset_pk": self.dataset_id, "pk": self.pk},
        )

preview_url property

Return a URL to preview the resource using titiler NOTE this only works for cloud optimized geotiffs

Returns:

Name	Type	Description
url	str	the url of the preview as a TMS endpoint

infer_metadata(deferred=True)

Infer the metadata of the resuurce using GDAL.

Parameters:

Name	Type	Description	Default
deferred	bool	should the inference be executed in a deferred task?	True

Source code in src/dms/datasets/models.py

def infer_metadata(self, deferred=True):
    """
    Infer the metadata of the resuurce using GDAL.

    Args:
        deferred (bool): should the inference be executed in a deferred task?
    """
    if self.is_metadata_manual:
        return

    if deferred:
        app.configure_task(name="dms.datasets.tasks.infer_metadata_task").defer(
            resource_id=self.pk
        )
        return

    if not re.search(r"^https?://", self.uri):
        self.last_sync = {
            "status": "not started",
            "timestamp": now(),
            "error": "resource is not reachable",
            "warnings": [],
        }
        self.metadata = {}
        self.save(update_fields=["last_sync", "metadata"])
        return

    try:
        # Disable permenent auxillary files to prevent GDAL
        # creating a stats file with a remote resource
        uri = f"/vsicurl/{self.uri}" if self.uri.startswith("http") else self.uri
        with gdal.config_options({"GDAL_PAM_ENABLED": "NO"}):
            with gdal.Run(
                "raster",
                "info",
                input=uri,
                approx_stats=True,
            ) as alg:
                metadata = alg.Output()
                # Add HTTP headers to metadata if present
                http_headers = self._get_http_headers()
                if http_headers:
                    metadata["http_headers"] = http_headers

                self.metadata = metadata
                self.last_sync = {"timestamp": now(), "status": "ok"}
                self.save(update_fields=["metadata", "last_sync"])
    except Exception as e:
        self.last_sync = {
            "status": "fail",
            "timestamp": now(),
            "error": str(e),
            "warnings": [],
        }
        self.metadata = {}
        self.save(update_fields=["last_sync", "metadata"])

dms.datasets.models.TabularResource

Bases: Resource

Source code in src/dms/datasets/models.py

class TabularResource(Resource):
    @property
    def type(self):
        return "tabular"

    def infer_metadata(self, deferred=True):
        """
        Infer the metadata of the resuurce using GDAL.

        Args:
            deferred (bool): should the inference be executed in a deferred task?
        """
        if self.is_metadata_manual:
            return

        if deferred:
            app.configure_task(name="dms.datasets.tasks.infer_metadata_task").defer(
                resource_id=self.pk
            )
            return

        if not re.search(r"^https?://", self.uri):
            self.last_sync = {
                "status": "not started",
                "timestamp": now(),
                "error": "resource is not reachable",
                "warnings": [],
            }
            self.metadata = {}
            self.save(update_fields=["last_sync", "metadata"])
            return

        try:
            with gdal.Run(
                "vector",
                "info",
                input=self.uri,
            ) as alg:
                metadata = alg.Output()

                # Add HTTP headers to metadata if present
                http_headers = self._get_http_headers()
                if http_headers:
                    metadata["http_headers"] = http_headers

                self.metadata = metadata
                self.last_sync = {"timestamp": now(), "status": "ok", "warnings": []}

                self.save(
                    update_fields=[
                        "metadata",
                        "last_sync",
                    ]
                )

        except Exception as e:
            self.last_sync = {
                "status": "fail",
                "timestamp": now(),
                "error": str(e),
                "warnings": [],
            }
            self.metadata = {}
            self.save(update_fields=["last_sync", "metadata"])

    def get_edit_url(self):
        return reverse(
            "datasets:tabularresource_update",
            kwargs={"dataset_pk": self.dataset_id, "pk": self.pk},
        )

    @hook(
        AFTER_SAVE,
        condition=WhenFieldHasChanged("metadata", has_changed=True),
    )
    def _extract_from_metadata(self):
        """
        Extract from GDAL's ogrinfo output the:
        - layers, with structure and CRS
        - extent
        """
        if not self.last_sync:
            self.last_sync = {
                "status": "never",
                "warnings": [],
            }

        with transaction.atomic():
            layers = self.metadata.get("layers", [])

            tables = []
            coverage = None
            for layer in layers:
                geometries = layer.get("geometryFields", [])
                extent = None

                try:
                    if len(geometries):
                        geom = geometries[0]

                        extent = Polygon.from_bbox(geom.get("extent"))
                        extent.srid = int(
                            geom.get("coordinateSystem")
                            .get("projjson")
                            .get("id")
                            .get("code")
                        )

                        if extent.srid != 4326:
                            extent.transform(4326)

                        coverage = extent if not coverage else coverage.union(extent)
                except Exception:
                    self.last_sync["warnings"].append(traceback.format_exc())

                tables.append(
                    DataTable(
                        name=layer.get("name").replace("_gdal_http_", ""),
                        fields=layer.get("fields", []),
                        metadata=layer.get("metadata", {}),
                        resource=self,
                        count=layer.get("featureCount"),
                        geometryFields=geometries,
                        extent=extent,
                    )
                )

            DataTable.objects.filter(resource=self).delete()
            DataTable.objects.bulk_create(tables)

            self.extent = coverage
            self.save(update_fields=["extent", "last_sync"])

infer_metadata(deferred=True)

Infer the metadata of the resuurce using GDAL.

Parameters:

Name	Type	Description	Default
deferred	bool	should the inference be executed in a deferred task?	True

Source code in src/dms/datasets/models.py

def infer_metadata(self, deferred=True):
    """
    Infer the metadata of the resuurce using GDAL.

    Args:
        deferred (bool): should the inference be executed in a deferred task?
    """
    if self.is_metadata_manual:
        return

    if deferred:
        app.configure_task(name="dms.datasets.tasks.infer_metadata_task").defer(
            resource_id=self.pk
        )
        return

    if not re.search(r"^https?://", self.uri):
        self.last_sync = {
            "status": "not started",
            "timestamp": now(),
            "error": "resource is not reachable",
            "warnings": [],
        }
        self.metadata = {}
        self.save(update_fields=["last_sync", "metadata"])
        return

    try:
        with gdal.Run(
            "vector",
            "info",
            input=self.uri,
        ) as alg:
            metadata = alg.Output()

            # Add HTTP headers to metadata if present
            http_headers = self._get_http_headers()
            if http_headers:
                metadata["http_headers"] = http_headers

            self.metadata = metadata
            self.last_sync = {"timestamp": now(), "status": "ok", "warnings": []}

            self.save(
                update_fields=[
                    "metadata",
                    "last_sync",
                ]
            )

    except Exception as e:
        self.last_sync = {
            "status": "fail",
            "timestamp": now(),
            "error": str(e),
            "warnings": [],
        }
        self.metadata = {}
        self.save(update_fields=["last_sync", "metadata"])

Project

dms.projects.models.Project

Bases: RulesModel

Source code in src/dms/projects/models.py

class Project(RulesModel):
    class Status(models.TextChoices):
        ACTIVE = "N", "Active"
        COMPLETED = "T", "Completed"
        PARKED = "P", "Parked"
        NOT_COMPLETED = "C", "Not completed"

    number = models.CharField(primary_key=True)
    name = models.CharField(null=True, blank=True)
    description = models.TextField(blank=True, null=True)
    memberships = models.ManyToManyField(
        "users.User", through="ProjectMembership", blank=True
    )
    start_date = models.DateTimeField()
    end_date = models.DateTimeField(blank=True, null=True)
    status = models.CharField(null=True, blank=True, choices=Status.choices)
    category = models.ForeignKey(
        "Category",
        on_delete=models.SET_NULL,
        null=True,
        blank=True,
        db_constraint=False,
    )
    section = models.ForeignKey(
        "Section", on_delete=models.SET_NULL, null=True, blank=True, db_constraint=False
    )
    customer = models.CharField(null=True, blank=True)
    budget = models.DecimalField(decimal_places=2, max_digits=16, null=True, blank=True)

    topics = models.ManyToManyField("ProjectTopic", blank=True)
    tags = TaggableManager(through=GenericStringTaggedItem, blank=True)

    def __str__(self) -> str:
        if self.name:
            return f"{self.number} {self.name}"

        return self.number

    def get_absolute_url(self):
        return reverse("projects:project_detail", kwargs={"pk": self.pk})

    @property
    def leaders(self):
        return User.objects.filter(
            memberships__role=ProjectMembership.Role.OWNER,
            memberships__project_id=self.number,
        )

    class Meta:
        rules_permissions = {
            "add": rules.is_staff,
            "view": rules.always_allow,
            "change": project_role_is(ProjectMembership.Role.OWNER),
            "delete": rules.is_staff,
        }

dms.projects.models.DMP

Bases: RulesModel

Source code in src/dms/projects/models.py

class DMP(RulesModel):
    name = models.CharField()
    data = models.JSONField(null=True, blank=True)
    schema = models.JSONField(null=True, blank=True)
    schema_from = models.ForeignKey(
        "DMPSchema", on_delete=models.SET_NULL, null=True, blank=True
    )
    project = models.OneToOneField(
        "Project", null=True, blank=True, on_delete=models.SET_NULL, related_name="dmp"
    )
    owner = models.ForeignKey(User, on_delete=models.CASCADE)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    external_reference = models.CharField(
        null=True,
        blank=True,
        help_text="URL or reference to an external DMP",
    )
    external_file = models.FileField(
        upload_to=upload_external_dmp,
        null=True,
        blank=True,
        help_text="Upload the external DMP",
    )
    featured_at = models.DateTimeField(null=True, blank=True)

    def get_absolute_url(self):
        return reverse("projects:dmp_detail", kwargs={"pk": self.pk})

    @property
    def is_external(self):
        return bool(self.external_reference or self.external_file)

    class Meta:
        rules_permissions = {
            "add": rules.is_authenticated,
            "view": rules.always_allow,
            "change": is_owner,
            "delete": is_owner,
        }

    def __str__(self):
        return self.name

Uploads

dms.uploads.models.HookRequest

Bases: LifecycleModel

A class that stores requests send by the Tusd webhooks. On upload success it implements a method to move the uploaded file according to the upload metadata.

Source code in src/dms/uploads/models.py

class HookRequest(LifecycleModel):
    """
    A class that stores requests send by the Tusd webhooks.
    On upload success it implements a method to move the uploaded file
    according to the upload metadata.
    """

    id = models.CharField(primary_key=True, default=uuid.uuid4)
    event = models.JSONField(encoder=DjangoJSONEncoder)
    created_at = models.DateTimeField(auto_now_add=True)
    type = models.CharField()
    completed_at = models.DateTimeField(null=True, blank=True)
    user = models.ForeignKey(
        "users.User", on_delete=models.SET_NULL, null=True, blank=True
    )

    def process(self):
        """
        This is intended to be used in a queue.

        Move the file in the s3 bucket to a different position and
        register a new resource.
        If a resource with the same file name already exists just the file is updated
        """
        metadata = self.event.get("Upload").get("MetaData")
        dataset = Dataset.objects.get(id=metadata.get("dataset"))
        source_key = self.event.get("Upload").get("Storage").get("Key")
        dest_key = f"project/{dataset.project_id}/datasets/{dataset.id}/{metadata.get('filename')}"  # noqa: E501

        if settings.AWS_ACCESS_KEY_ID:
            from dms.core.helpers.s3client import s3client

            s3client.copy_object(
                Bucket=settings.AWS_STORAGE_BUCKET_NAME,
                CopySource={
                    "Bucket": settings.AWS_STORAGE_BUCKET_NAME,
                    "Key": source_key,
                },
                Key=f"{settings.MEDIA_BASE_LOCATION}/{dest_key}",
            )

            Resource.objects.get_or_create(
                dataset=dataset,
                uri=settings.MEDIA_URL + dest_key,
                defaults={
                    "title": metadata.get("filename"),
                    "metadata": {},
                    "id": self.event.get("Upload").get("ID"),
                },
            )

            self.completed_at = now()
            self.save(update_fields=["completed_at"])
            s3client.delete_object(
                Bucket=settings.AWS_STORAGE_BUCKET_NAME, Key=source_key
            )

process()

This is intended to be used in a queue.

Move the file in the s3 bucket to a different position and register a new resource. If a resource with the same file name already exists just the file is updated

Source code in src/dms/uploads/models.py

def process(self):
    """
    This is intended to be used in a queue.

    Move the file in the s3 bucket to a different position and
    register a new resource.
    If a resource with the same file name already exists just the file is updated
    """
    metadata = self.event.get("Upload").get("MetaData")
    dataset = Dataset.objects.get(id=metadata.get("dataset"))
    source_key = self.event.get("Upload").get("Storage").get("Key")
    dest_key = f"project/{dataset.project_id}/datasets/{dataset.id}/{metadata.get('filename')}"  # noqa: E501

    if settings.AWS_ACCESS_KEY_ID:
        from dms.core.helpers.s3client import s3client

        s3client.copy_object(
            Bucket=settings.AWS_STORAGE_BUCKET_NAME,
            CopySource={
                "Bucket": settings.AWS_STORAGE_BUCKET_NAME,
                "Key": source_key,
            },
            Key=f"{settings.MEDIA_BASE_LOCATION}/{dest_key}",
        )

        Resource.objects.get_or_create(
            dataset=dataset,
            uri=settings.MEDIA_URL + dest_key,
            defaults={
                "title": metadata.get("filename"),
                "metadata": {},
                "id": self.event.get("Upload").get("ID"),
            },
        )

        self.completed_at = now()
        self.save(update_fields=["completed_at"])
        s3client.delete_object(
            Bucket=settings.AWS_STORAGE_BUCKET_NAME, Key=source_key
        )