Added mongoeingin pylint plugin. First version of API and users module with respective tests.

.vscode/settings.json

@@ -10,6 +10,7 @@
     "eslint.autoFixOnSave": true,
     "python.linting.pylintArgs": [
         "--disable=all",
+        "--load-plugins=pylint_mongoengine",
         "--enable=F,E,unreachable,duplicate-key,unnecessary-semicolon,global-variable-not-assigned,unused-variable,binary-op-exception,bad-format-string,anomalous-backslash-in-string,bad-open-mode,unused-import"
     ],
     "python.linting.pep8Enabled": true,

infrastructure/docker-compose.yml

@@ -63,6 +63,19 @@ services:
         ports:
             - 9200:9200
 
+    # the user data db
+    mongo:
+        image: mongo:latest
+        container_name: nomad-xt-mongo
+        environment:
+          - MONGO_DATA_DIR=/data/db
+          - MONGO_LOG_DIR=/dev/null
+        volumes:
+          - '../.volumes/mongo:/data/db'
+        ports:
+            - 27017:27017
+        command: mongod --smallfiles --logpath=/dev/null # --quiet
+
     # used for centralized logging
     elk:
         restart: always

nomad/api.py

+from flask import Flask
+from flask_restful import Resource, Api, abort
+from datetime import datetime
+from threading import Thread
+import logging
+import mongoengine.errors
+
+from nomad import users, files, processing
+
+logger = logging.getLogger(__name__)
+
+app = Flask(__name__)
+api = Api(app)
+
+
+# provida a fake user for testing
+me = users.User.objects(email='me@gmail.com').first()
+if me is None:
+    me = users.User(email='me@gmail.com', name='Me Meyer')
+    me.save()
+
+
+class Uploads(Resource):
+
+    @staticmethod
+    def _render(upload: users.Upload):
+        data = {
+            'id': upload.upload_id,
+            'presigned_url': upload.presigned_url,
+            'create_time': upload.create_time.isoformat() if upload.create_time is not None else None,
+            'upload_time': upload.upload_time.isoformat() if upload.upload_time is not None else None
+        }
+
+        return {key: value for key, value in data.items() if value is not None}
+
+    def get(self):
+        return [Uploads._render(user) for user in users.Upload.objects()], 200
+
+    def post(self):
+        upload = users.Upload(user=me)
+        upload.save()
+
+        upload.presigned_url = files.get_presigned_upload_url(upload.upload_id)
+        upload.create_time = datetime.now()
+        upload.save()
+
+        return Uploads._render(upload), 200
+
+
+class Upload(Resource):
+    def get(self, upload_id):
+        try:
+            upload = users.Upload.objects(id=upload_id).first()
+        except mongoengine.errors.ValidationError:
+            abort(400, message='%s is not a valid upload id.' % upload_id)
+
+        if upload is None:
+            abort(404, message='Upload with id %s does not exist.' % upload_id)
+
+        return Uploads._render(upload), 200
+
+
+api.add_resource(Uploads, '/uploads')
+api.add_resource(Upload, '/uploads/<string:upload_id>')
+
+if __name__ == '__main__':
+    @files.upload_put_handler
+    def handle_upload_put(received_upload_id: str):
+        upload = users.Upload.objects(id=received_upload_id).first()
+        if upload is None:
+            logger.error(
+                'Received upload put event on non existing upload %s.' %
+                received_upload_id)
+            return
+
+        upload.upload_time = datetime.now()
+        try:
+            proc = processing.UploadProcessing(received_upload_id)
+            proc.start()
+            upload.processing = proc.to_json()
+        except Exception as e:
+            logger.error(
+                'Unexpected exception while starting processing of upload %s.' %
+                received_upload_id, exc_info=e)
+
+        upload.save()
+
+    def handle_uploads():
+        handle_upload_put(received_upload_id='provided by decorator')
+
+    handle_uploads_thread = Thread(target=handle_uploads)
+    handle_uploads_thread.start()
+
+    app.run(debug=True)
+    handle_uploads_thread.join()

nomad/config.py

@@ -34,6 +34,12 @@ MinioConfig = namedtuple('Minio', ['host', 'port', 'accesskey', 'secret'])
 FSConfig = namedtuple('FSConfig', ['tmp'])
 """ Used to configure file stystem access. """
 
+ElasticConfig = namedtuple('ElasticConfig', ['host', 'calc_index'])
+""" Used to configure elastic search. """
+
+MongoConfig = namedtuple('MongoConfig', ['host', 'users_db'])
+""" Used to configure mongo db. """
+
 LogstashConfig = namedtuple('LogstashConfig', ['enabled', 'host', 'tcp_port'])
 """ Used to configure and enable/disable the ELK based centralized logging. """
 
@@ -59,6 +65,14 @@ minio = MinioConfig(
 fs = FSConfig(
     tmp='.volumes/fs'
 )
+elastic = ElasticConfig(
+    host='localhost',
+    calc_index='calcs'
+)
+mongo = MongoConfig(
+    host='localhost',
+    users_db='users'
+)
 logstash = LogstashConfig(
     enabled=False,
     host=os.environ.get('NOMAD_LOGSTASH_HOST', 'localhost'),

nomad/processing.py

@@ -46,8 +46,8 @@ from nomad.files import Upload, UploadError
 from nomad import files, utils
 from nomad.parsing import parsers, parser_dict
 from nomad.normalizing import normalizers
-from nomad.search import Calc
-import nomad.patch
+from nomad import search, users
+import nomad.patch  # pylint: disable=ununsed-import
 
 # The legacy nomad code uses a logger called 'nomad'. We do not want that this
 # logger becomes a child of this logger due to its module name starting with 'nomad.'
@@ -355,7 +355,7 @@ def parse(processing: UploadProcessing, parse_spec: ParseSpec) -> ProcessingTask
 
     # update search
     try:
-        Calc.add_from_backend(
+        search.Calc.add_from_backend(
             parser_backend,
             upload_hash=upload_hash,
             calc_hash=calc_hash,
@@ -371,7 +371,7 @@ def parse(processing: UploadProcessing, parse_spec: ParseSpec) -> ProcessingTask
     archive_id = '%s/%s' % (upload_hash, calc_hash)
     logger.debug('Written results of %s for %s to %s.' % (parser, mainfile, archive_id))
 
-    # persistence
+    # calc data persistence
     try:
         with files.write_archive_json(archive_id) as out:
             parser_backend.write_json(out, pretty=True)

nomad/search.py

+# Copyright 2018 Markus Scheidgen
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an"AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+The interface towards our search engine. It allows to store calculations as documents
+of search relevant properties.
+
+..autoclass:: nomad.search.Calc:
+"""
+
 from elasticsearch_dsl import Document, Date, Keyword, connections
 import logging
-import inspect
 
+from nomad import config
 from nomad.parsing import LocalBackend
 
 logger = logging.getLogger(__name__)
-connections.create_connection(hosts=['localhost'])
+
+# ensure elastic connection
+connections.create_connection(hosts=[config.elastic.host])
 
 
 class Calc(Document):
@@ -31,7 +54,7 @@ class Calc(Document):
     XC_functional_name = Keyword()
 
     class Index:
-        name = 'calcs'
+        name = config.elastic.calc_index
 
     @staticmethod
     def add_from_backend(backend: LocalBackend, **kwargs) -> 'Calc':

nomad/users.py

+# Copyright 2018 Markus Scheidgen
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an"AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+This module comprises a set of persistent document classes that hold all user related
+data. These are information about users, their uploads and datasets, and the
+associations between users and the assets stored in nomad-xt.
+
+..autoclass:: nomad.users.User
+..autoclass:: nomad.users.Upload
+..autoclass:: nomad.users.DataSet
+"""
+
+from mongoengine import \
+    Document, EmailField, StringField, BooleanField, DateTimeField, ListField, \
+    DictField, ReferenceField, connect
+
+from nomad import config
+
+# ensure mongo connection
+connect(db=config.mongo.users_db, host=config.mongo.host)
+
+
+class User(Document):
+    """ Represents users in the database. """
+    email = EmailField(primary=True)
+    name = StringField()
+
+
+class Upload(Document):
+    """
+    Represents uploads in the databases. Provides persistence access to the files storage,
+    and processing system.
+
+    Attributes:
+        upload_hash: The UUID hash of the uploaded file. Used to id the extracted upload
+                     in the repository files storage.
+        in_staging: True if the upload is still in staging and can be edited by the uploader.
+        is_private: True if the upload and its derivitaves are only visible to the uploader.
+        procssing: The serialized instance of :class:`nomad.processing.UploadProcessing`.
+        upload_time: The timestamp when the system realised the upload.
+    """
+
+    upload_hash = StringField()
+    in_staging = BooleanField(default=True)
+    is_private = BooleanField(default=False)
+    processing = DictField()
+    upload_time = DateTimeField()
+    create_time = DateTimeField()
+    presigned_url = StringField()
+
+    user = ReferenceField(User, required=True)
+
+    meta = {
+        'indexes': [
+            'upload_hash',
+            'user'
+        ]
+    }
+
+    @property
+    def upload_id(self):
+        return self.id.__str__()
+
+
+class DataSet(Document):
+    name = StringField()
+    description = StringField()
+    doi = StringField()
+
+    user = ReferenceField(User)
+    calcs = ListField(StringField)
+
+    meta = {
+        'indexes': [
+            'user',
+            'doi',
+            'calcs'
+        ]
+    }

pylint_mongoengine.py

+from astroid import scoped_nodes
+from astroid import MANAGER
+
+
+def register(linter):
+    # Needed for registering the plugin.
+    pass
+
+
+def transform(cls: scoped_nodes.ClassDef):
+    if any(getattr(base, 'name', None) == 'Document' for base in cls.bases):
+        cls.locals['objects'] = [scoped_nodes.FunctionDef('objects')]
+
+MANAGER.register_transform(scoped_nodes.ClassDef, transform)

requirements.txt

@@ -2,6 +2,8 @@ minio
 requests
 celery[redis]
 elasticsearch-dsl>=6.0.0,<7.0.0
+mongoengine
+flask-restful
 python-logstash
 gitpython
 ase

tests/test_api.py

+import pytest
+from threading import Thread
+import subprocess
+import shlex
+import time
+import json
+from mongoengine import connect
+from mongoengine.connection import disconnect
+from minio.error import ResponseError
+
+from nomad import config, api, files
+
+from tests.test_files import example_file
+
+@pytest.fixture
+def client():
+    disconnect()
+    connect('users_test', host=config.mongo.host, is_mock=True)
+
+    api.app.config['TESTING'] = True
+    client = api.app.test_client()
+
+    yield client
+
+
+def assert_uploads(upload_json_str, count=0, **kwargs):
+    data = json.loads(upload_json_str)
+    assert isinstance(data, list)
+    assert len(data) == count
+
+    if count > 0:
+        assert_upload(json.dumps(data[0]), **kwargs)
+
+
+def assert_upload(upload_json_str, id=None):
+    data = json.loads(upload_json_str)
+    assert 'id' in data
+    if id is not None:
+        assert id == data['id']
+    assert 'create_time' in data
+    assert 'presigned_url' in data
+
+    return data
+
+
+def test_no_uploads(client):
+    rv = client.get('/uploads')
+
+    assert rv.status_code == 200
+    assert_uploads(rv.data, count=0)
+
+
+def test_bad_upload_id(client):
+    rv = client.get('/uploads/bad_id')
+    assert rv.status_code == 400
+
+
+def test_not_existing_upload(client):
+    rv = client.get('/uploads/123456789012123456789012')
+    assert rv.status_code == 404
+
+
+def test_create_upload(client):
+    rv = client.post('/uploads')
+
+    assert rv.status_code == 200
+    upload_id = assert_upload(rv.data)['id']
+
+    rv = client.get('/uploads/%s' % upload_id)
+    assert rv.status_code == 200
+    assert_upload(rv.data, id=upload_id)
+
+    rv = client.get('/uploads')
+    assert rv.status_code == 200
+    assert_uploads(rv.data, count=1, id=upload_id)
+
+
+@pytest.mark.timeout(10)
+def test_upload_to_upload(client):
+    rv = client.post('/uploads')
+    assert rv.status_code == 200
+    upload = assert_upload(rv.data)
+
+    @files.upload_put_handler
+    def handle_upload_put(received_upload_id: str):
+        assert upload['id'] == received_upload_id
+        raise StopIteration
+
+    def handle_uploads():
+        handle_upload_put(received_upload_id='provided by decorator')
+
+    handle_uploads_thread = Thread(target=handle_uploads)
+    handle_uploads_thread.start()
+
+    time.sleep(1)
+    upload_url = upload['presigned_url']
+    cmd = files.create_curl_upload_cmd(upload_url).replace('<ZIPFILE>', example_file)
+    subprocess.call(shlex.split(cmd))
+
+    handle_uploads_thread.join()
+
+    try:
+        files._client.remove_object(config.files.uploads_bucket, upload['id'])
+    except ResponseError:
+        assert False