Merge pull request #52 from blockchain-etl/add_blockchainetl_from_eth…

…ereumetl

Add files from ethereum-etl as workaround for https://github.com/bloc…

blockchainetl/jobs/exporters/composite_item_exporter.py

@@ -24,17 +24,20 @@
 from blockchainetl.atomic_counter import AtomicCounter
 from blockchainetl.exporters import CsvItemExporter, JsonLinesItemExporter
 from blockchainetl.file_utils import get_file_handle, close_silently
+from blockchainetl.jobs.exporters.converters.composite_item_converter import CompositeItemConverter
 
 
 class CompositeItemExporter:
-    def __init__(self, filename_mapping, field_mapping=None):
+    def __init__(self, filename_mapping, field_mapping=None, converters=()):
         self.filename_mapping = filename_mapping
         self.field_mapping = field_mapping or {}
 
         self.file_mapping = {}
         self.exporter_mapping = {}
         self.counter_mapping = {}
 
+        self.converter = CompositeItemConverter(converters)
+
         self.logger = logging.getLogger('CompositeItemExporter')
 
     def open(self):
@@ -62,7 +65,7 @@ def export_item(self, item):
         exporter = self.exporter_mapping.get(item_type)
         if exporter is None:
             raise ValueError('Exporter for item type {} not found'.format(item_type))
-        exporter.export_item(item)
+        exporter.export_item(self.converter.convert_item(item))
 
         counter = self.counter_mapping.get(item_type)
         if counter is not None:

blockchainetl/jobs/exporters/converters/composite_item_converter.py

@@ -0,0 +1,45 @@
+#  MIT License
+#
+#  Copyright (c) 2018 Evgeny Medvedev, evge.medvedev@gmail.com
+#
+#  Permission is hereby granted, free of charge, to any person obtaining a copy
+#  of this software and associated documentation files (the "Software"), to deal
+#  in the Software without restriction, including without limitation the rights
+#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+#  copies of the Software, and to permit persons to whom the Software is
+#  furnished to do so, subject to the following conditions:
+#
+#  The above copyright notice and this permission notice shall be included in all
+#  copies or substantial portions of the Software.
+#
+#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+#  SOFTWARE.
+
+# MIT License
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+
+
+class CompositeItemConverter:
+
+    def __init__(self, converters=()):
+        self.converters = converters
+
+    def convert_item(self, item):
+        if self.converters is None:
+            return item
+
+        for converter in self.converters:
+            item = converter.convert_item(item)
+        return item

blockchainetl/jobs/exporters/converters/int_to_decimal_item_converter.py

@@ -0,0 +1,47 @@
+#  MIT License
+#
+#  Copyright (c) 2018 Evgeny Medvedev, evge.medvedev@gmail.com
+#
+#  Permission is hereby granted, free of charge, to any person obtaining a copy
+#  of this software and associated documentation files (the "Software"), to deal
+#  in the Software without restriction, including without limitation the rights
+#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+#  copies of the Software, and to permit persons to whom the Software is
+#  furnished to do so, subject to the following conditions:
+#
+#  The above copyright notice and this permission notice shall be included in all
+#  copies or substantial portions of the Software.
+#
+#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+#  SOFTWARE.
+
+# MIT License
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+#
+
+from decimal import Decimal
+
+from blockchainetl.jobs.exporters.converters.simple_item_converter import SimpleItemConverter
+
+# Large ints are not handled correctly by pg8000 so we use Decimal instead:
+# https://github.com/mfenniak/pg8000/blob/412eace074514ada824e7a102765e37e2cda8eaa/pg8000/core.py#L1703
+class IntToDecimalItemConverter(SimpleItemConverter):
+
+    def convert_field(self, key, value):
+        if isinstance(value, int):
+            return Decimal(value)
+        else:
+            return value

blockchainetl/jobs/exporters/converters/int_to_string_item_converter.py

@@ -0,0 +1,46 @@
+#  MIT License
+#
+#  Copyright (c) 2018 Evgeny Medvedev, evge.medvedev@gmail.com
+#
+#  Permission is hereby granted, free of charge, to any person obtaining a copy
+#  of this software and associated documentation files (the "Software"), to deal
+#  in the Software without restriction, including without limitation the rights
+#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+#  copies of the Software, and to permit persons to whom the Software is
+#  furnished to do so, subject to the following conditions:
+#
+#  The above copyright notice and this permission notice shall be included in all
+#  copies or substantial portions of the Software.
+#
+#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+#  SOFTWARE.
+
+# MIT License
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+
+
+from blockchainetl.jobs.exporters.converters.simple_item_converter import SimpleItemConverter
+
+
+class IntToStringItemConverter(SimpleItemConverter):
+
+    def __init__(self, keys=None):
+        self.keys = set(keys) if keys else None
+
+    def convert_field(self, key, value):
+        if isinstance(value, int) and (self.keys is None or key in self.keys):
+            return str(value)
+        else:
+            return value

blockchainetl/jobs/exporters/converters/list_field_item_converter.py

@@ -0,0 +1,56 @@
+#  MIT License
+#
+#  Copyright (c) 2018 Evgeny Medvedev, evge.medvedev@gmail.com
+#
+#  Permission is hereby granted, free of charge, to any person obtaining a copy
+#  of this software and associated documentation files (the "Software"), to deal
+#  in the Software without restriction, including without limitation the rights
+#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+#  copies of the Software, and to permit persons to whom the Software is
+#  furnished to do so, subject to the following conditions:
+#
+#  The above copyright notice and this permission notice shall be included in all
+#  copies or substantial portions of the Software.
+#
+#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+#  SOFTWARE.
+
+# MIT License
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+
+
+class ListFieldItemConverter:
+
+    def __init__(self, field, new_field_prefix, fill=0, fill_with=None):
+        self.field = field
+        self.new_field_prefix = new_field_prefix
+        self.fill = fill
+        self.fill_with = fill_with
+
+    def convert_item(self, item):
+        if not item:
+            return item
+
+        lst = item.get(self.field)
+        result = item
+        if lst is not None and isinstance(lst, list):
+            result = item.copy()
+            del result[self.field]
+            for lst_item_index, lst_item in enumerate(lst):
+                result[self.new_field_prefix + str(lst_item_index)] = lst_item
+            if len(lst) < self.fill:
+                for i in range(len(lst), self.fill):
+                    result[self.new_field_prefix + str(i)] = self.fill_with
+        return result

blockchainetl/jobs/exporters/converters/simple_item_converter.py

@@ -0,0 +1,42 @@
+#  MIT License
+#
+#  Copyright (c) 2018 Evgeny Medvedev, evge.medvedev@gmail.com
+#
+#  Permission is hereby granted, free of charge, to any person obtaining a copy
+#  of this software and associated documentation files (the "Software"), to deal
+#  in the Software without restriction, including without limitation the rights
+#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+#  copies of the Software, and to permit persons to whom the Software is
+#  furnished to do so, subject to the following conditions:
+#
+#  The above copyright notice and this permission notice shall be included in all
+#  copies or substantial portions of the Software.
+#
+#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+#  SOFTWARE.
+
+# MIT License
+#
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+
+
+class SimpleItemConverter:
+
+    def convert_item(self, item):
+        return {
+            key: self.convert_field(key, value) for key, value in item.items()
+        }
+
+    def convert_field(self, key, value):
+        return value

blockchainetl/jobs/exporters/converters/unix_timestamp_item_converter.py

@@ -0,0 +1,41 @@
+# MIT License
+#
+# Copyright (c) 2020 Evgeny Medvedev, evge.medvedev@gmail.com
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+from datetime import datetime
+
+from blockchainetl.jobs.exporters.converters.simple_item_converter import SimpleItemConverter
+
+
+class UnixTimestampItemConverter(SimpleItemConverter):
+
+    def convert_field(self, key, value):
+        if key is not None and key.endswith('timestamp'):
+            return to_timestamp(value)
+        else:
+            return value
+
+
+def to_timestamp(value):
+    if isinstance(value, int):
+        return datetime.utcfromtimestamp(value).strftime('%Y-%m-%d %H:%M:%S')
+    else:
+        return value

blockchainetl/jobs/exporters/postgres_item_exporter.py

@@ -0,0 +1,70 @@
+# MIT License
+#
+# Copyright (c) 2020 Evgeny Medvedev, evge.medvedev@gmail.com
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+import collections
+
+from sqlalchemy import create_engine
+
+from blockchainetl.jobs.exporters.converters.composite_item_converter import CompositeItemConverter
+
+
+class PostgresItemExporter:
+
+    def __init__(self, connection_url, item_type_to_insert_stmt_mapping, converters=(), print_sql=True):
+        self.connection_url = connection_url
+        self.item_type_to_insert_stmt_mapping = item_type_to_insert_stmt_mapping
+        self.converter = CompositeItemConverter(converters)
+        self.print_sql = print_sql
+
+        self.engine = self.create_engine()
+
+    def open(self):
+        pass
+
+    def export_items(self, items):
+        items_grouped_by_type = group_by_item_type(items)
+
+        for item_type, insert_stmt in self.item_type_to_insert_stmt_mapping.items():
+            item_group = items_grouped_by_type.get(item_type)
+            if item_group:
+                connection = self.engine.connect()
+                converted_items = list(self.convert_items(item_group))
+                connection.execute(insert_stmt, converted_items)
+
+    def convert_items(self, items):
+        for item in items:
+            yield self.converter.convert_item(item)
+
+    def create_engine(self):
+        engine = create_engine(self.connection_url, echo=self.print_sql, pool_recycle=3600)
+        return engine
+
+    def close(self):
+        pass
+
+
+def group_by_item_type(items):
+    result = collections.defaultdict(list)
+    for item in items:
+        result[item.get('type')].append(item)
+
+    return result