api/tasks for bulk import using OclFleximporter (ocldev-py3) get/post

core/common/tasks.py

@@ -98,3 +98,15 @@ def populate_indexes(app_names=None):  # app_names has to be an iterable of stri
         call_command('search_index', '--populate', '-f', '--models', *app_names, '--parallel')
     else:
         call_command('search_index', '--populate', '-f', '--parallel')
+
+
+@app.task(base=QueueOnce)
+def bulk_import(to_import, username, update_if_exists):
+    from core.importers.models import BulkImport
+    return BulkImport(content=to_import, username=username, update_if_exists=update_if_exists).run()
+
+
+@app.task(base=QueueOnce)
+def bulk_priority_import(to_import, username, update_if_exists):
+    from core.importers.models import BulkImport
+    return BulkImport(content=to_import, username=username, update_if_exists=update_if_exists).run()

core/common/utils.py

@@ -5,6 +5,7 @@
 from urllib import parse
 
 from dateutil import parser
+from django.conf import settings
 from django.urls import NoReverseMatch, reverse, get_resolver, resolve, Resolver404
 from djqscsv import csv_file_for
 from pydash import flatten
@@ -262,3 +263,10 @@ def write_export_file(
     uploaded_path = S3.url_for(s3_key)
     logger.info('Uploaded to %s.' % uploaded_path)
     os.chdir(cwd)
+
+
+def get_base_url():
+    if settings.ENV == 'development':
+        return "http://localhost:8000"
+
+    return "https://api.{}2.openconceptlab.org".format(settings.ENV.lower())

core/importers/models.py

@@ -0,0 +1,47 @@
+import json
+
+from django.utils.text import compress_string
+from ocldev.oclfleximporter import OclFlexImporter
+
+from core.common.utils import get_base_url
+from core.users.models import UserProfile
+
+
+class ImportResults:
+    def __init__(self, importer):
+        self.json = compress_string(importer.import_results.to_json())
+        self.detailed_summary = importer.import_results.get_detailed_summary()
+        self.report = importer.import_results.display_report()
+
+
+class BulkImport:
+    def __init__(self, content, username, update_if_exists):
+        self.input_list = []
+        self.user = None
+        self.result = None
+        self.importer = None
+        self.content = content
+        self.username = username
+        self.update_if_exists = update_if_exists
+        self.populate_input_list()
+        self.set_user()
+        self.prepare_importer()
+
+    def populate_input_list(self):
+        for line in self.content.splitlines():
+            self.input_list.append(json.loads(line))
+
+    def set_user(self):
+        self.user = UserProfile.objects.get(username=self.username)
+
+    def prepare_importer(self):
+        self.importer = OclFlexImporter(
+            input_list=self.input_list,
+            api_url_root=get_base_url(),
+            api_token=self.user.auth_token.key,
+            do_update_if_exists=self.update_if_exists
+        )
+
+    def run(self):
+        self.importer.process()
+        self.result = ImportResults(self.importer)

core/importers/urls.py

@@ -0,0 +1,7 @@
+from django.urls import re_path
+
+from core.importers import views
+
+urlpatterns = [
+    re_path(r'^bulk-import/$', views.BulkImportView.as_view(), name='bulk-import'),
+]

core/importers/views.py

@@ -0,0 +1,59 @@
+import uuid
+
+from celery.result import AsyncResult
+from rest_framework import status
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from core.common.tasks import bulk_import, bulk_priority_import
+
+
+class BulkImportView(APIView):
+    permission_classes = (IsAuthenticated,)
+
+    def post(self, request):
+        username = self.request.user.username
+        update_if_exists = request.GET.get('update_if_exists', 'true')
+        if update_if_exists not in ['true', 'false']:
+            return Response(
+                {'exception': 'update_if_exists must be either \'true\' or \'false\''},
+                status=status.HTTP_400_BAD_REQUEST
+            )
+        update_if_exists = update_if_exists == 'true'
+
+        task_id = str(uuid.uuid4()) + '-' + username
+        if username == 'root':
+            task = bulk_priority_import.apply_async((request.body, username, update_if_exists), task_id=task_id)
+        else:
+            task = bulk_import.apply_async((request.body, username, update_if_exists), task_id=task_id)
+
+        return Response(dict(task=task.id, state=task.state))
+
+    def get(self, request):  # pylint: disable=too-many-return-statements
+        task_id = request.GET.get('task')
+        result_format = request.GET.get('result')
+        if not task_id:
+            return Response(dict(exception='Required task id'), status=status.HTTP_400_BAD_REQUEST)
+        username = task_id[37:]
+        user = self.request.user
+
+        if not user.is_staff and user.username != username:
+            return Response(status=status.HTTP_403_FORBIDDEN)
+
+        task = AsyncResult(task_id)
+
+        if task.successful():
+            result = task.get()
+            if result_format == 'json':
+                response = Response(result.json, content_type="application/json")
+                response['Content-Encoding'] = 'gzip'
+                return response
+            if result_format == 'report':
+                return Response(result.report)
+            return Response(result.detailed_summary)
+
+        if task.failed():
+            return Response({'exception': str(task.result)}, status=status.HTTP_400_BAD_REQUEST)
+
+        return Response(dict(task=task.id, state=task.state))

core/settings.py

@@ -49,6 +49,7 @@
     'core.collections',
     'core.concepts',
     'core.mappings',
+    'core.importers',
 ]
 
 REST_FRAMEWORK = {

core/urls.py

@@ -13,7 +13,6 @@
     1. Import the include() function: from django.urls import include, path
     2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))
 """
-from django.conf import settings
 from django.conf.urls import url
 from django.contrib import admin
 from django.urls import path, include
@@ -23,12 +22,7 @@
 
 import core.concepts.views as concept_views
 import core.mappings.views as mapping_views
-
-if settings.ENV == 'development':
-    BASE_URL = "http://localhost:8000"
-else:
-    BASE_URL = "https://api.{}2.openconceptlab.org".format(settings.ENV.lower())
-
+from core.common.utils import get_base_url
 
 SchemaView = get_schema_view(
     openapi.Info(
@@ -38,7 +32,7 @@
     ),
     public=True,
     permission_classes=(permissions.AllowAny,),
-    url=BASE_URL
+    url=get_base_url()
 )
 
 urlpatterns = [
@@ -52,4 +46,5 @@
     path('collections/', include('core.collections.urls')),
     path('concepts/', concept_views.ConceptVersionListAllView.as_view(), name='all-concepts'),
     path('mappings/', mapping_views.MappingVersionListAllView.as_view(), name='all-mappings'),
+    path('importers/', include('core.importers.urls')),
 ]