diff --git a/examples/get_tenant_token.py b/examples/get_tenant_token.py index 20d1157..3824092 100644 --- a/examples/get_tenant_token.py +++ b/examples/get_tenant_token.py @@ -1,6 +1,7 @@ import json import sys from getpass import getpass + from onekey_client import Client API_URL = "https://app.eu.onekey.com/api" @@ -15,7 +16,7 @@ print("Tenants:", ", ".join([tenant.name for tenant in tenants])) -if len(sys.argv) > 2: +if len(sys.argv) > 2: # noqa: PLR2004 (magic constant) # Filter tenants that matches the provided pattern tenants = filter(lambda tenant: sys.argv[2] in tenant.name, tenants) diff --git a/examples/upload_firmware.py b/examples/upload_firmware.py index ef255fa..5309756 100644 --- a/examples/upload_firmware.py +++ b/examples/upload_firmware.py @@ -1,5 +1,4 @@ import sys - from getpass import getpass from pathlib import Path diff --git a/onekey_client/__init__.py b/onekey_client/__init__.py index 5bf88f6..b4afc12 100644 --- a/onekey_client/__init__.py +++ b/onekey_client/__init__.py @@ -1,2 +1,3 @@ from .client import Client as Client -from .models import Tenant as Tenant, FirmwareMetadata as FirmwareMetadata +from .models import FirmwareMetadata as FirmwareMetadata +from .models import Tenant as Tenant diff --git a/onekey_client/cli/ci.py b/onekey_client/cli/ci.py index 4c63ac7..ea4d6f5 100644 --- a/onekey_client/cli/ci.py +++ b/onekey_client/cli/ci.py @@ -1,13 +1,11 @@ import sys import time from pathlib import Path -from typing import Optional from uuid import UUID import click import httpx - -from junit_xml import TestSuite, TestCase +from junit_xml import TestCase, TestSuite from onekey_client import Client from onekey_client.queries import load_query @@ -42,9 +40,7 @@ def get_result(self): except httpx.HTTPError as e: if error_count <= self.retry_count: click.echo( - "Error communicating with ONEKEY platform, retrying; error='{}'".format( - str(e) - ) + f"Error communicating with ONEKEY platform, retrying; error='{e!s}'" ) time.sleep(self.retry_wait * error_count) error_count += 1 @@ -138,7 +134,7 @@ def wait_for_analysis_finish(self): ) break except Exception as e: - click.echo(f"Error fetching results {str(e)}") + click.echo(f"Error fetching results {e!s}") sys.exit(10) def get_recent_firmware_id(self): @@ -156,11 +152,11 @@ def get_recent_firmware_id(self): click.echo( f"Latest firmware upload is not the current firmware, skipping comparison with previous, latest={latest_id}" ) - return + return None if not firmware_ids: click.echo("No previous firmware") - return + return None return firmware_ids[0] @@ -303,10 +299,9 @@ def ci_result( retry_count: int, retry_wait: int, check_interval: int, - junit_path: Optional[Path], + junit_path: Path | None, ): - """Fetch analysis results for CI""" - + """Fetch analysis results for CI.""" handler = ResultHandler( client, firmware_id, diff --git a/onekey_client/cli/cli.py b/onekey_client/cli/cli.py index 06cda79..61eefc9 100644 --- a/onekey_client/cli/cli.py +++ b/onekey_client/cli/cli.py @@ -4,9 +4,10 @@ import httpx from onekey_client import Client -from .firmware_upload import upload_firmware -from .misc import list_tenants, get_tenant_token + from .ci import ci_result +from .firmware_upload import upload_firmware +from .misc import get_tenant_token, list_tenants @click.group() diff --git a/onekey_client/cli/firmware_upload.py b/onekey_client/cli/firmware_upload.py index b498e8e..52a9fba 100644 --- a/onekey_client/cli/firmware_upload.py +++ b/onekey_client/cli/firmware_upload.py @@ -1,10 +1,9 @@ import sys from pathlib import Path -from typing import Optional import click -from onekey_client import FirmwareMetadata, Client +from onekey_client import Client, FirmwareMetadata from onekey_client.errors import QueryError @@ -41,12 +40,11 @@ def upload_firmware( vendor_name: str, product_group_name: str, analysis_configuration_name: str, - version: Optional[str], - name: Optional[str], + version: str | None, + name: str | None, filename: Path, ): - """Uploads a firmware to the ONEKEY platform""" - + """Upload a firmware to the ONEKEY platform.""" product_group_id = _get_product_group_id_by_name(client, product_group_name) analysis_configuration_id = _get_analysis_configuration_id_by_name( client, analysis_configuration_name @@ -73,7 +71,7 @@ def upload_firmware( click.echo(res["id"]) except QueryError as e: click.echo("Error during firmware upload:") - for error in e._errors: + for error in e.errors: click.echo(f"- {error['message']}") sys.exit(11) @@ -86,7 +84,7 @@ def _get_product_group_id_by_name(client: Client, product_group_name: str): except KeyError: click.echo(f"Missing product group: {product_group_name}") click.echo("Available product groups:") - for pg in product_groups.keys(): + for pg in product_groups: click.echo(f"- {pg}") sys.exit(10) @@ -101,6 +99,6 @@ def _get_analysis_configuration_id_by_name( except KeyError: click.echo(f"Missing analysis configuration {analysis_configuration_name}") click.echo("Available analysis configurations:") - for config in analysis_configurations.keys(): + for config in analysis_configurations: click.echo(f"- {config}") sys.exit(12) diff --git a/onekey_client/cli/misc.py b/onekey_client/cli/misc.py index 2b55376..ffe8926 100644 --- a/onekey_client/cli/misc.py +++ b/onekey_client/cli/misc.py @@ -8,8 +8,7 @@ @click.command() @click.pass_obj def list_tenants(client: Client): - """List available tenants""" - + """List available tenants.""" tenants = client.get_all_tenants() for tenant in tenants: click.echo(f"{tenant.name} ({tenant.id}") @@ -18,6 +17,5 @@ def list_tenants(client: Client): @click.command() @click.pass_obj def get_tenant_token(client: Client): - """Get tenant specific Bearer token""" - + """Get tenant specific Bearer token.""" click.echo(json.dumps(client.get_auth_headers())) diff --git a/onekey_client/client.py b/onekey_client/client.py index e618215..26a8a8c 100644 --- a/onekey_client/client.py +++ b/onekey_client/client.py @@ -2,9 +2,6 @@ import gc import secrets from pathlib import Path -from typing import Optional, List, Dict - -from httpx import URL try: from importlib import resources @@ -12,17 +9,17 @@ import importlib_resources as resources import httpx -from pydantic import parse_obj_as -from authlib.oidc.core import IDToken from authlib.jose import jwt -from .queries import load_query -from . import errors -from . import models as m -from . import keys +from authlib.oidc.core import IDToken +from httpx import URL +from pydantic import parse_obj_as +from . import errors, keys +from . import models as m +from .queries import load_query CLIENT_ID = "ONEKEY Python SDK" -TOKEN_NAMESPACE = "https://www.onekey.com/" +TOKEN_NAMESPACE = "https://www.onekey.com/" # noqa: S105 (hardcoded credential) def _login_required(func): @@ -51,8 +48,8 @@ class Client: def __init__( self, api_url: str, - ca_bundle: Optional[Path] = None, - disable_tls_verify: Optional[bool] = False, + ca_bundle: Path | None = None, + disable_tls_verify: bool | None = False, ): self._api_url = URL(api_url) self._client = self._setup_httpx_client(api_url, ca_bundle, disable_tls_verify) @@ -66,11 +63,11 @@ def __init__( def _setup_httpx_client( self, api_url: str, - ca_bundle: Optional[Path] = None, - disable_tls_verify: Optional[bool] = False, + ca_bundle: Path | None = None, + disable_tls_verify: bool | None = False, ): if disable_tls_verify: - return httpx.Client(base_url=api_url, verify=False) + return httpx.Client(base_url=api_url, verify=False) # noqa: S501 (TLS certificate validation disabled) if ca_bundle is not None: ca = ca_bundle.expanduser() @@ -78,17 +75,15 @@ def _setup_httpx_client( raise errors.InvalidCABundle return httpx.Client(base_url=api_url, verify=str(ca)) - else: - with resources.path(keys, "ca.pem") as ca: - return httpx.Client(base_url=api_url, verify=str(ca)) + with resources.path(keys, "ca.pem") as ca: + return httpx.Client(base_url=api_url, verify=str(ca)) - def _load_key(self, key_name: str, path: Optional[Path] = None): + def _load_key(self, key_name: str, path: Path | None = None): if path is not None: return path.read_bytes() - else: - response = self._client.get(f"/{key_name}.pem") - response.raise_for_status() - return response.read() + response = self._client.get(f"/{key_name}.pem") + response.raise_for_status() + return response.read() @property def api_url(self) -> URL: @@ -111,7 +106,7 @@ def login(self, email: str, password: str): claims_cls=IDToken, ) tenants = id_token[TOKEN_NAMESPACE + "tenants"] - tenants = parse_obj_as(List[m.Tenant], tenants) + tenants = parse_obj_as(list[m.Tenant], tenants) self._state.tenants = {e.name: e for e in tenants} self._state.email = email self._state.raw_id_token = json_res["id_token"] @@ -120,7 +115,7 @@ def use_token(self, token: str): try: tenant_id, _ = token.split("/", 1) except ValueError: - raise errors.InvalidAPIToken() + raise errors.InvalidAPIToken from None self._state.raw_tenant_token = token @@ -131,7 +126,7 @@ def use_token(self, token: str): self._state.tenants = {tenant.name: tenant} self._state.tenant = tenant - def _post(self, path: str, headers: Optional[Dict] = None, **kwargs): + def _post(self, path: str, headers: dict | None = None, **kwargs): response = self._client.post(path, headers=headers, **kwargs) response.raise_for_status() return response.json() @@ -152,7 +147,7 @@ def get_tenant(self, name: str): return self._state.tenants[name] @_login_required - def get_all_tenants(self) -> List[m.Tenant]: + def get_all_tenants(self) -> list[m.Tenant]: """Get the list of Tenants you have access to.""" return list(self._state.tenants.values()) @@ -182,8 +177,8 @@ def refresh_tenant_token(self): self.use_tenant(self._state.tenant) @_tenant_required - def query(self, query: str, variables: Optional[Dict] = None, timeout=60): - """Issues a GraphQL query and returns the results""" + def query(self, query: str, variables: dict | None = None, timeout=60): + """Issues a GraphQL query and returns the results.""" res = self._post_with_token( "/graphql", json={"query": query, "variables": variables}, timeout=timeout ) @@ -224,10 +219,9 @@ def upload_firmware( raise errors.QueryError(res["createFirmwareUpload"]["errors"]) upload_url = res["createFirmwareUpload"]["uploadUrl"] - res = self._post_with_token( + return self._post_with_token( upload_url, files={"firmware": path.open("rb")}, timeout=timeout ) - return res @_tenant_required def get_product_groups(self): @@ -271,6 +265,7 @@ def _verify_token( class _LoginState: """Keeps state after login. + Client.logout() will simply delete the instance from memory. """ diff --git a/onekey_client/errors.py b/onekey_client/errors.py index 5eaead9..5647ddf 100644 --- a/onekey_client/errors.py +++ b/onekey_client/errors.py @@ -1,11 +1,10 @@ import json -from typing import Optional class ClientError(Exception): """Base class for all Client errors.""" - def __init__(self, message: Optional[str] = None): + def __init__(self, message: str | None = None): super().__init__(message or self.MESSAGE) @@ -35,7 +34,7 @@ class QueryError(ClientError): """raised when a GraphQL query returns errors.""" def __init__(self, errors_json: dict): - self._errors = errors_json + self.errors = errors_json def __str__(self): - return json.dumps(self._errors, indent=4) + return json.dumps(self.errors, indent=4) diff --git a/onekey_client/models.py b/onekey_client/models.py index 97def41..9342999 100644 --- a/onekey_client/models.py +++ b/onekey_client/models.py @@ -1,6 +1,6 @@ import datetime as dt -from typing import Optional from uuid import UUID + from pydantic import BaseModel @@ -11,11 +11,11 @@ class Tenant(BaseModel): class FirmwareMetadata(BaseModel): name: str - version: Optional[str] = None - release_date: Optional[dt.datetime] = None - notes: Optional[str] = None + version: str | None = None + release_date: dt.datetime | None = None + notes: str | None = None vendor_name: str product_name: str - product_category: Optional[str] = None + product_category: str | None = None product_group_id: UUID analysis_configuration_id: UUID diff --git a/onekey_client/queries/utils.py b/onekey_client/queries/utils.py index f073911..0226dee 100644 --- a/onekey_client/queries/utils.py +++ b/onekey_client/queries/utils.py @@ -1,4 +1,5 @@ import functools + from .. import queries try: @@ -7,7 +8,7 @@ import importlib_resources as resources -@functools.lru_cache() +@functools.lru_cache def load_query(query_name) -> str: """Load a predefined GraphQL query and cache it.""" assert query_name.endswith(".graphql") diff --git a/pyproject.toml b/pyproject.toml index 680a331..b660eea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,3 +38,52 @@ dev = ["ruff==0.14.10"] [build-system] requires = ["hatchling"] build-backend = "hatchling.build" + +[tool.ruff] +target-version = "py310" + +[tool.ruff.lint] +select = [ + "A", # flake8-builtins + "ARG", # flake8-unused-arguments + "B", # flake8-bugbear + "C4", # flake8-comprehensions + "C90", # mccabe + "COM818", # flake8-commas; trailing-comma-on-bare-tuple + "D", # pydocstyle + "E", # pycodestyle (errors) + "F", # pyflakes + "I", # isort + "N", # pep8-naming + "PIE", # flake8-pie + "PL", # pylint + "PT", # flake8-pytest-style + "PTH", # flake8-use-pathlib + "RET", # flake8-return + "RSE", # flake8-raise + "RUF", # ruff's own lints + "S", # flake8-bandit + "SIM", # flake8-simplify + "SLF", # flake8-self + "T10", # flake8-debugger + "T20", # flake8-print + "TC", # flake8-type-checking + "TRY", # tryceratops + "UP", # pyupgrade + "W", # pycodestyle (warnings) +] + +ignore = [ + "D1", # undocumented-*: We are not documenting every public symbol + "D203", # one-blank-line-before-class: D211 (no-blank-line-before-class) is used instead + "D213", # multi-line-summary-second-line: D212 (multi-line-summary-first-line) is used instead + "E501", # line-too-long: Let ruff format handle line length violations + "N818", # error-suffix-on-exception-name: Exception names can be meaningful without smurfs + "PLR09", # too-many-{arguments,branches,...}: We do not want to impose hard limits + "S101", # assert: Enable usage of asserts +] + +[tool.ruff.lint.per-file-ignores] +"examples/*" = [ + "T201", # print +]