[FEATURE][SEARCH-1] File indexing

.env

+CELERY_LOG_LEVEL=error
+
 CERN_SEARCH_INSTANCE=test
 CERN_SEARCH_REMOTE_APP_RESOURCE=localhost
 CERN_SEARCH_USE_EGROUPS='False'
+CERN_SEARCH_FILES_PROCESSOR_QUEUE=files_processor
+CERN_SEARCH_FILES_PROCESSOR_QUEUE_DLX=files_processor_dlx
+CERN_SEARCH_FILES_PROCESSOR_EXCHANGE=default
+CERN_SEARCH_FILES_PROCESSOR_EXCHANGE_DLX=dlx
 
 CONTAINER_NAME=cern-search-rest-api
+WORKER_APP=invenio_app.celery
 
 DEFAULT_RECORDS_FILES_LOCATION=/usr/share/cern-search-api/files
 
 ENV=dev
 
 FLASK_SKIP_DOTENV=1
+FLASK_DEBUG='False'
+
+FLOWER_PASS=password
 
 INVENIO_ACCOUNTS_SESSION_REDIS_URL=redis://redis:6379/1
 INVENIO_ADMIN_ACCESS_GROUPS=CernSearch-Administrators@cern.ch
 INVENIO_ADMIN_USER=test@example.com
 INVENIO_ADMIN_VIEW_ACCESS_GROUPS=CernSearch-Administrators@cern.ch
 INVENIO_APP_ALLOWED_HOSTS=['localhost', 'nginx']
+INVENIO_BROKER_URL=amqp://guest:password@rabbitmq:5672
 INVENIO_CACHE_REDIS_HOST=redis
 INVENIO_CACHE_REDIS_URL=redis://redis:6379/0
 INVENIO_CERN_APP_CREDENTIALS={'consumer_key':'bah'}
 INVENIO_CERN_APP_CREDENTIALS_CONSUMER_KEY=xxx
+INVENIO_CELERY_BROKER_URL=amqp://guest:password@rabbitmq:5672
+INVENIO_CELERY_RESULT_BACKEND=redis://redis:6379/2
 INVENIO_COLLECT_STORAGE=flask_collect.storage.file
 INVENIO_INDEXER_DEFAULT_DOC_TYPE=doc_v0.0.2
 INVENIO_INDEXER_DEFAULT_INDEX=test-doc_v0.0.2
 INVENIO_LOGGING_CONSOLE='True'
-INVENIO_LOGGING_CONSOLE_LEVEL=DEBUG
+INVENIO_LOGGING_CONSOLE_LEVEL=WARNING
 INVENIO_RATELIMIT_STORAGE_URL='redis://redis:6379/3'
-INVENIO_RATELIMIT_AUTHENTICATED_USER=5000/hour
+INVENIO_RATELIMIT_AUTHENTICATED_USER=100000/hour
 INVENIO_SEARCH_ELASTIC_HOSTS=elasticsearch
 INVENIO_SEARCH_INDEX_PREFIX=cernsearch-
 INVENIO_SQLALCHEMY_DATABASE_URI=postgresql+psycopg2://uservice:itsjust1234@postgres/uservice
@@ -34,6 +47,13 @@ INVENIO_THEME_LOGO=/images/cernsearchicon.png
 INVENIO_THEME_LOGO_ADMIN=/images/cernsearchicon.png
 INVENIO_THEME_SITENAME='CERN Search DEV'
 
+RABBITMQ_DEFAULT_PASS=password
+
 POSTGRESQL_DATABASE=uservice
 POSTGRESQL_PASSWORD=itsjust1234
 POSTGRESQL_USER=uservice
+
+INVENIO_FILES_PROCESSOR_TIKA_SERVER_ENDPOINT=http://tika:9998
+
+SQLALCHEMY_POOL_SIZE=10
+SQLALCHEMY_MAX_OVERFLOW=15

.gitignore

@@ -40,3 +40,7 @@ builds/
 
 # Others
 .DS_Store
+
+
+# Pip source folder
+src

.gitlab-ci.yml

@@ -84,6 +84,15 @@ validate-base-image-updated:
 lint:
   services:
     - docker:dind
+  variables:
+    # As of GitLab 12.5, privileged runners at CERN mount a /certs/client docker volume that enables use of TLS to
+    # communicate with the docker daemon. This avoids a warning about the docker service possibly not starting
+    # successfully.
+    # See https://docs.gitlab.com/ee/ci/docker/using_docker_build.html#tls-enabled
+    DOCKER_TLS_CERTDIR: "/certs"
+    # Note that we do not need to set DOCKER_HOST when using the official docker client image: it automatically
+    # defaults to tcp://docker:2376 upon seeing the TLS certificate directory.
+    #DOCKER_HOST: tcp://docker:2376/
   stage: test
   only:
     refs:
@@ -92,6 +101,7 @@ lint:
   script: make build-env lint MODE=test
   before_script:
     - docker info
+    - docker login -u "$CI_REGISTRY_USER" -p "$CI_REGISTRY_PASSWORD" $CI_REGISTRY
     - docker-compose --version
     - apk add make
   allow_failure: true
@@ -102,14 +112,24 @@ lint:
 test:
   services:
     - docker:dind
+  variables:
+    # As of GitLab 12.5, privileged runners at CERN mount a /certs/client docker volume that enables use of TLS to
+    # communicate with the docker daemon. This avoids a warning about the docker service possibly not starting
+    # successfully.
+    # See https://docs.gitlab.com/ee/ci/docker/using_docker_build.html#tls-enabled
+    DOCKER_TLS_CERTDIR: "/certs"
+    # Note that we do not need to set DOCKER_HOST when using the official docker client image: it automatically
+    # defaults to tcp://docker:2376 upon seeing the TLS certificate directory.
+    #DOCKER_HOST: tcp://docker:2376/
   stage: test
   only:
     refs:
       - merge_requests
   image: tmaier/docker-compose:latest
-  script: make generate-certificates build-env populate-instance load-fixtures test MODE=test
+  script: make ci-test MODE=test
   before_script:
     - docker info
+    - docker login -u "$CI_REGISTRY_USER" -p "$CI_REGISTRY_PASSWORD" $CI_REGISTRY
     - docker-compose --version
     - apk add make
     - apk add openssl

.pipenv.env

+CELERY_LOG_LEVEL=INFO
+
 CERN_SEARCH_INSTANCE=test
 CERN_SEARCH_REMOTE_APP_RESOURCE=localhost
 CERN_SEARCH_USE_EGROUPS='False'
+CERN_SEARCH_FILES_PROCESSOR_QUEUE=files_processor
+CERN_SEARCH_FILES_PROCESSOR_QUEUE_DLX=files_processor_dlx
+CERN_SEARCH_FILES_PROCESSOR_EXCHANGE=default
+CERN_SEARCH_FILES_PROCESSOR_EXCHANGE_DLX=dlx
 
 CONTAINER_NAME=cern-search-rest-api
 
-DEFAULT_RECORDS_FILES_LOCATION=/usr/share/cern-search-api/files
 
 ENV=dev
 
+FLASK_DEBUG='True'
 FLASK_SKIP_DOTENV=1
 
+FLOWER_PASS=password
+
 INVENIO_ACCOUNTS_SESSION_REDIS_URL=redis://localhost:6379/1
 INVENIO_ADMIN_ACCESS_GROUPS=CernSearch-Administrators@cern.ch
 INVENIO_ADMIN_USER=test@example.com
 INVENIO_ADMIN_VIEW_ACCESS_GROUPS=CernSearch-Administrators@cern.ch
 INVENIO_APP_ALLOWED_HOSTS=['localhost','127.0.0.1']
+INVENIO_BROKER_URL=amqp://guest:password@localhost:5672
 INVENIO_CACHE_REDIS_HOST=localhost
 INVENIO_CACHE_REDIS_URL=redis://localhost:6379/0
 INVENIO_CERN_APP_CREDENTIALS={'consumer_key':'bah'}
 INVENIO_CERN_APP_CREDENTIALS_CONSUMER_KEY=xxx
+INVENIO_CELERY_BROKER_URL=amqp://guest:password@localhost:5672
+INVENIO_CELERY_RESULT_BACKEND=redis://localhost:6379/2
 INVENIO_COLLECT_STORAGE=flask_collect.storage.file
 INVENIO_INDEXER_DEFAULT_DOC_TYPE=doc_v0.0.2
 INVENIO_INDEXER_DEFAULT_INDEX=test-doc_v0.0.2
 INVENIO_LOGGING_CONSOLE='True'
 INVENIO_LOGGING_CONSOLE_LEVEL=DEBUG
 INVENIO_RATELIMIT_STORAGE_URL='redis://localhost:6379/3'
-INVENIO_RATELIMIT_AUTHENTICATED_USER=5000/hour
+INVENIO_RATELIMIT_AUTHENTICATED_USER=100000/hour
 INVENIO_SEARCH_ELASTIC_HOSTS=localhost
 INVENIO_SEARCH_INDEX_PREFIX=cernsearch-
 INVENIO_SQLALCHEMY_DATABASE_URI=postgresql+psycopg2://uservice:itsjust1234@localhost/uservice
@@ -34,6 +45,13 @@ INVENIO_THEME_LOGO=/images/cernsearchicon.png
 INVENIO_THEME_LOGO_ADMIN:=/images/cernsearchicon.png
 INVENIO_THEME_SITENAME='CERN Search DEV'
 
+RABBITMQ_DEFAULT_PASS=password
+
 POSTGRESQL_DATABASE=uservice
 POSTGRESQL_PASSWORD=itsjust1234
 POSTGRESQL_USER=uservice
+
+INVENIO_FILES_PROCESSOR_TIKA_SERVER_ENDPOINT=http://localhost:9998
+
+SQLALCHEMY_POOL_SIZE=10
+SQLALCHEMY_MAX_OVERFLOW=15

.pre-commit-config.yaml

@@ -6,7 +6,7 @@ repos:
     -   id: end-of-file-fixer
 
 -   repo: https://gitlab.com/pycqa/flake8
-    rev: 3.7.8
+    rev: 3.7.9
     hooks:
     -   id: flake8
         additional_dependencies: [flake8-docstrings]

CONTRIBUTING.md

 # How to contribute
 
-Guidelines on how to contribute are explained in the official documentation, they can be found [here](http://cernsearchdocs.web.cern.ch/cernsearchdocs/contributing/).
\ No newline at end of file
+Guidelines on how to contribute are explained in the official documentation, they can be found [here](http://cern-search.docs.cern.ch/cernsearchdocs/contributing/).
\ No newline at end of file

Dockerfile

@@ -3,11 +3,11 @@
 # This file is part of CERN Search.
 # Copyright (C) 2018-2019 CERN.
 #
-# CERN Search is free software; you can redistribute it and/or modify it
+# Citadel Search is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
 
 # Use CentOS7:
-FROM gitlab-registry.cern.ch/webservices/cern-search/cern-search-rest-api/cern-search-rest-api-base:f4fe630858f6626c39f0fbdcb9d31e99227987a2
+FROM gitlab-registry.cern.ch/webservices/cern-search/cern-search-rest-api/cern-search-rest-api-base:d7589fe287aa752ac5ee0b62b27bb6e8befa4fcb
 ARG build_env
 
 # CERN Search installation

Makefile

@@ -16,6 +16,7 @@
 # make lint                   # runs linting tools
 
 SERVICE_NAME :=  cern-search-api
+WORKER_NAME :=  worker
 API_TOKEN := .api_token
 MODE?=full
 TEST_MODE=test
@@ -49,6 +50,10 @@ destroy-env:
 	docker-compose -f $(DOCKER_FILE) rm -f
 .PHONY: destroy-env
 
+stop-env:
+	docker-compose -f $(DOCKER_FILE) down --volumes
+.PHONY: destroy-env
+
 reload-env: destroy-env env
 .PHONY: reload-env
 
@@ -56,6 +61,10 @@ shell-env:
 	docker-compose -f $(DOCKER_FILE) exec $(SERVICE_NAME) /bin/bash
 .PHONY: shell-env
 
+shell-worker:
+	docker-compose -f $(DOCKER_FILE) exec $(WORKER_NAME) /bin/bash
+.PHONY: shell-env
+
 env: generate-certificates build-env populate-instance load-fixtures shell-env
 .PHONY: env
 
@@ -63,11 +72,17 @@ generate-certificates:
 	sh scripts/gen-cert.sh
 .PHONY: generate-certificates
 
-test:
+pytest:
 	docker-compose -f $(DOCKER_FILE) exec -T $(SERVICE_NAME) /bin/bash -c \
-	"API_TOKEN=$$(cat $(API_TOKEN)) pytest tests -vv;"
+	"pytest tests -vv;"
 .PHONY: test
 
+ci-test: build-env pytest
+.PHONY: ci-test
+
+test: stop-env build-env pytest
+.PHONY: local-test
+
 lint:
 	docker-compose -f $(DOCKER_FILE) exec -T $(SERVICE_NAME) /bin/bash -c \
 		"echo running isort...; \
@@ -112,7 +127,7 @@ build-local-env: check-requirements-local
 .PHONY: build-local-env
 
 populate-instance-local:
-	PIPENV_DOTENV_LOCATION=$(PIPENV_DOTENV) pipenv run sh scripts/populate-instance.sh
+	PIPENV_DOTENV_LOCATION=$(PIPENV_DOTENV) pipenv run sh scripts/pipenv/populate-instance.sh
 .PHONY: populate-instance-local
 
 load-fixtures-local:
@@ -140,8 +155,8 @@ reload-local-env: destroy-local-env local-env
 .PHONY: reload-local-env
 
 local-test:
-	@echo todo
-#	python pytest
+	@echo running tests...;
+	pipenv run pytest tests -v;
 .PHONY: test
 
 local-lint:

Pipfile

@@ -4,11 +4,12 @@ url = "https://pypi.org/simple"
 verify_ssl = true
 
 [dev-packages]
-flake8 = ">=3.7.8"
+flake8 = ">=3.7.9"
 flake8-docstrings = ">=1.5.0"
 isort = "==4.3.21"
 pytest-invenio = ">=1.2.0"
 pytest-mock = ">=1.6.0"
+pytest-dotenv = "*"
 
 [packages]
 invenio-access = ">=1.0.0,<1.1.0"
@@ -21,7 +22,7 @@ invenio-db = {version = ">=1.0.0,<1.1.0",extras = ["postgresql", "versioning"]}
 invenio-indexer = {extras = ["elasticsearch6"],version = ">=1.1.0,<1.2.0"}
 invenio-jsonschemas = ">=1.0.0,<1.1.0"
 invenio-logging = ">=1.0.0,<1.1.0"
-invenio-records-rest = {extras = ["elasticsearch6"],version = "<1.6.0,>=1.4.0"}
+invenio-records-rest = {extras = ["elasticsearch6"],version = "<1.7.0,>=1.6.0"}
 invenio-records = {version = ">=1.0.0,<1.1.0",extras = ["postgresql"]}
 invenio-rest = {extras = ["cors"],version = "<1.2.0,>=1.1.0"}
 invenio-oauthclient = ">=1.0.0,<1.1.0"
@@ -37,7 +38,14 @@ uwsgi-tools = ">=1.1.1,<1.2.0"
 Flask = "*"
 uWSGI = ">=2.0.16"
 marshmallow = "<3"
-invenio-records-files = ">=1.1.0,<1.2.0"
+invenio-records-files = "<1.3.0,>=1.2.0"
+invenio-celery = "<1.2.0,>=1.1.0"
+invenio-files-rest = ">=1.0.5,<1.1.0"
+gevent = ">=1.4.0"
+invenio-files-processor = {extras = ["tika"],git = "https://github.com/carantunes/invenio-files-processor.git",ref = "global/implement-processor-tika"}
+Werkzeug = ">=0.15,<1.0.0"
+pip = "*"
+eventlet = "*"
 
 [requires]
 python_version = "3.6"

README.md

@@ -3,5 +3,5 @@
 CERN Search provides enterprise search capabilities on demand. You can set up your own search instance, submit your 
 documents and search among them when needed!
 
-- User documentation [here](http://cernsearchdocs.web.cern.ch/cernsearchdocs/).
-- Administration documentation [here](https://cernsearch-admin-docs.web.cern.ch/cernsearch-admin-docs/).
\ No newline at end of file
+- User documentation [here](http://cern-search.docs.cern.ch/cernsearchdocs/).
+- Administration documentation [here](https://cern-search-admin.docs.cern.ch/cernsearch-admin-docs/).
\ No newline at end of file

cern_search_rest_api/config.py

@@ -14,14 +14,16 @@ import ast
 import copy
 import os
 
+from cern_search_rest_api.modules.cernsearch.permissions import (record_create_permission_factory,
+                                                                 record_delete_permission_factory,
+                                                                 record_list_permission_factory,
+                                                                 record_read_permission_factory,
+                                                                 record_update_permission_factory)
 from flask import request
 from invenio_oauthclient.contrib import cern
 from invenio_records_rest import config as irr_config
 from invenio_records_rest.facets import terms_filter
-
-from .modules.cernsearch.permissions import (record_create_permission_factory, record_delete_permission_factory,
-                                             record_list_permission_factory, record_read_permission_factory,
-                                             record_update_permission_factory)
+from kombu import Exchange, Queue
 
 
 def _(x):
@@ -202,3 +204,50 @@ SECURITY_CONFIRMABLE = False
 SECURITY_REGISTERABLE = False  # Avoid user registration outside of CERN SSO
 SECURITY_RECOVERABLE = False  # Avoid user password recovery
 SESSION_COOKIE_SECURE = True
+
+# Celery Configuration
+# ====================
+FILES_PROCESSOR_QUEUE = os.getenv("CERN_SEARCH_FILES_PROCESSOR_QUEUE", 'files_processor')
+FILES_PROCESSOR_QUEUE_DLX = os.getenv("CERN_SEARCH_FILES_PROCESSOR_QUEUE_DLX", 'files_processor_dlx')
+FILES_PROCESSOR_EXCHANGE = os.getenv("CERN_SEARCH_FILES_PROCESSOR_EXCHANGE", 'default')
+FILES_PROCESSOR_EXCHANGE_DLX = os.getenv("CERN_SEARCH_FILES_PROCESSOR_EXCHANGE_DLX", 'dlx')
+
+#: URL of message broker for Celery (default is RabbitMQ).
+CELERY_BROKER_URL = os.getenv('INVENIO_CELERY_BROKER_URL', 'amqp://guest:guest@localhost:5672')
+#: URL of backend for result storage (default is Redis).
+CELERY_RESULT_BACKEND = os.getenv('INVENIO_CELERY_RESULT_BACKEND', 'redis://localhost:6379/2')
+
+CELERY_TASK_QUEUES = {
+    Queue(
+        name=FILES_PROCESSOR_QUEUE,
+        exchange=Exchange(FILES_PROCESSOR_EXCHANGE, type='direct'),
+        routing_key=FILES_PROCESSOR_QUEUE,
+        queue_arguments={
+            'x-dead-letter-exchange': FILES_PROCESSOR_EXCHANGE_DLX,
+            'x-dead-letter-routing-key': FILES_PROCESSOR_QUEUE_DLX
+        }
+    ),
+    Queue('celery', Exchange('celery'), routing_key='celery')
+}
+
+CELERY_TASK_ROUTES = {
+    'cern_search_rest_api.modules.cernsearch.tasks.process_file_async': {
+        'queue': FILES_PROCESSOR_QUEUE,
+        'routing_key': FILES_PROCESSOR_QUEUE,
+    }
+}
+
+CELERY_TASK_DEFAULT_QUEUE = 'celery'
+
+CELERY_BROKER_POOL_LIMIT = os.getenv("BROKER_POOL_LIMIT", None)
+
+SQLALCHEMY_ENGINE_OPTIONS = {
+    'pool_size': int(os.getenv("SQLALCHEMY_POOL_SIZE", 5)),
+    'max_overflow': int(os.getenv("SQLALCHEMY_MAX_OVERFLOW", 10)),
+    'pool_recycle': int(os.getenv("SQLALCHEMY_POOL_RECYCLE", 300)),  # in seconds
+}
+
+SEARCH_CLIENT_CONFIG = dict(
+    # allow up to 25 connections to each node
+    maxsize=int(os.getenv("ELASTICSEARCH_MAX_SIZE", 5)),
+)

cern_search_rest_api/modules/cernsearch/api.py

@@ -8,30 +8,122 @@
 # under the terms of the MIT License; see LICENSE file for more details.
 
 """Record API."""
+from cern_search_rest_api.modules.cernsearch.fetchers import recid_fetcher
 from cern_search_rest_api.modules.cernsearch.utils import default_record_to_mapping
+from invenio_files_rest.models import Bucket
 from invenio_pidstore.models import PersistentIdentifier
-from invenio_records_files.api import Record
+from invenio_records.errors import MissingModelError
+from invenio_records_files.api import FilesMixin, Record
+from invenio_records_files.models import RecordsBuckets
 
-from .fetchers import recid_fetcher
+BUCKET_KEY = '_bucket'
+BUCKET_CONTENT_KEY = '_bucket_content'
 
 
-class CernSearchRecord(Record):
-    """CERN Search Record."""
+class CernSearchFilesMixin(FilesMixin):
+    """Metafiles for CernSearchRecord models."""
 
-    record_fetcher = staticmethod(recid_fetcher)
+    @property
+    def files_content(self):
+        """Get files' extracted content iterator.
 
-    @classmethod
-    def create(cls, data, id_=None, with_bucket=True, **kwargs):
-        """Create a record and the associated bucket.
+        :returns: Files iterator.
+        """
+        if self.model is None:
+            raise MissingModelError()
+
+        bucket_content = None
+        records_buckets = RecordsBuckets.query.filter_by(record_id=self.id)
+        for record_bucket in records_buckets:
+            if self.get(BUCKET_CONTENT_KEY) == str(record_bucket.bucket.id):
+                bucket_content = record_bucket.bucket
+
+        return self.files_iter_cls(self, bucket=bucket_content, file_cls=self.file_cls)
+
+    @files_content.setter
+    def files_content(self, data):
+        """Set files' extracted content from data."""
+        current_files = self.files_content
+        if current_files:
+            raise RuntimeError('Can not update existing files.')
+        for key in data:
+            current_files[key] = data[key]
 
-        :param with_bucket: Create a bucket automatically on record creation if mapping allows
+    @property
+    def files(self):
+        """Get files iterator.
+
+        :returns: Files iterator.
         """
-        bucket_allowed = False
+        if self.model is None:
+            raise MissingModelError()
+
+        records_buckets = RecordsBuckets.query.filter_by(record_id=self.id)
+        bucket = None
+        for record_bucket in records_buckets:
+            if self.get(BUCKET_KEY) == str(record_bucket.bucket.id):
+                bucket = record_bucket.bucket
+
+        return self.files_iter_cls(self, bucket=bucket, file_cls=self.file_cls)
+
+
+class CernSearchRecord(Record, CernSearchFilesMixin):
+    """CERN Search Record.
+
+    The record class implements a one-to-one relationship between a bucket and
+    a record (to store a record's file).
+    It also implements a one-to-one relationship between a bucket_content and
+    a record. The purpose of this bucket is to store the result of file's content extraction.
+
+    Both buckets are automatically created and associated with the record when the record is created
+    with :py:data:`CernSearchRecord.create()`.
+
+    The buckets id are stored in the record metadata, in the keys``_bucket`` and ``_bucket_content_id``.
+    """
+
+    record_fetcher = staticmethod(recid_fetcher)
+
+    def __init__(self, *args, **kwargs):
+        """Initialize the record."""
+        self._bucket_content = None
+        super(CernSearchRecord, self).__init__(*args, **kwargs)
+
+    @staticmethod
+    def __buckets_allowed(data):
+        buckets_allowed = False
+
         mapping = default_record_to_mapping(data)
         if mapping is not None:
-            bucket_allowed = '_bucket' in mapping['properties']
+            buckets_allowed = BUCKET_KEY in mapping['properties']
+
+        return buckets_allowed
+
+    @classmethod
+    def create(cls, data, id_=None, with_bucket=False, **kwargs):
+        """Create a record and the associated buckets.
+
+         Creates buckets:
+          - ``bucket`` for files
+          - ``bucket_content`` for files' extracted content.
 
-        return super(CernSearchRecord, cls).create(data, id_=id_, with_bucket=bucket_allowed, **kwargs)
+        :param with_bucket: Create both buckets automatically on record creation if mapping allows.
+        """
+        bucket_content = None
+
+        bucket_allowed = with_bucket or cls.__buckets_allowed(data)
+        if bucket_allowed:
+            bucket_content = cls.create_bucket(data)
+            if bucket_content:
+                cls.dump_bucket_content(data, bucket_content)
+
+        record = super(CernSearchRecord, cls).create(data, id_=id_, with_bucket=bucket_allowed, **kwargs)
+
+        # Create link between record and file content bucket
+        if bucket_allowed and bucket_content:
+            RecordsBuckets.create(record=record.model, bucket=bucket_content)
+            record._bucket_content = bucket_content
+
+        return record
 
     @property
     def pid(self):
@@ -39,3 +131,70 @@ class CernSearchRecord(Record):
         pid = self.record_fetcher(self.id, self)
 
         return PersistentIdentifier.get(pid.pid_type, pid.pid_value)
+
+    @classmethod
+    def dump_bucket_content(cls, data, bucket):
+        """Dump the file content bucket id into the record metadata..
+
+        :param data: A dictionary of the record metadata.
+        :param bucket: The created bucket for the record.
+        """
+        data[BUCKET_CONTENT_KEY] = str(bucket.id)
+
+    @classmethod
+    def load_bucket_content(cls, record):
+        """Load the file content bucket id from the record metadata.
+
+        :param record: A record instance.
+        """
+        return record.get(BUCKET_CONTENT_KEY, "")
+
+    @property
+    def bucket_content_id(self):
+        """Get file content bucket id from record metadata."""
+        return self.load_bucket_content(self)
+
+    @property
+    def file_content_bucket(self):
+        """Get file content bucket instance."""
+        if self._bucket_content is None:
+            if self.bucket_content_id:
+                self._bucket_content = Bucket.get(self.bucket_content_id)
+        return self._bucket_content
+
+    def clear(self) -> None:
+        """Clear that keeps protected fields."""
+        # extract protected fields
+        protected_fields = {
+            BUCKET_CONTENT_KEY: self.bucket_content_id,
+            BUCKET_KEY: self.bucket_id,
+        }
+
+        super(CernSearchRecord, self).clear()
+
+        # keep protected fields
+        self.update({k: v for k, v in protected_fields.items() if v})
+
+    def update(self, other=None, **kwargs):
+        """Update that keeps protected fields."""
+        # extract protected fields
+        protected_fields = {
+            BUCKET_CONTENT_KEY: self.bucket_content_id,
+            BUCKET_KEY: self.bucket_id,
+        }
+
+        super(CernSearchRecord, self).update(other, **kwargs)
+
+        # keep protected fields
+        super(CernSearchRecord, self).update({k: v for k, v in protected_fields.items() if v})
+
+    def delete(self, force=False):
+        """Delete a record and also remove the RecordsBuckets if necessary.
+
+        :param force: True to remove also the
+            :class:`~invenio_records_files.models.RecordsBuckets` object.
+        :returns: Deleted record.
+        """
+        if force:
+            RecordsBuckets.query.filter_by(record=self.model, bucket=self.files_content.bucket).delete()
+        return super(CernSearchRecord, self).delete(force)

cern_search_rest_api/modules/cernsearch/celery.py

+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+#
+# This file is part of CERN Search.
+# Copyright (C) 2018-2019 CERN.
+#
+# Citadel Search is free software; you can redistribute it and/or modify it
+# under the terms of the MIT License; see LICENSE file for more details.
+"""Celery utilities."""
+
+from celery import bootsteps
+from cern_search_rest_api.config import FILES_PROCESSOR_EXCHANGE_DLX, FILES_PROCESSOR_QUEUE_DLX
+from kombu import Exchange, Queue
+
+
+class DeclareDeadletter(bootsteps.StartStopStep):
+    """Celery Bootstep to declare the DLX and DLQ before the worker starts processing tasks."""
+
+    requires = {'celery.worker.components:Pool'}
+
+    def start(self, worker):
+        """Declare deadletter queue and exchange on worker pool start."""