Merge pull request #50 from socrata/cduranti/major-version

Cduranti/major version

examples/create.py

@@ -26,10 +26,7 @@ def create(name, filepath):
             name = name
         ).csv(csv_file)
 
-        (ok, job) = revision.apply(output_schema = output)
-        assert ok, job
-
+        job = revision.apply(output_schema = output)
         revision.open_in_browser()
 
-
 create(args.name, args.csv)

examples/update_metadata.py

@@ -27,28 +27,23 @@ def create(name, filepath):
             name = name
         ).csv(csv_file)
 
-        (ok, job) = initial_rev.apply(output_schema = output)
-        assert ok, job
-        (ok, job) = job.wait_for_finish()
-        assert ok, job
+        job = initial_rev.apply(output_schema = output)
+        job = job.wait_for_finish()
 
-        (ok, view) = socrata.views.lookup(initial_rev.attributes['fourfour'])
-        assert ok, view
+        view = socrata.views.lookup(initial_rev.attributes['fourfour'])
         update(view)
 
 
 def update(view):
-    (ok, revision) = view.revisions.create_replace_revision()
-    assert ok, revision
-    (ok, source) = revision.source_from_dataset()
-    assert ok, source
+    revision = view.revisions.create_replace_revision()
+    source = revision.source_from_dataset()
 
     output_schema = source.get_latest_input_schema().get_latest_output_schema()
 
     print(output_schema)
     random_column = output_schema.attributes['output_columns'][0]['field_name']
 
-    (ok, new_output) = output_schema\
+    new_output = output_schema\
         .change_column_metadata(random_column, 'description').to('this is the updated description')\
         .change_column_metadata(random_column, 'display_name').to('updated display name')\
         .run()

setup.py

@@ -11,7 +11,7 @@
     # Versions should comply with PEP440.  For a discussion on single-sourcing
     # the version across setup.py and the project code, see
     # https://packaging.python.org/en/latest/single_source_version.html
-    version='0.4.21',
+    version='1.0.0',
 
     description="SDK For publishing to Socrata",
     long_description="""

socrata/__init__.py

@@ -5,8 +5,6 @@
 from socrata.revisions import Revisions
 from socrata.operations.configured_job import ConfiguredJob
 from socrata.operations.create import Create
-from socrata.operations.utils import SocrataException
-
 
 class Socrata(Collection):
     """
@@ -57,9 +55,7 @@ def using_config(self, config_name, view):
         ```
 
         """
-        (ok, config) = result = self.configs.lookup(config_name)
-        if not ok:
-            raise SocrataException("Failed to lookup config %s" % config_name, result)
+        config = self.configs.lookup(config_name)
         return ConfiguredJob(self, view=view, config=config)
 
     def create(self, **kwargs):
@@ -76,7 +72,7 @@ def create(self, **kwargs):
             description = "a description"
         ).csv(file)
 
-        (ok, job) = revision.apply(output_schema = output_schema)
+        job = revision.apply(output_schema = output_schema)
         ```
 
         Args:
@@ -119,7 +115,7 @@ def new(self, metadata):
 
         Examples:
         ```python
-            (ok, rev) = Socrata(auth).new({
+            rev = Socrata(auth).new({
                 'name': 'hi',
                 'description': 'foo!',
                 'metadata': {

socrata/builders/parse_options.py

@@ -54,7 +54,7 @@ def change_parse_option(self, name):
 
         Examples:
         ```python
-            (ok, source) = source\
+            source = source\
             .change_parse_option('header_count').to(2)\
             .change_parse_option('column_header').to(2)\
             .run()

socrata/configs.py

@@ -74,10 +74,8 @@ def create_revision(self, uri, fourfour):
         # Because of circular dependencies ;_;
         from socrata.revisions import Revision
 
-        (ok, res) = result = post(
+        res = post(
             self.path(uri).format(fourfour = fourfour),
             auth = self.auth
         )
-        if not ok:
-            return result
-        return (ok, Revision(self.auth, res))
+        return Revision(self.auth, res)

socrata/http.py

@@ -8,6 +8,12 @@
 class TimeoutException(Exception):
     pass
 
+class UnexpectedResponseException(Exception):
+    def __init__(self, status, body):
+        super(UnexpectedResponseException, self).__init__("Unexpected status {status} {body}".format(status=status, body=body))
+        self.body = body
+        self.status = status
+
 def noop(*args, **kwargs):
     pass
 
@@ -37,21 +43,20 @@ def is_json(response):
     return 'application/json' in response.headers['Content-Type']
 
 def respond(response, request_id = None):
-    try:
-        if response.status_code in [200, 201, 202]:
-            if is_json(response):
-                return (True, response.json())
-            else:
-                return (True, response)
+    if response.status_code in [200, 201, 202]:
+        if is_json(response):
+            return response.json()
+        else:
+            return response
+    else:
+        log.warning("Request failed with %s, request_id was %s", response.status_code, request_id)
+        if is_json(response):
+            raise UnexpectedResponseException(response.status_code, response.json())
         else:
-            log.warning("Request failed with %s, request_id was %s", response.status_code, request_id)
-            if is_json(response):
-                return (False, response.json())
-            else:
-                return (False, response)
-    except Exception: # json.decoder.JSONDecodeError isn't always a thing???? WHY PYTHON
-        log.error("Request raised an exception, request_id was %s", request_id)
-        return (False, {'error': 'json', 'content': response.content})
+            raise UnexpectedResponseException(response.status_code, response)
+
+def pluck_resource(body):
+    return body['resource']
 
 def post(path, auth = None, data = None, headers = {}):
     (headers, request_id) = prepare(headers, auth)

socrata/lazy_pool.py

@@ -0,0 +1,80 @@
+from threading import Thread, Lock, Semaphore, Event
+from queue import Queue
+
+
+ONE_YEAR = 365 * 24 * 60 * 60
+
+THREAD_DONE = object()
+
+
+class LazyThreadPoolExecutor(object):
+    def __init__(self, num_workers=1):
+        self.num_workers = num_workers
+        self.result_queue = Queue()
+        self.thread_sem = Semaphore(num_workers)
+        self._shutdown = Event()
+        self.threads = []
+
+    def map(self, predicate, iterable):
+        self._shutdown.clear()
+        self.iterable = ThreadSafeIterator(iterable)
+        self._start_threads(predicate)
+        return self._result_iterator()
+
+    def shutdown(self, wait=True):
+        self._shutdown.set()
+        if wait:
+            for t in self.threads:
+                t.join()
+
+    def _start_threads(self, predicate):
+        for i in range(self.num_workers):
+            t = Thread(
+                name="LazyChild #{0}".format(i),
+                target=self._make_worker(predicate)
+            )
+            t.daemon = True
+            self.threads.append(t)
+            t.start()
+
+    def _make_worker(self, predicate):
+        def _w():
+            with self.thread_sem:
+                for thing in self.iterable:
+                    self.result_queue.put(predicate(thing))
+                    if self._shutdown.is_set():
+                        break
+            self.result_queue.put(THREAD_DONE)
+        return _w
+
+    def _result_iterator(self):
+        while 1:
+            # Queue.get is not interruptable w/ ^C unless you specify a
+            # timeout.
+            # Hopefully one year is long enough...
+            # See http://bugs.python.org/issue1360
+            result = self.result_queue.get(True, ONE_YEAR)
+            if result is not THREAD_DONE:
+                yield result
+            else:
+                # if all threads have exited
+                # sorry, this is kind of a gross way to use semaphores
+                # break
+                if self.thread_sem._value == self.num_workers:
+                    break
+                else:
+                    continue
+
+
+
+class ThreadSafeIterator(object):
+    def __init__(self, it):
+        self._it = iter(it)
+        self.lock = Lock()
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        with self.lock:
+            return self._it.__next__()

socrata/operations/configured_job.py

@@ -1,31 +1,15 @@
-from socrata.operations.utils import get_filename, SocrataException
+from socrata.operations.utils import get_filename
 from socrata.operations.operation import Operation
 
 class ConfiguredJob(Operation):
     def run(self, data, put_bytes, filename = None):
         filename = get_filename(data, filename)
-
-        (ok, rev) = self.properties['view'].revisions.create_using_config(
+        rev = self.properties['view'].revisions.create_using_config(
             self.properties['config']
         )
-        if not ok:
-            raise SocrataException("Failed to create the revision", rev)
-
-        (ok, source) = rev.create_upload(filename)
-        if not ok:
-            raise SocrataException("Failed to create the upload", source)
-
-        (ok, source) = put_bytes(source)
-        if not ok:
-            raise SocrataException("Failed to upload the file", source)
-
+        source = rev.create_upload(filename)
+        source = put_bytes(source)
         output_schema = source.get_latest_input_schema().get_latest_output_schema()
-
-        (ok, output_schema) = output_schema.wait_for_finish()
-        if not ok:
-            raise SocrataException("The dataset failed to validate", output_schema)
-
-        (ok, job) = rev.apply(output_schema = output_schema)
-        if not ok:
-            raise SocrataException("Failed to apply the change", job)
+        output_schema = output_schema.wait_for_finish()
+        job = rev.apply(output_schema = output_schema)
         return (rev, job)

socrata/operations/create.py

@@ -1,26 +1,12 @@
-from socrata.operations.utils import get_filename, SocrataException
+from socrata.operations.utils import get_filename
 from socrata.operations.operation import Operation
 
 class Create(Operation):
     def run(self, data, put_bytes, filename = None):
         filename = get_filename(data, filename)
-
-        (ok, rev) = self.publish.new(self.properties['metadata'])
-        if not ok:
-            raise SocrataException("Failed to create the view and revision", view)
-
-        (ok, source) = rev.create_upload(filename)
-        if not ok:
-            raise SocrataException("Failed to create the upload", source)
-
-        (ok, source) = put_bytes(source)
-        if not ok:
-            raise SocrataException("Failed to upload the file", source)
-
+        rev = self.publish.new(self.properties['metadata'])
+        source = rev.create_upload(filename)
+        source = put_bytes(source)
         output_schema = source.get_latest_input_schema().get_latest_output_schema()
-
-        (ok, output_schema) = output_schema.wait_for_finish()
-        if not ok:
-            raise SocrataException("The dataset failed to validate")
-
+        output_schema = output_schema.wait_for_finish()
         return (rev, output_schema)