Add docker-compose file and bump pylint (#142)

Author: Samira-El

.github/workflows/ci.yaml

@@ -15,20 +15,13 @@ jobs:
       matrix:
         python-version: [ 3.6, 3.7, 3.8 ]
 
-    services:
-      test_postgres:
-        image: postgres:12
-        env:
-          POSTGRES_USER: test_user
-          POSTGRES_PASSWORD: my-secret-passwd
-          POSTGRES_DB: tap_postgres_test
-        ports:
-          - 5432:5432
-
     steps:
       - name: Checkout repository
         uses: actions/checkout@v2
 
+      - name: Build PG DB container
+        run: make start_db
+
       - name: Set up Python ${{ matrix.python-version }}
         uses: actions/setup-python@v2
         with:

Makefile

@@ -1,13 +1,16 @@
 venv:
-	python3 -m venv venv
-	. ./venv/bin/activate
-	pip install --upgrade pip setuptools wheel
+	python3 -m venv venv ;\
+	. ./venv/bin/activate ;\
+	pip install --upgrade pip setuptools wheel;\
 	pip install -e .[test]
 
 pylint:
-	. ./venv/bin/activate
+	. ./venv/bin/activate ;\
 	pylint --rcfile .pylintrc --disable duplicate-code tap_postgres/
 
+start_db:
+	docker-compose up -d --build db
+
 test:
-	. ./venv/bin/activate
+	. ./venv/bin/activate ;\
 	pytest --cov=tap_postgres  --cov-fail-under=59 tests -v

README.md

@@ -31,10 +31,7 @@ It's recommended to use a virtualenv:
 or
 
 ```bash
-  python3 -m venv venv
-  . venv/bin/activate
-  pip install --upgrade pip
-  pip install .
+  make venv
 ```
 
 ### Create a config.json
@@ -168,10 +165,7 @@ to the tap for the next sync.
 
 1. Install python test dependencies in a virtual env:
 ```
-  python3 -m venv venv
-  . venv/bin/activate
-  pip install --upgrade pip
-  pip install .[test]
+ make venv
 ```
 
 2. You need to have a postgres database to run the tests and export its credentials:
@@ -182,6 +176,8 @@ to the tap for the next sync.
   export TAP_POSTGRES_PASSWORD=<postgres-password>
 ```
 
+You can make use of the local docker-compose to spin up a test database by running `make start_db`
+
 Test objects will be created in the `postgres` database.
 
 3. To run the tests:
@@ -193,11 +189,6 @@ Test objects will be created in the `postgres` database.
 
 1. Install python dependencies and run python linter
 ```
-  python3 -m venv venv
-  . venv/bin/activate
-  pip install --upgrade pip
-  pip install .[test]
+  make venv
   make pylint
 ```
-
----

docker-compose.yml

@@ -0,0 +1,13 @@
+version: "3.3"
+
+services:
+  db:
+    image: "debezium/postgres:12-alpine"
+    container_name: ""
+    ports:
+      - "5432:5432"
+    environment:
+      - POSTGRES_USER=test_user
+      - POSTGRES_PASSWORD=my-secret-passwd
+      - POSTGRES_DB=tap_postgres_test
+    command: -c "wal_level=logical" -c "max_replication_slots=5" -c "max_wal_senders=5"

setup.py

@@ -24,7 +24,7 @@
       extras_require={
           "test": [
               'pytest==6.2.5',
-              'pylint==2.10.2',
+              'pylint==2.12.*',
               'pytest-cov==3.0.0'
           ]
       },

tap_postgres/__init__.py

@@ -78,7 +78,7 @@ def do_sync_incremental(conn_config, stream, state, desired_columns, md_map):
     illegal_bk_keys = set(stream_state.keys()).difference(
         {'replication_key', 'replication_key_value', 'version', 'last_replication_method'})
     if len(illegal_bk_keys) != 0:
-        raise Exception("invalid keys found in state: {}".format(illegal_bk_keys))
+        raise Exception(f"invalid keys found in state: {illegal_bk_keys}")
 
     state = singer.write_bookmark(state, stream['tap_stream_id'], 'replication_key', replication_key)
 
@@ -104,7 +104,7 @@ def sync_method_for_streams(streams, state, default_replication_method):
         state = clear_state_on_replication_change(state, stream['tap_stream_id'], replication_key, replication_method)
 
         if replication_method not in {'LOG_BASED', 'FULL_TABLE', 'INCREMENTAL'}:
-            raise Exception("Unrecognized replication_method {}".format(replication_method))
+            raise Exception(f"Unrecognized replication_method {replication_method}")
 
         md_map = metadata.to_map(stream['metadata'])
         desired_columns = [c for c in stream['schema']['properties'].keys() if
@@ -187,7 +187,7 @@ def sync_traditional_stream(conn_config, stream, state, sync_method, end_lsn):
         sync_common.send_schema_message(stream, [])
         state = full_table.sync_table(conn_config, stream, state, desired_columns, md_map)
     else:
-        raise Exception("unknown sync method {} for stream {}".format(sync_method, stream['tap_stream_id']))
+        raise Exception(f"unknown sync method {sync_method} for stream {stream['tap_stream_id']}")
 
     state = singer.set_currently_syncing(state, None)
     singer.write_message(singer.StateMessage(value=copy.deepcopy(state)))
@@ -209,7 +209,7 @@ def sync_logical_streams(conn_config, logical_streams, state, end_lsn, state_fil
         # This is to avoid sending very old starting and flushing positions to source
         selected_streams = set()
         for stream in logical_streams:
-            selected_streams.add("{}".format(stream['tap_stream_id']))
+            selected_streams.add(stream['tap_stream_id'])
 
         new_state = dict(currently_syncing=state['currently_syncing'], bookmarks={})
 
@@ -269,7 +269,7 @@ def register_type_adapters(conn_config):
                 enum_oid = oid[0]
                 psycopg2.extensions.register_type(
                     psycopg2.extensions.new_array_type(
-                        (enum_oid,), 'ENUM_{}[]'.format(enum_oid), psycopg2.STRING))
+                        (enum_oid,), f'ENUM_{enum_oid}[]', psycopg2.STRING))
 
 
 def do_sync(conn_config, catalog, default_replication_method, state, state_file=None):

tap_postgres/db.py

@@ -18,7 +18,7 @@
 
 # pylint: disable=invalid-name,missing-function-docstring
 def calculate_destination_stream_name(stream, md_map):
-    return "{}-{}".format(md_map.get((), {}).get('schema-name'), stream['stream'])
+    return f"{md_map.get((), {}).get('schema-name')}-{stream['stream']}"
 
 
 # from the postgres docs:
@@ -29,13 +29,11 @@ def canonicalize_identifier(identifier):
 
 
 def fully_qualified_column_name(schema, table, column):
-    return '"{}"."{}"."{}"'.format(canonicalize_identifier(schema),
-                                   canonicalize_identifier(table),
-                                   canonicalize_identifier(column))
+    return f'"{canonicalize_identifier(schema)}"."{canonicalize_identifier(table)}"."{canonicalize_identifier(column)}"'
 
 
 def fully_qualified_table_name(schema, table):
-    return '"{}"."{}"'.format(canonicalize_identifier(schema), canonicalize_identifier(table))
+    return f'"{canonicalize_identifier(schema)}"."{canonicalize_identifier(table)}"'
 
 
 def open_connection(conn_config, logical_replication=False):
@@ -60,7 +58,7 @@ def open_connection(conn_config, logical_replication=False):
     return conn
 
 def prepare_columns_for_select_sql(c, md_map):
-    column_name = ' "{}" '.format(canonicalize_identifier(c))
+    column_name = f' "{canonicalize_identifier(c)}" '
 
     if ('properties', c) in md_map:
         sql_datatype = md_map[('properties', c)]['sql-datatype']
@@ -73,17 +71,17 @@ def prepare_columns_for_select_sql(c, md_map):
     return column_name
 
 def prepare_columns_sql(c):
-    column_name = """ "{}" """.format(canonicalize_identifier(c))
+    column_name = f""" "{canonicalize_identifier(c)}" """
     return column_name
 
 
 def filter_dbs_sql_clause(sql, filter_dbs):
-    in_clause = " AND datname in (" + ",".join(["'{}'".format(b.strip(' ')) for b in filter_dbs.split(',')]) + ")"
+    in_clause = " AND datname in (" + ",".join([f"'{b.strip(' ')}'" for b in filter_dbs.split(',')]) + ")"
     return sql + in_clause
 
 
 def filter_schemas_sql_clause(sql, filer_schemas):
-    in_clause = " AND n.nspname in (" + ",".join(["'{}'".format(b.strip(' ')) for b in filer_schemas.split(',')]) + ")"
+    in_clause = " AND n.nspname in (" + ",".join([f"'{b.strip(' ')}'" for b in filer_schemas.split(',')]) + ")"
     return sql + in_clause
 
 
@@ -152,11 +150,10 @@ def selected_value_to_singer_value_impl(elem, sql_datatype):
         if sql_datatype == 'hstore':
             cleaned_elem = elem
         else:
-            raise Exception("do not know how to marshall a dict if its not an hstore or json: {}".format(sql_datatype))
+            raise Exception(f"do not know how to marshall a dict if its not an hstore or json: {sql_datatype}")
     else:
         raise Exception(
-            "do not know how to marshall value of class( {} ) and sql_datatype ( {} )".format(elem.__class__,
-                                                                                              sql_datatype))
+            f"do not know how to marshall value of class( {elem.__class__} ) and sql_datatype ( {sql_datatype} )")
 
     return cleaned_elem
 
@@ -249,7 +246,7 @@ def numeric_min(precision, scale):
 
 
 def filter_tables_sql_clause(sql, tables: List[str]):
-    in_clause = " AND pg_class.relname in (" + ",".join(["'{}'".format(b.strip(' ')) for b in tables]) + ")"
+    in_clause = " AND pg_class.relname in (" + ",".join([f"'{b.strip(' ')}'" for b in tables]) + ")"
     return sql + in_clause
 
 def get_database_name(connection):

tap_postgres/discovery_utils.py

@@ -315,7 +315,7 @@ def schema_for_column(col_info):
         scale = post_db.numeric_scale(col_info)
         precision = post_db.numeric_precision(col_info)
         schema_name = schema_name_for_numeric_array(precision, scale)
-        column_schema['items'] = {'$ref': '#/definitions/{}'.format(schema_name)}
+        column_schema['items'] = {'$ref': f'#/definitions/{schema_name}'}
     elif col_info.sql_data_type == 'double precision[]':
         column_schema['items'] = {'$ref': '#/definitions/sdc_recursive_number_array'}
     elif col_info.sql_data_type == 'hstore[]':
@@ -364,7 +364,7 @@ def nullable_column(col_type, pk):
 
 
 def schema_name_for_numeric_array(precision, scale):
-    schema_name = 'sdc_recursive_decimal_{}_{}_array'.format(precision, scale)
+    schema_name = f'sdc_recursive_decimal_{precision}_{scale}_array'
     return schema_name
 
 
@@ -382,7 +382,7 @@ def include_array_schemas(columns, schema):
                                               'maximum': post_db.numeric_max(precision, scale),
                                               'exclusiveMinimum': True,
                                               'minimum': post_db.numeric_min(precision, scale),
-                                              'items': {'$ref': '#/definitions/{}'.format(schema_name)}}
+                                              'items': {'$ref': f'#/definitions/{schema_name}'}}
 
     return schema
 
@@ -391,7 +391,7 @@ def write_sql_data_type_md(mdata, col_info):
     c_name = col_info.column_name
     if col_info.sql_data_type == 'bit' and col_info.character_maximum_length > 1:
         mdata = metadata.write(mdata, ('properties', c_name),
-                               'sql-datatype', "bit({})".format(col_info.character_maximum_length))
+                               'sql-datatype', f"bit({col_info.character_maximum_length})")
     else:
         mdata = metadata.write(mdata, ('properties', c_name), 'sql-datatype', col_info.sql_data_type)
 

tap_postgres/sync_strategies/full_table.py

@@ -44,9 +44,8 @@ def sync_view(conn_info, stream, state, desired_columns, md_map):
         with post_db.open_connection(conn_info) as conn:
             with conn.cursor(cursor_factory=psycopg2.extras.DictCursor, name='stitch_cursor') as cur:
                 cur.itersize = post_db.CURSOR_ITER_SIZE
-                select_sql = 'SELECT {} FROM {}'.format(','.join(escaped_columns),
-                                                        post_db.fully_qualified_table_name(schema_name,
-                                                                                           stream['table_name']))
+                select_sql = f"SELECT {','.join(escaped_columns)} FROM " \
+                             f"{post_db.fully_qualified_table_name(schema_name,stream['table_name'])}"
 
                 LOGGER.info("select %s with itersize %s", select_sql, cur.itersize)
                 cur.execute(select_sql)
@@ -60,7 +59,7 @@ def sync_view(conn_info, stream, state, desired_columns, md_map):
                                                                             time_extracted,
                                                                             md_map)
                     singer.write_message(record_message)
-                    rows_saved = rows_saved + 1
+                    rows_saved += 1
                     if rows_saved % UPDATE_BOOKMARK_PERIOD == 0:
                         singer.write_message(singer.StateMessage(value=copy.deepcopy(state)))
 
@@ -128,17 +127,15 @@ def sync_table(conn_info, stream, state, desired_columns, md_map):
                 xmin = singer.get_bookmark(state, stream['tap_stream_id'], 'xmin')
                 if xmin:
                     LOGGER.info("Resuming Full Table replication %s from xmin %s", nascent_stream_version, xmin)
-                    select_sql = """SELECT {}, xmin::text::bigint
-                                      FROM {} where age(xmin::xid) <= age('{}'::xid)
-                                     ORDER BY xmin::text ASC""".format(','.join(escaped_columns),
-                                                                       fq_table_name,
-                                                                       xmin)
+                    select_sql = f"""
+                        SELECT {','.join(escaped_columns)}, xmin::text::bigint
+                        FROM {fq_table_name} where age(xmin::xid) <= age('{xmin}'::xid)
+                        ORDER BY xmin::text ASC"""
                 else:
                     LOGGER.info("Beginning new Full Table replication %s", nascent_stream_version)
-                    select_sql = """SELECT {}, xmin::text::bigint
-                                      FROM {}
-                                     ORDER BY xmin::text ASC""".format(','.join(escaped_columns),
-                                                                       fq_table_name)
+                    select_sql = f"""SELECT {','.join(escaped_columns)}, xmin::text::bigint
+                                      FROM {fq_table_name}
+                                     ORDER BY xmin::text ASC"""
 
                 LOGGER.info("select %s with itersize %s", select_sql, cur.itersize)
                 cur.execute(select_sql)
@@ -155,7 +152,7 @@ def sync_table(conn_info, stream, state, desired_columns, md_map):
                                                                             md_map)
                     singer.write_message(record_message)
                     state = singer.write_bookmark(state, stream['tap_stream_id'], 'xmin', xmin)
-                    rows_saved = rows_saved + 1
+                    rows_saved += 1
                     if rows_saved % UPDATE_BOOKMARK_PERIOD == 0:
                         singer.write_message(singer.StateMessage(value=copy.deepcopy(state)))