chore: linters

Author: angrybayblade

operate/cli.py

@@ -20,7 +20,7 @@
 """Operate app CLI module."""
 
 import os
-import shutil
+import typing as t
 from pathlib import Path
 
 from aea_ledger_ethereum.ethereum import EthereumCrypto
@@ -37,6 +37,7 @@
 from operate.keys import Keys
 from operate.services.manage import Services
 
+
 DEFAULT_HARDHAT_KEY = (
     "0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80"
 ).encode()
@@ -45,7 +46,7 @@
 class App(Resource):
     """App resource."""
 
-    def __init__(self, home: Path) -> None:
+    def __init__(self, home: t.Optional[Path] = None) -> None:
         """Initialize object."""
         super().__init__()
         self._path = (home or (Path.home() / OPERATE)).resolve()
@@ -76,7 +77,7 @@ def make(self) -> None:
             )
 
     @property
-    def json(self) -> None:
+    def json(self) -> dict:
         """Json representation of the app."""
         return {
             "name": "Operate HTTP server",
@@ -98,7 +99,7 @@ def _daemon(
     host: Annotated[str, params.String(help="HTTP server host string")] = "localhost",
     port: Annotated[int, params.Integer(help="HTTP server port")] = 8000,
     home: Annotated[
-        Path, params.Directory(long_flag="--home", help="Home directory")
+        t.Optional[Path], params.Directory(long_flag="--home", help="Home directory")
     ] = None,
 ) -> None:
     """Launch operate daemon."""
@@ -125,28 +126,6 @@ def _daemon(
     )
 
 
-@_operate.command(name="prune")
-def _prune(
-    home: Annotated[
-        Path, params.Directory(long_flag="--home", help="Home directory")
-    ] = None,
-) -> None:
-    """Delete unused/cached data."""
-    app = App(home=home)
-    for service in app.services.json:
-        if service["active"]:
-            continue
-        try:
-            shutil.rmtree(app.services.path / service["hash"])
-            print("Removed service " + service["hash"])
-        except PermissionError:
-            print(
-                "Error removing "
-                + service["hash"]
-                + " please try with admin previledges"
-            )
-
-
 def main() -> None:
     """CLI entry point."""
     run(cli=_operate)

operate/data/__init__.py

@@ -21,4 +21,5 @@
 
 from pathlib import Path
 
+
 DATA_DIR = Path(__file__).parent

operate/data/contracts/uniswap_v2_erc20/contract.py

@@ -26,6 +26,7 @@
 from aea.contracts.base import Contract
 from aea_ledger_ethereum import EthereumApi
 
+
 PUBLIC_ID = PublicId.from_str("valory/uniswap_v2_erc20:0.1.0")
 
 _logger = logging.getLogger(
@@ -197,7 +198,6 @@ def get_transaction_transfer_logs(  # type: ignore  # pylint: disable=too-many-a
         transfer_logs: List = []
 
         if transfer_logs_data:
-
             transfer_logs = cast(
                 List,
                 transfer_logs_data["logs"],

operate/http/__init__.py

@@ -29,6 +29,9 @@
 
 from operate.http.exceptions import NotAllowed, ResourceException
 
+
+# pylint: disable=no-self-use
+
 GenericResource = t.TypeVar("GenericResource")
 PostPayload = t.TypeVar("PostPayload")
 PostResponse = t.TypeVar("PostResponse")
@@ -52,7 +55,7 @@ class Resource(
 ):
     """Web<->Local resource object."""
 
-    _handlers: t.Dict[str, t.Callable[[t.Dict], t.Dict]]
+    _handlers: t.Dict[str, t.Callable]
 
     def __init__(self) -> None:
         """Initialize object."""
@@ -119,7 +122,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> t.Any:
             return
 
         try:
-            handler = self._handlers.get(request.method)
+            handler = self._handlers[request.method]
             try:
                 data = await request.json()
             except json.decoder.JSONDecodeError:
@@ -134,8 +137,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> t.Any:
                 content={"error": e.args[0]},
                 status_code=e.code,
             )
-        except Exception as e:
-            raise
+        except Exception as e:  # pylint: disable=broad-except
             response = JSONResponse(
                 content={"error": str(e)},
                 status_code=500,

operate/keys.py

@@ -20,7 +20,6 @@
 """Keys manager."""
 import json
 import os
-import typing as t
 from pathlib import Path
 
 from aea_ledger_ethereum.ethereum import EthereumCrypto

operate/ledger/__init__.py

@@ -26,6 +26,7 @@
 from operate.ledger.solana import Solana
 from operate.types import ChainType, LedgerType
 
+
 ETHEREUM_RPC = "https://ethereum.publicnode.com"
 GNOSIS_RPC = "https://rpc.gnosischain.com"
 GOERLI_RPC = "https://ethereum-goerli.publicnode.com"
@@ -60,16 +61,20 @@
 
 
 def get_default_rpc(chain: ChainType) -> str:
+    """Get default RPC chain type."""
     return DEFAULT_RPCS.get(chain, ETHEREUM_RPC)
 
 
 def get_ledger_helper_by_chain(rpc: str, chain: ChainType) -> LedgerHelper:
+    """Get ledger helper by chain type."""
     return CHAIN_HELPERS.get(chain, Ethereum)(rpc=rpc)
 
 
 def get_ledger_helper_by_ledger(rpc: str, ledger: LedgerHelper) -> LedgerHelper:
-    return LEDGER_HELPERS.get(ledger, Ethereum)(rpc=rpc)
+    """Get ledger helper by ledger type."""
+    return LEDGER_HELPERS.get(ledger, Ethereum)(rpc=rpc)  # type: ignore
 
 
-def get_currency_denom(chain: ChainType):
+def get_currency_denom(chain: ChainType) -> str:
+    """Get currency denom by chain type."""
     return CURRENCY_DENOMS.get(chain, "Wei")

operate/ledger/base.py

@@ -23,7 +23,7 @@
 from abc import ABC, abstractmethod
 
 
-class LedgerHelper(ABC):
+class LedgerHelper(ABC):  # pylint: disable=too-few-public-methods
     """Base ledger helper."""
 
     api: t.Any

operate/ledger/ethereum.py

@@ -33,6 +33,7 @@ class Ethereum(LedgerHelper):
     api: EthereumApi
 
     def __init__(self, rpc: str) -> None:
+        """Initialize object."""
         super().__init__(rpc)
         self.api = EthereumApi(address=self.rpc)
 

operate/ledger/profiles.py

@@ -21,6 +21,7 @@
 
 from operate.types import ChainType, ContractAddresses
 
+
 CONTRACTS = {
     ChainType.GNOSIS: ContractAddresses(
         {

operate/ledger/solana.py

@@ -28,9 +28,6 @@
 class Solana(LedgerHelper):
     """Solana ledger helper."""
 
-    def __init__(self, rpc: str) -> None:
-        super().__init__(rpc)
-
     def create_key(self) -> t.Dict:
         """Create key."""
         return {

operate/services/manage.py

@@ -47,11 +47,12 @@
 from operate.types import (
     ChainData,
     ConfigurationTemplate,
-    ServicesType,
     ServiceTemplate,
     ServiceType,
+    ServicesType,
 )
 
+
 OPERATE = ".operate"
 CONFIG = "config.json"
 SERVICES = "services"
@@ -83,10 +84,6 @@ def build_dirs(build_dir: Path) -> None:
             continue
 
 
-class GetServices(ServicesType):
-    """Get payload."""
-
-
 class PostServices(ServiceTemplate):
     """Create payload."""
 
@@ -112,7 +109,7 @@ class DeleteServicesResponse(TypedDict):
 
 class Services(
     Resource[
-        GetServices,
+        ServicesType,
         PostServices,
         ServiceType,
         PutServices,
@@ -142,7 +139,7 @@ async def access(
         await resource(scope=scope, receive=receive, send=send)
 
     @property
-    def json(self) -> GetServices:
+    def json(self) -> ServicesType:
         """Returns the list of available services."""
         data = []
         for path in self.path.iterdir():
@@ -279,7 +276,7 @@ def _create(
 
         return service
 
-    def create(self, data: PostServices) -> PostServices:
+    def create(self, data: PostServices) -> ServiceType:
         """Create a service."""
         service = self._create(
             phash=data["hash"],
@@ -326,7 +323,7 @@ def update(self, data: PutServices) -> ServiceType:
 
         # Swap owners on the old safe
         owner, *_ = old.chain_data["instances"]
-        owner_key = self.keys.get(owner).get("private_key")
+        owner_key = str(self.keys.get(owner).get("private_key"))
         ocm.swap(
             service_id=old.chain_data["token"],
             multisig=old.chain_data["multisig"],

operate/services/protocol.py

@@ -48,7 +48,8 @@
 from operate.data.contracts.service_staking_token.contract import (
     ServiceStakingTokenContract,
 )
-from operate.ledger.profiles import CONTRACTS
+from operate.types import ContractAddresses
+
 
 ZERO_ETH = 0
 
@@ -80,7 +81,7 @@ class MultiSendOperation(Enum):
     DELEGATE_CALL = 1
 
 
-def hash_payload_to_hex(
+def hash_payload_to_hex(  # pylint: disable=too-many-arguments,too-many-locals
     safe_tx_hash: str,
     ether_value: int,
     safe_tx_gas: int,
@@ -245,7 +246,9 @@ def stake(
         # since it has the same interface as ERC721 we might want to create
         # a ERC721 contract package
 
-        def _build_approval_tx(*args, **kargs) -> t.Dict:
+        def _build_approval_tx(  # pylint: disable=unused-argument
+            *args: t.Any, **kargs: t.Any
+        ) -> t.Dict:
             return registry_contracts.erc20.get_approve_tx(
                 ledger_api=self.ledger_api,
                 contract_address=service_registry,
@@ -254,15 +257,17 @@ def _build_approval_tx(*args, **kargs) -> t.Dict:
                 amount=service_id,
             )
 
-        setattr(tx_settler, "build", _build_approval_tx)
+        setattr(tx_settler, "build", _build_approval_tx)  # noqa: B010
         tx_settler.transact(
             method=lambda: {},
             contract="",
             kwargs={},
             dry_run=False,
         )
 
-        def _build_staking_tx(*args, **kargs) -> t.Dict:
+        def _build_staking_tx(  # pylint: disable=unused-argument
+            *args: t.Any, **kargs: t.Any
+        ) -> t.Dict:
             return self.ledger_api.build_transaction(
                 contract_instance=self.staking_ctr.get_instance(
                     ledger_api=self.ledger_api,
@@ -276,7 +281,7 @@ def _build_staking_tx(*args, **kargs) -> t.Dict:
                 raise_on_try=True,
             )
 
-        setattr(tx_settler, "build", _build_staking_tx)
+        setattr(tx_settler, "build", _build_staking_tx)  # noqa: B010
         tx_settler.transact(
             method=lambda: {},
             contract="",
@@ -328,7 +333,9 @@ def unstake(self, service_id: int, staking_contract: str) -> None:
             sleep=self.sleep,
         )
 
-        def _build_unstaking_tx(*args, **kargs) -> t.Dict:
+        def _build_unstaking_tx(  # pylint: disable=unused-argument
+            *args: t.Any, **kargs: t.Any
+        ) -> t.Dict:
             return self.ledger_api.build_transaction(
                 contract_instance=self.staking_ctr.get_instance(
                     ledger_api=self.ledger_api,
@@ -342,7 +349,7 @@ def _build_unstaking_tx(*args, **kargs) -> t.Dict:
                 raise_on_try=True,
             )
 
-        setattr(tx_settler, "build", _build_unstaking_tx)
+        setattr(tx_settler, "build", _build_unstaking_tx)  # noqa: B010
         tx_settler.transact(
             method=lambda: {},
             contract="",
@@ -354,7 +361,7 @@ def _build_unstaking_tx(*args, **kargs) -> t.Dict:
 class OnChainManager:
     """On chain service management."""
 
-    def __init__(self, rpc: str, key: Path, contracts: t.Dict) -> None:
+    def __init__(self, rpc: str, key: Path, contracts: ContractAddresses) -> None:
         """On chain manager."""
         self.rpc = rpc
         self.key = key
@@ -427,7 +434,7 @@ def info(self, token_id: int) -> t.Dict:
             instances=instances,
         )
 
-    def mint(
+    def mint(  # pylint: disable=too-many-arguments,too-many-locals
         self,
         package_path: Path,
         agent_id: int,
@@ -437,8 +444,8 @@ def mint(
         nft: Optional[Union[Path, IPFSHash]],
         update_token: t.Optional[int] = None,
         token: t.Optional[str] = None,
-    ):
-        "Mint service."
+    ) -> t.Dict:
+        """Mint service."""
         # TODO: Support for update
         self._patch()
         manager = MintManager(
@@ -538,7 +545,7 @@ def deploy(
                 reuse_multisig=reuse_multisig,
             )
 
-    def swap(
+    def swap(  # pylint: disable=too-many-arguments,too-many-locals
         self,
         service_id: int,
         multisig: str,
@@ -554,7 +561,7 @@ def swap(
         )
         with tempfile.TemporaryDirectory() as temp_dir:
             key_file = Path(temp_dir, "key.txt")
-            key_file.write_text(owner_key)
+            key_file.write_text(owner_key, encoding="utf-8")
             owner_crypto = EthereumCrypto(private_key_path=str(key_file))
         owner_cryptos: list[EthereumCrypto] = [owner_crypto]
         owners = [