netbox/netbox/core/models/files.py

import logging
import os
from functools import cached_property

from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.files.storage import storages
from django.db import models
from django.utils.translation import gettext as _

from extras.storage import ScriptFileSystemStorage
from netbox.models.features import SyncedDataMixin
from utilities.querysets import RestrictedQuerySet

from ..choices import ManagedFileRootPathChoices

__all__ = (
    'ManagedFile',
)

logger = logging.getLogger('netbox.core.files')


class ManagedFile(SyncedDataMixin, models.Model):
    """
    Database representation for a file on disk. This class is typically wrapped by a proxy class (e.g. ScriptModule)
    to provide additional functionality.
    """
    created = models.DateTimeField(
        verbose_name=_('created'),
        auto_now_add=True
    )
    last_updated = models.DateTimeField(
        verbose_name=_('last updated'),
        editable=False,
        blank=True,
        null=True
    )
    file_root = models.CharField(
        verbose_name=_('file root'),
        max_length=1000,
        choices=ManagedFileRootPathChoices
    )
    file_path = models.FilePathField(
        verbose_name=_('file path'),
        editable=False,
        help_text=_('File path relative to the designated root path')
    )

    objects = RestrictedQuerySet.as_manager()
    _netbox_private = True

    class Meta:
        ordering = ('file_root', 'file_path')
        constraints = (
            models.UniqueConstraint(
                fields=('file_root', 'file_path'),
                name='%(app_label)s_%(class)s_unique_root_path'
            ),
        )
        verbose_name = _('managed file')
        verbose_name_plural = _('managed files')

    def __str__(self):
        return self.name

    @property
    def name(self):
        return self.file_path

    @property
    def full_path(self):
        return os.path.join(self._resolve_root_path(), self.file_path)

    def _resolve_root_path(self):
        storage = self.storage
        if isinstance(storage, ScriptFileSystemStorage):
            return {
                'scripts': settings.SCRIPTS_ROOT,
                'reports': settings.REPORTS_ROOT,
            }[self.file_root]
        else:
            return ""

    def sync_data(self):
        if self.data_file:
            self.file_path = os.path.basename(self.data_path)

            storage = self.storage

            with storage.open(self.full_path, 'wb+') as new_file:
                new_file.write(self.data_file.data)
    sync_data.alters_data = True

    @cached_property
    def storage(self):
        return storages.create_storage(storages.backends["scripts"])

    def clean(self):
        super().clean()

        if self.data_file and not self.file_path:
            self.file_path = os.path.basename(self.data_path)

        # Ensure that the file root and path make a unique pair
        if self._meta.model.objects.filter(
                file_root=self.file_root, file_path=self.file_path
        ).exclude(pk=self.pk).exists():
            raise ValidationError(
                _("A {model} with this file path already exists ({path}).").format(
                    model=self._meta.verbose_name.lower(),
                    path=f"{self.file_root}/{self.file_path}"
                ))

    def delete(self, *args, **kwargs):
        # Delete file from disk
        storage = self.storage
        try:
            storage.delete(self.full_path)
        except FileNotFoundError:
            pass

        return super().delete(*args, **kwargs)