Merge pull request #284 from datosgobar/277-import-analytics

Importar analytics desde API mgmt

conf/settings/base.py

@@ -163,7 +163,8 @@ def export_vars(_):
     'sendfile',
     'des',
     'scheduler',
-    'django_datajsonar'
+    'django_datajsonar',
+    'solo',
 )
 
 APPS = (

requirements/base.txt

@@ -18,4 +18,4 @@ python-crontab==2.2.8
 django-des==2.2.0
 django-rq-scheduler==1.1.1
 -e git+git://github.com/datosgobar/django-datajsonar#egg=django_datajsonar
-
+django-solo==1.1.2

series_tiempo_ar_api/apps/analytics/admin.py

@@ -2,8 +2,10 @@
 from __future__ import unicode_literals
 
 from django.contrib import admin
+from solo.admin import SingletonModelAdmin
 
-from series_tiempo_ar_api.apps.analytics.models import Query
+from .models import Query, ImportConfig, AnalyticsImportTask
+from .tasks import import_last_day_analytics_from_api_mgmt
 
 
 class QueryAdmin(admin.ModelAdmin):
@@ -13,4 +15,18 @@ class QueryAdmin(admin.ModelAdmin):
     search_fields = ('timestamp', 'params', 'ip_address', 'args', 'ids')
 
 
+class ImportConfigAdmin(SingletonModelAdmin):
+    # django-des overridea el change_form_template de la clase padre(!), volvemos al default de django
+    change_form_template = 'admin/change_form.html'
+
+
+class ImportTaskAdmin(admin.ModelAdmin):
+    readonly_fields = ('status', 'logs', 'timestamp')
+
+    def save_model(self, request, obj, form, change):
+        import_last_day_analytics_from_api_mgmt.delay()
+
+
 admin.site.register(Query, QueryAdmin)
+admin.site.register(ImportConfig, ImportConfigAdmin)
+admin.site.register(AnalyticsImportTask, ImportTaskAdmin)

series_tiempo_ar_api/apps/analytics/management/commands/import_analytics.py

@@ -0,0 +1,9 @@
+#! coding: utf-8
+from django.core.management import BaseCommand
+
+from series_tiempo_ar_api.apps.analytics.tasks import import_last_day_analytics_from_api_mgmt
+
+
+class Command(BaseCommand):
+    def handle(self, *args, **options):
+        import_last_day_analytics_from_api_mgmt()

series_tiempo_ar_api/apps/analytics/migrations/0005_importconfig.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+# Generated by Django 1.11.6 on 2018-06-07 19:14
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('analytics', '0004_auto_20180117_1045'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='ImportConfig',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('endpoint', models.URLField()),
+                ('token', models.CharField(max_length=64)),
+                ('kong_api_id', models.CharField(max_length=64)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+    ]

series_tiempo_ar_api/apps/analytics/migrations/0006_analyticsimporttask.py

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+# Generated by Django 1.11.6 on 2018-06-08 18:09
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('analytics', '0005_importconfig'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='AnalyticsImportTask',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('status', models.CharField(choices=[('running', 'Corriendo'), ('finished', 'Finalizada')], max_length=64)),
+                ('logs', models.TextField(blank=True)),
+                ('timestamp', models.DateTimeField()),
+            ],
+        ),
+    ]

series_tiempo_ar_api/apps/analytics/models.py

@@ -1,7 +1,10 @@
 # -*- coding: utf-8 -*-
 from __future__ import unicode_literals
 
+import requests
 from django.db import models
+from django.core.exceptions import ValidationError
+from solo.models import SingletonModel
 
 
 class Query(models.Model):
@@ -14,3 +17,59 @@ class Query(models.Model):
 
     def __unicode__(self):
         return u'Query at %s: %s' % (self.timestamp, self.ids)
+
+
+class ImportConfig(SingletonModel):
+    endpoint = models.URLField()
+    token = models.CharField(max_length=64)
+    kong_api_id = models.CharField(max_length=64)
+
+    def clean(self):
+        status_code = requests.head(
+            self.endpoint,
+            headers={'Authorization': 'Token {}'.format(self.token)}
+        ).status_code
+
+        if status_code != 200:
+            raise ValidationError('URL / Token inválido')
+
+    def get_results(self, from_date=None, to_date=None, limit=1000, offset=0):
+        """Wrapper sobre requests para pegarle al endpoint configurado"""
+        return requests.get(
+            self.endpoint,
+            headers=self.get_authorization_header(),
+            params={'from_date': from_date,
+                    'to_date': to_date,
+                    'limit': limit,
+                    'offset': offset,
+                    'kong_api_id': self.kong_api_id}
+        ).json()
+
+    def get_authorization_header(self):
+        """Devuelve el header de auth formateado para usar en la libreria de requests"""
+        return {'Authorization': 'Token {}'.format(self.token)}
+
+
+class AnalyticsImportTask(models.Model):
+
+    RUNNING = 'running'
+    FINISHED = 'finished'
+
+    STATUS_CHOICES = (
+        (RUNNING, "Corriendo"),
+        (FINISHED, "Finalizada"),
+    )
+
+    status = models.CharField(max_length=64, choices=STATUS_CHOICES)
+    logs = models.TextField(blank=True)
+    timestamp = models.DateTimeField()
+
+    def __str__(self):
+        return "Analytics import task at {}".format(self.timestamp)
+
+    def write_logs(self, text):
+        if not self.logs:
+            self.logs = ''
+
+        self.logs += text + '\n'
+        self.save()

series_tiempo_ar_api/apps/analytics/tasks.py

@@ -1,12 +1,18 @@
 #! coding: utf-8
 import os
 import json
+from urllib.parse import parse_qs
 
+import iso8601
+import requests
 import unicodecsv
+from dateutil.relativedelta import relativedelta
 from django_rq import job
 from django.conf import settings
-from django.utils.timezone import localtime
-from .models import Query
+from django.utils import timezone
+from django.core.exceptions import FieldError
+
+from .models import Query, ImportConfig, AnalyticsImportTask
 from .utils import kong_milliseconds_to_tzdatetime
 
 
@@ -24,7 +30,7 @@ def export(path=None):
     filepath = path or os.path.join(settings.PROTECTED_MEDIA_DIR, settings.ANALYTICS_CSV_FILENAME)
 
     fields = {
-        'timestamp': lambda x: localtime(x.timestamp),
+        'timestamp': lambda x: timezone.localtime(x.timestamp),
         'ip_address': lambda x: x.ip_address,
         'ids': lambda x: x.ids,
         'params': lambda x: x.params,
@@ -37,3 +43,60 @@ def export(path=None):
         for query in queryset.iterator():
 
             writer.writerow([val(query) for val in fields.values()])
+
+
+@job('default', timeout=1000)
+def import_last_day_analytics_from_api_mgmt(limit=1000):
+    task = AnalyticsImportTask(status=AnalyticsImportTask.RUNNING,
+                               timestamp=timezone.now())
+    import_config_model = ImportConfig.get_solo()
+    task.write_logs("Usando config: endpoint {}, api_id {}, token {}".format(
+        import_config_model.endpoint,
+        import_config_model.kong_api_id,
+        import_config_model.token
+    ))
+    try:
+        count = _run_import(limit)
+        task.write_logs("Todo OK. Queries importadas: {}".format(count))
+    except Exception as e:
+        task.write_logs("Error importando analytics: {}".format(e))
+    task.status = task.FINISHED
+    task.save()
+
+
+def _run_import(limit):
+    today = timezone.now().date()
+    yesterday = today - relativedelta(days=1)
+    import_config_model = ImportConfig.get_solo()
+    if (not import_config_model.endpoint or
+            not import_config_model.token or
+            not import_config_model.kong_api_id):
+        raise FieldError("Configuración de importación de analytics no inicializada")
+    response = import_config_model.get_results(from_date=yesterday, limit=limit)
+    count = response['count']
+    _load_queries_into_db(response)
+    next_results = response['next']
+    while next_results:
+        response = requests.get(
+            next_results,
+            headers=import_config_model.get_authorization_header()
+        ).json()
+        _load_queries_into_db(response)
+        next_results = response['next']
+
+    return count
+
+
+def _load_queries_into_db(query_results):
+    queries = []
+    for result in query_results['results']:
+        parsed_querystring = parse_qs(result['querystring'], keep_blank_values=True)
+        queries.append(Query(
+            ip_address=result['ip_address'],
+            args=result['querystring'],
+            timestamp=iso8601.parse_date(result['start_time']),
+            ids=parsed_querystring.get('ids', ''),
+            params=parsed_querystring,
+        ))
+
+    Query.objects.bulk_create(queries)

series_tiempo_ar_api/apps/analytics/tests/import_tests.py

@@ -0,0 +1,94 @@
+#!coding=utf8
+
+import mock
+from django.test import TestCase
+from series_tiempo_ar_api.apps.analytics.tasks import import_last_day_analytics_from_api_mgmt
+from series_tiempo_ar_api.apps.analytics.models import ImportConfig, Query, AnalyticsImportTask
+
+
+class UninitializedImportConfigTests(TestCase):
+
+    def test_not_initialized_model(self):
+        import_last_day_analytics_from_api_mgmt()
+        self.assertTrue('Error' in AnalyticsImportTask.objects.last().logs)
+
+
+def mocked_requests_get():
+    class MockResponse:
+        def __init__(self, json_data):
+            self.json_data = json_data
+
+        def json(self):
+            return self.json_data
+
+        def __call__(self, *args, **kwargs):
+            return self
+
+    return MockResponse(
+        {
+            'next': None,
+            'count': 2,
+            'results': [
+                {
+                    'ip_address': '127.0.0.1',
+                    'querystring': '',
+                    'start_time': '2018-06-07T05:00:00-03:00',
+                }
+            ]
+        }
+    )
+
+
+class ImportTests(TestCase):
+
+    def setUp(self):
+        config_model = ImportConfig.get_solo()
+        config_model.endpoint = 'http://localhost:80/fake_endpoint'
+        config_model.token = 'fake-token'
+        config_model.kong_api_id = 'fake_id'
+        config_model.save()
+
+    def test_single_empty_result(self):
+        with mock.patch.object(ImportConfig, 'get_results', return_value={
+            'next': None,
+            'count': 0,
+            'results': []
+        }):
+            import_last_day_analytics_from_api_mgmt()
+
+        self.assertEqual(Query.objects.count(), 0)
+
+    def test_single_page_results(self):
+        with mock.patch.object(ImportConfig, 'get_results', return_value={
+            'next': None,
+            'count': 1,
+            'results': [
+                {
+                    'ip_address': '127.0.0.1',
+                    'querystring': '',
+                    'start_time': '2018-06-07T05:00:00-03:00',
+                }
+            ]
+        }):
+            import_last_day_analytics_from_api_mgmt()
+            self.assertEqual(Query.objects.count(), 1)
+
+    def test_multiple_page_results(self):
+        """Emula una query con dos páginas de resultados, cada una con una query"""
+        return_value = {
+            'next': 'next_page_url',
+            'count': 2,
+            'results': [
+                {
+                    'ip_address': '127.0.0.1',
+                    'querystring': '',
+                    'start_time': '2018-06-07T05:00:00-03:00',
+                }
+            ]
+        }
+        with mock.patch.object(ImportConfig, 'get_results', return_value=return_value):
+            with mock.patch('series_tiempo_ar_api.apps.analytics.tasks.requests.get',
+                            new_callable=mocked_requests_get):
+                import_last_day_analytics_from_api_mgmt()
+
+        self.assertEqual(Query.objects.count(), 2)