Add tests

.github/workflows/github-actions.yml

@@ -0,0 +1,41 @@
+name: Pytest package
+
+on: [push]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    services:
+      es-1:
+        image: udata/elasticsearch:7.16.2
+        env:
+          node.name: es01
+          cluster.name: es-docker-cluster
+          cluster.initial_master_nodes: es01
+        ports:
+          - 9200:9200
+    strategy:
+      matrix:
+        python-version: [3.9]
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install flake8 pytest
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+      - name: Lint with flake8
+        run: |
+          # stop the build if there are Python syntax errors or undefined names
+          flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+          # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
+          flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+      - name: Test with pytest
+        run: |
+          pytest

app/__init__.py

@@ -1,7 +1,6 @@
 from flask import Flask
 from app.config import Config
 from app.container import Container
-from app.infrastructure import kafka_consumer
 from app.presentation import api, commands
 
 

app/infrastructure/kafka_consumer.py

@@ -37,8 +37,8 @@ def create_kafka_consumer():
     consumer = KafkaConsumer(
         bootstrap_servers=f'{KAFKA_HOST}:{KAFKA_PORT}',
         group_id='elastic',
-        reconnect_backoff_max_ms=100000, # TODO: what value to set here?
-        
+        reconnect_backoff_max_ms=100000,  # TODO: what value to set here?
+
         # API Version is needed in order to prevent api version guessing leading to an error
         # on startup if Kafka Broker isn't ready yet
         api_version=tuple([int(value) for value in KAFKA_API_VERSION.split('.')])
@@ -79,7 +79,7 @@ def consume_messages(consumer, es):
     for message in consumer:
         value = message.value
         val_utf8 = value.decode('utf-8').replace('NaN', 'null')
-        
+
         key = message.key
         index = message.topic
 

app/infrastructure/search_clients.py

@@ -1,6 +1,6 @@
 from typing import Tuple, Optional, List
 from elasticsearch.exceptions import NotFoundError
-from elasticsearch_dsl import Index, Document, Integer, Text, tokenizer, token_filter, analyzer, query, Date
+from elasticsearch_dsl import Index, Document, Integer, Text, tokenizer, token_filter, analyzer, query
 from elasticsearch_dsl.connections import connections
 from app.domain.entities import Dataset, Organization, Reuse
 from app.config import Config
@@ -28,7 +28,7 @@ class SearchableOrganization(Document):
     description = Text(analyzer=dgv_analyzer)
     url = Text()
     orga_sp = Integer()
-    created_at = Date()
+    created_at = Text()
     followers = Integer()
     datasets = Integer()
 
@@ -39,7 +39,7 @@ class Index:
 class SearchableReuse(Document):
     title = Text(analyzer=dgv_analyzer)
     url = Text()
-    created_at = Date()
+    created_at = Text()
     orga_followers = Integer()
     views = Integer()
     followers = Integer()
@@ -57,7 +57,7 @@ class SearchableDataset(Document):
     title = Text(analyzer=dgv_analyzer)
     acronym = Text()
     url = Text()
-    created_at = Date()
+    created_at = Text()
     orga_sp = Integer()
     orga_followers = Integer()
     views = Integer()
@@ -67,8 +67,8 @@ class SearchableDataset(Document):
     resources_count = Integer()
     concat_title_org = Text(analyzer=dgv_analyzer)
     organization_id = Text()
-    temporal_coverage_start = Date()
-    temporal_coverage_end = Date()
+    temporal_coverage_start = Text()
+    temporal_coverage_end = Text()
     granularity = Text()
     geozones = Text()
     description = Text(analyzer=dgv_analyzer)
@@ -103,69 +103,98 @@ def index_dataset(self, to_index: Dataset) -> None:
     def index_reuse(self, to_index: Reuse) -> None:
         SearchableReuse(meta={'id': to_index.id}, **to_index.to_dict()).save(skip_empty=False)
 
-    def query_organizations(self, query_text: str, offset: int, page_size: int) -> Tuple[int, List[dict]]:
-        s = SearchableOrganization.search().query('bool', should=[
-                query.Q(
-                    'function_score',
-                    query=query.Bool(should=[query.MultiMatch(query=query_text, type='phrase', fields=['title^15','acronym^15','description^8'])]),
-                    functions=[
-                        query.SF("field_value_factor", field="orga_sp", factor=8, modifier='sqrt', missing=1),
-                        query.SF("field_value_factor", field="followers", factor=4, modifier='sqrt', missing=1),
-                        query.SF("field_value_factor", field="datasets", factor=1, modifier='sqrt', missing=1),
-                    ],
-                ),
-            query.Match(title={"query": query_text, 'fuzziness': 'AUTO'})
-        ])
+    def query_organizations(self, query_text: str, offset: int, page_size: int, filters: dict) -> Tuple[int, List[dict]]:
+        s = SearchableOrganization.search()
+
+        for key, value in filters.items():
+            s = s.filter('term', **{key: value})
+
+        if query_text:
+            s = s.query('bool', should=[
+                    query.Q(
+                        'function_score',
+                        query=query.Bool(should=[query.MultiMatch(query=query_text, type='phrase', fields=['name^15', 'acronym^15', 'description^8'])]),
+                        functions=[
+                            query.SF("field_value_factor", field="orga_sp", factor=8, modifier='sqrt', missing=1),
+                            query.SF("field_value_factor", field="followers", factor=4, modifier='sqrt', missing=1),
+                            query.SF("field_value_factor", field="datasets", factor=1, modifier='sqrt', missing=1),
+                        ],
+                    ),
+                query.Match(title={"query": query_text, 'fuzziness': 'AUTO'})
+            ])
+
         s = s[offset:(offset + page_size)]
+
         response = s.execute()
         results_number = response.hits.total.value
         res = [hit.to_dict(skip_empty=False) for hit in response.hits]
         return results_number, res
 
-    def query_datasets(self, query_text: str, offset: int, page_size: int) -> Tuple[int, List[dict]]:
-        datasets_score_functions = [
-            query.SF("field_value_factor", field="orga_sp", factor=8, modifier='sqrt', missing=1),
-            query.SF("field_value_factor", field="views", factor=4, modifier='sqrt', missing=1),
-            query.SF("field_value_factor", field="followers", factor=4, modifier='sqrt', missing=1),
-            query.SF("field_value_factor", field="orga_followers", factor=1, modifier='sqrt', missing=1),
-            query.SF("field_value_factor", field="featured", factor=1, modifier='sqrt', missing=1),
-        ]
-        s = SearchableDataset.search().query(
-            'bool',
-            should=[
-                query.Q(
-                    'function_score',
-                    query=query.Bool(should=[query.MultiMatch(query=query_text, type='phrase', fields=['title^15','acronym^15','description^8','organization^8'])]),
-                    functions=datasets_score_functions,
-                ),
-                query.Q(
-                    'function_score',
-                    query=query.Bool(must=[query.Match(concat_title_org={"query": query_text, "operator": "and", "boost": 8})]),
-                    functions=datasets_score_functions,
-                ),
-                query.MultiMatch(query=query_text, type='most_fields', fields=['title', 'organization'], fuzziness='AUTO')
-            ])
+    def query_datasets(self, query_text: str, offset: int, page_size: int, filters: dict) -> Tuple[int, List[dict]]:
+        s = SearchableDataset.search()
+
+        for key, value in filters.items():
+            if key == 'temporal_coverage_start':
+                s = s.filter('range', **{'temporal_coverage_start': {'lte': value}})
+            elif key == 'temporal_coverage_end':
+                s = s.filter('range', **{'temporal_coverage_end': {'gte': value}})
+            else:
+                s = s.filter('term', **{key: value})
+
+        if query_text:
+            datasets_score_functions = [
+                query.SF("field_value_factor", field="orga_sp", factor=8, modifier='sqrt', missing=1),
+                query.SF("field_value_factor", field="views", factor=4, modifier='sqrt', missing=1),
+                query.SF("field_value_factor", field="followers", factor=4, modifier='sqrt', missing=1),
+                query.SF("field_value_factor", field="orga_followers", factor=1, modifier='sqrt', missing=1),
+                query.SF("field_value_factor", field="featured", factor=1, modifier='sqrt', missing=1),
+            ]
+            s = s.query(
+                'bool',
+                should=[
+                    query.Q(
+                        'function_score',
+                        query=query.Bool(should=[query.MultiMatch(query=query_text, type='phrase', fields=['title^15', 'acronym^15', 'description^8', 'organization^8'])]),
+                        functions=datasets_score_functions,
+                    ),
+                    query.Q(
+                        'function_score',
+                        query=query.Bool(must=[query.Match(concat_title_org={"query": query_text, "operator": "and", "boost": 8})]),
+                        functions=datasets_score_functions,
+                    ),
+                    query.MultiMatch(query=query_text, type='most_fields', fields=['title', 'organization'], fuzziness='AUTO')
+                ])
+
         s = s[offset:(offset + page_size)]
+
         response = s.execute()
         results_number = response.hits.total.value
         res = [hit.to_dict(skip_empty=False) for hit in response.hits]
         return results_number, res
 
-    def query_reuses(self, query_text: str, offset: int, page_size: int) -> Tuple[int, List[dict]]:
-        s = SearchableReuse.search().query('bool', should=[
-                query.Q(
-                    'function_score',
-                    query=query.Bool(should=[query.MultiMatch(query=query_text, type='phrase', fields=['title^15','description^8','organization^8'])]),
-                    functions=[
-                        query.SF("field_value_factor", field="views", factor=4, modifier='sqrt', missing=1),
-                        query.SF("field_value_factor", field="followers", factor=4, modifier='sqrt', missing=1),
-                        query.SF("field_value_factor", field="orga_followers", factor=1, modifier='sqrt', missing=1),
-                        query.SF("field_value_factor", field="featured", factor=1, modifier='sqrt', missing=1),
-                    ],
-                ),
-                query.MultiMatch(query=query_text, type='most_fields', fields=['title', 'organization'], fuzziness='AUTO')
-            ])
+    def query_reuses(self, query_text: str, offset: int, page_size: int, filters: dict) -> Tuple[int, List[dict]]:
+        s = SearchableReuse.search()
+
+        for key, value in filters.items():
+            s = s.filter('term', **{key: value})
+
+        if query_text:
+            s = s.query('bool', should=[
+                    query.Q(
+                        'function_score',
+                        query=query.Bool(should=[query.MultiMatch(query=query_text, type='phrase', fields=['title^15', 'description^8', 'organization^8'])]),
+                        functions=[
+                            query.SF("field_value_factor", field="views", factor=4, modifier='sqrt', missing=1),
+                            query.SF("field_value_factor", field="followers", factor=4, modifier='sqrt', missing=1),
+                            query.SF("field_value_factor", field="orga_followers", factor=1, modifier='sqrt', missing=1),
+                            query.SF("field_value_factor", field="featured", factor=1, modifier='sqrt', missing=1),
+                        ],
+                    ),
+                    query.MultiMatch(query=query_text, type='most_fields', fields=['title', 'organization'], fuzziness='AUTO')
+                ])
+
         s = s[offset:(offset + page_size)]
+
         response = s.execute()
         results_number = response.hits.total.value
         res = [hit.to_dict(skip_empty=False) for hit in response.hits]

app/infrastructure/services.py

@@ -11,13 +11,17 @@ def __init__(self, search_client: ElasticClient):
     def feed(self, organization: Organization) -> None:
         self.search_client.index_organization(organization)
 
-    def search(self, search_text: str, page: int, page_size: int) -> Tuple[List[Organization], int, int]:
+    def search(self, filters: dict) -> Tuple[List[Organization], int, int]:
+        page = filters.pop('page')
+        page_size = filters.pop('page_size')
+        search_text = filters.pop('q')
+
         if page > 1:
             offset = page_size * (page - 1)
         else:
             offset = 0
 
-        results_number, search_results = self.search_client.query_organizations(search_text, offset, page_size)
+        results_number, search_results = self.search_client.query_organizations(search_text, offset, page_size, filters)
         results = [Organization.load_from_dict(hit) for hit in search_results]
         total_pages = round(results_number / page_size) or 1
         return results, results_number, total_pages
@@ -37,13 +41,20 @@ def __init__(self, search_client: ElasticClient):
     def feed(self, dataset: Dataset) -> None:
         self.search_client.index_dataset(dataset)
 
-    def search(self, search_text: str, page: int, page_size: int) -> Tuple[List[Dataset], int, int]:
+    def search(self, filters: dict) -> Tuple[List[Dataset], int, int]:
+        page = filters.pop('page')
+        page_size = filters.pop('page_size')
+        search_text = filters.pop('q')
+
         if page > 1:
             offset = page_size * (page - 1)
         else:
             offset = 0
 
-        results_number, search_results = self.search_client.query_datasets(search_text, offset, page_size)
+        if filters['temporal_coverage']:
+            self.parse_temporal_value(filters)
+
+        results_number, search_results = self.search_client.query_datasets(search_text, offset, page_size, filters)
         results = [Dataset.load_from_dict(hit) for hit in search_results]
         total_pages = round(results_number / page_size) or 1
         return results, results_number, total_pages
@@ -54,6 +65,12 @@ def find_one(self, dataset_id: str) -> Optional[Dataset]:
         except TypeError:
             return None
 
+    @staticmethod
+    def parse_temporal_value(filters):
+        parts = filters.pop('temporal_coverage')
+        filters['temporal_coverage_start'] = parts[:10]
+        filters['temporal_coverage_end'] = parts[11:]
+
 
 class ReuseService:
 
@@ -63,13 +80,17 @@ def __init__(self, search_client: ElasticClient):
     def feed(self, reuse: Reuse) -> None:
         self.search_client.index_reuse(reuse)
 
-    def search(self, search_text: str, page: int, page_size: int) -> Tuple[List[Reuse], int, int]:
+    def search(self, filters: dict) -> Tuple[List[Reuse], int, int]:
+        page = filters.pop('page')
+        page_size = filters.pop('page_size')
+        search_text = filters.pop('q')
+
         if page > 1:
             offset = page_size * (page - 1)
         else:
             offset = 0
 
-        results_number, search_results = self.search_client.query_reuses(search_text, offset, page_size)
+        results_number, search_results = self.search_client.query_reuses(search_text, offset, page_size, filters)
         results = [Reuse.load_from_dict(hit) for hit in search_results]
         total_pages = round(results_number / page_size) or 1
         return results, results_number, total_pages