Type hints

.github/workflows/build.yml

@@ -3,15 +3,14 @@ on:
     push:
         branches: [main, develop]
     pull_request:
-        branches: [main, develop]
     workflow_dispatch:
 jobs:
   build_python:
     runs-on: ubuntu-latest
     timeout-minutes: 10
     strategy:
       matrix:
-        python-version: ["3.8", "3.9", "3.10", "3.11"]
+        python-version: ["3.9", "3.10", "3.11", "3.12"]
 
     steps:
     - uses: actions/checkout@v2

.gitignore

@@ -7,3 +7,4 @@ pyenv
 dist/
 examples/voting50/
 examples/.ipynb_checkpoints/
+.mypy_cache/

.pre-commit-config.yaml

@@ -8,3 +8,8 @@ repos:
     rev: 6.0.0
     hooks:
       - id: flake8
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.10.0
+    hooks:
+      - id: mypy
+        args: [swissparlpy]

CHANGELOG.md

@@ -4,11 +4,14 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/) and this p
 
 ## [Unreleased]
 ### Changed
-- BC-Break: Python 3.8 is now required
+- BC-Break: Python 3.9 is now required
+- BC-Break: Renamed the callable filter from `swissparlpy.filter` to `swissparlpy.Filter`
 - Switched from `flit` to `uv` to install dependencies, build and publish the package
 
 ### Added
 - Added the plotting functionality from [`ggswissparl`](https://github.com/zumbov2/swissparl#ggswissparl) (Issue #9)
+- Added type hints in the library
+- Add to_dataframe() convenience method to SwissParlResponse to directly transform a result to a DataFrame
 
 ## [0.3.0] - 2023-08-31
 ### Added

README.md

@@ -109,6 +109,8 @@ To create a pandas DataFrame from `get_data` simply pass the return value to the
 
 The `plot_voting` function allows you to visualize voting results of the Swiss National Council according to the seating order, similar to the `ggswissparl` function from the R package.
 
+NOTE: the mapping from seats to persons is currently not historized, so "older" votes might not be displayed correctly. You can provide your own mapping with the `seats` parameter.
+
 **Note**: This feature requires matplotlib. Install with: `pip install swissparlpy[visualization]`
 
 ```python
@@ -217,15 +219,15 @@ print(df[['FirstName', 'LastName']])
 
 You can provide a callable as a filter which allows for more advanced filters.
 
-`swissparlpy.filter` provides `or_` and `and_`.
+`swissparlpy.Filter` provides `or_` and `and_`.
 
 ```python
 import swissparlpy as spp
 import pandas as pd
 
 # filter by FirstName = 'Stefan' OR LastName == 'Seiler'
 def filter_by_name(ent):
-   return spp.filter.or_(
+   return spp.Filter.or_(
       ent.FirstName == 'Stefan',
       ent.LastName == 'Seiler'
    )

examples/filter_advanced.py

@@ -3,7 +3,7 @@
 
 
 def name_filter(e):
-    return spp.filter.or_(e.FirstName == "Stefan", e.LastName == "Seiler")
+    return spp.Filter.or_(e.FirstName == "Stefan", e.LastName == "Seiler")
 
 
 persons = spp.get_data("Person", filter=name_filter, Language="DE")

pyproject.toml

@@ -1,12 +1,8 @@
-[build-system]
-requires = ["hatchling"]
-build-backend = "hatchling.build"
-
 [project]
 name = "swissparlpy"
 authors = [{name = "Stefan Oderbolz", email = "odi@metaodi.ch"}]
 readme = "README.md"
-requires-python = ">=3.8"
+requires-python = ">=3.9"
 classifiers = ["License :: OSI Approved :: MIT License"]
 dynamic = ["version"]
 description = "Client for Swiss parliament API"
@@ -26,18 +22,35 @@ test = [
 ]
 dev = [
     "jupyter",
-    "pandas",
     "black[jupyter]>=24.0.0,<25.0.0",
     "pre-commit",
+    "mypy",
+    "types-requests",
+    "pandas-stubs",
 ]
 visualization = [
     "matplotlib>=3.0.0",
     "pandas",
 ]
 
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
 
 [project.urls]
 Home = "https://github.com/metaodi/swissparlpy"
 
 [tool.hatch.version]
 path = "swissparlpy/__init__.py"
+
+[tool.pytest.ini_options]
+addopts = "--cov=swissparlpy"
+
+[tool.mypy]
+python_version = "3.9"
+ignore_missing_imports = true
+disallow_untyped_defs = true
+check_untyped_defs = true
+warn_return_any = true
+show_error_codes = true
+pretty = true

setup.cfg

@@ -6,7 +6,4 @@ disable_noqa = False
 
 # adaptions for black
 max-line-length = 88
-extend-ignore = E203,W503
-
-[tool:pytest]
-addopts = --cov=swissparlpy
+extend-ignore = E203,W503

swissparlpy/__init__.py

@@ -1,11 +1,13 @@
 """Client for Swiss parliament API"""
 
-__version__ = "0.3.0"
+__version__ = "1.0.0"
 __all__ = ["client", "errors", "visualization"]
 
 from .errors import SwissParlError  # noqa
-from .client import SwissParlClient
-from pyodata.v2.service import GetEntitySetFilter as filter  # noqa
+from .client import SwissParlClient  # noqa
+from .client import SwissParlResponse  # noqa
+from pyodata.v2.service import GetEntitySetFilter as Filter  # noqa
+from typing import Callable, Union, Any
 
 # Import visualization function if matplotlib is available
 try:
@@ -15,26 +17,28 @@
     pass
 
 
-def get_tables():
+def get_tables() -> list[str]:
     client = SwissParlClient()
     return client.get_tables()
 
 
-def get_variables(table):
+def get_variables(table: str) -> list[str]:
     client = SwissParlClient()
     return client.get_variables(table)
 
 
-def get_overview():
+def get_overview() -> dict[str, list[str]]:
     client = SwissParlClient()
     return client.get_overview()
 
 
-def get_glimpse(table, rows=5):
+def get_glimpse(table: str, rows: int = 5) -> "SwissParlResponse":
     client = SwissParlClient()
     return client.get_glimpse(table, rows)
 
 
-def get_data(table, filter=None, **kwargs):  # noqa
+def get_data(
+    table: str, filter: Union[str, Callable, None] = None, **kwargs: Any
+) -> "SwissParlResponse":
     client = SwissParlClient()
     return client.get_data(table, filter, **kwargs)

swissparlpy/client.py

@@ -3,31 +3,42 @@
 import requests
 import pyodata
 from . import errors
+from typing import Optional, Union, Callable, Iterator, Any
+
+try:
+    import pandas as pd
+
+    PANDAS_AVAILABLE = True
+except ImportError:
+    PANDAS_AVAILABLE = False
 
 SERVICE_URL = "https://ws.parlament.ch/odata.svc/"
 log = logging.getLogger(__name__)
 
 
 class SwissParlClient(object):
-    def __init__(self, session=None, url=SERVICE_URL):
+    def __init__(
+        self, session: Optional[requests.Session] = None, url: str = SERVICE_URL
+    ) -> None:
         if not session:
             session = requests.Session()
         self.url = url
         self.client = pyodata.Client(url, session)
-        self.cache = {}
+        self.cache: dict[str, list[str]] = {}
         self.get_overview()
 
-    def get_tables(self):
+    def get_tables(self) -> list[str]:
         if self.cache:
             return list(self.cache.keys())
         return [es.name for es in self.client.schema.entity_sets]
 
-    def get_variables(self, table):
+    def get_variables(self, table: str) -> list[str]:
         if self.cache and table in self.cache:
             return self.cache[table]
-        return [p.name for p in self.client.schema.entity_type(table).proprties()]
+        entity_type = self.client.schema.entity_type(table)
+        return [p.name for p in entity_type.proprties()]
 
-    def get_overview(self):
+    def get_overview(self) -> dict[str, list[str]]:
         log.debug("Load tables and variables from OData...")
         if self.cache:
             return self.cache
@@ -36,50 +47,58 @@ def get_overview(self):
             self.cache[t] = self.get_variables(t)
         return self.cache
 
-    def get_glimpse(self, table, rows=5):
+    def get_glimpse(self, table: str, rows: int = 5) -> "SwissParlResponse":
         entities = self._get_entities(table)
         return SwissParlResponse(
-            entities.top(rows).count(inline=True), self.get_variables(table)
+            entities.top(rows).count(inline=True),  # type: ignore
+            self.get_variables(table),
         )
 
-    def get_data(self, table, filter=None, **kwargs):
+    def get_data(
+        self, table: str, filter: Union[str, Callable, None] = None, **kwargs: Any
+    ) -> "SwissParlResponse":
         entities = self._get_entities(table)
         if filter and callable(filter):
-            entities = entities.filter(filter(entities))
+            entities = entities.filter(filter(entities))  # type: ignore
         elif filter:
-            entities = entities.filter(filter)
+            entities = entities.filter(filter)  # type: ignore
 
         if kwargs:
-            entities = entities.filter(**kwargs)
-        return SwissParlResponse(entities.count(inline=True), self.get_variables(table))
+            entities = entities.filter(**kwargs)  # type: ignore
+        return SwissParlResponse(
+            entities.count(inline=True),  # type: ignore
+            self.get_variables(table),
+        )
 
-    def _get_entities(self, table):
+    def _get_entities(self, table: str) -> object:
         return getattr(self.client.entity_sets, table).get_entities()
 
 
 class SwissParlResponse(object):
-    def __init__(self, entity_request, variables):
+    def __init__(self, entity_request: object, variables: list[str]) -> None:
         self.variables = variables
-        self.data = []
+        self.data: list[SwissParlDataProxy] = []
+        self.count = 0
         self.entity_request = entity_request
         entities = self.load()
         self._parse_data(entities)
 
-    def load(self, next_url=None):
+    def load(self, next_url: Union[str, None] = None) -> object:
         log.debug(f"Load data, next_url={next_url}")
         if next_url:
-            entities = self.entity_request.next_url(next_url).execute()
+            entities = self.entity_request.next_url(next_url).execute()  # type: ignore
         else:
-            entities = self.entity_request.execute()
+            entities = self.entity_request.execute()  # type: ignore
 
         return entities
 
-    def _load_new_data_until(self, limit):
+    def _load_new_data_until(self, limit: int) -> None:
         if limit >= 10000:
             warnings.warn(
                 """
-                More than 10'000 items are loaded, this will use a lot of memory.
-                Consider to query a subset of the data to improve performance.
+                More than 10'000 items are loaded, this will use a lot
+                of memory. Consider to query a subset of the data to
+                improve performance.
                 """,
                 errors.ResultVeryLargeWarning,
             )
@@ -98,25 +117,25 @@ def _load_new_data_until(self, limit):
             except errors.NoMoreRecordsError:
                 break
 
-    def _load_new_data(self):
+    def _load_new_data(self) -> None:
         if self.next_url is None:
             raise errors.NoMoreRecordsError()
         entities = self.load(next_url=self.next_url)
         self._parse_data(entities)
 
-    def _parse_data(self, entities):
-        self.count = entities.total_count
+    def _parse_data(self, entities: object) -> None:
+        self.count = entities.total_count  # type: ignore
         self._setup_proxies(entities)
-        self.next_url = entities.next_url
+        self.next_url = entities.next_url  # type: ignore
 
-    def _setup_proxies(self, entities):
-        for e in entities:
+    def _setup_proxies(self, entities: object) -> None:
+        for e in entities:  # type: ignore
             self.data.append(SwissParlDataProxy(e))
 
-    def __len__(self):
+    def __len__(self) -> int:
         return self.count
 
-    def __iter__(self):
+    def __iter__(self) -> Iterator[dict[str, object]]:
         # use while loop since self.data could grow while iterating
         i = 0
         while True:
@@ -129,7 +148,7 @@ def __iter__(self):
             yield {k: self.data[i](k) for k in self.variables}
             i += 1
 
-    def __getitem__(self, key):
+    def __getitem__(self, key: Union[int, slice]) -> object:
         if isinstance(key, slice):
             limit = max(key.start or 0, key.stop or self.count)
             self._load_new_data_until(limit)
@@ -149,10 +168,20 @@ def __getitem__(self, key):
         self._load_new_data_until(limit)
         return {k: self.data[key](k) for k in self.variables}
 
+    def to_dataframe(self) -> "pd.DataFrame":
+        if not PANDAS_AVAILABLE:
+            raise ImportError(
+                "pandas is not installed. Install it with "
+                "'pip install pandas' to use to_dataframe()."
+            )
+
+        self._load_new_data_until(self.count)
+        return pd.DataFrame(list(self))
+
 
 class SwissParlDataProxy(object):
-    def __init__(self, proxy):
+    def __init__(self, proxy: object) -> None:
         self.proxy = proxy
 
-    def __call__(self, attribute):
+    def __call__(self, attribute: str) -> object:
         return getattr(self.proxy, attribute)