ci: Run mypy in CI

.pre-commit-config.yaml

@@ -1,32 +1,25 @@
 repos:
 - repo: https://github.com/pre-commit/pre-commit-hooks
-  rev: v2.3.0
+  rev: v4.5.0
   hooks:
   - id: check-yaml
   - id: end-of-file-fixer
   - id: trailing-whitespace
 - repo: https://github.com/psf/black
-  rev: 22.10.0
+  rev: 23.12.1
   hooks:
   - id: black
 - repo: https://github.com/pycqa/isort
-  rev: 5.12.0
+  rev: 5.13.2
   hooks:
   - id: isort
 - repo: https://github.com/python-poetry/poetry
-  rev: 1.3.2
+  rev: 1.7.0
   hooks:
   - id: poetry-check
   - id: poetry-lock
     args: [--no-update]
-- repo: https://github.com/pre-commit/mirrors-mypy
-  rev: 'v0.991'
-  hooks:
-  - id: mypy
-    exclude: tests
-    additional_dependencies:
-    - types-paramiko
 - repo: https://github.com/pycqa/flake8
-  rev: 6.1.0
+  rev: 7.0.0
   hooks:
-  - id: flake8
+  - id: flake8

README.md

@@ -261,4 +261,4 @@ Note also that using log-based replication will cause the replication key for al
     "*":
       replication_method: LOG_BASED
       replication_key: _sdc_lsn
-    ```
+    ```

log_based/init.sql

@@ -1 +1 @@
-SELECT * FROM pg_create_logical_replication_slot('tappostgres', 'wal2json');
+SELECT * FROM pg_create_logical_replication_slot('tappostgres', 'wal2json');

pyproject.toml

@@ -47,8 +47,10 @@ mypy = "1.8.0"
 pendulum = "~=3.0"
 pre-commit = "^3.0.4"
 pydocstyle = "^6.1.1"
-singer-sdk = {version = "*", extras = ["testing"]}
+singer-sdk = { version = "*", extras = ["testing"] }
+sqlalchemy = { version = "<2", extras = ["mypy"] }
 tox = "^4"
+types-paramiko = "^3.3.0.2"
 
 [tool.isort]
 profile = "black"
@@ -57,13 +59,14 @@ src_paths = "tap_postgres"
 
 [tool.mypy]
 exclude = "tests"
-python_version = "3.9"
+python_version = "3.11"
 warn_unused_configs = true
 warn_unused_ignores = true
 
 [[tool.mypy.overrides]]
 ignore_missing_imports = true
 module = [
+    "psycopg2",
     "sshtunnel",
 ]
 

tap_postgres/client.py

@@ -117,7 +117,7 @@ def __init__(
 
     # Note super is static, we can get away with this because this is called once
     # and is luckily referenced via the instance of the class
-    def to_jsonschema_type(
+    def to_jsonschema_type(  # type: ignore[override]
         self,
         sql_type: Union[
             str,
@@ -178,6 +178,7 @@ def sdk_typing_object(
         | th.DateType
         | th.StringType
         | th.BooleanType
+        | th.CustomType
     ):
         """Return the JSON Schema dict that describes the sql type.
 
@@ -209,7 +210,8 @@ def sdk_typing_object(
             | th.IntegerType
             | th.DateType
             | th.StringType
-            | th.BooleanType,
+            | th.BooleanType
+            | th.CustomType,
         ] = {
             "jsonb": th.CustomType(
                 {"type": ["string", "number", "integer", "array", "object", "boolean"]}
@@ -379,7 +381,7 @@ def _increment_stream_state(
                     f"stream(replication method={self.replication_method})"
                 )
                 raise ValueError(msg)
-            treat_as_sorted = self.is_sorted()
+            treat_as_sorted = self.is_sorted
             if not treat_as_sorted and self.state_partitioning_keys is not None:
                 # Streams with custom state partitioning are not resumable.
                 treat_as_sorted = False
@@ -529,6 +531,7 @@ def logical_replication_connection(self):
     # TODO: Make this change upstream in the SDK?
     # I'm not sure if in general SQL databases don't guarantee order of records log
     # replication, but at least Postgres does not.
+    @property
     def is_sorted(self) -> bool:
         """Return True if the stream is sorted by the replication key."""
         return self.replication_method == REPLICATION_INCREMENTAL

tap_postgres/tap.py

@@ -1,4 +1,5 @@
 """Postgres tap class."""
+
 from __future__ import annotations
 
 import atexit
@@ -10,7 +11,7 @@
 from typing import Any, Dict, cast
 
 import paramiko
-from singer_sdk import SQLTap, Stream
+from singer_sdk import SQLTap
 from singer_sdk import typing as th
 from singer_sdk._singerlib import (  # JSON schema typing helpers
     Catalog,
@@ -405,7 +406,7 @@ def connector(self) -> PostgresConnector:
 
         """
         # We mutate this url to use the ssh tunnel if enabled
-        url = make_url(self.get_sqlalchemy_url(config=self.config))
+        url = make_url(self.get_sqlalchemy_url(config=self.config))  # type: ignore[arg-type] # noqa: E501
         ssh_config = self.config.get("ssh_tunnel", {})
 
         if ssh_config.get("enable", False):
@@ -529,7 +530,7 @@ def catalog(self) -> Catalog:  # noqa: C901
             stream_modified = False
             new_stream = copy.deepcopy(stream)
             if new_stream.replication_method == "LOG_BASED":
-                for property in new_stream.schema.properties.values():
+                for property in new_stream.schema.properties.values():  # type: ignore[union-attr] # noqa: E501
                     if "null" not in property.type:
                         if isinstance(property.type, list):
                             property.type.append("null")
@@ -538,29 +539,37 @@ def catalog(self) -> Catalog:  # noqa: C901
                 if new_stream.schema.required:
                     stream_modified = True
                     new_stream.schema.required = None
-                if "_sdc_deleted_at" not in new_stream.schema.properties:
+                if "_sdc_deleted_at" not in new_stream.schema.properties:  # type: ignore[operator] # noqa: E501
                     stream_modified = True
-                    new_stream.schema.properties.update(
-                        {"_sdc_deleted_at": Schema(type=["string", "null"])}
+                    new_stream.schema.properties.update(  # type: ignore[union-attr]
+                        {
+                            "_sdc_deleted_at": Schema(type=["string", "null"]),
+                        },
                     )
                     new_stream.metadata.update(
                         {
                             ("properties", "_sdc_deleted_at"): Metadata(
-                                Metadata.InclusionType.AVAILABLE, True, None
-                            )
-                        }
+                                Metadata.InclusionType.AVAILABLE,
+                                True,
+                                None,
+                            ),
+                        },
                     )
-                if "_sdc_lsn" not in new_stream.schema.properties:
+                if "_sdc_lsn" not in new_stream.schema.properties:  # type: ignore[operator] # noqa: E501
                     stream_modified = True
-                    new_stream.schema.properties.update(
-                        {"_sdc_lsn": Schema(type=["integer", "null"])}
+                    new_stream.schema.properties.update(  # type: ignore[union-attr]
+                        {
+                            "_sdc_lsn": Schema(type=["integer", "null"]),
+                        },
                     )
                     new_stream.metadata.update(
                         {
                             ("properties", "_sdc_lsn"): Metadata(
-                                Metadata.InclusionType.AVAILABLE, True, None
-                            )
-                        }
+                                Metadata.InclusionType.AVAILABLE,
+                                True,
+                                None,
+                            ),
+                        },
                     )
             if stream_modified:
                 modified_streams.append(new_stream.tap_stream_id)
@@ -573,13 +582,13 @@ def catalog(self) -> Catalog:  # noqa: C901
             )
         return new_catalog
 
-    def discover_streams(self) -> list[Stream]:
+    def discover_streams(self) -> list[PostgresStream | PostgresLogBasedStream]:  # type: ignore[override] # noqa: E501
         """Initialize all available streams and return them as a list.
 
         Returns:
             List of discovered Stream objects.
         """
-        streams = []
+        streams: list[PostgresStream | PostgresLogBasedStream] = []
         for catalog_entry in self.catalog_dict["streams"]:
             if catalog_entry["replication_method"] == "LOG_BASED":
                 streams.append(

tests/test_core.py

@@ -11,13 +11,13 @@
 from singer_sdk.testing.runners import TapTestRunner
 from sqlalchemy import Column, DateTime, Integer, MetaData, Numeric, String, Table, text
 from sqlalchemy.dialects.postgresql import (
+    ARRAY,
     BIGINT,
     DATE,
     JSON,
     JSONB,
     TIME,
     TIMESTAMP,
-    ARRAY,
 )
 from tests.settings import DB_SCHEMA_NAME, DB_SQLALCHEMY_URL
 from tests.test_replication_key import TABLE_NAME, TapTestReplicationKey
@@ -418,13 +418,13 @@ def run_sync_dry_run(self) -> bool:
         return True
 
 
-def test_invalid_python_dates():
+def test_invalid_python_dates():  # noqa: C901
     """Some dates are invalid in python, but valid in Postgres
 
     Check out https://www.psycopg.org/psycopg3/docs/advanced/adapt.html#example-handling-infinity-date
     for more information.
 
-    """
+    """  # noqa: E501
     table_name = "test_invalid_python_dates"
     engine = sqlalchemy.create_engine(SAMPLE_CONFIG["sqlalchemy_url"], future=True)
 

tests/test_log_based.py

@@ -3,9 +3,9 @@
 import sqlalchemy
 from sqlalchemy import Column, MetaData, Table
 from sqlalchemy.dialects.postgresql import BIGINT, TEXT
-from tap_postgres.tap import TapPostgres
 from tests.test_core import PostgresTestRunner
 
+from tap_postgres.tap import TapPostgres
 
 LOG_BASED_CONFIG = {
     "host": "localhost",
@@ -15,6 +15,7 @@
     "database": "postgres",
 }
 
+
 def test_null_append():
     """LOG_BASED syncs failed with string property types. (issue #294).
 
@@ -23,14 +24,16 @@ def test_null_append():
     LOG_BASED replication can still append the "null" option to a property's type.
     """
     table_name = "test_null_append"
-    engine = sqlalchemy.create_engine("postgresql://postgres:postgres@localhost:5434/postgres")
+    engine = sqlalchemy.create_engine(
+        "postgresql://postgres:postgres@localhost:5434/postgres"
+    )
 
     metadata_obj = MetaData()
     table = Table(
         table_name,
         metadata_obj,
-        Column("id", BIGINT, primary_key = True),
-        Column("data", TEXT, nullable = True)
+        Column("id", BIGINT, primary_key=True),
+        Column("data", TEXT, nullable=True),
     )
     with engine.connect() as conn:
         table.drop(conn, checkfirst=True)

tox.ini

@@ -42,7 +42,7 @@ commands =
     poetry run flake8 tap_postgres
     poetry run pydocstyle tap_postgres
     # refer to pyproject.toml for specific settings
-    # poetry run mypy .
+    poetry run mypy .
 
 [flake8]
 ignore = W503