From d983dcaecd1ef9f3821de26f6557227393cc2cd3 Mon Sep 17 00:00:00 2001 From: tronikos Date: Wed, 16 Aug 2023 00:47:18 -0700 Subject: [PATCH] mypy --- .github/workflows/python-package.yml | 7 ++-- README.md | 8 +++-- mypy.ini | 31 ++++++++++++++++ pyproject.toml | 2 +- src/demo.py | 6 ++-- src/opower/opower.py | 53 ++++++++++++++-------------- src/opower/utilities/base.py | 12 ++++--- src/opower/utilities/coned.py | 15 ++++---- src/opower/utilities/evergy.py | 10 +++--- src/opower/utilities/exelon.py | 36 +++++++++++++------ src/opower/utilities/pge.py | 4 +-- src/opower/utilities/pse.py | 10 +++--- tests/test_opower.py | 2 +- 13 files changed, 127 insertions(+), 69 deletions(-) create mode 100644 mypy.ini diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index a959058..1a6de79 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -18,7 +18,7 @@ jobs: strategy: fail-fast: false matrix: - python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"] + python-version: ["3.9", "3.10", "3.11"] steps: - uses: actions/checkout@v3 @@ -30,7 +30,7 @@ jobs: run: | python -m pip install --upgrade pip python -m pip install . - python -m pip install flake8 pytest ruff + python -m pip install flake8 pytest ruff mypy pydantic if [ -f requirements.txt ]; then pip install -r requirements.txt; fi - name: Lint with flake8 run: | @@ -40,6 +40,9 @@ jobs: - name: Lint with ruff run: | ruff . + - name: Static typing with mypy + run: | + mypy --install-types --non-interactive --no-warn-unused-ignores . - name: Test with pytest run: | pytest diff --git a/README.md b/README.md index ebac9c9..22b2c4c 100644 --- a/README.md +++ b/README.md @@ -60,8 +60,12 @@ python -m pip install flake8 ruff flake8 . ruff . --fix -# Run formatter and lint -isort . ; black . ; flake8 . ; ruff . --fix +# Run type checking +python -m pip install mypy pydantic +mypy . + +# Run formatter, lint, and type checking +isort . ; black . ; flake8 . ; ruff . --fix ; mypy . # Run tests python -m pip install pytest diff --git a/mypy.ini b/mypy.ini new file mode 100644 index 0000000..b034465 --- /dev/null +++ b/mypy.ini @@ -0,0 +1,31 @@ +[mypy] +exclude = (venv|build) +python_version = 3.9 +plugins = pydantic.mypy +show_error_codes = true +follow_imports = silent +ignore_missing_imports = true +local_partial_types = true +strict_equality = true +no_implicit_optional = true +warn_incomplete_stub = true +warn_redundant_casts = true +warn_unused_configs = true +warn_unused_ignores = true +enable_error_code = ignore-without-code, redundant-self, truthy-iterable +disable_error_code = annotation-unchecked +extra_checks = false +check_untyped_defs = true +disallow_incomplete_defs = true +disallow_subclassing_any = true +disallow_untyped_calls = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +warn_return_any = true +warn_unreachable = true + +[pydantic-mypy] +init_forbid_extra = true +init_typed = true +warn_required_dynamic_aliases = true +warn_untyped_fields = true diff --git a/pyproject.toml b/pyproject.toml index c042417..f8d0da0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -7,7 +7,7 @@ authors = [ ] description = "A Python library for getting historical and forecasted usage/cost from utilities that use opower.com such as PG&E" readme = "README.md" -requires-python = ">=3.7" +requires-python = ">=3.9" dependencies = [ "aiohttp>=3.8", "arrow>=1.2", diff --git a/src/demo.py b/src/demo.py index ebeaaa5..adc7002 100644 --- a/src/demo.py +++ b/src/demo.py @@ -5,13 +5,14 @@ from datetime import datetime, timedelta from getpass import getpass import logging +from typing import Optional import aiohttp from opower import AggregateType, Opower, ReadResolution, get_supported_utilities -async def _main(): +async def _main() -> None: supported_utilities = [ utility.__name__.lower() for utility in get_supported_utilities(supports_mfa=True) @@ -101,6 +102,7 @@ async def _main(): "end_date=", args.end_date, ) + prev_end: Optional[datetime] = None if args.usage_only: usage_data = await opower.async_get_usage_reads( account, @@ -108,7 +110,6 @@ async def _main(): args.start_date, args.end_date, ) - prev_end = None print( "start_time\tend_time\tconsumption" "\tstart_minus_prev_end\tend_minus_prev_end" @@ -135,7 +136,6 @@ async def _main(): args.start_date, args.end_date, ) - prev_end = None print( "start_time\tend_time\tconsumption\tprovided_cost" "\tstart_minus_prev_end\tend_minus_prev_end" diff --git a/src/opower/opower.py b/src/opower/opower.py index 724ee1b..75647d3 100644 --- a/src/opower/opower.py +++ b/src/opower/opower.py @@ -5,7 +5,7 @@ from enum import Enum import json import logging -from typing import Any, Optional +from typing import Any, Optional, Union from urllib.parse import urlencode import aiohttp @@ -26,7 +26,7 @@ class MeterType(Enum): ELEC = "ELEC" GAS = "GAS" - def __str__(self): + def __str__(self) -> str: """Return the value of the enum.""" return self.value @@ -38,7 +38,7 @@ class UnitOfMeasure(Enum): THERM = "THERM" CCF = "CCF" - def __str__(self): + def __str__(self) -> str: """Return the value of the enum.""" return self.value @@ -53,7 +53,7 @@ class AggregateType(Enum): # Home Assistant only has hourly data in the energy dashboard and # some utilities (e.g. PG&E) claim QUARTER_HOUR but they only provide HOUR. - def __str__(self): + def __str__(self) -> str: """Return the value of the enum.""" return self.value @@ -67,7 +67,7 @@ class ReadResolution(Enum): HALF_HOUR = "HALF_HOUR" QUARTER_HOUR = "QUARTER_HOUR" - def __str__(self): + def __str__(self) -> str: """Return the value of the enum.""" return self.value @@ -144,14 +144,14 @@ class UsageRead: # TODO: remove supports_mfa and accepts_mfa from all files after ConEd is released to Home Assistant -def get_supported_utilities(supports_mfa=False) -> list[type["UtilityBase"]]: +def get_supported_utilities(supports_mfa: bool = False) -> list[type["UtilityBase"]]: """Return a list of all supported utilities.""" return [ cls for cls in UtilityBase.subclasses if supports_mfa or not cls.accepts_mfa() ] -def get_supported_utility_names(supports_mfa=False) -> list[str]: +def get_supported_utility_names(supports_mfa: bool = False) -> list[str]: """Return a sorted list of names of all supported utilities.""" return sorted( [ @@ -184,13 +184,13 @@ def __init__( """Initialize.""" # Note: Do not modify default headers since Home Assistant that uses this library needs to use # a default session for all integrations. Instead specify the headers for each request. - self.session = session + self.session: aiohttp.ClientSession = session self.utility: type[UtilityBase] = _select_utility(utility) - self.username = username - self.password = password - self.optional_mfa_secret = optional_mfa_secret - self.access_token = None - self.customers = [] + self.username: str = username + self.password: str = password + self.optional_mfa_secret: Optional[str] = optional_mfa_secret + self.access_token: Optional[str] = None + self.customers: list[Any] = [] async def async_login(self) -> None: """Login to the utility website and authorize opower.com for access. @@ -310,8 +310,8 @@ async def async_get_cost_reads( self, account: Account, aggregate_type: AggregateType, - start_date: datetime | None = None, - end_date: datetime | None = None, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, usage_only: bool = False, ) -> list[CostRead]: """Get usage and cost data for the selected account in the given date range aggregated by bill/day/hour. @@ -328,7 +328,9 @@ async def async_get_cost_reads( CostRead( start_time=datetime.fromisoformat(read["startTime"]), end_time=datetime.fromisoformat(read["endTime"]), - consumption=read["value"] if "value" in read else read["consumption"]["value"], + consumption=read["value"] + if "value" in read + else read["consumption"]["value"], provided_cost=read.get("providedCost", 0) or 0, ) ) @@ -351,8 +353,8 @@ async def async_get_usage_reads( self, account: Account, aggregate_type: AggregateType, - start_date: datetime | None = None, - end_date: datetime | None = None, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, ) -> list[UsageRead]: """Get usage data for the selected account in the given date range aggregated by bill/day/hour. @@ -377,16 +379,15 @@ async def _async_get_dated_data( self, account: Account, aggregate_type: AggregateType, - start_date: datetime | None = None, - end_date: datetime | None = None, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, usage_only: bool = False, ) -> list[Any]: """Wrap _async_fetch by breaking requests for big date ranges to smaller ones to satisfy opower imposed limits.""" # TODO: remove not None check after a Home Assistant release if ( account.read_resolution is not None - and aggregate_type - not in SUPPORTED_AGGREGATE_TYPES.get(account.read_resolution) + and aggregate_type not in SUPPORTED_AGGREGATE_TYPES[account.read_resolution] ): raise ValueError( f"Requested aggregate_type: {aggregate_type} " @@ -433,8 +434,8 @@ async def _async_fetch( self, account: Account, aggregate_type: AggregateType, - start_date: datetime | arrow.Arrow | None = None, - end_date: datetime | arrow.Arrow | None = None, + start_date: Union[datetime, arrow.Arrow, None] = None, + end_date: Union[datetime, arrow.Arrow, None] = None, usage_only: bool = False, ) -> list[Any]: if usage_only: @@ -476,7 +477,7 @@ async def _async_fetch( result = await resp.json() if DEBUG_LOG_RESPONSE: _LOGGER.debug("Fetched: %s", json.dumps(result, indent=2)) - return result["reads"] + return list(result["reads"]) except ClientResponseError as err: # Ignore server errors for BILL requests # that can happen if end_date is before account activation @@ -484,7 +485,7 @@ async def _async_fetch( return [] raise err - def _get_headers(self): + def _get_headers(self) -> dict[str, str]: headers = {"User-Agent": USER_AGENT} if self.access_token: headers["authorization"] = f"Bearer {self.access_token}" diff --git a/src/opower/utilities/base.py b/src/opower/utilities/base.py index bd1dcc3..9a57ba1 100644 --- a/src/opower/utilities/base.py +++ b/src/opower/utilities/base.py @@ -1,7 +1,7 @@ """Base class that each utility needs to extend.""" -from typing import Optional +from typing import Any, Optional import aiohttp @@ -11,7 +11,7 @@ class UtilityBase: subclasses: list[type["UtilityBase"]] = [] - def __init_subclass__(cls, **kwargs) -> None: + def __init_subclass__(cls, **kwargs: Any) -> None: """Keep track of all subclass implementations.""" super().__init_subclass__(**kwargs) cls.subclasses.append(cls) @@ -35,7 +35,7 @@ def timezone() -> str: raise NotImplementedError @staticmethod - def accepts_mfa() -> str: + def accepts_mfa() -> bool: """Check if Utility implementations supports MFA.""" return False @@ -45,8 +45,10 @@ async def async_login( username: str, password: str, optional_mfa_secret: Optional[str], - ) -> str | None: - """Login to the utility website and authorize opower. + ) -> Optional[str]: + """Login to the utility website. + + Return the Opower access token or None if this function authorizes with Opower in other ways. :raises InvalidAuth: if login information is incorrect """ diff --git a/src/opower/utilities/coned.py b/src/opower/utilities/coned.py index 3776a0d..6a20660 100644 --- a/src/opower/utilities/coned.py +++ b/src/opower/utilities/coned.py @@ -36,7 +36,7 @@ def timezone() -> str: return "America/New_York" @staticmethod - def accepts_mfa() -> str: + def accepts_mfa() -> bool: """Check if Utility implementations supports MFA.""" return True @@ -46,7 +46,7 @@ async def async_login( username: str, password: str, optional_mfa_secret: Optional[str], - ) -> None: + ) -> str: """Login to the utility website.""" # Double-logins are somewhat broken if cookies stay around. # Let's clear everything except device tokens (which allow skipping 2FA) @@ -76,10 +76,12 @@ async def async_login( redirectUrl = result["authRedirectUrl"] else: if result["newDevice"]: - if not result["noMfa"] and not optional_mfa_secret: - raise InvalidAuth("TOTP secret is required for MFA accounts") - if not result["noMfa"]: + if not optional_mfa_secret: + raise InvalidAuth( + "TOTP secret is required for MFA accounts" + ) + mfaCode = TOTP(optional_mfa_secret).now() async with session.post( @@ -101,6 +103,7 @@ async def async_login( else: raise InvalidAuth("Login Failed") + assert redirectUrl async with session.get( redirectUrl, headers={ @@ -116,4 +119,4 @@ async def async_login( headers={"User-Agent": USER_AGENT}, raise_for_status=True, ) as resp: - return await resp.json() + return await resp.text() diff --git a/src/opower/utilities/evergy.py b/src/opower/utilities/evergy.py index 75e1c91..7110911 100644 --- a/src/opower/utilities/evergy.py +++ b/src/opower/utilities/evergy.py @@ -19,9 +19,9 @@ class EvergyLoginParser(HTMLParser): def __init__(self) -> None: """Initialize.""" super().__init__() - self.verification_token = None + self.verification_token: Optional[str] = None - def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None: """Try to extract the verification token from the login input.""" if tag == "input" and ("name", "evrgaf") in attrs: _, token = next(filter(lambda attr: attr[0] == "value", attrs)) @@ -31,7 +31,7 @@ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None class Evergy(UtilityBase): """Evergy.""" - _subdomain = None + _subdomain: Optional[str] = None @staticmethod def name() -> str: @@ -56,7 +56,7 @@ async def async_login( password: str, optional_mfa_secret: Optional[str], ) -> str: - """Login to the utility website and authorize opower.""" + """Login to the utility website.""" login_parser = EvergyLoginParser() async with session.get( @@ -91,7 +91,7 @@ async def async_login( if resp.status == 200: raise InvalidAuth("Username and password failed") - opower_access_token = None + opower_access_token: Optional[str] = None async with session.get( "https://www.evergy.com/api/sso/jwt", diff --git a/src/opower/utilities/exelon.py b/src/opower/utilities/exelon.py index 72b9b89..d1ad2ac 100644 --- a/src/opower/utilities/exelon.py +++ b/src/opower/utilities/exelon.py @@ -16,7 +16,7 @@ class Exelon: """Base class for Exelon subsidiaries.""" - _subdomain = None + _subdomain: Optional[str] = None # Can find the opower.com subdomain using the GetConfiguration endpoint # e.g. https://secure.bge.com/api/Services/MyAccountService.svc/GetConfiguration @@ -35,8 +35,19 @@ def login_domain() -> str: @classmethod def subdomain(cls) -> str: """Return the opower.com subdomain for this utility.""" + assert Exelon._subdomain, "async_login not called" return Exelon._subdomain + @staticmethod + def primary_subdomain() -> str: + """Return the opower.com subdomain for this utility.""" + raise NotImplementedError + + @staticmethod + def secondary_subdomain() -> str: + """Return the opower.com secondary subdomain for this utility.""" + raise NotImplementedError + @classmethod async def async_account( cls, @@ -104,8 +115,11 @@ async def async_login( # policy = "B2C_1A_SignIn" # tenant = "/euazurebge.onmicrosoft.com/B2C_1A_SignIn" # api = "CombinedSigninAndSignup" - settings = json.loads(re.search(r"var SETTINGS = ({.*});", result).group(1)) + settings_match = re.search(r"var SETTINGS = ({.*});", result) + assert settings_match + settings = json.loads(settings_match.group(1)) login_post_domain = resp.real_url.host + assert login_post_domain async with session.post( "https://" @@ -127,10 +141,10 @@ async def async_login( }, raise_for_status=True, ) as resp: - result = json.loads(await resp.text(encoding="utf-8")) + result_json = json.loads(await resp.text(encoding="utf-8")) - if result["status"] != "200": - raise InvalidAuth(result["message"]) + if result_json["status"] != "200": + raise InvalidAuth(result_json["message"]) async with session.get( "https://" @@ -166,11 +180,11 @@ async def async_login( headers={"User-Agent": USER_AGENT}, raise_for_status=True, ) as resp: - result = await resp.json() + result_json = await resp.json() # confirm no account number is set - if result["accountNumber"] is None: - bearer_token = result["token"] + if result_json["accountNumber"] is None: + bearer_token = result_json["token"] # if we don't yet have an account, look one up and set it if account is None: account = await cls.async_account(session, bearer_token) @@ -198,13 +212,13 @@ async def async_login( headers={"User-Agent": USER_AGENT}, raise_for_status=True, ) as resp: - result = await resp.json() + result_json = await resp.json() # If pepco or delmarva, determine if we should use secondary subdomain if cls.login_domain() in ["secure.pepco.com", "secure.delmarva.com"]: # Get the account type & state if account is None: - account = await cls.async_account(session, result["access_token"]) + account = await cls.async_account(session, result_json["access_token"]) isResidential = account["isResidential"] state = account["PremiseInfo"][0]["mainAddress"]["townDetail"][ @@ -220,4 +234,4 @@ async def async_login( _LOGGER.debug("detected exelon subdomain to be: %s", Exelon._subdomain) - return result["access_token"] + return str(result_json["access_token"]) diff --git a/src/opower/utilities/pge.py b/src/opower/utilities/pge.py index b653062..ee60cbb 100644 --- a/src/opower/utilities/pge.py +++ b/src/opower/utilities/pge.py @@ -10,11 +10,11 @@ from .base import UtilityBase -def _get_form_action_url_and_hidden_inputs(html: str): +def _get_form_action_url_and_hidden_inputs(html: str) -> tuple[str, dict[str, str]]: """Return the URL and hidden inputs from the single form in a page.""" match = re.search(r'action="([^"]*)"', html) if not match: - return None, None + return "", {} action_url = match.group(1) inputs = {} for match in re.finditer( diff --git a/src/opower/utilities/pse.py b/src/opower/utilities/pse.py index d00219a..380f83c 100644 --- a/src/opower/utilities/pse.py +++ b/src/opower/utilities/pse.py @@ -17,9 +17,9 @@ class PSELoginParser(HTMLParser): def __init__(self) -> None: """Initialize.""" super().__init__() - self.verification_token = None + self.verification_token: Optional[str] = None - def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None: """Try to extract the verification token from the login input.""" if tag == "input" and ("name", "__RequestVerificationToken") in attrs: _, token = next(filter(lambda attr: attr[0] == "value", attrs)) @@ -34,10 +34,10 @@ class PSEUsageParser(HTMLParser): def __init__(self) -> None: """Initialize.""" super().__init__() - self.opower_access_token = None + self.opower_access_token: Optional[str] = None self._in_inline_script = False - def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + def handle_starttag(self, tag: str, attrs: list[tuple[str, Optional[str]]]) -> None: """Recognizes inline scripts.""" if ( tag == "script" @@ -83,7 +83,7 @@ async def async_login( password: str, optional_mfa_secret: Optional[str], ) -> str: - """Login to the utility website and authorize opower.""" + """Login to the utility website.""" login_parser = PSELoginParser() async with session.get( diff --git a/tests/test_opower.py b/tests/test_opower.py index 42b2c5f..142c411 100644 --- a/tests/test_opower.py +++ b/tests/test_opower.py @@ -1,6 +1,6 @@ """Tests for Opower.""" -def test_dummy(): +def test_dummy() -> None: """Test dummy.""" assert True