Merge pull request #25 from AASHE/chunk-get-orgs

Chunk get orgs

Author: baronvonvaderham

membersuite_api_client/memberships/services.py

@@ -6,20 +6,21 @@
 """
 from zeep.exceptions import TransportError
 
+from ..mixins import ChunkQueryMixin
 from .models import Membership, MembershipProduct
 from ..utils import convert_ms_object
 import datetime
 
 
-class MembershipService(object):
+class MembershipService(ChunkQueryMixin, object):
 
     def __init__(self, client):
         """
         Accepts a ConciergeClient to connect with MemberSuite
         """
         self.client = client
 
-    def get_memberships_for_org(self, account_num):
+    def get_memberships_for_org(self, account_num, verbose=False):
         """
         Retrieve all memberships associated with an organization
         """
@@ -28,18 +29,13 @@ def get_memberships_for_org(self, account_num):
 
         query = "SELECT Objects() FROM Membership " \
                 "WHERE Owner = '%s'" % account_num
-        result = self.client.runSQL(query)
-        msql_result = result["body"]["ExecuteMSQLResult"]
-        obj_result = msql_result["ResultValue"]["ObjectSearchResult"]
-        if obj_result['Objects']:
-            objects = obj_result['Objects']['MemberSuiteObject']
-            if not msql_result["Errors"] and len(objects):
-                return self.package_memberships(objects)
-        return None
-
-    def get_all_memberships(self, since_when=None, results=None,
-                            start_record=0, limit_to=200,
-                            depth=1, max_depth=None):
+
+        membership_list = self.get_long_query(query, verbose=verbose)
+        return membership_list
+
+    def get_all_memberships(
+            self, limit_to=100, max_calls=None, parameters=None,
+            since_when=None, start_record=0, verbose=False):
         """
         Retrieve all memberships updated since "since_when"
 
@@ -51,100 +47,65 @@ def get_all_memberships(self, since_when=None, results=None,
             self.client.request_session()
 
         query = "SELECT Objects() FROM Membership"
+
+        # collect all where parameters into a list of
+        # (key, operator, value) tuples
+        where_params = []
+
+        if parameters:
+            for k, v in parameters.items():
+                where_params.append((k, "=", v))
+
         if since_when:
-            query += " WHERE LastModifiedDate > '{since_when} 00:00:00'" \
-                     " ORDER BY LocalID".format(
-                        since_when=datetime.date.today() -
-                        datetime.timedelta(days=since_when))
-        else:
-            query += " ORDER BY LocalID"
-
-        try:
-            result = self.client.runSQL(
-                query=query,
-                start_record=start_record,
-                limit_to=limit_to,
-            )
-
-        except TransportError:
-            # API Intermittently fails and kicks a 504,
-            # this is a way to retry if that happens.
-            result = self.get_all_memberships(
-                since_when=since_when,
-                results=results,
-                start_record=start_record,
-                limit_to=limit_to,
-                depth=depth,
-                max_depth=max_depth,
-            )
-
-        msql_result = result['body']["ExecuteMSQLResult"]
-        if (not msql_result['Errors'] and msql_result["ResultValue"]
-                ["ObjectSearchResult"]["Objects"]):
-            new_results = self.package_memberships(msql_result
-                                                   ["ResultValue"]
-                                                   ["ObjectSearchResult"]
-                                                   ["Objects"]
-                                                   ["MemberSuiteObject"]
-                                                   ) + (results or [])
-
-            # Check if the queryset was completely full. If so, there may be
-            # More results we need to query
-            if len(new_results) >= limit_to and not depth == max_depth:
-                new_results = self.get_all_memberships(
-                    since_when=since_when,
-                    results=new_results,
-                    start_record=start_record + limit_to,
-                    limit_to=limit_to,
-                    depth=depth+1,
-                    max_depth=max_depth
-                )
-            return new_results
-        else:
-            return results
-
-    def get_all_membership_products(self):
+            d = datetime.date.today() - datetime.timedelta(days=since_when)
+            where_params.append(
+                ('LastModifiedDate', ">", "'%s 00:00:00'" % d))
+
+        if where_params:
+            query += " WHERE "
+            query += " AND ".join(
+                ["%s %s %s" % (p[0], p[1], p[2]) for p in where_params])
+
+        query += " ORDER BY LocalID"
+
+        # note, get_long_query is overkill when just looking at
+        # one org, but it still only executes once
+        # `get_long_query` uses `ms_object_to_model` to return Organizations
+        membership_list = self.get_long_query(
+            query, limit_to=limit_to, max_calls=max_calls,
+            start_record=start_record, verbose=verbose)
+
+        return membership_list
+
+    def ms_object_to_model(self, ms_obj):
+        " Converts an individual result to a Subscription Model "
+        sane_obj = convert_ms_object(
+            ms_obj['Fields']['KeyValueOfstringanyType'])
+        return Membership(sane_obj)
+
+
+class MembershipProductService(ChunkQueryMixin, object):
+
+    def __init__(self, client):
+        """
+        Accepts a ConciergeClient to connect with MemberSuite
+        """
+        self.client = client
+
+    def get_all_membership_products(self, verbose=False):
         """
         Retrieves membership product objects
         """
         if not self.client.session_id:
             self.client.request_session()
 
         query = "SELECT Objects() FROM MembershipDuesProduct"
-        result = self.client.runSQL(query)
-        msql_result = result['body']['ExecuteMSQLResult']
-        if not msql_result['Errors']:
-            return self.package_membership_products(msql_result)
-        else:
-            return None
-
-    def package_memberships(self, object_list):
-        """
-        Loops through MS objects returned from queries to turn them into
-        Membership objects and pack them into a list for later use.
-        """
-        membership_list = []
-        for obj in object_list:
-            if type(obj) != str:
-                sane_obj = convert_ms_object(
-                    obj['Fields']['KeyValueOfstringanyType'])
-                membership = Membership(sane_obj)
-                membership_list.append(membership)
 
-        return membership_list
+        membership_product_list = self.get_long_query(query, verbose=verbose)
+        return membership_product_list
 
-    def package_membership_products(self, msql_result):
-        """
-        Loops through MS objects returned from queries to turn them into
-        MembershipProduct objects and pack them into a list for later use.
-        """
-        obj_result = msql_result['ResultValue']['ObjectSearchResult']
-        objects = obj_result['Objects']['MemberSuiteObject']
-        product_list = []
-        for obj in objects:
-            sane_obj = convert_ms_object(
-                obj['Fields']['KeyValueOfstringanyType']
-            )
-            product = MembershipProduct(sane_obj)
-            product_list.append(product)
-        return product_list
+    def ms_object_to_model(self, ms_obj):
+        " Converts an individual result to a Subscription Model "
+        sane_obj = convert_ms_object(
+            ms_obj['Fields']['KeyValueOfstringanyType'])
+        return MembershipProduct(sane_obj)

membersuite_api_client/mixins.py

@@ -1,4 +1,9 @@
 from retrying import retry
+import datetime
+
+from .exceptions import ExecuteMSQLError
+
+RETRY_ATTEMPTS = 10
 
 
 class ChunkQueryMixin():
@@ -15,28 +20,41 @@ class ChunkQueryMixin():
     """
 
     def get_long_query(
-            self, base_query, retry_attempts=2, limit_to=200, max_calls=None):
+            self, base_query, limit_to=100, max_calls=None,
+            start_record=0, verbose=False):
         """
         Takes a base query for all objects and recursively requests them
 
-        @base_query - the base query to be executed
-        @retry_attempts - the number of times to retry a query when it fails
-        @limit_to - how many rows to query for in each chunk
-        @max_recursion_depth - None is infinite
+        :param str base_query: the base query to be executed
+        :param int limit_to: how many rows to query for in each chunk
+        :param int max_calls: the max calls(chunks to request) None is infinite
+        :param int start_record: the first record to return from the query
+        :param bool verbose: print progress to stdout
+        :return: a list of Organization objects
         """
 
-        @retry(stop_max_attempt_number=retry_attempts)
-        def run_query(base_query, start_record, limit_to):
+        @retry(stop_max_attempt_number=RETRY_ATTEMPTS, wait_fixed=2000)
+        def run_query(base_query, start_record, limit_to, verbose):
             # inline method to take advantage of retry
+
+            if verbose:
+                print("[start: %d limit: %d]" % (start_record, limit_to))
+            start = datetime.datetime.now()
             result = self.client.runSQL(
                 query=base_query,
                 start_record=start_record,
                 limit_to=limit_to,
             )
+            end = datetime.datetime.now()
+            if verbose:
+                print("[%s - %s]" % (start, end))
             return self.result_to_models(result)
 
-        record_index = 0
-        result = run_query(base_query, record_index, limit_to)
+        if verbose:
+            print(base_query)
+
+        record_index = start_record
+        result = run_query(base_query, record_index, limit_to, verbose)
         all_objects = result
         call_count = 1
         """
@@ -49,7 +67,32 @@ def run_query(base_query, start_record, limit_to):
                 len(result) >= limit_to):
 
             record_index += len(result)  # should be `limit_to`
-            all_objects += run_query(base_query, record_index, limit_to)
+            result = run_query(
+                base_query, record_index, limit_to, verbose)
+            all_objects += result
             call_count += 1
 
         return all_objects
+
+    def result_to_models(self, result):
+        """
+            this is the 'transorm' part of ETL:
+            converts the result of the SQL to Models
+        """
+        mysql_result = result['body']['ExecuteMSQLResult']
+
+        if not mysql_result['Errors']:
+            obj_result = mysql_result['ResultValue']['ObjectSearchResult']
+            if not obj_result['Objects']:
+                return []
+            objects = obj_result['Objects']['MemberSuiteObject']
+
+            model_list = []
+            for obj in objects:
+                model = self.ms_object_to_model(obj)
+                model_list.append(model)
+
+            return model_list
+
+        else:
+            raise ExecuteMSQLError(result)