CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [1.1.6]
+
+### Changed
+- mechanism to fetch all results from atoka, exceeding the maximum limit improved
+
+
 ## [1.1.5]
 
 ### Fixed

project/__init__.py

@@ -3,4 +3,4 @@
 Openpolis Data Manager service package (backend)
 """
 
-__version__ = '1.1.5'
+__version__ = '1.1.6'

project/atoka/connections.py

@@ -99,7 +99,7 @@ class AtokaConn(object):
         return result['items'][0]
 
     def get_items_from_ids(
-        self, ids: list, item_type: str, ids_field_name: str = 'ids', batch_size: int = 50, limit=50, **kwargs
+        self, ids: list, item_type: str, ids_field_name: str = 'ids', batch_size: int = 50, **kwargs
     ) -> list:
         """Transform a request for a list of ids larger than batch_size,
          to a batch request of enough rows with a limit of batch_size, so that all results
@@ -111,7 +111,6 @@ class AtokaConn(object):
         :param item_type: str
         :param ids_field_name: ids, tax_ids
         :param batch_size: size of the number of ids searched by row of the batch IO
-        :param limit: max n. of results for eache batch request
         :param kwargs: - more atoka parameters for filtering results
             (ex: packages=base,shares, active='true', ccia='*')
         :return: results as a list of dicts
@@ -141,13 +140,11 @@ class AtokaConn(object):
 
         # build fileIO to upload form batch execution
         file_io = StringIO()
-        for n, r in enumerate(chunks(ids, batch_size)):
-            for offset in range(0, limit, 50):
-                print(json.dumps({
-                    "reqId": "r{0:05d}".format(int(n*10e4 + offset)),
-                    ids_field_name: ','.join(r),
-                    'offset': offset
-                }), file=file_io)
+        for n, r in enumerate(chunks(ids, batch_size), start=1):
+            print(json.dumps({
+                "reqId": ','.join(r),
+                ids_field_name: ','.join(r),
+            }), file=file_io)
 
         # batch API request
         fields = {
@@ -181,8 +178,42 @@ class AtokaConn(object):
         total_response = []
 
         if 'responses' in json_response:
-            for r in json_response['responses'].values():
+            for req_id, r in json_response['responses'].items():
                 total_response.extend(r['items'])
+
+                # if more than 50 results, get the others
+                if r['meta']['count'] > 50:
+                    file_io = StringIO()
+                    for offset in range(50, r['meta']['count'], 50):
+                        print(json.dumps({
+                            "reqId": "r{0:05d}".format(offset),
+                            ids_field_name: req_id,
+                            'offset': offset
+                        }), file=file_io)
+
+                    # batch API request
+                    fields = {
+                        'batch': ('batch.json', file_io),
+                        'limit': '50'
+                    }
+                    fields.update(kwargs)
+                    m = MultipartEncoder(
+                        fields=fields
+                    )
+                    refined_response = requests.post(
+                        api_endpoint,
+                        params={'token': self.key},
+                        data=m,
+                        headers={'Content-Type': m.content_type}
+                    )
+
+                    # destroy fileIO
+                    file_io.close()
+
+                    refined_json_response = refined_response.json()
+                    for _r in refined_json_response['responses'].values():
+                        total_response.extend(_r['items'])
+
         else:
             total_response.extend(json_response['items'])
 

project/atoka/etl/extractors.py

@@ -131,7 +131,12 @@ class AtokaOwnershipsExtractor(Extractor):
                     return True
             return False
 
-        res_tot = list(filter(check_owner_has_shares_in_active_companies, res_tot))
+        res_tot = list(
+            sorted(
+                filter(check_owner_has_shares_in_active_companies, res_tot),
+                key=lambda x: x['base']['taxId']
+            )
+        )
         self.logger.debug(
             "- {0} istituzioni hanno partecipazioni in aziende attive".format(
                 len(res_tot)
@@ -142,18 +147,22 @@ class AtokaOwnershipsExtractor(Extractor):
         # merging results from multiple records corresponding to the same tax_id
         res_dict = {}
 
-        r_dict = {}
         for r in res_tot:
-            r_dict.setdefault('atoka_id', r['id'])
-            r_dict.setdefault('other_atoka_ids', [
-                    atoka_id for atoka_id in res_doubles[r['base']['taxId']] if atoka_id != r['id']
-                ] if r['base']['taxId'] in res_doubles else []
-            )
-            r_dict.setdefault('name', r['name'])
-            r_dict.setdefault('legal_form', [x['name'] for x in r['base']['legalForms'] if x['level'] == 2][0])
-            r_dict.setdefault('rea', r['base'].get('rea', None))
-            r_dict.setdefault('cciaa', r['base'].get('cciaa', None))
-            r_dict.setdefault('shares_owned', [])
+            if r['base']['taxId'] not in res_dict:
+                r_dict = {
+                    'atoka_id': r['id'],
+                    'other_atoka_ids': [
+                        atoka_id for atoka_id in res_doubles[r['base']['taxId']] if atoka_id != r['id']
+                    ] if r['base']['taxId'] in res_doubles else [],
+                    'name': r['name'],
+                    'legal_form': [x['name'] for x in r['base']['legalForms'] if x['level'] == 2][0],
+                    'rea': r['base'].get('rea', None),
+                    'cciaa': r['base'].get('cciaa', None),
+                    'shares_owned': []
+                }
+            else:
+                r_dict = res_dict[r['base']['taxId']]
+
             r_dict['shares_owned'].extend(
                 {
                     'name': sho['name'],
@@ -166,7 +175,7 @@ class AtokaOwnershipsExtractor(Extractor):
                     r['shares']['sharesOwned']
                 )
             )
-        res_dict[r['base']['taxId']] = r_dict
+            res_dict[r['base']['taxId']] = r_dict
 
         # extract all atoka_ids from shares_owned elements and returns flat list
         # then apply list(set(x)) to remove duplicates, if any
@@ -195,7 +204,7 @@ class AtokaOwnershipsExtractor(Extractor):
 
         # extract all people's atoka_ids from res_owned elements and returns flat list, removing duplicates
         people = atoka_conn.get_roles_from_atoka_ids(
-            owned_atoka_ids, packages='base,companies', limit=150,
+            owned_atoka_ids, packages='base,companies',
             companiesRolesOfficial='true', companiesRoles=atoka_conn.allowed_roles
         )
         atoka_people_requests += len(people)

project/atoka/management/commands/import_atoka_all.py

 # -*- coding: utf-8 -*-
-import os
-
 from django.core import management
 from django.db.models import Sum
 from models import Classification, Organization
@@ -40,7 +38,8 @@ class Command(LoggingBaseCommand):
             help="Maximum level of the public share"
         )
 
-    def compute_and_update_public_interest(self, c_level, threshold=25.):
+    @staticmethod
+    def compute_and_update_public_interest(c_level, threshold=25.):
         """Compute and update the total percentage of public partecipation
            for all organizations having given level"""
 
@@ -110,7 +109,6 @@ class Command(LoggingBaseCommand):
             # calcolo percentuali e definizione di pubblico interesse per l1vello successivo
             self.compute_and_update_public_interest(next_c)
 
-
         management.call_command(
             'import_atoka_economics',
             verbosity=verbosity,
@@ -118,5 +116,4 @@ class Command(LoggingBaseCommand):
             batchsize=batchsize
         )
 
-
         self.logger.info("End overall procedure")

setup.cfg

 [bumpversion]
-current_version = 1.1.5
+current_version = 1.1.6
 commit = True
 tag = True
 tag_name = v{new_version}