mypy linting and type checking

.github/workflows/main.yml

@@ -37,3 +37,9 @@ jobs:
 
       - name: Runs tests with coverage
         run: nosetests --with-doctest -v --nocapture
+
+      - name: Installs mypy types where available
+        run: mypy --install-types
+
+      - name: Runs mypy linter and type checks
+        run: mypy --config-file ./mypy.ini singer

.gitignore

@@ -6,6 +6,9 @@ __pycache__/
 # C extensions
 *.so
 
+# Developer IDE settings
+.vscode/settings.json
+
 # Distribution / packaging
 .Python
 env/

mypy.ini

@@ -0,0 +1,17 @@
+[mypy]
+python_version = 3.8
+warn_unused_configs = True
+warn_return_any = True
+exclude = tests
+
+# No typing provided in current version of jsonschema, backoff, and cisco8601.
+# This supresses the related mypy warnings:
+
+[mypy-jsonschema.*]
+ignore_missing_imports = True
+
+[mypy-backoff.*]
+ignore_missing_imports = True
+
+[mypy-ciso8601.*]
+ignore_missing_imports = True

setup.py

@@ -30,6 +30,7 @@
               'ipython',
               'ipdb',
               'nose',
+              'mypy',
               'unify==0.5'
           ]
       },

singer/bookmarks.py

@@ -1,4 +1,8 @@
-def ensure_bookmark_path(state, path):
+from typing import Dict, List, Optional, Any, cast
+
+StateDict = Dict[str, Any]
+
+def ensure_bookmark_path(state: StateDict, path: List[str]) -> StateDict:
     submap = state
     for path_component in path:
         if submap.get(path_component) is None:
@@ -7,40 +11,46 @@ def ensure_bookmark_path(state, path):
         submap = submap[path_component]
     return state
 
-def write_bookmark(state, tap_stream_id, key, val):
+def write_bookmark(
+    state: StateDict, tap_stream_id: str, key: str, val: Any
+) -> StateDict:
     state = ensure_bookmark_path(state, ['bookmarks', tap_stream_id])
     state['bookmarks'][tap_stream_id][key] = val
     return state
 
-def clear_bookmark(state, tap_stream_id, key):
+def clear_bookmark(state: StateDict, tap_stream_id: str, key: str) -> StateDict:
     state = ensure_bookmark_path(state, ['bookmarks', tap_stream_id])
     state['bookmarks'][tap_stream_id].pop(key, None)
     return state
 
-def reset_stream(state, tap_stream_id):
+def reset_stream(state: StateDict, tap_stream_id: str) -> StateDict:
     state = ensure_bookmark_path(state, ['bookmarks', tap_stream_id])
     state['bookmarks'][tap_stream_id] = {}
     return state
 
-def get_bookmark(state, tap_stream_id, key, default=None):
+def get_bookmark(
+    state: StateDict, tap_stream_id: str, key: str, default: Optional[Any] = None
+) -> Optional[Any]:
     return state.get('bookmarks', {}).get(tap_stream_id, {}).get(key, default)
 
-def set_offset(state, tap_stream_id, offset_key, offset_value):
+def set_offset(
+    state: StateDict, tap_stream_id: str, offset_key: Any, offset_value: Any
+):
     state = ensure_bookmark_path(state, ['bookmarks', tap_stream_id, 'offset', offset_key])
     state['bookmarks'][tap_stream_id]['offset'][offset_key] = offset_value
     return state
 
-def clear_offset(state, tap_stream_id):
+def clear_offset(state: StateDict, tap_stream_id: str):
     state = ensure_bookmark_path(state, ['bookmarks', tap_stream_id, 'offset'])
     state['bookmarks'][tap_stream_id]['offset'] = {}
     return state
 
-def get_offset(state, tap_stream_id, default=None):
+def get_offset(state: StateDict, tap_stream_id: str, default: Optional[Any] = None):
     return state.get('bookmarks', {}).get(tap_stream_id, {}).get('offset', default)
 
-def set_currently_syncing(state, tap_stream_id):
+def set_currently_syncing(state: StateDict, tap_stream_id: str):
     state['currently_syncing'] = tap_stream_id
     return state
 
-def get_currently_syncing(state, default=None):
-    return state.get('currently_syncing', default)
+def get_currently_syncing(state: StateDict, default: str = None) -> Optional[str]:
+    return cast(str, state.get('currently_syncing', default))

singer/catalog.py

@@ -1,6 +1,7 @@
 '''Provides an object model for a Singer Catalog.'''
 import orjson
 import sys
+from typing import Iterable, Optional, List, Any, cast
 
 from . import metadata as metadata_module
 from .bookmarks import get_currently_syncing
@@ -10,7 +11,7 @@
 LOGGER = get_logger()
 
 
-def write_catalog(catalog):
+def write_catalog(catalog: "Catalog") -> None:
     # If the catalog has no streams, log a warning
     if not catalog.streams:
         LOGGER.warning('Catalog being written with no streams.')
@@ -22,10 +23,21 @@ def write_catalog(catalog):
 # pylint: disable=too-many-instance-attributes
 class CatalogEntry():
 
-    def __init__(self, tap_stream_id=None, stream=None,
-                 key_properties=None, schema=None, replication_key=None,
-                 is_view=None, database=None, table=None, row_count=None,
-                 stream_alias=None, metadata=None, replication_method=None):
+    def __init__(
+        self,
+        tap_stream_id: Optional[str] = None,
+        stream: Optional[str] = None,
+        key_properties: Optional[List[str]] = None,
+        schema: Optional[Schema] = None,
+        replication_key: Optional[str] = None,
+        is_view: Optional[bool] = None,
+        database: Optional[str] = None,
+        table: Optional[str] = None,
+        row_count: Optional[int] = None,
+        stream_alias: Optional[str] = None,
+        metadata: Optional[dict] = None,
+        replication_method: Optional[str] = None
+    ) -> None:
 
         self.tap_stream_id = tap_stream_id
         self.stream = stream
@@ -83,22 +95,22 @@ def to_dict(self):
 
 class Catalog():
 
-    def __init__(self, streams):
+    def __init__(self, streams: List[CatalogEntry]) -> None:
         self.streams = streams
 
-    def __str__(self):
+    def __str__(self) -> str:
         return str(self.__dict__)
 
-    def __eq__(self, other):
-        return self.__dict__ == other.__dict__
+    def __eq__(self, other: Any) -> bool:
+        return cast(bool, self.__dict__ == other.__dict__)
 
     @classmethod
-    def load(cls, filename):
+    def load(cls, filename: str) -> "Catalog":
         with open(filename, encoding='utf-8') as fp:  # pylint: disable=invalid-name
             return Catalog.from_dict(orjson.loads(fp.read()))
 
     @classmethod
-    def from_dict(cls, data):
+    def from_dict(cls, data: dict) -> "Catalog":
         # TODO: We may want to store streams as a dict where the key is a
         # tap_stream_id and the value is a CatalogEntry. This will allow
         # faster lookup based on tap_stream_id. This would be a breaking
@@ -121,19 +133,19 @@ def from_dict(cls, data):
             streams.append(entry)
         return Catalog(streams)
 
-    def to_dict(self):
+    def to_dict(self) -> dict:
         return {'streams': [stream.to_dict() for stream in self.streams]}
 
-    def dump(self):
+    def dump(self) -> None:
         write_catalog(self)
 
-    def get_stream(self, tap_stream_id):
+    def get_stream(self, tap_stream_id: str) -> Optional[CatalogEntry]:
         for stream in self.streams:
             if stream.tap_stream_id == tap_stream_id:
                 return stream
         return None
 
-    def _shuffle_streams(self, state):
+    def _shuffle_streams(self, state: dict) -> List[CatalogEntry]:
         currently_syncing = get_currently_syncing(state)
 
         if currently_syncing is None:
@@ -149,7 +161,7 @@ def _shuffle_streams(self, state):
         return top_half + bottom_half
 
 
-    def get_selected_streams(self, state):
+    def get_selected_streams(self, state: dict) -> Iterable[CatalogEntry]:
         for stream in self._shuffle_streams(state):
             if not stream.is_selected():
                 LOGGER.info('Skipping stream: %s', stream.tap_stream_id)

singer/logger.py

@@ -3,7 +3,7 @@
 import os
 
 
-def get_logger(name='singer'):
+def get_logger(name: str = 'singer') -> logging.Logger:
     """Return a Logger instance to use in singer."""
     # Use custom logging config provided by environment variable
     if 'LOGGING_CONF_FILE' in os.environ and os.environ['LOGGING_CONF_FILE']: