class DuplicateTagError(InvalidOperationError):
    """Invalid tag key and/or value."""

    code = 400
    description = "Duplicate tag key" class ExhaustedStreamError(FilesException):
    """The incoming file stream has been already consumed."""

    code = 500 class FileSizeError(StorageError):
    """Exception raised when a file larger than allowed."""

    code = 400 class InvalidTagError(InvalidOperationError):
    """Invalid tag key and/or value."""

    code = 400
    description = "Too long/short tag key or tag value." class MissingQueryParameter(FilesException):
    """Exception raised when missing a query parameter."""

    code = 400
    description = "Missing required query argument '{arg_name}'"

    def __init__(self, arg_name, **kwargs):
        """Initialize RESTException."""
        self.arg_name = arg_name
        super(MissingQueryParameter, self).__init__(**kwargs)

    def get_description(self, environ=None):
        """Get the description."""
        return self.description.format(arg_name=self.arg_name) class MultipartInvalidChunkSize(MultipartException):
    """Exception raised when multipart object is already completed."""

    code = 400
    description = "Invalid part size." class Bucket(db.Model, Timestamp):
    """Model for storing buckets.

    A bucket is a container of objects. Buckets have a default location and
    storage class. Individual objects in the bucket can however have different
    locations and storage classes.

    A bucket can be marked as deleted. A bucket can also be marked as locked
    to prevent operations on the bucket.

    Each bucket can also define a quota. The size of a bucket is the size
    of all objects in the bucket (including all versions).
    """

    __tablename__ = 'files_bucket'

    id = db.Column(
        UUIDType,
        primary_key=True,
        default=uuid.uuid4,
    )
    """Bucket identifier."""

    default_location = db.Column(
        db.Integer,
        db.ForeignKey(Location.id, ondelete='RESTRICT'),
        nullable=False)
    """Default location."""

    default_storage_class = db.Column(
        db.String(1), nullable=False,
        default=lambda: current_app.config['FILES_REST_DEFAULT_STORAGE_CLASS'])
    """Default storage class."""

    size = db.Column(db.BigInteger, default=0, nullable=False)
    """Size of bucket.

    This is a computed property which can rebuilt any time from the objects
    inside the bucket.
    """

    quota_size = db.Column(
        db.BigInteger,
        nullable=True,
        default=lambda: current_app.config['FILES_REST_DEFAULT_QUOTA_SIZE']
    )
    """Quota size of bucket.

    Usage of this property depends on which file size limiters are installed.
    """

    max_file_size = db.Column(
        db.BigInteger,
        nullable=True,
        default=lambda: current_app.config['FILES_REST_DEFAULT_MAX_FILE_SIZE']
    )
    """Maximum size of a single file in the bucket.

    Usage of this property depends on which file size limiters are installed.
    """

    locked = db.Column(db.Boolean(name='locked'),
                       default=False,
                       nullable=False)
    """Lock state of bucket.

    Modifications are not allowed on a locked bucket.
    """

    deleted = db.Column(db.Boolean(name='deleted'),
                        default=False,
                        nullable=False)
    """Delete state of bucket."""

    location = db.relationship(Location, backref='buckets')
    """Location associated with this bucket."""

    def __repr__(self):
        """Return representation of location."""
        return str(self.id)

    @property
    def quota_left(self):
        """Get how much space is left in the bucket."""
        if self.quota_size:
            return max(self.quota_size - self.size, 0)

    @property
    def size_limit(self):
        """Get size limit for this bucket.

        The limit is based on the minimum output of the file size limiters.
        """
        limits = [
            lim for lim in current_files_rest.file_size_limiters(
                self)
            if lim.limit is not None
        ]
        return min(limits) if limits else None

    @validates('default_storage_class')
    def validate_storage_class(self, key, default_storage_class):
        """Validate storage class."""
        if default_storage_class not in \
                current_app.config['FILES_REST_STORAGE_CLASS_LIST']:
            raise ValueError('Invalid storage class.')
        return default_storage_class

    @ensure_not_deleted()
    def snapshot(self, lock=False):
        """Create a snapshot of latest objects in bucket.

        :param lock: Create the new bucket in a locked state.
        :returns: Newly created bucket containing copied ObjectVersion.
        """
        with db.session.begin_nested():
            bucket = Bucket(
                default_location=self.default_location,
                default_storage_class=self.default_storage_class,
                quota_size=self.quota_size,
            )
            db.session.add(bucket)

        for o in ObjectVersion.get_by_bucket(self):
            o.copy(bucket=bucket)

        bucket.locked = True if lock else self.locked

        return bucket

    @ensure_not_deleted(msg=[BucketError('Cannot sync a deleted bucket.')])
    def sync(self, bucket, delete_extras=False):
        """Sync self bucket ObjectVersions to the destination bucket.

        The bucket is fully mirrored with the destination bucket following the
        logic:

         * same ObjectVersions are not touched
         * new ObjectVersions are added to destination
         * deleted ObjectVersions are deleted in destination
         * extra ObjectVersions in dest are deleted if `delete_extras` param is
           True

        :param bucket: The destination bucket.
        :param delete_extras: Delete extra ObjectVersions in destination if
            True.
        :returns: The bucket with an exact copy of ObjectVersions in self.
        """
        assert not bucket.locked

        src_ovs = ObjectVersion.get_by_bucket(bucket=self, with_deleted=True)
        dest_ovs = ObjectVersion.get_by_bucket(bucket=bucket,
                                               with_deleted=True)
        # transform into a dict { key: object version }
        src_dict_ovs = {}
        for ov in src_ovs:
            # keep only the latest version of an ObjectVersion and don't
            # override with previous
            if ov.key not in src_dict_ovs:
                src_dict_ovs[ov.key] = ov

        dest_dict_ovs = {}
        for ov in dest_ovs:
            # keep only the latest version of an ObjectVersion and don't
            # override with previous
            if ov.key not in dest_dict_ovs:
                dest_dict_ovs[ov.key] = ov

        for key, ov in src_dict_ovs.items():
            key_in_dest = key in dest_dict_ovs
            if ov.deleted:
                if key_in_dest and not dest_dict_ovs[key].deleted:
                    ObjectVersion.delete(bucket, key)
            else:
                if not key_in_dest:
                    ov.copy(bucket=bucket)
                else:
                    dest_ov = dest_dict_ovs[key]
                    file_in_dest_differs = ov.file_id != dest_ov.file_id
                    ov_deleted_in_dest = dest_ov.deleted
                    if file_in_dest_differs or ov_deleted_in_dest:
                        ov.copy(bucket=bucket)

        if delete_extras:
            for key, ov in dest_dict_ovs.items():
                if key not in src_dict_ovs:
                    ObjectVersion.delete(bucket, key)

        return bucket

    def get_tags(self):
        """Get tags for bucket as dictionary."""
        return {t.key: t.value for t in self.tags}

    @classmethod
    def create(cls, location=None, storage_class=None, **kwargs):
        r"""Create a bucket.

        :param location: Location of a bucket (instance or name).
            Default: Default location.
        :param storage_class: Storage class of a bucket.
            Default: Default storage class.
        :param \**kwargs: Keyword arguments are forwarded to the class
        :param \**kwargs: Keyword arguments are forwarded to the class
            constructor.
        :returns: Created bucket.
        """
        with db.session.begin_nested():
            if location is None:
                location = Location.get_default()
            elif isinstance(location, str):
                location = Location.get_by_name(location)

            obj = cls(
                default_location=location.id,
                default_storage_class=storage_class or current_app.config[
                    'FILES_REST_DEFAULT_STORAGE_CLASS'],
                **kwargs
            )
            db.session.add(obj)
        return obj

    @classmethod
    def get(cls, bucket_id):
        """Get a bucket object (excluding deleted).

        :param bucket_id: Bucket identifier.
        :returns: Bucket instance.
        """
        return cls.query.filter_by(
            id=bucket_id,
            deleted=False
        ).one_or_none()

    @classmethod
    def all(cls):
        """Return query of all buckets (excluding deleted)."""
        return cls.query.filter_by(
            deleted=False
        )

    @classmethod
    def delete(cls, bucket_id):
        """Delete a bucket.

        Does not actually delete the Bucket, just marks it as deleted.
        """
        bucket = cls.get(bucket_id)
        if not bucket or bucket.deleted:
            return False

        bucket.deleted = True
        return True

    @ensure_unlocked()
    def remove(self):
        """Permanently remove a bucket and all objects (including versions).

        .. warning::

           This by-passes the normal versioning and should only be used when
           you want to permanently delete a bucket and its objects. Otherwise
           use :py:data:`Bucket.delete()`.

           Note the method does not remove the associated file instances which
           must be garbage collected.

        :returns: ``self``.
        """
        with db.session.begin_nested():
            ObjectVersion.query.filter_by(
                bucket_id=self.id
            ).delete()
            self.query.filter_by(id=self.id).delete()
        return self class MultipartObject(db.Model, Timestamp):
    """Model for storing files in chunks.

    A multipart object belongs to a specific bucket and key and is identified
    by an upload id. You can have multiple multipart uploads for the same
    bucket and key. Once all parts of a multipart object is uploaded, the state
    is changed to ``completed``. Afterwards it is not possible to upload
    new parts. Once completed, the multipart object is merged, and added as
    a new version in the current object/bucket.

    All parts for a multipart upload must be of the same size, except for the
    last part.
    """

    __tablename__ = 'files_multipartobject'

    __table_args__ = (
        db.UniqueConstraint('upload_id', 'bucket_id', 'key', name='uix_item'),
    )

    upload_id = db.Column(
        UUIDType,
        default=uuid.uuid4,
        primary_key=True,
    )
    """Identifier for the specific version of an object."""

    bucket_id = db.Column(
        UUIDType,
        db.ForeignKey(Bucket.id, ondelete='RESTRICT'),
    )
    """Bucket identifier."""

    key = db.Column(
        db.Text().with_variant(mysql.VARCHAR(255), 'mysql'),
    )
    """Key identifying the object."""

    file_id = db.Column(
        UUIDType,
        db.ForeignKey(FileInstance.id, ondelete='RESTRICT'), nullable=False)
    """File instance for this multipart object."""

    chunk_size = db.Column(db.Integer, nullable=True)
    """Size of chunks for file."""

    size = db.Column(db.BigInteger, nullable=True)
    """Size of file."""

    completed = db.Column(db.Boolean(name='completed'),
                          nullable=False,
                          default=False)
    """Defines if object is the completed."""

    # Relationships definitions
    bucket = db.relationship(Bucket, backref='multipart_objects')
    """Relationship to buckets."""

    file = db.relationship(FileInstance, backref='multipart_objects')
    """Relationship to buckets."""

    def __repr__(self):
        """Return representation of the multipart object."""
        return "{0}:{2}:{1}".format(
            self.bucket_id, self.key, self.upload_id)

    @property
    def last_part_number(self):
        """Get last part number."""
        return int(self.size / self.chunk_size) \
            if self.size % self.chunk_size else \
            int(self.size / self.chunk_size) - 1

    @property
    def last_part_size(self):
        """Get size of last part."""
        return self.size % self.chunk_size

    @validates('key')
    def validate_key(self, key, key_):
        """Validate key."""
        return validate_key(key_)

    @staticmethod
    def is_valid_chunksize(chunk_size):
        """Check if size is valid."""
        min_csize = current_app.config['FILES_REST_MULTIPART_CHUNKSIZE_MIN']
        max_csize = current_app.config['FILES_REST_MULTIPART_CHUNKSIZE_MAX']
        return chunk_size >= min_csize and chunk_size <= max_csize

    @staticmethod
    def is_valid_size(size, chunk_size):
        """Validate max theoretical size."""
        min_csize = current_app.config['FILES_REST_MULTIPART_CHUNKSIZE_MIN']
        max_size = \
            chunk_size * current_app.config['FILES_REST_MULTIPART_MAX_PARTS']
        return size > min_csize and size <= max_size

    def expected_part_size(self, part_number):
        """Get expected part size for a particular part number."""
        last_part = self.multipart.last_part_number

        if part_number == last_part:
            return self.multipart.last_part_size
        elif part_number >= 0 and part_number < last_part:
            return self.multipart.chunk_size
        else:
            raise MultipartInvalidPartNumber()

    @ensure_uncompleted()
    def complete(self):
        """Mark a multipart object as complete."""
        if Part.count(self) != self.last_part_number + 1:
            raise MultipartMissingParts()

        with db.session.begin_nested():
            self.completed = True
            self.file.readable = True
            self.file.writable = False
        return self

    @ensure_completed()
    def merge_parts(self, version_id=None, **kwargs):
        """Merge parts into object version."""
        self.file.update_checksum(**kwargs)
        with db.session.begin_nested():
            obj = ObjectVersion.create(
                self.bucket,
                self.key,
                _file_id=self.file_id,
                version_id=version_id
            )
            self.delete()
        return obj

    def delete(self):
        """Delete a multipart object."""
        # Update bucket size.
        self.bucket.size -= self.size
        # Remove parts
        Part.query_by_multipart(self).delete()
        # Remove self
        self.query.filter_by(upload_id=self.upload_id).delete()

    @classmethod
    def create(cls, bucket, key, size, chunk_size):
        """Create a new object in a bucket."""
        bucket = as_bucket(bucket)

        if bucket.locked:
            raise BucketLockedError()

        # Validate chunk size.
        if not cls.is_valid_chunksize(chunk_size):
            raise MultipartInvalidChunkSize()

        # Validate max theoretical size.
        if not cls.is_valid_size(size, chunk_size):
            raise MultipartInvalidSize()

        # Validate max bucket size.
        bucket_limit = bucket.size_limit
        if bucket_limit and size > bucket_limit:
            desc = 'File size limit exceeded.' \
                if isinstance(bucket_limit, int) else bucket_limit.reason
            raise FileSizeError(description=desc)

        with db.session.begin_nested():
            file_ = FileInstance.create()
            file_.size = size
            obj = cls(
                upload_id=uuid.uuid4(),
                bucket=bucket,
                key=key,
                chunk_size=chunk_size,
                size=size,
                completed=False,
                file=file_,
            )
            bucket.size += size
            db.session.add(obj)
        file_.init_contents(
            size=size,
            default_location=bucket.location.uri,
            default_storage_class=bucket.default_storage_class,
        )
        return obj

    @classmethod
    def get(cls, bucket, key, upload_id, with_completed=False):
        """Fetch a specific multipart object."""
        q = cls.query.filter_by(
            upload_id=upload_id,
            bucket_id=as_bucket_id(bucket),
            key=key,
        )
        if not with_completed:
            q = q.filter(cls.completed.is_(False))

        return q.one_or_none()

    @classmethod
    def query_expired(cls, dt, bucket=None):
        """Query all uncompleted multipart uploads."""
        q = cls.query.filter(cls.created < dt).filter_by(completed=True)
        if bucket:
            q = q.filter(cls.bucket_id == as_bucket_id(bucket))
        return q

    @classmethod
    def query_by_bucket(cls, bucket):
        """Query all uncompleted multipart uploads."""
        return cls.query.filter(cls.bucket_id == as_bucket_id(bucket)) class ObjectVersion(db.Model, Timestamp):
    """Model for storing versions of objects.

    A bucket stores one or more objects identified by a key. Each object is
    versioned where each version is represented by an ``ObjectVersion``.

    An object version can either be 1) a *normal version* which is linked to
    a file instance, or 2) a *delete marker*, which is *not* linked to a file
    instance.

    An normal object version is linked to a physical file on disk via a file
    instance. This allows for multiple object versions to point to the same
    file on disk, to optimize storage efficiency (e.g. useful for snapshotting
    an entire bucket without duplicating the files).

    A delete marker object version represents that the object at hand was
    deleted.

    The latest version of an object is marked using the ``is_head`` property.
    If the latest object version is a delete marker the object will not be
    shown in the bucket.
    """

    __tablename__ = 'files_object'

    version_id = db.Column(
        UUIDType,
        primary_key=True,
        default=uuid.uuid4)
    """Identifier for the specific version of an object."""

    key = db.Column(
        db.Text().with_variant(mysql.VARCHAR(255), 'mysql'), nullable=False)
    """Key identifying the object."""

    bucket_id = db.Column(
        UUIDType,
        db.ForeignKey(Bucket.id, ondelete='RESTRICT'),
        default=uuid.uuid4, nullable=False)
    """Bucket identifier."""

    file_id = db.Column(
        UUIDType,
        db.ForeignKey(FileInstance.id, ondelete='RESTRICT'), nullable=True)
    """File instance for this object version.

    A null value in this column defines that the object has been deleted.
    """

    _mimetype = db.Column(
        db.String(255),
        index=True,
        nullable=True, )
    """MIME type of the object."""

    is_head = db.Column(db.Boolean(name='is_head'),
                        nullable=False,
                        default=True)
    """Defines if object is the latest version."""

    # Relationships definitions
    bucket = db.relationship(Bucket, backref='objects')
    """Relationship to buckets."""

    file = db.relationship(FileInstance, backref='objects')
    """Relationship to file instance."""

    __table_args__ = (
        db.UniqueConstraint('bucket_id', 'version_id', 'key'),
    )

    @validates('key')
    def validate_key(self, key, key_):
        """Validate key."""
        return validate_key(key_)

    def __unicode__(self):
        """Return unicoded object."""
        return u"{0}:{1}:{2}".format(
            self.bucket_id, self.version_id, self.key)

    # https://docs.python.org/3.3/howto/pyporting.html#str-unicode
    if sys.version_info[0] >= 3:  # Python 3
        def __repr__(self):
            """Return representation of location."""
            return self.__unicode__()
    else:  # Python 2
        def __repr__(self):
            """Return representation of location."""
            return self.__unicode__().encode('utf8')

    @hybrid_property
    def mimetype(self):
        """Get MIME type of object."""
        return self._mimetype if self._mimetype else guess_mimetype(self.key)

    @mimetype.setter
    def mimetype(self, value):
        """Setter for MIME type."""
        self._mimetype = value

    @property
    def basename(self):
        """Return filename of the object."""
        return basename(self.key)

    @property
    def deleted(self):
        """Determine if object version is a delete marker."""
        return self.file_id is None

    @ensure_no_file()
    @update_bucket_size
    def set_contents(self, stream, chunk_size=None, size=None, size_limit=None,
                     progress_callback=None):
        """Save contents of stream to file instance.

        If a file instance has already been set, this methods raises an
        ``FileInstanceAlreadySetError`` exception.

        :param stream: File-like stream.
        :param size: Size of stream if known.
        :param chunk_size: Desired chunk size to read stream in. It is up to
            the storage interface if it respects this value.
        """
        if size_limit is None:
            size_limit = self.bucket.size_limit

        self.file = FileInstance.create()
        self.file.set_contents(
            stream, size_limit=size_limit, size=size, chunk_size=chunk_size,
            progress_callback=progress_callback,
            default_location=self.bucket.location.uri,
            default_storage_class=self.bucket.default_storage_class,
        )

        return self

    @ensure_no_file()
    @update_bucket_size
    def set_location(self, uri, size, checksum, storage_class=None):
        """Set only URI location of for object.

        Useful to link files on externally controlled storage. If a file
        instance has already been set, this methods raises an
        ``FileInstanceAlreadySetError`` exception.

        :param uri: Full URI to object (which can be interpreted by the storage
            interface).
        :param size: Size of file.
        :param checksum: Checksum of file.
        :param storage_class: Storage class where file is stored ()
        """
        self.file = FileInstance()
        self.file.set_uri(
            uri, size, checksum, storage_class=storage_class
        )
        db.session.add(self.file)
        return self

    @ensure_no_file()
    @update_bucket_size
    def set_file(self, fileinstance):
        """Set a file instance."""
        self.file = fileinstance
        return self

    def send_file(self, restricted=True, trusted=False, **kwargs):
        """Wrap around FileInstance's send file."""
        return self.file.send_file(
            self.basename,
            restricted=restricted,
            mimetype=self.mimetype,
            trusted=trusted,
            **kwargs
        )

    @ensure_is_previous_version()
    def restore(self):
        """Restore this object version to become the latest version.

        Raises an exception if the object is the latest version.
        """
        # Note, copy calls create which will fail if bucket is locked.
        return self.copy()

    @ensure_not_deleted(
        msg=[ObjectVersionError('Cannot copy a delete marker.')])
    def copy(self, bucket=None, key=None):
        """Copy an object version to a given bucket + object key.

        The copy operation is handled completely at the metadata level. The
        actual data on disk is not copied. Instead, the two object versions
        will point to the same physical file (via the same FileInstance).

        All the tags associated with the current object version are copied over
        to the new instance.

        .. warning::

           If the destination object exists, it will be replaced by the new
           object version which will become the latest version.

        :param bucket: The bucket (instance or id) to copy the object to.
            Default: current bucket.
        :param key: Key name of destination object.
            Default: current object key.
        :returns: The copied object version.
        """
        new_ob = ObjectVersion.create(
            self.bucket if bucket is None else as_bucket(bucket),
            key or self.key,
            _file_id=self.file_id
        )

        for tag in self.tags:
            ObjectVersionTag.create_or_update(object_version=new_ob,
                                              key=tag.key,
                                              value=tag.value)

        return new_ob

    @ensure_unlocked(getter=lambda o: not o.bucket.locked)
    def remove(self):
        """Permanently remove a specific object version from the database.

        .. warning::

           This by-passes the normal versioning and should only be used when
           you want to permanently delete a specific object version. Otherwise
           use :py:data:`ObjectVersion.delete()`.

           Note the method does not remove the associated file instance which
           must be garbage collected.

        :returns: ``self``.
        """
        with db.session.begin_nested():
            if self.file_id:
                self.bucket.size -= self.file.size
            self.query.filter_by(
                bucket_id=self.bucket_id,
                key=self.key,
                version_id=self.version_id,
            ).delete()

        return self

    @classmethod
    def create(cls, bucket, key, _file_id=None, stream=None, mimetype=None,
               version_id=None, **kwargs):
        """Create a new object in a bucket.

        The created object is by default created as a delete marker. You must
        use ``set_contents()`` or ``set_location()`` in order to change this.

        :param bucket: The bucket (instance or id) to create the object in.
        :param key: Key of object.
        :param _file_id: For internal use.
        :param stream: File-like stream object. Used to set content of object
            immediately after being created.
        :param mimetype: MIME type of the file object if it is known.
        :param kwargs: Keyword arguments passed to ``Object.set_contents()``.
        """
        bucket = as_bucket(bucket)

        if bucket.locked:
            raise BucketLockedError()

        with db.session.begin_nested():
            latest_obj = cls.query.filter(
                cls.bucket == bucket, cls.key == key, cls.is_head.is_(True)
            ).one_or_none()
            if latest_obj is not None:
                latest_obj.is_head = False
                db.session.add(latest_obj)

            # By default objects are created in a deleted state (i.e.
            # file_id is null).
            obj = cls(
                bucket=bucket,
                key=key,
                version_id=version_id or uuid.uuid4(),
                is_head=True,
                mimetype=mimetype,
            )
            if _file_id:
                file_ = _file_id if isinstance(_file_id, FileInstance) else \
                    FileInstance.get(_file_id)
                obj.set_file(file_)
            db.session.add(obj)
        if stream:
            obj.set_contents(stream, **kwargs)
        return obj

    @classmethod
    def get(cls, bucket, key, version_id=None):
        """Fetch a specific object.

        By default the latest object version is returned, if
        ``version_id`` is not set.

        :param bucket: The bucket (instance or id) to get the object from.
        :param key: Key of object.
        :param version_id: Specific version of an object.
        """
        filters = [
            cls.bucket_id == as_bucket_id(bucket),
            cls.key == key,
        ]

        if version_id:
            filters.append(cls.version_id == version_id)
        else:
            filters.append(cls.is_head.is_(True))
            filters.append(cls.file_id.isnot(None))

        return cls.query.filter(*filters).one_or_none()

    @classmethod
    def get_versions(cls, bucket, key, desc=True):
        """Fetch all versions of a specific object.

        :param bucket: The bucket (instance or id) to get the object from.
        :param key: Key of object.
        :param desc: Sort results desc if True, asc otherwise.
        :returns: The query to execute to fetch all versions.
        """
        filters = [
            cls.bucket_id == as_bucket_id(bucket),
            cls.key == key,
        ]

        order = cls.created.desc() if desc else cls.created.asc()

        return cls.query.filter(*filters).order_by(cls.key, order)

    @classmethod
    def delete(cls, bucket, key):
        """Delete an object.

        Technically works by creating a new version which works as a delete
        marker.

        :param bucket: The bucket (instance or id) to delete the object from.
        :param key: Key of object.
        :returns: Created delete marker object if key exists else ``None``.
        """
        bucket_id = as_bucket_id(bucket)

        obj = cls.get(bucket_id, key)
        if obj:
            return cls.create(as_bucket(bucket), key)
        return None

    @classmethod
    def get_by_bucket(cls, bucket, versions=False, with_deleted=False):
        """Return query that fetches all the objects in a bucket.

        :param bucket: The bucket (instance or id) to query.
        :param versions: Select all versions if True, only heads otherwise.
        :param with_deleted: Select also deleted objects if True.
        :returns: The query to retrieve filtered objects in the given bucket.
        """
        bucket_id = bucket.id if isinstance(bucket, Bucket) else bucket

        filters = [
            cls.bucket_id == bucket_id,
        ]

        if not versions:
            filters.append(cls.is_head.is_(True))

        if not with_deleted:
            filters.append(cls.file_id.isnot(None))

        return cls.query.filter(*filters).order_by(cls.key, cls.created.desc())

    @classmethod
    def relink_all(cls, old_file, new_file):
        """Relink all object versions (for a given file) to a new file.

        .. warning::

           Use this method with great care.
        """
        assert old_file.checksum == new_file.checksum
        assert old_file.id
        assert new_file.id

        with db.session.begin_nested():
            ObjectVersion.query.filter_by(file_id=str(old_file.id)).update({
                ObjectVersion.file_id: str(new_file.id)})

    def get_tags(self):
        """Get tags for object version as dictionary."""
        return {t.key: t.value for t in self.tags}

    def __eq__(self, other):
        """Check if the two object are equals."""
        return other and isinstance(other, self.__class__) and \
            self.key == other.key and self.file_id == other.file_id

    def __ne__(self, other):
        """Check if are not equal."""
        return not self.__eq__(other=other)

    # DDL string used to avoid automap in mysql until sqlalchemy 2.0
    # https://github.com/sqlalchemy/sqlalchemy/discussions/7597
    @classmethod
    def ix_uq_partial_files_object_is_head_dll(cls):
        """Return DDL instruction for ix_uq_partial_files_object_is_head."""
        return db.DDL(
            "CREATE UNIQUE INDEX ix_uq_partial_files_object_is_head "
            "ON %(table)s (bucket_id, key) WHERE is_head") class ObjectVersionTag(db.Model):
    """Model for storing tags associated to object versions.

    Used for storing extra technical information for an object version.
    """

    __tablename__ = 'files_objecttags'

    version_id = db.Column(
        UUIDType,
        db.ForeignKey(ObjectVersion.version_id, ondelete='CASCADE'),
        default=uuid.uuid4,
        primary_key=True, )
    """Object version id."""

    key = db.Column(db.String(255), primary_key=True)
    """Tag key."""

    value = db.Column(db.Text, nullable=False)
    """Tag value."""

    object_version = db.relationship(ObjectVersion, backref='tags')
    """Relationship to object versions."""

    def copy(self, object_version=None, key=None):
        """Copy a tag to a given object version.

        :param object_version: The object version instance to copy the tag to.
            Default: current object version.
        :param key: Key of destination tag.
            Default: current tag key.
        :return: The copied object version tag.
        """
        return ObjectVersionTag.create(
            self.object_version if object_version is None else object_version,
            key or self.key,
            self.value
        )

    @classmethod
    def get(cls, object_version, key):
        """Get the tag object."""
        return cls.query.filter_by(
            version_id=as_object_version_id(object_version),
            key=key,
        ).one_or_none()

    @classmethod
    def create(cls, object_version, key, value):
        """Create a new tag for a given object version."""
        assert len(key) < 256
        assert len(value) < 256
        with db.session.begin_nested():
            obj = cls(version_id=as_object_version_id(object_version),
                      key=key,
                      value=value)
            db.session.add(obj)
        return obj

    @classmethod
    def create_or_update(cls, object_version, key, value):
        """Create or update a new tag for a given object version."""
        assert len(key) < 256
        assert len(value) < 256
        obj = cls.get(object_version, key)
        if obj:
            obj.value = value
            db.session.merge(obj)
        else:
            obj = cls.create(object_version, key, value)
        return obj

    @classmethod
    def get_value(cls, object_version, key):
        """Get the tag value."""
        obj = cls.get(object_version, key)
        return obj.value if obj else None

    @classmethod
    def delete(cls, object_version, key=None):
        """Delete tags.

        :param object_version: The object version instance or id.
        :param key: Key of the tag to delete.
            Default: delete all tags.
        """
        with db.session.begin_nested():
            q = cls.query.filter_by(
                version_id=as_object_version_id(object_version))
            if key:
                q = q.filter_by(key=key)
            q.delete() class Part(db.Model, Timestamp):
    """Part object."""

    __tablename__ = 'files_multipartobject_part'

    upload_id = db.Column(
        UUIDType,
        db.ForeignKey(MultipartObject.upload_id, ondelete='RESTRICT'),
        primary_key=True,
    )
    """Multipart object identifier."""

    part_number = db.Column(db.Integer, primary_key=True, autoincrement=False)
    """Part number."""

    checksum = db.Column(db.String(255), nullable=True)
    """String representing the checksum of the part."""

    # Relationships definitions
    multipart = db.relationship(MultipartObject, backref='parts')
    """Relationship to multipart objects."""

    @property
    def start_byte(self):
        """Get start byte in file of this part."""
        return self.part_number * self.multipart.chunk_size

    @property
    def end_byte(self):
        """Get end byte in file for this part."""
        return min(
            (self.part_number + 1) * self.multipart.chunk_size,
            self.multipart.size
        )

    @property
    def part_size(self):
        """Get size of this part."""
        return self.end_byte - self.start_byte

    @classmethod
    def create(cls, mp, part_number, stream=None, **kwargs):
        """Create a new part object in a multipart object."""
        if part_number < 0 or part_number > mp.last_part_number:
            raise MultipartInvalidPartNumber()

        with db.session.begin_nested():
            obj = cls(
                multipart=mp,
                part_number=part_number,
            )
            db.session.add(obj)
        if stream:
            obj.set_contents(stream, **kwargs)
        return obj

    @classmethod
    def get_or_none(cls, mp, part_number):
        """Get part number."""
        return cls.query.filter_by(
            upload_id=mp.upload_id,
            part_number=part_number
        ).one_or_none()

    @classmethod
    def get_or_create(cls, mp, part_number):
        """Get or create a part."""
        obj = cls.get_or_none(mp, part_number)
        if obj:
            return obj
        return cls.create(mp, part_number)

    @classmethod
    def delete(cls, mp, part_number):
        """Get part number."""
        return cls.query.filter_by(
            upload_id=mp.upload_id,
            part_number=part_number
        ).delete()

    @classmethod
    def query_by_multipart(cls, multipart):
        """Get all parts for a specific multipart upload.

        :param multipart: A :class:`invenio_files_rest.models.MultipartObject`
            instance.
        :returns: A :class:`invenio_files_rest.models.Part` instance.
        """
        upload_id = multipart.upload_id \
            if isinstance(multipart, MultipartObject) else multipart
        return cls.query.filter_by(
            upload_id=upload_id
        )

    @classmethod
    def count(cls, mp):
        """Count number of parts for a given multipart object."""
        return cls.query_by_multipart(mp).count()

    @ensure_uncompleted(getter=lambda o: not o.multipart.completed)
    def set_contents(self, stream, progress_callback=None):
        """Save contents of stream to part of file instance.

        If a the MultipartObject is completed this methods raises an
        ``MultipartAlreadyCompleted`` exception.

        :param stream: File-like stream.
        :param size: Size of stream if known.
        :param chunk_size: Desired chunk size to read stream in. It is up to
            the storage interface if it respects this value.
        """
        size, checksum = self.multipart.file.update_contents(
            stream, seek=self.start_byte, size=self.part_size,
            progress_callback=progress_callback,
        )
        self.checksum = checksum
        return self def json_serializer(data=None, code=200, headers=None, context=None,
                    etag=None, task_result=None,
                    serializer_mapping=serializer_mapping, view_name=None):
    """Build a json flask response using the given data.

    :param data: The data to serialize. (Default: ``None``)
    :param code: The HTTP status code. (Default: ``200``)
    :param headers: The HTTP headers to include. (Default: ``None``)
    :param context: The schema class context. (Default: ``None``)
    :param etag: The ETag header. (Default: ``None``)
    :param task_result: Optionally you can pass async task to wait for.
        (Default: ``None``)
    :param serializer_mapping: Optionally provide the serializer with a
        different mapping.
    :param view_name: Optionally push to the marshmallow context the view
        name prefix,usefull in case of multiple routes pointing to the same
        blueprint.
    :returns: A Flask response with json data.
    :rtype: :py:class:`flask.Response`
    """
    context = context or {}
    if view_name:
        context.update({'view_name': view_name})
    schema_class, many = schema_from_context(context, serializer_mapping)

    if data is not None:
        # Generate JSON response
        data = json.dumps(
            schema_class(context=context).dump(data, many=many).data,
            **_format_args()
        )

        interval = current_app.config['FILES_REST_TASK_WAIT_INTERVAL']
        max_rounds = int(
            current_app.config['FILES_REST_TASK_WAIT_MAX_SECONDS'] // interval
        )

        response = current_app.response_class(
            # Stream response if waiting for task result.
            data if task_result is None else wait_for_taskresult(
                task_result, data, interval, max_rounds, ),
            mimetype='application/json'
        )
    else:
        response = current_app.response_class(mimetype='application/json')

    response.status_code = code
    if headers is not None:
        response.headers.extend(headers)

    if etag:
        response.set_etag(etag)

    return response @shared_task()
def merge_multipartobject(upload_id, version_id=None):
    """Merge multipart object.

    :param upload_id: The :class:`invenio_files_rest.models.MultipartObject`
        upload ID.
    :param version_id: Optionally you can define which file version.
        (Default: ``None``)
    :returns: The :class:`invenio_files_rest.models.ObjectVersion` version
        ID.
    """
    mp = MultipartObject.query.filter_by(upload_id=upload_id).one_or_none()
    if not mp:
        raise RuntimeError('Upload ID does not exists.')
    if not mp.completed:
        raise RuntimeError('MultipartObject is not completed.')

    try:
        obj = mp.merge_parts(
            version_id=version_id,
            progress_callback=progress_updater
        )
        db.session.commit()
        file_uploaded.send(current_app._get_current_object(), obj=obj)
        return str(obj.version_id)
    except Exception:
        db.session.rollback()
        raise @shared_task(ignore_result=True)
def remove_file_data(file_id, silent=True):
    """Remove file instance and associated data.

    :param file_id: The :class:`invenio_files_rest.models.FileInstance` ID.
    :param silent: It stops propagation of a possible raised IntegrityError
        exception. (Default: ``True``)
    :raises sqlalchemy.exc.IntegrityError: Raised if the database removal goes
        wrong and silent is set to ``False``.
    """
    try:
        # First remove FileInstance from database and commit transaction to
        # ensure integrity constraints are checked and enforced.
        f = FileInstance.get(file_id)
        if not f.writable:
            return
        f.delete()
        db.session.commit()
        # Next, remove the file on disk. This leaves the possibility of having
        # a file on disk dangling in case the database removal works, and the
        # disk file removal doesn't work.
        f.storage().delete()
    except IntegrityError:
        if not silent:
            raise
import uuid
from flask import Blueprint, abort, current_app, json, request
from flask_login import current_user
from functools import partial, wraps
from invenio_db import db
from invenio_rest import ContentNegotiatedMethodView
from marshmallow import missing
from urllib.parse import parse_qsl
from webargs import fields
from webargs.flaskparser import use_kwargs
from .errors import DuplicateTagError, ExhaustedStreamError, FileSizeError, \
    InvalidTagError, MissingQueryParameter, MultipartInvalidChunkSize
from .models import Bucket, MultipartObject, ObjectVersion, ObjectVersionTag, \
    Part
from .proxies import current_files_rest, current_permission_factory
from .serializer import json_serializer
from .signals import file_deleted, file_downloaded, file_uploaded
from .tasks import merge_multipartobject, remove_file_data

    delete_args = {
        'version_id': fields.UUID(
            location='query',
            load_from='versionId',
            data_key='versionId',
            missing=None,
        ),
        'upload_id': fields.UUID(
            location='query',
            load_from='uploadId',
            data_key='uploadId',
            missing=None,
        ),
        'uploads': fields.Raw(
            location='query',
            validate=invalid_subresource_validator,
        ),
    }

    get_args = dict(
        delete_args,
        download=fields.Raw(
            location='query',
            missing=None,
        )
    )

    post_args = {
        'uploads': fields.Raw(
            location='query',
        ),
        'upload_id': fields.UUID(
            location='query',
            load_from='uploadId',
            data_key='uploadId',
            missing=None,
        )
    }

    put_args = {
        'upload_id': fields.UUID(
            location='query',
            load_from='uploadId',
            data_key='uploadId',
            missing=None,
        ),
    }

    multipart_init_args = {
        'size': fields.Int(
            locations=('query', 'json'),
            missing=None,
        ),
        'part_size': fields.Int(
            locations=('query', 'json'),
            missing=None,
            load_from='partSize',
            data_key='partSize',
        ),
    }

    def __init__(self, *args, **kwargs):
        """Instantiate content negotiated view."""
        super(ObjectResource, self).__init__(*args, **kwargs)

    #
    # ObjectVersion helpers
    #
    @staticmethod
    def check_object_permission(obj):
        """Retrieve object and abort if it doesn't exists."""
        check_permission(current_permission_factory(
            obj,
            'object-read'
        ))
        if not obj.is_head:
            check_permission(
                current_permission_factory(obj, 'object-read-version'),
                hidden=False
            )

    @classmethod
    def get_object(cls, bucket, key, version_id):
        """Retrieve object and abort if it doesn't exist.

        If the file is not found, the connection is aborted and the 404
        error is returned.

        :param bucket: The bucket (instance or id) to get the object from.
        :param key: The file key.
        :param version_id: The version ID.
        :returns: A :class:`invenio_files_rest.models.ObjectVersion` instance.
        """
        obj = ObjectVersion.get(bucket, key, version_id=version_id)
        if not obj:
            abort(404, 'Object does not exists.')

        cls.check_object_permission(obj)

        return obj

    def create_object(self, bucket, key):
        """Create a new object.

        :param bucket: The bucket (instance or id) to get the object from.
        :param key: The file key.
        :returns: A Flask response.
        """
        # Initial validation of size based on Content-Length.
        # User can tamper with Content-Length, so this is just an initial up
        # front check. The storage subsystem must validate the size limit as
        # well.
        stream, content_length, content_md5, tags = \
            current_files_rest.upload_factory()

        size_limit = bucket.size_limit
        if content_length and size_limit and content_length > size_limit:
            desc = 'File size limit exceeded.' \
                if isinstance(size_limit, int) else size_limit.reason

            raise FileSizeError(description=desc)