diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 83983c50..91ce0e9e 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -15,17 +15,17 @@ jobs: include: - { python-version: "3.11", os: ubuntu-latest, session: "pre-commit" } - { python-version: "3.11", os: ubuntu-latest, session: "safety" } - # - { python-version: "3.11", os: ubuntu-latest, session: "mypy" } - # - { python-version: "3.10", os: ubuntu-latest, session: "mypy" } - # - { python-version: "3.9", os: ubuntu-latest, session: "mypy" } - # - { python-version: "3.8", os: ubuntu-latest, session: "mypy" } + - { python-version: "3.11", os: ubuntu-latest, session: "mypy" } + - { python-version: "3.10", os: ubuntu-latest, session: "mypy" } + - { python-version: "3.9", os: ubuntu-latest, session: "mypy" } + - { python-version: "3.8", os: ubuntu-latest, session: "mypy" } - { python-version: "3.11", os: ubuntu-latest, session: "tests" } - { python-version: "3.10", os: ubuntu-latest, session: "tests" } - { python-version: "3.9", os: ubuntu-latest, session: "tests" } - { python-version: "3.8", os: ubuntu-latest, session: "tests" } - { python-version: "3.11", os: windows-latest, session: "tests" } - { python-version: "3.11", os: macos-latest, session: "tests" } - # - { python-version: "3.11", os: ubuntu-latest, session: "typeguard" } + - { python-version: "3.11", os: ubuntu-latest, session: "typeguard" } - { python-version: "3.11", os: ubuntu-latest, session: "xdoctest" } - { python-version: "3.11", os: ubuntu-latest, session: "docs-build" } diff --git a/mypy.ini b/mypy.ini deleted file mode 100644 index 01202bca..00000000 --- a/mypy.ini +++ /dev/null @@ -1,26 +0,0 @@ -[mypy] -check_untyped_defs = True -disallow_any_generics = True -disallow_incomplete_defs = True -disallow_subclassing_any = True -disallow_untyped_calls = True -disallow_untyped_decorators = True -disallow_untyped_defs = True -no_implicit_optional = True -no_implicit_reexport = True -pretty = True -show_column_numbers = True -show_error_codes = True -show_error_context = True -strict_equality = True -warn_redundant_casts = True -warn_return_any = True -warn_unreachable = True -warn_unused_configs = True -warn_unused_ignores = True - -[mypy-pytest] -ignore_missing_imports = True - -[mypy-tests.*] -disallow_untyped_decorators = False diff --git a/noxfile.py b/noxfile.py index 904edc5f..9c647cdb 100644 --- a/noxfile.py +++ b/noxfile.py @@ -25,9 +25,9 @@ nox.options.sessions = ( "pre-commit", "safety", - # "mypy", + "mypy", "tests", - # "typeguard", + "typeguard", "xdoctest", "docs-build", ) @@ -172,7 +172,7 @@ def docs_build(session: Session) -> None: """Build the documentation.""" args = session.posargs or ["docs", "docs/_build"] session.install(".") - session.install("sphinx", "sphinx-click", "sphinx-rtd-theme") + session.install("sphinx", "sphinx-click") build_dir = Path("docs", "_build") if build_dir.exists(): @@ -186,7 +186,7 @@ def docs(session: Session) -> None: """Build and serve the documentation with live reloading on file changes.""" args = session.posargs or ["--open-browser", "docs", "docs/_build"] session.install(".") - session.install("sphinx", "sphinx-autobuild", "sphinx-click", "sphinx-rtd-theme") + session.install("sphinx", "sphinx-autobuild", "sphinx-click") build_dir = Path("docs", "_build") if build_dir.exists(): diff --git a/poetry.lock b/poetry.lock index 6b308d88..7a27d74a 100644 --- a/poetry.lock +++ b/poetry.lock @@ -712,20 +712,20 @@ smmap = ">=3.0.1,<6" [[package]] name = "gitpython" -version = "3.1.37" +version = "3.1.40" description = "GitPython is a Python library used to interact with Git repositories" optional = false python-versions = ">=3.7" files = [ - {file = "GitPython-3.1.37-py3-none-any.whl", hash = "sha256:5f4c4187de49616d710a77e98ddf17b4782060a1788df441846bddefbb89ab33"}, - {file = "GitPython-3.1.37.tar.gz", hash = "sha256:f9b9ddc0761c125d5780eab2d64be4873fc6817c2899cbcb34b02344bdc7bc54"}, + {file = "GitPython-3.1.40-py3-none-any.whl", hash = "sha256:cf14627d5a8049ffbf49915732e5eddbe8134c3bdb9d476e6182b676fc573f8a"}, + {file = "GitPython-3.1.40.tar.gz", hash = "sha256:22b126e9ffb671fdd0c129796343a02bf67bf2994b35449ffc9321aa755e18a4"}, ] [package.dependencies] gitdb = ">=4.0.1,<5" [package.extras] -test = ["black", "coverage[toml]", "ddt (>=1.1.1,!=1.4.3)", "mypy", "pre-commit", "pytest", "pytest-cov", "pytest-sugar"] +test = ["black", "coverage[toml]", "ddt (>=1.1.1,!=1.4.3)", "mock", "mypy", "pre-commit", "pytest", "pytest-cov", "pytest-instafail", "pytest-subtests", "pytest-sugar"] [[package]] name = "identify" @@ -1435,20 +1435,20 @@ jupyter = ["ipywidgets (>=7.5.1,<9)"] [[package]] name = "ruamel-yaml" -version = "0.17.35" +version = "0.17.36" description = "ruamel.yaml is a YAML parser/emitter that supports roundtrip preservation of comments, seq/map flow style, and map key order" optional = false python-versions = ">=3" files = [ - {file = "ruamel.yaml-0.17.35-py3-none-any.whl", hash = "sha256:b105e3e6fc15b41fdb201ba1b95162ae566a4ef792b9f884c46b4ccc5513a87a"}, - {file = "ruamel.yaml-0.17.35.tar.gz", hash = "sha256:801046a9caacb1b43acc118969b49b96b65e8847f29029563b29ac61d02db61b"}, + {file = "ruamel.yaml-0.17.36-py3-none-any.whl", hash = "sha256:9d0a8ae7050ce0065bab51d4f0b000aa8afa91e8e04d1364bc304491f4f25943"}, + {file = "ruamel.yaml-0.17.36.tar.gz", hash = "sha256:497d9f12c32e3201b12e897f1c7756bea59a7c50461daff66781598c48a7f289"}, ] [package.dependencies] "ruamel.yaml.clib" = {version = ">=0.2.7", markers = "platform_python_implementation == \"CPython\" and python_version < \"3.13\""} [package.extras] -docs = ["ryd"] +docs = ["mercurial (>5.7)", "ryd"] jinja2 = ["ruamel.yaml.jinja2 (>=0.2)"] [[package]] @@ -1788,13 +1788,13 @@ files = [ [[package]] name = "urllib3" -version = "1.26.17" +version = "1.26.18" description = "HTTP library with thread-safe connection pooling, file post, and more." optional = false python-versions = ">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*, !=3.5.*" files = [ - {file = "urllib3-1.26.17-py2.py3-none-any.whl", hash = "sha256:94a757d178c9be92ef5539b8840d48dc9cf1b2709c9d6b588232a055c524458b"}, - {file = "urllib3-1.26.17.tar.gz", hash = "sha256:24d6a242c28d29af46c3fae832c36db3bbebcc533dd1bb549172cd739c82df21"}, + {file = "urllib3-1.26.18-py2.py3-none-any.whl", hash = "sha256:34b97092d7e0a3a8cf7cd10e386f401b3737364026c45e622aa02903dffe0f07"}, + {file = "urllib3-1.26.18.tar.gz", hash = "sha256:f8ecc1bba5667413457c529ab955bf8c67b45db799d159066261719e328580a0"}, ] [package.extras] @@ -1956,4 +1956,4 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p [metadata] lock-version = "2.0" python-versions = "^3.8.18" -content-hash = "333533b5550618d39e562e93ec97e13152504bee40e9e8f5c58bbe4c73b3bfeb" +content-hash = "375962797763381b04d5041c257096882e3ba5af8279ac68034b6291a863e804" diff --git a/pyproject.toml b/pyproject.toml index 2c8acb0f..e510a838 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,7 @@ Changelog = "https://github.com/hacf-fr/freebox-api/releases" [tool.poetry.dependencies] python = "^3.8.18" -urllib3 = "^1.26.17" +urllib3 = "^1.26.18" aiohttp = ">=3,<4" importlib-metadata = {version = ">=3.3,<5.0", python = "<3.12"} @@ -39,7 +39,7 @@ coverage = {extras = ["toml"], version = "^7.2"} safety = "^2.3.5" mypy = "^1.6" typeguard = "^4.1.5" -xdoctest = {extras = ["colors"], version = "^1.6.1"} +xdoctest = {extras = ["colors"], version = "^1.1.1"} sphinx = "^4.2.0" sphinx-autobuild = "^2021.3.14" pre-commit = "^3.5.0" @@ -75,6 +75,12 @@ pretty = true show_column_numbers = true show_error_codes = true show_error_context = true +# TODO: Work on removing that +allow_untyped_defs = true +allow_untyped_calls = true +exclude = [ + '^tests/example\.py$', +] [build-system] requires = ["poetry-core>=1.0.0"] diff --git a/src/freebox_api/__init__.py b/src/freebox_api/__init__.py index b13a41bc..111d8b1d 100644 --- a/src/freebox_api/__init__.py +++ b/src/freebox_api/__init__.py @@ -6,13 +6,13 @@ # importlib.metadata available from Python 3.8 use importlib_metadata for # earlier versions. try: - from importlib.metadata import version, PackageNotFoundError # type: ignore + from importlib.metadata import version, PackageNotFoundError except ImportError: # pragma: no cover from importlib_metadata import version, PackageNotFoundError # type: ignore try: - __version__ = version(__name__) + __version__: str = version(__name__) except PackageNotFoundError: # pragma: no cover __version__ = "unknown" diff --git a/src/freebox_api/access.py b/src/freebox_api/access.py index f62a340a..e23fad45 100644 --- a/src/freebox_api/access.py +++ b/src/freebox_api/access.py @@ -1,8 +1,13 @@ import hmac import json import logging +from typing import Any +from typing import Dict +from typing import Optional from urllib.parse import urljoin +from aiohttp import ClientSession + from freebox_api.exceptions import AuthorizationError from freebox_api.exceptions import HttpRequestError from freebox_api.exceptions import InsufficientPermissionsError @@ -11,30 +16,39 @@ class Access: - def __init__(self, session, base_url, app_token, app_id, http_timeout): + def __init__( + self, + session: ClientSession, + base_url: str, + app_token: str, + app_id: str, + http_timeout: int, + ): self.session = session self.base_url = base_url self.app_token = app_token self.app_id = app_id self.timeout = http_timeout - self.session_token = None - self.session_permissions = None + self.session_token: Optional[str] = None + self.session_permissions: Optional[Dict[str, bool]] = None async def _get_challenge(self, base_url, timeout=10): """ Return challenge from freebox API """ url = urljoin(base_url, "login") - r = await self.session.get(url, timeout=timeout) - resp = await r.json() + resp = await self.session.get(url, timeout=timeout) + resp_data = await resp.json() # raise exception if resp.success != True - if not resp.get("success"): + if not resp_data.get("success"): raise AuthorizationError( - "Getting challenge failed (APIResponse: {})".format(json.dumps(resp)) + "Getting challenge failed (APIResponse: {})".format( + json.dumps(resp_data) + ) ) - return resp["result"]["challenge"] + return resp_data["result"]["challenge"] async def _get_session_token(self, base_url, app_token, app_id, timeout=10): """ @@ -50,17 +64,19 @@ async def _get_session_token(self, base_url, app_token, app_id, timeout=10): url = urljoin(base_url, "login/session/") data = json.dumps({"app_id": app_id, "password": password}) - r = await self.session.post(url, data=data, timeout=timeout) - resp = await r.json() + resp = await self.session.post(url, data=data, timeout=timeout) + resp_data = await resp.json() # raise exception if resp.success != True - if not resp.get("success"): + if not resp_data.get("success"): raise AuthorizationError( - "Starting session failed (APIResponse: {})".format(json.dumps(resp)) + "Starting session failed (APIResponse: {})".format( + json.dumps(resp_data) + ) ) - session_token = resp.get("result").get("session_token") - session_permissions = resp.get("result").get("permissions") + session_token = resp_data["result"].get("session_token") + session_permissions = resp_data["result"].get("permissions") return (session_token, session_permissions) @@ -75,7 +91,7 @@ async def _refresh_session_token(self): self.session_token = session_token self.session_permissions = session_permissions - def _get_headers(self): + def _get_headers(self) -> Dict[str, Optional[str]]: return {"X-Fbx-App-Auth": self.session_token} async def _perform_request(self, verb, end_url, **kwargs): @@ -91,58 +107,64 @@ async def _perform_request(self, verb, end_url, **kwargs): "headers": self._get_headers(), "timeout": self.timeout, } - r = await verb(url, **request_params) + resp = await verb(url, **request_params) # Return response if content is not json - if r.content_type != "application/json": - return r - else: - resp = await r.json() + if resp.content_type != "application/json": + return resp - if resp.get("error_code") in ["auth_required", "invalid_session"]: - logger.debug("Invalid session") - await self._refresh_session_token() - request_params["headers"] = self._get_headers() - r = await verb(url, **request_params) - resp = await r.json() + resp_data = await resp.json() + if resp_data.get("error_code") in ["auth_required", "invalid_session"]: + logger.debug("Invalid session") + await self._refresh_session_token() + request_params["headers"] = self._get_headers() + resp = await verb(url, **request_params) + resp_data = await resp.json() - if not resp["success"]: - err_msg = "Request failed (APIResponse: {})".format(json.dumps(resp)) - if resp.get("error_code") == "insufficient_rights": - raise InsufficientPermissionsError(err_msg) - else: - raise HttpRequestError(err_msg) + if not resp_data["success"]: + err_msg = "Request failed (APIResponse: {})".format(json.dumps(resp_data)) + if resp_data.get("error_code") == "insufficient_rights": + raise InsufficientPermissionsError(err_msg) + raise HttpRequestError(err_msg) - return resp["result"] if "result" in resp else None + return resp_data.get("result") - async def get(self, end_url): + async def get( + self, end_url: str + ) -> Any: # Union[Dict[str, Any], List[Dict[str, Any]]]: """ Send get request and return results """ return await self._perform_request(self.session.get, end_url) - async def post(self, end_url, payload=None): + async def post( + self, end_url: str, payload: Optional[Dict[str, Any]] = None + ) -> Dict[str, Any]: """ Send post request and return results """ - data = json.dumps(payload) if payload is not None else None - return await self._perform_request(self.session.post, end_url, data=data) + data = json.dumps(payload) if payload else None + return await self._perform_request(self.session.post, end_url, data=data) # type: ignore - async def put(self, end_url, payload=None): + async def put( + self, end_url: str, payload: Optional[Dict[str, Any]] = None + ) -> Dict[str, Any]: """ Send post request and return results """ - data = json.dumps(payload) if payload is not None else None - return await self._perform_request(self.session.put, end_url, data=data) + data = json.dumps(payload) if payload else None + return await self._perform_request(self.session.put, end_url, data=data) # type: ignore - async def delete(self, end_url, payload=None): + async def delete( + self, end_url: str, payload: Optional[Dict[str, Any]] = None + ) -> Optional[Dict[str, bool]]: """ Send delete request and return results """ - data = json.dumps(payload) if payload is not None else None - return await self._perform_request(self.session.delete, end_url, data=data) + data = json.dumps(payload) if payload else None + return await self._perform_request(self.session.delete, end_url, data=data) # type: ignore - async def get_permissions(self): + async def get_permissions(self) -> Optional[Dict[str, bool]]: """ Returns the permissions for this session/app. """ diff --git a/src/freebox_api/aiofreepybox.py b/src/freebox_api/aiofreepybox.py index 1da93c59..33fc4a48 100644 --- a/src/freebox_api/aiofreepybox.py +++ b/src/freebox_api/aiofreepybox.py @@ -4,9 +4,15 @@ import os import socket import ssl +from typing import Any +from typing import Dict +from typing import Optional +from typing import Tuple +from typing import Union from urllib.parse import urljoin -import aiohttp +from aiohttp import ClientSession +from aiohttp import TCPConnector import freebox_api from freebox_api.access import Access @@ -40,12 +46,12 @@ from freebox_api.exceptions import NotOpenError # Token file default location -token_filename = "app_auth" # noqa S105 -token_dir = os.path.dirname(os.path.abspath(__file__)) -token_file = os.path.join(token_dir, token_filename) +DEFAULT_TOKEN_FILENAME = "app_auth" # noqa S105 +DEFAULT_TOKEN_DIRECTORY = os.path.dirname(os.path.abspath(__file__)) +DEFAULT_TOKEN_FILE = os.path.join(DEFAULT_TOKEN_DIRECTORY, DEFAULT_TOKEN_FILENAME) # Default application descriptor -app_desc = { +DEFAULT_APP_DESC: Dict[str, str] = { "app_id": "aiofpbx", "app_name": "freebox-api", "app_version": freebox_api.__version__, @@ -57,15 +63,20 @@ class Freepybox: def __init__( - self, app_desc=app_desc, token_file=token_file, api_version="v3", timeout=10 + self, + app_desc: Dict[str, str] = DEFAULT_APP_DESC, + token_file: str = DEFAULT_TOKEN_FILE, + api_version: str = "v3", + timeout: int = 10, ): - self.token_file = token_file - self.api_version = api_version - self.timeout = timeout - self.app_desc = app_desc - self._access = None - - async def open(self, host, port): + self.app_desc: Dict[str, str] = app_desc + self.token_file: str = token_file + self.api_version: str = api_version + self.timeout: int = timeout + self._session: ClientSession + self._access: Access + + async def open(self, host: str, port: str) -> None: """ Open a session to the freebox, get a valid access module and instantiate freebox modules @@ -77,8 +88,8 @@ async def open(self, host, port): ssl_ctx = ssl.create_default_context() ssl_ctx.load_verify_locations(cafile=cert_path) - conn = aiohttp.TCPConnector(ssl_context=ssl_ctx) - self._session = aiohttp.ClientSession(connector=conn) + conn = TCPConnector(ssl_context=ssl_ctx) + self._session = ClientSession(connector=conn) self._access = await self._get_freebox_access( host, port, self.api_version, self.token_file, self.app_desc, self.timeout @@ -111,17 +122,17 @@ async def open(self, host, port): self.upnpav = Upnpav(self._access) self.upnpigd = Upnpigd(self._access) - async def close(self): + async def close(self) -> None: """ Close the freebox session """ - if self._access is None: + if not self._access: raise NotOpenError("Freebox is not open") await self._access.post("login/logout") await self._session.close() - async def get_permissions(self): + async def get_permissions(self) -> Optional[Dict[str, bool]]: """ Returns the permissions for this app. @@ -137,17 +148,22 @@ async def get_permissions(self): """ if self._access: return await self._access.get_permissions() - else: - return None + return None async def _get_freebox_access( - self, host, port, api_version, token_file, app_desc, timeout=10 - ): + self, + host: str, + port: str, + api_version: str, + token_file: str, + app_desc: Dict[str, str], + timeout: int = 10, + ) -> Access: """ Returns an access object used for HTTP requests. """ - base_url = self._get_base_url(host, port, api_version) + base_url: str = self._get_base_url(host, port, api_version) # Read stored application token logger.info("Read application authorization file") @@ -163,7 +179,7 @@ async def _get_freebox_access( # Check the authorization status out_msg_flag = False - status = None + status: Optional[str] = None while status != "granted": status = await self._get_authorization_status( base_url, track_id, timeout @@ -199,7 +215,9 @@ async def _get_freebox_access( return fbx_access - async def _get_authorization_status(self, base_url, track_id, timeout): + async def _get_authorization_status( + self, base_url: str, track_id: int, timeout: int + ) -> str: """ Get authorization status of the application token @@ -210,12 +228,14 @@ async def _get_authorization_status(self, base_url, track_id, timeout): granted: the app_token is valid and can be used to open a session denied: the user denied the authorization request """ - url = urljoin(base_url, "login/authorize/{0}".format(track_id)) - r = await self._session.get(url, timeout=timeout) - resp = await r.json() - return resp["result"]["status"] - - async def _get_app_token(self, base_url, app_desc, timeout=10): + url = urljoin(base_url, f"login/authorize/{track_id}") + resp = await self._session.get(url, timeout=timeout) + resp_data = await resp.json() + return str(resp_data["result"]["status"]) + + async def _get_app_token( + self, base_url: str, app_desc: Dict[str, str], timeout: int = 10 + ) -> Tuple[str, int]: """ Get the application token from the freebox Returns (app_token, track_id) @@ -223,40 +243,48 @@ async def _get_app_token(self, base_url, app_desc, timeout=10): # Get authentification token url = urljoin(base_url, "login/authorize/") data = json.dumps(app_desc) - r = await self._session.post(url, data=data, timeout=timeout) - resp = await r.json() + resp = await self._session.post(url, data=data, timeout=timeout) + resp_data = await resp.json() # raise exception if resp.success != True - if not resp.get("success"): + if not resp_data.get("success"): raise AuthorizationError( - "Authorization failed (APIResponse: {0})".format(json.dumps(resp)) + "Authorization failed (APIResponse: {0})".format(json.dumps(resp_data)) ) - app_token = resp["result"]["app_token"] - track_id = resp["result"]["track_id"] + app_token: str = resp_data["result"]["app_token"] + track_id: int = resp_data["result"]["track_id"] return (app_token, track_id) - def _writefile_app_token(self, app_token, track_id, app_desc, file): + def _writefile_app_token( + self, app_token: str, track_id: int, app_desc: Dict[str, str], token_file: str + ) -> None: """ Store the application token in g_app_auth_file file """ - d = {**app_desc, "app_token": app_token, "track_id": track_id} - - with open(file, "w") as f: - json.dump(d, f) - - def _readfile_app_token(self, file): + file_content: Dict[str, Union[str, int]] = { + **app_desc, + "app_token": app_token, + "track_id": track_id, + } + + with open(token_file, "w") as f: + json.dump(file_content, f) + + def _readfile_app_token( + self, token_file: str + ) -> Union[Tuple[str, int, Dict[str, Any]], Tuple[None, None, None]]: """ Read the application token in the authentication file. Returns (app_token, track_id, app_desc) """ try: - with open(file, "r") as f: + with open(token_file, "r") as f: d = json.load(f) - app_token = d["app_token"] - track_id = d["track_id"] - app_desc = { + app_token: str = d["app_token"] + track_id: int = d["track_id"] + app_desc: Dict[str, str] = { k: d[k] for k in ("app_id", "app_name", "app_version", "device_name") if k in d @@ -266,14 +294,13 @@ def _readfile_app_token(self, file): except FileNotFoundError: return (None, None, None) - def _get_base_url(self, host, port, freebox_api_version): + def _get_base_url(self, host: str, port: str, api_version: str) -> str: """ Returns base url for HTTPS requests - :return: """ - return "https://{0}:{1}/api/{2}/".format(host, port, freebox_api_version) + return f"https://{host}:{port}/api/{api_version}/" - def _is_app_desc_valid(self, app_desc): + def _is_app_desc_valid(self, app_desc: Dict[str, str]) -> bool: """ Check validity of the application descriptor """ diff --git a/src/freebox_api/api/__init__.py b/src/freebox_api/api/__init__.py index a9a2c5b3..78a3ae3a 100644 --- a/src/freebox_api/api/__init__.py +++ b/src/freebox_api/api/__init__.py @@ -1 +1,4 @@ -__all__ = [] +"""Freebox APIs.""" +from typing import List + +__all__: List[str] = [] diff --git a/src/freebox_api/api/airmedia.py b/src/freebox_api/api/airmedia.py index 8f4b0330..22bc9575 100644 --- a/src/freebox_api/api/airmedia.py +++ b/src/freebox_api/api/airmedia.py @@ -1,3 +1,7 @@ +""" +AirMedia API. +https://dev.freebox.fr/sdk/os/airmedia/ +""" from typing import Any from typing import Dict from typing import List @@ -26,11 +30,11 @@ def __init__(self, access: Access) -> None: "position": 0, } - async def get_airmedia_receivers(self) -> Optional[List[Dict[str, Any]]]: + async def get_airmedia_receivers(self) -> List[Dict[str, Any]]: """ Get AirMedia receivers """ - return await self._access.get("airmedia/receivers/") + return await self._access.get("airmedia/receivers/") # type: ignore async def send_airmedia( self, receiver_name: str, airmedia_data: Dict[str, Any] @@ -43,15 +47,15 @@ async def send_airmedia( """ await self._access.post(f"airmedia/receivers/{receiver_name}/", airmedia_data) - async def get_airmedia_configuration(self) -> Optional[Dict[str, bool]]: + async def get_airmedia_configuration(self) -> Dict[str, bool]: """ Get AirMedia configuration """ - return await self._access.get("airmedia/config/") + return await self._access.get("airmedia/config/") # type: ignore async def set_airmedia_configuration( self, airmedia_config: Dict[str, Any] - ) -> Optional[Dict[str, Any]]: + ) -> Dict[str, Any]: """ Set AirMedia configuration @@ -79,6 +83,7 @@ async def update_airmedia_configuration( config: Dict[str, Any] = {} if enabled is not None: config.update({"enabled": enabled}) - if password is not None: + if password: config.update({"password": password}) + return await self.set_airmedia_configuration(config) diff --git a/src/freebox_api/api/call.py b/src/freebox_api/api/call.py index 70226926..9568ce72 100644 --- a/src/freebox_api/api/call.py +++ b/src/freebox_api/api/call.py @@ -1,6 +1,10 @@ -import logging +""" +Call API. +https://dev.freebox.fr/sdk/os/call/ +""" +from typing import Dict -_LOGGER = logging.getLogger(__name__) +from freebox_api.access import Access class Call: @@ -8,24 +12,24 @@ class Call: Call """ - def __init__(self, access): + def __init__(self, access: Access): self._access = access mark_call_log_as_read_data_schema = {"new": False} - async def delete_call_log(self, log_id): + async def delete_call_log(self, log_id: int) -> Dict[str, bool]: """ Delete call log log_id : `int` """ - await self._access.delete(f"call/log/{log_id}") + return await self._access.delete(f"call/log/{log_id}") # type: ignore - async def delete_calls_log(self): + async def delete_calls_log(self) -> None: """ Delete calls log """ - await self._access.delete("call/log/delete_all/") + return await self._access.delete("call/log/delete_all/") # type: ignore async def get_call_log(self, log_id): """ @@ -39,16 +43,6 @@ async def get_calls_log(self): """ return await self._access.get("call/log/") - # TODO: remove - async def get_call_list(self): - """ - Returns the collection of all call entries - """ - _LOGGER.warning( - "Using deprecated get_call_list, please use get_calls_log instead" - ) - return await self.get_calls_log() - async def mark_calls_log_as_read(self): """ Mark calls log as read diff --git a/src/freebox_api/api/connection.py b/src/freebox_api/api/connection.py index 948abd1d..0734b03c 100644 --- a/src/freebox_api/api/connection.py +++ b/src/freebox_api/api/connection.py @@ -1,5 +1,16 @@ +""" +Connection API. +https://dev.freebox.fr/sdk/os/connection/ +""" +from freebox_api.access import Access + + class Connection: - def __init__(self, access): + """ + Connection + """ + + def __init__(self, access: Access): self._access = access lte_configuration_data_schema = {"enabled": True} @@ -40,11 +51,11 @@ async def get_xdsl(self): """ return await self._access.get("connection/xdsl/") - async def remove_connection_logs(self): + async def remove_connection_logs(self) -> None: """ Remove connection logs """ - return await self._access.delete("connection/logs/") + return await self._access.delete("connection/logs/") # type: ignore async def set_config(self, connection_configuration): """ diff --git a/src/freebox_api/api/dhcp.py b/src/freebox_api/api/dhcp.py index 2a5bf8ba..f9d1e122 100644 --- a/src/freebox_api/api/dhcp.py +++ b/src/freebox_api/api/dhcp.py @@ -1,4 +1,9 @@ +""" +DHCP API. +https://dev.freebox.fr/sdk/os/dhcp/ +""" import logging +from typing import Dict from freebox_api.access import Access @@ -7,7 +12,7 @@ class Dhcp: """ - Dhcp + DHCP """ def __init__(self, access: Access): @@ -36,11 +41,11 @@ async def create_dhcp_static_lease(self, static_lease): """ return await self._access.post("dhcp/static_lease/", static_lease) - async def delete_dhcp_static_lease(self, lease_id): + async def delete_dhcp_static_lease(self, lease_id: str) -> Dict[str, bool]: """ Delete dhcp static lease """ - await self._access.delete(f"dhcp/static_lease/{lease_id}") + return await self._access.delete(f"dhcp/static_lease/{lease_id}") # type: ignore async def edit_dhcp_static_lease(self, lease_id, static_lease): """ @@ -83,25 +88,3 @@ async def get_dhcp_static_leases(self): Get the list of DHCP static leases """ return await self._access.get("dhcp/static_lease/") - - # TODO: remove - async def get_dynamic_dhcp_lease(self): - """ - Get the list of DHCP dynamic leases - """ - logger.warning( - "Using deprecated call get_dynamic_dhcp_lease, please use " - "get_dhcp_dynamic_leases instead" - ) - return await self.get_dhcp_dynamic_leases() - - # TODO: remove - async def get_static_dhcp_lease(self): - """ - Get the list of DHCP static leases - """ - logger.warning( - "Using deprecated call get_static_dhcp_lease, please use " - "get_dhcp_static_leases instead" - ) - return await self.get_dhcp_static_leases() diff --git a/src/freebox_api/api/freeplug.py b/src/freebox_api/api/freeplug.py index 7fadb26a..bb77371d 100644 --- a/src/freebox_api/api/freeplug.py +++ b/src/freebox_api/api/freeplug.py @@ -1,5 +1,16 @@ +""" +Freeplug API. +https://dev.freebox.fr/sdk/os/freeplug/ +""" +from freebox_api.access import Access + + class Freeplug: - def __init__(self, access): + """ + Freeplug + """ + + def __init__(self, access: Access): self._access = access async def get_freeplug_networks(self): diff --git a/src/freebox_api/api/fs.py b/src/freebox_api/api/fs.py index e03bd4b8..fda340c7 100644 --- a/src/freebox_api/api/fs.py +++ b/src/freebox_api/api/fs.py @@ -1,19 +1,24 @@ +""" +File System API. +https://dev.freebox.fr/sdk/os/fs/ +""" import base64 import logging import os +from typing import Dict import freebox_api.exceptions - +from freebox_api.access import Access logger = logging.getLogger(__name__) class Fs: """ - Fs + File System """ - def __init__(self, access): + def __init__(self, access: Access): self._access = access self._path = "/" @@ -83,11 +88,11 @@ async def cp(self, copy): """ return await self._access.post("fs/copy/", copy) - async def delete_file_task(self, task_id): + async def delete_file_task(self, task_id: int) -> Dict[str, bool]: """ Delete file task """ - return await self._access.delete(f"fs/tasks/{task_id}") + return await self._access.delete(f"fs/tasks/{task_id}") # type: ignore async def extract_archive(self, extract): """ @@ -150,7 +155,7 @@ async def ls(self): """ List directory """ - return [i["name"] for i in await self.list_file(self._path)] + return [i["name"] for i in await self.list_files(self._path)] async def mkdir(self, create_directory=create_directory_schema): """ diff --git a/src/freebox_api/api/ftp.py b/src/freebox_api/api/ftp.py index ead834b4..5c3c0556 100644 --- a/src/freebox_api/api/ftp.py +++ b/src/freebox_api/api/ftp.py @@ -1,9 +1,13 @@ +""" +FTP API. +https://dev.freebox.fr/sdk/os/ftp/ +""" from freebox_api.access import Access class Ftp: """ - Ftp + FTP """ def __init__(self, access: Access): diff --git a/src/freebox_api/api/fw.py b/src/freebox_api/api/fw.py index 8e462c3b..afd978e2 100644 --- a/src/freebox_api/api/fw.py +++ b/src/freebox_api/api/fw.py @@ -1,5 +1,18 @@ +""" +Port Forwarding API. +https://dev.freebox.fr/sdk/os/nat/#port-forwarding +""" +from typing import Dict + +from freebox_api.access import Access + + class Fw: - def __init__(self, access): + """ + Port Forwarding + """ + + def __init__(self, access: Access): self._access = access ip_proto = ["tcp", "udp"] @@ -27,13 +40,15 @@ async def create_port_forwarding_configuration(self, port_forwarding_config): """ return await self._access.post("fw/redir/", port_forwarding_config) - async def delete_port_forwarding_configuration(self, config_id): + async def delete_port_forwarding_configuration( + self, config_id: int + ) -> Dict[str, bool]: """ Delete port forwarding configuration config_id : `int` """ - await self._access.delete(f"fw/redir/{config_id}") + return await self._access.delete(f"fw/redir/{config_id}") # type: ignore async def get_port_forwarding_configuration(self, redir_id): """ diff --git a/src/freebox_api/api/home.py b/src/freebox_api/api/home.py index 3d8c9d85..8fb03d40 100644 --- a/src/freebox_api/api/home.py +++ b/src/freebox_api/api/home.py @@ -1,7 +1,13 @@ -# Home structure : adapter > node > endpoint +""" +Home API. +No public documentation available yet. +""" +from typing import Dict + from freebox_api.access import Access +# Home structure : adapter > node > endpoint class Home: """ Home @@ -145,11 +151,11 @@ async def get_home_links(self): """ return await self._access.get("home/links") - async def del_home_node(self, node_id): + async def del_home_node(self, node_id: int) -> Dict[str, bool]: """ Delete home node id """ - return await self._access.delete(f"home/nodes/{node_id}") + return await self._access.delete(f"home/nodes/{node_id}") # type: ignore async def get_home_node(self, node_id): """ diff --git a/src/freebox_api/api/lan.py b/src/freebox_api/api/lan.py index 99ab2b10..70ed213f 100644 --- a/src/freebox_api/api/lan.py +++ b/src/freebox_api/api/lan.py @@ -1,9 +1,13 @@ +""" +LAN API. +https://dev.freebox.fr/sdk/os/lan/ +""" from freebox_api.access import Access class Lan: """ - Lan + LAN """ def __init__(self, access: Access): @@ -33,12 +37,6 @@ def __init__(self, access: Access): wol_schema = {"mac": "", "password": ""} - async def delete_lan_host(self, host_id, interface="pub"): - """ - Delete lan host - """ - await self._access.delete(f"lan/browser/{interface}/{host_id}/") - async def get_config(self): """ Get Lan configuration @@ -69,6 +67,12 @@ async def get_host_information(self, host_id, interface="pub"): """ return await self._access.get(f"lan/browser/{interface}/{host_id}") + async def delete_lan_host(self, host_id, interface="pub"): + """ + Delete lan host + """ + await self._access.delete(f"lan/browser/{interface}/{host_id}/") + async def set_host_information( self, host_id, lan_host_data=lan_host_data_schema, interface="pub" ): diff --git a/src/freebox_api/api/lcd.py b/src/freebox_api/api/lcd.py index 1b848eed..469f1b7e 100644 --- a/src/freebox_api/api/lcd.py +++ b/src/freebox_api/api/lcd.py @@ -1,5 +1,16 @@ +""" +LCD API. +https://dev.freebox.fr/sdk/os/lcd/ +""" +from freebox_api.access import Access + + class Lcd: - def __init__(self, access): + """ + LCD + """ + + def __init__(self, access: Access): self._access = access lcd_config_schema = { diff --git a/src/freebox_api/api/netshare.py b/src/freebox_api/api/netshare.py index 65d30b58..e4465233 100644 --- a/src/freebox_api/api/netshare.py +++ b/src/freebox_api/api/netshare.py @@ -1,9 +1,16 @@ +""" +Network Share API. +https://dev.freebox.fr/sdk/os/network_share/ +""" +from freebox_api.access import Access + + class Netshare: """ - Netshare + Network Share """ - def __init__(self, access): + def __init__(self, access: Access): self._access = access server_type = [ diff --git a/src/freebox_api/api/notifications.py b/src/freebox_api/api/notifications.py index 189ca00d..8b328c2c 100644 --- a/src/freebox_api/api/notifications.py +++ b/src/freebox_api/api/notifications.py @@ -1,5 +1,18 @@ +""" +Notification API. +No public documentation available yet. +""" +from typing import Dict + +from freebox_api.access import Access + + class Notifications: - def __init__(self, access): + """ + Notification + """ + + def __init__(self, access: Access): self._access = access os_type = ["android", "ios"] @@ -22,11 +35,11 @@ async def create_notification_target( """ return await self._access.post("notif/targets/", notification_target_data) - async def delete_notification_target(self, target_id): + async def delete_notification_target(self, target_id: str) -> Dict[str, bool]: """ Delete notification target """ - await self._access.delete(f"notif/targets/{target_id}") + return await self._access.delete(f"notif/targets/{target_id}") # type: ignore async def edit_notification_target(self, target_id, notification_target_data): """ diff --git a/src/freebox_api/api/parental.py b/src/freebox_api/api/parental.py index b08dee14..a89744c0 100644 --- a/src/freebox_api/api/parental.py +++ b/src/freebox_api/api/parental.py @@ -1,5 +1,18 @@ +""" +Parental Control API. +https://dev.freebox.fr/sdk/os/parental/ +""" +from typing import Dict + +from freebox_api.access import Access + + class Parental: - def __init__(self, access): + """ + Parental Control + """ + + def __init__(self, access: Access): self._access = access # valid values are: allowed, denied or webonly @@ -13,11 +26,11 @@ async def create_parental_filter(self, parental_filter): """ return await self._access.post("parental/filter/", parental_filter) - async def delete_parental_filter(self, filter_id): + async def delete_parental_filter(self, filter_id: int) -> Dict[str, bool]: """ Delete parental filter """ - return await self._access.delete(f"parental/filter/{filter_id}") + return await self._access.delete(f"parental/filter/{filter_id}") # type: ignore async def edit_parental_filter(self, filter_id, parental_filter): """ diff --git a/src/freebox_api/api/phone.py b/src/freebox_api/api/phone.py index f8e73b24..8a72c5ba 100644 --- a/src/freebox_api/api/phone.py +++ b/src/freebox_api/api/phone.py @@ -1,5 +1,16 @@ +""" +Phone API. +No public documentation available yet. +""" +from freebox_api.access import Access + + class Phone: - def __init__(self, access): + """ + Phone + """ + + def __init__(self, access: Access): self._access = access dect_configuration_schema = {"dect_enabled": True, "dect_registration": True} diff --git a/src/freebox_api/api/player.py b/src/freebox_api/api/player.py index 99ce86c0..7a62c951 100644 --- a/src/freebox_api/api/player.py +++ b/src/freebox_api/api/player.py @@ -1,3 +1,7 @@ +""" +Player API. +No public documentation available yet. +""" from typing import Any from typing import Dict from typing import List @@ -14,14 +18,10 @@ class Player: """ def __init__( - self, access: Access, player_api_version: Optional[str] = None + self, access: Access, player_api_version: str = _DEFAULT_PLAYER_API_VERSION ) -> None: self._access = access - self._player_api_version = ( - _DEFAULT_PLAYER_API_VERSION - if not player_api_version - else player_api_version - ) + self._player_api_version = player_api_version media_control_seek_args = {"seek_position": 0, "type": "seek_position"} media_control_stream = {"quality": "", "source": ""} @@ -52,11 +52,11 @@ def __init__( } media_control_data_schema = {"args": media_control_stream_args, "cmd": "pause"} - async def get_players(self) -> Optional[List[Dict[str, Any]]]: + async def get_players(self) -> List[Dict[str, Any]]: """ Get players """ - return await self._access.get("player") + return await self._access.get("player") # type: ignore async def _get_default_player_id(self) -> int: """ @@ -64,7 +64,7 @@ async def _get_default_player_id(self) -> int: """ players = await self.get_players() - return players[0]["id"] + return int(players[0]["id"]) async def get_player_status( self, player_id: Optional[int] = None @@ -79,7 +79,7 @@ async def get_player_status( if player_id is None: player_id = await self._get_default_player_id() - return await self._access.get( + return await self._access.get( # type: ignore f"player/{player_id}/api/{self._player_api_version}/status/" ) @@ -96,7 +96,7 @@ async def get_player_volume( if player_id is None: player_id = await self._get_default_player_id() - return await self._access.get( + return await self._access.get( # type: ignore f"player/{player_id}/api/{self._player_api_version}/control/volume" ) diff --git a/src/freebox_api/api/remote.py b/src/freebox_api/api/remote.py index 9482090d..885da479 100644 --- a/src/freebox_api/api/remote.py +++ b/src/freebox_api/api/remote.py @@ -1,3 +1,7 @@ +""" +Remote API. +No public documentation available yet. +""" import asyncio from asyncio import TimeoutError as Timeout from typing import Any @@ -5,7 +9,7 @@ from typing import List from typing import Optional -from aiohttp import client_exceptions as cl_ex +from aiohttp import ServerDisconnectedError from freebox_api.access import Access @@ -95,7 +99,7 @@ def build_key( repeat : `int`, optional Default to 0 """ - key_data = {"code": code, "key": key} + key_data: Dict[str, Any] = {"code": code, "key": key} if long_press: key_data["long"] = "True" if repeat: @@ -108,7 +112,7 @@ async def send_key( key: str, long_press: bool = _DEFAULT_LONG_PRESS, repeat: int = _DEFAULT_REPEAT, - ): + ) -> bool: """ Send Key @@ -171,10 +175,10 @@ async def set_key( resp = await self._access.session.get( f"http://{self.player_host}{_REMOTE_CONTROL}", params=self.build_key( - code=key_data.get("code"), - key=key_data.get("key"), - long_press=key_data.get("long"), - repeat=key_data.get("repeat"), + code=key_data.get("code"), # type: ignore + key=key_data.get("key"), # type: ignore + long_press=key_data.get("long"), # type: ignore + repeat=key_data.get("repeat"), # type: ignore ), timeout=_DEFAULT_TIMEOUT, skip_auto_headers=[ @@ -188,7 +192,7 @@ async def set_key( await resp.read() if resp.status == _PL_STATUS and resp.content_length == 0: return True - except (Timeout, cl_ex.ServerDisconnectedError): + except (Timeout, ServerDisconnectedError): pass return False diff --git a/src/freebox_api/api/rrd.py b/src/freebox_api/api/rrd.py index 3c50ca7f..2aea61d4 100644 --- a/src/freebox_api/api/rrd.py +++ b/src/freebox_api/api/rrd.py @@ -1,8 +1,18 @@ +""" +RRD API [UNSTABLE]. +https://dev.freebox.fr/sdk/os/rrd/ +""" import time +from freebox_api.access import Access + class Rrd: - def __init__(self, access): + """ + RRD + """ + + def __init__(self, access: Access): self._access = access db = ["net", "temp", "dsl", "switch"] diff --git a/src/freebox_api/api/storage.py b/src/freebox_api/api/storage.py index b2d3f898..31bdb1a8 100644 --- a/src/freebox_api/api/storage.py +++ b/src/freebox_api/api/storage.py @@ -1,9 +1,16 @@ +""" +Storage API [UNSTABLE]. +https://dev.freebox.fr/sdk/os/storage/ +""" +from freebox_api.access import Access + + class Storage: """ Storage """ - def __init__(self, access): + def __init__(self, access: Access): self._access = access eject_schema = {"state": "disabled"} diff --git a/src/freebox_api/api/switch.py b/src/freebox_api/api/switch.py index db6d805f..c8da132c 100644 --- a/src/freebox_api/api/switch.py +++ b/src/freebox_api/api/switch.py @@ -1,5 +1,16 @@ +""" +Switch API. +https://dev.freebox.fr/sdk/os/switch/ +""" +from freebox_api.access import Access + + class Switch: - def __init__(self, access): + """ + Switch + """ + + def __init__(self, access: Access): self._access = access switch_duplex = ["auto", "full", "half"] diff --git a/src/freebox_api/api/system.py b/src/freebox_api/api/system.py index 8f6bd730..303a6a2a 100644 --- a/src/freebox_api/api/system.py +++ b/src/freebox_api/api/system.py @@ -1,5 +1,16 @@ +""" +System API. +https://dev.freebox.fr/sdk/os/system/ +""" +from freebox_api.access import Access + + class System: - def __init__(self, access): + """ + System + """ + + def __init__(self, access: Access): self._access = access async def get_config(self): diff --git a/src/freebox_api/api/tv.py b/src/freebox_api/api/tv.py index 1bb732a2..03b048a2 100644 --- a/src/freebox_api/api/tv.py +++ b/src/freebox_api/api/tv.py @@ -1,8 +1,20 @@ +""" +PVR API [UNSTABLE]. +PVR = Program Video Recording ? +https://dev.freebox.fr/sdk/os/pvr/ +""" import time +from typing import Dict + +from freebox_api.access import Access class Tv: - def __init__(self, access): + """ + TV + """ + + def __init__(self, access: Access): self._access = access async def archive_tv_record(self, record_id): @@ -23,17 +35,17 @@ async def create_tv_record_generator(self, tv_record_generator): """ return await self._access.post("pvr/generator/", tv_record_generator) - async def delete_finished_tv_record(self, record_id): + async def delete_finished_tv_record(self, record_id: int) -> Dict[str, bool]: """ Delete finished tv record """ - await self._access.delete(f"pvr/finished/{record_id}") + return await self._access.delete(f"pvr/finished/{record_id}") # type: ignore - async def delete_programmed_tv_record(self, record_id): + async def delete_programmed_tv_record(self, record_id: int) -> Dict[str, bool]: """ Delete programmed tv record """ - await self._access.delete(f"pvr/programmed/{record_id}") + return await self._access.delete(f"pvr/programmed/{record_id}") # type: ignore async def delete_tv_record_generator(self, generator_id): """ diff --git a/src/freebox_api/api/upnpav.py b/src/freebox_api/api/upnpav.py index 92483f39..4565e974 100644 --- a/src/freebox_api/api/upnpav.py +++ b/src/freebox_api/api/upnpav.py @@ -1,5 +1,16 @@ +""" +UPnP AV API. +https://dev.freebox.fr/sdk/os/upnpav/ +""" +from freebox_api.access import Access + + class Upnpav: - def __init__(self, access): + """ + UPnP AV + """ + + def __init__(self, access: Access): self._access = access async def get_configuration(self): diff --git a/src/freebox_api/api/upnpigd.py b/src/freebox_api/api/upnpigd.py index 9395abc1..5eb429a0 100644 --- a/src/freebox_api/api/upnpigd.py +++ b/src/freebox_api/api/upnpigd.py @@ -1,12 +1,25 @@ +""" +UPnP IGD API. +https://dev.freebox.fr/sdk/os/igd/ +""" +from typing import Dict + +from freebox_api.access import Access + + class Upnpigd: - def __init__(self, access): + """ + UPnP IGD + """ + + def __init__(self, access: Access): self._access = access - async def delete_redir(self, id): + async def delete_redir(self, id: str) -> Dict[str, bool]: """ Deletes the given upnpigd redirection """ - return await self._access.delete(f"upnpigd/redir/{id}") + return await self._access.delete(f"upnpigd/redir/{id}") # type: ignore async def get_configuration(self): """ diff --git a/src/freebox_api/api/wifi.py b/src/freebox_api/api/wifi.py index 14a8f211..fa083af0 100644 --- a/src/freebox_api/api/wifi.py +++ b/src/freebox_api/api/wifi.py @@ -1,5 +1,18 @@ +""" +Wi-Fi API. +https://dev.freebox.fr/sdk/os/wifi/ +""" +from typing import Dict + +from freebox_api.access import Access + + class Wifi: - def __init__(self, access): + """ + Wi-Fi + """ + + def __init__(self, access: Access): self._access = access # accessType can be full or net_only @@ -41,23 +54,23 @@ async def create_wifi_mac_filter(self, wifi_mac_filter=wifi_mac_filter_schema): """ return await self._access.post("wifi/mac_filter/", wifi_mac_filter) - async def delete_wifi_custom_key(self, key_id): + async def delete_wifi_custom_key(self, key_id: int) -> Dict[str, bool]: """ Delete wifi custom key """ - return await self._access.delete(f"wifi/custom_key/{key_id}") + return await self._access.delete(f"wifi/custom_key/{key_id}") # type: ignore - async def delete_wifi_mac_filter(self, filter_id): + async def delete_wifi_mac_filter(self, filter_id: str) -> Dict[str, bool]: """ Delete wifi mac filter """ - return await self._access.delete(f"wifi/mac_filter/{filter_id}") + return await self._access.delete(f"wifi/mac_filter/{filter_id}") # type: ignore - async def delete_wps_sessions(self): + async def delete_wps_sessions(self) -> Dict[str, bool]: """ Delete wps sessions """ - return await self._access.delete("wifi/wps/sessions") + return await self._access.delete("wifi/wps/sessions") # type: ignore async def edit_wifi_access_point(self, ap_id, wifi_ap_configuration_data): """