Source code for ferrosoft.apps.ferrobase.models.file

import mimetypes
import re
from dataclasses import dataclass
from pathlib import PurePosixPath

from django.core.exceptions import ValidationError
from django.core.files.storage import default_storage
from django.db import models
from django.urls import reverse
from django.utils.translation import gettext_lazy as _

from ferrosoft.apps.ferrobase.models.base import CommonLimit, UUIDModel
from ferrosoft.apps.ferrobase.models.shared import LocalUser


[docs] class AccessSubject(models.IntegerChoices): """Subject who can access a file.""" OWNER = 1, _("Owner") """Owner is the creator of the file.""" ORGANIZATION = 2, _("Organization") """Organization of the owner of the file.""" PUBLIC = 3, _("Public") """Any logged-in user."""
[docs] class AccessMode(models.IntegerChoices): """Access mode (read or write) for a ``FileAccessGrant``.""" READ = 1, _("Read") WRITE = 2, _("Write")
[docs] @dataclass class FileInfo: """Lightweight value object carrying the display name and media type of a file. ``file_name`` is the original name given by the user; ``media_type`` is the MIME type string (e.g. ``"application/pdf"``). """ file_name: str media_type: str
[docs] class Directory(UUIDModel): """Hierarchical folder structure for organising ``StoredFile`` objects. Directories can be nested via ``parent_directory``; a ``null`` parent indicates a root directory. Names must be unique within the same parent. """ class Meta: verbose_name = _("Folder") verbose_name_plural = _("Folders") constraints = [ models.UniqueConstraint( name="uniq_dir_per_dir", fields=["parent_directory", "name"], ) ] indexes = [ models.Index(fields=["name"]), models.Index(fields=["parent_directory"]), ] name = models.CharField( max_length=int(CommonLimit.NORMAL_CHAR), verbose_name=_("Name"), ) parent_directory = models.ForeignKey( "Directory", on_delete=models.CASCADE, verbose_name=_("Parent directory"), null=True, ) def __str__(self): return self.name
[docs] @classmethod def find_by_name(cls, name): return cls.objects.filter(name=name).first()
[docs] @classmethod def get_or_create(cls, name: str) -> "Directory": it = cls.find_by_name(name) return it or cls.objects.create(name=name)
[docs] def get_absolute_url(self): return reverse("ferrobase:directory_listing", args=[self.pk])
def _stored_file_upload_to(instance, filename): id_str = str(instance.pk) return "stored/%s/%s/%s/%s" % ( id_str[0:2], id_str[2:4], id_str[4:6], id_str, )
[docs] class StoredFileManager(models.Manager): """Manager for ``StoredFile`` that pre-fetches owner and directory relations. Also provides ``create_with_access_grants`` for atomically creating a file together with its access control entries. """
[docs] def get_queryset(self): return super().get_queryset().select_related("owner", "directory")
[docs] def create_with_access_grants( self, grants: dict[AccessSubject, list[AccessMode]], **kwargs ) -> "StoredFile": file = self.create(**kwargs) FileAccessGrant.objects.bulk_create( [ FileAccessGrant(file=file, subject=subject, mode=mode) for subject, modes in grants.items() for mode in modes ], ignore_conflicts=True, unique_fields=["file", "subject", "mode"], ) return file
[docs] class StoredFile(UUIDModel): """ StoredFile wraps a Django file and enriches it with some metadata. * Name the file name given by the user. * Content is the file content. * Media type is the media/MIME type guessed by file name extension. * Directory is where the file is placed into (it has nothing to do with the location in STORAGES backend). * Creation time is when the file was created. * Owner is the user who can manage the file. Beyond the owner attribute, no access control has been defined. """ class Meta: verbose_name = _("File") verbose_name_plural = _("Files") indexes = [ models.Index(fields=["directory"]), models.Index(fields=["owner"]), ] objects = StoredFileManager() name = models.CharField( max_length=int(CommonLimit.NORMAL_CHAR), verbose_name=_("Name"), ) content = models.FileField( verbose_name=_("Content"), upload_to=_stored_file_upload_to, null=True, ) media_type = models.CharField( max_length=int(CommonLimit.NORMAL_CHAR), verbose_name=_("Media type"), ) directory = models.ForeignKey( Directory, on_delete=models.CASCADE, verbose_name=_("Directory"), ) creation_time = models.DateTimeField( verbose_name=_("Creation time"), ) owner = models.ForeignKey( LocalUser, on_delete=models.RESTRICT, verbose_name=_("Owner"), ) expire_on = models.DateTimeField( verbose_name=_("Expire on"), null=True, blank=True, db_index=True, ) def __str__(self): return self.name
[docs] @classmethod def guess_media_type(cls, file_name: str) -> str: p = PurePosixPath(file_name) if p.suffix in mimetypes.types_map: return mimetypes.types_map[p.suffix] return "application/octet-stream"
[docs] @classmethod def get_authenticated_file(cls, path: str, user: LocalUser) -> FileInfo | None: match = re.match( r"stored/../../../([0-9a-fA-F]{8}\b-[0-9a-fA-F]{4}\b-[0-9a-fA-F]{4}\b-[0-9a-fA-F]{4}\b-[0-9a-fA-F]{12}).*", path, ) if match is None: return None try: stored_file = cls.objects.filter(id=match.group(1), owner=user).first() if stored_file is None: return None return FileInfo(stored_file.name, stored_file.media_type) except ValidationError: # Matched path part is not an UUID. return None
[docs] def delete(self, *args, **kwargs): if self.content and self.content.name: default_storage.delete(self.content.name) super().delete(*args, **kwargs)
[docs] def get_absolute_url(self): return self.content.url
[docs] class FileAccessGrant(UUIDModel): """Records a single access permission for a ``StoredFile``. Each grant links a file to a subject (owner, organisation, or public) and a mode (read or write). The combination of file, subject, and mode is unique, preventing duplicate grants. """ class Meta: verbose_name = _("File Access Grant") verbose_name_plural = _("File Access Grants") constraints = [ models.UniqueConstraint( name="uniq_file_access_grant", fields=["file", "subject", "mode"], ) ] file = models.ForeignKey( StoredFile, on_delete=models.CASCADE, related_name="access_grants", verbose_name=_("File"), ) subject = models.IntegerField( choices=AccessSubject.choices, verbose_name=_("Subject"), ) mode = models.IntegerField( choices=AccessMode.choices, verbose_name=_("Mode"), )