More code updates

app/core/chialisp/tail.py

@@ -2,9 +2,9 @@
 
 from typing import Optional
 
-from chia_rs import G1Element
 from chia.types.blockchain_format.program import Program
 from chia.types.blockchain_format.sized_bytes import bytes32
+from chia_rs import G1Element
 
 from app.core.chialisp.load_clvm import load_clvm_locally
 from app.core.types import GatewayMode

app/core/climate_wallet/wallet.py

@@ -28,6 +28,7 @@
 from chia.wallet.util.compute_memos import compute_memos
 from chia.wallet.util.tx_config import DEFAULT_TX_CONFIG
 from chia.wallet.util.wallet_types import WalletType
+from chia.wallet.wallet_spend_bundle import WalletSpendBundle
 from chia_rs import AugSchemeMPL, G1Element, G2Element, PrivateKey
 
 from app.core.chialisp.gateway import create_gateway_puzzle, parse_gateway_spend
@@ -135,7 +136,7 @@ def has_wallet_client(self) -> bool:
         return self.wallet_client is not None
 
     @property
-    def delegated_signatures(self) -> Dict[Tuple[bytes, bytes], G2Element]:
+    def delegated_signatures(self) -> Dict[Tuple[G1Element, bytes], G2Element]:
         return {
             (self.root_public_key, message): signature
             for (
@@ -188,7 +189,7 @@ async def _create_transaction(
         to_puzzle_hash: Optional[bytes32] = None,
         key_value_pairs: Optional[List[Tuple[Any, Any]]] = None,
         gateway_public_key: Optional[G1Element] = None,
-        public_key_to_secret_key: Optional[Dict[bytes, PrivateKey]] = None,
+        public_key_to_secret_key: Optional[Dict[G1Element, PrivateKey]] = None,
         allow_missing_signature: bool = False,
         wallet_id: int = 1,
     ) -> Dict[str, Any]:
@@ -224,22 +225,22 @@ async def _create_transaction(
             aggregated_signature=signature,
         )
 
-        response = await get_created_signed_transactions(
+        transactions = await get_created_signed_transactions(
             transaction_request=transaction_request,
             wallet_id=wallet_id,
             wallet_client=self.wallet_client,
         )
         new_txs = []
-        for tx in response.transactions:
+        for tx in transactions:
             if unsigned_gateway_coin_spend.coin in tx.additions:
-                spend_bundle = SpendBundle.aggregate(
+                spend_bundle = WalletSpendBundle.aggregate(
                     [gateway_spend_bundle] + ([] if tx.spend_bundle is None else [tx.spend_bundle])
                 )
                 additions = [
                     add for add in tx.additions if add != unsigned_gateway_coin_spend.coin
                 ] + gateway_spend_bundle.additions()
             else:
-                spend_bundle = tx.spend_bundle
+                spend_bundle = WalletSpendBundle.aggregate(([] if tx.spend_bundle is None else [tx.spend_bundle]))
                 additions = tx.additions
             removals = [rem for rem in tx.removals if rem not in additions]
             new_tx = dataclasses.replace(
@@ -303,17 +304,17 @@ async def _create_client_transaction(
                     memos=[],
                 )
             ],
-            fee=0,
+            fee=uint64(0),
         )
-        response = await get_created_signed_transactions(
+        transactions = await get_created_signed_transactions(
             transaction_request=transaction_request,
             wallet_id=wallet_id,
             wallet_client=self.wallet_client,
         )
-        if len(response.transactions) != 1:
-            raise ValueError(f"Transaction record has unexpected length {len(response.transactions)}!")
+        if len(transactions) != 1:
+            raise ValueError(f"Transaction record has unexpected length {len(transactions)}!")
 
-        transaction_record = response.transactions[0]
+        transaction_record = transactions[0]
         if transaction_record.spend_bundle is None:
             raise ValueError("No spend bundle created!")
         coin_spend: CoinSpend = transaction_record.spend_bundle.coin_spends[0]
@@ -461,8 +462,8 @@ async def parse_detokenization_request(
             if puzzle_args is None:
                 continue
 
-            (_, asset_id, inner_puzzle) = puzzle_args
-            asset_id = asset_id.as_atom()
+            (_, asset_id_program, inner_puzzle) = puzzle_args
+            asset_id = asset_id_program.as_atom()
             inner_solution = solution.at("f")
 
             # check for gateway puzzle
@@ -553,10 +554,10 @@ async def sign_and_send_detokenization_request(
         if aggregated_signature == G2Element():
             raise ValueError("Invalid detokenization request!")
 
-        spend_bundle = SpendBundle.aggregate(
+        spend_bundle = WalletSpendBundle.aggregate(
             [
                 unsigned_spend_bundle,
-                SpendBundle(coin_spends=[], aggregated_signature=aggregated_signature),
+                WalletSpendBundle(coin_spends=[], aggregated_signature=aggregated_signature),
             ]
         )
         if gateway_coin_spend is None:
@@ -677,20 +678,20 @@ async def get_activities(
             delegated_solution: Program = tail_solution.at("r")
             key_value_pairs: Program = delegated_solution.at("f")
 
-            metadata: Dict[bytes, bytes] = {}
+            metadata: Dict[str, str] = {}
             for key_value_pair in key_value_pairs.as_iter():
                 if (not key_value_pair.listp()) or (key_value_pair.at("r").listp()):
                     logger.warning(f"Coin {coin.name()} has incorrect metadata structure")
                     continue
 
-                key = key_value_pair.at("f").as_atom()
-                value = key_value_pair.at("r").as_atom()
+                key_bytes = key_value_pair.at("f").as_atom()
+                value_bytes = key_value_pair.at("r").as_atom()
 
-                key = key.decode()
+                key = key_bytes.decode()
                 if key in ["bp"]:
-                    value = f"0x{value.hex()}"
+                    value = f"0x{value_bytes.hex()}"
                 elif key in ["ba", "bn"]:
-                    value = value.decode()
+                    value = value_bytes.decode()
                 else:
                     raise ValueError("Unknown key '{key}'!")
 

app/core/climate_wallet/wallet_utils.py

@@ -98,7 +98,7 @@ def create_gateway_request_and_spend(
     extra_delta: int = 0
     conditions = []
 
-    conditions.append([ConditionOpcode.CREATE_COIN, None, -113, tail_program, tail_solution])
+    conditions.append(Program.to([ConditionOpcode.CREATE_COIN, None, -113, tail_program, tail_solution]))
 
     if to_puzzle_hash is None:
         if mode in [GatewayMode.TOKENIZATION]:
@@ -107,7 +107,7 @@ def create_gateway_request_and_spend(
         extra_delta = -amount
 
     else:
-        conditions.append([ConditionOpcode.CREATE_COIN, to_puzzle_hash, amount, [to_puzzle_hash]])
+        conditions.append(Program.to([ConditionOpcode.CREATE_COIN, to_puzzle_hash, amount, [to_puzzle_hash]]))
 
     conditions_program = Program.to(conditions)
     gateway_announcement = create_gateway_announcement(
@@ -118,7 +118,7 @@ def create_gateway_request_and_spend(
         coins=coins,
         payments=[gateway_payment],
         coin_announcements=[gateway_announcement],
-        fee=fee,
+        fee=uint64(fee),
     )
 
     gateway_solution: Program = create_gateway_solution(

app/crud/chia.py

@@ -6,10 +6,10 @@
 from urllib.parse import urlencode, urlparse
 
 import requests
-from chia_rs import G1Element
 from chia.rpc.full_node_rpc_client import FullNodeRpcClient
 from chia.types.blockchain_format.coin import Coin
 from chia.types.coin_record import CoinRecord
+from chia_rs import G1Element
 from fastapi.encoders import jsonable_encoder
 
 from app import schemas
@@ -46,7 +46,7 @@ def _get_paginated_data(self, path: str, search_params: Dict[str, Any]) -> List[
         Returns:
             A list of all data retrieved from the paginated API.
         """
-        all_data = []
+        all_data: list[Any] = []
         page = 1
         limit = 10
 
@@ -68,16 +68,19 @@ def _get_paginated_data(self, path: str, search_params: Dict[str, Any]) -> List[
                 data = response.json()
                 if data is None:
                     # some cadt endpoints return null with no pagination info if no data is found
-                    # to prevent an infinite loop need to assume that there is no data matching the search from this iteration on
+                    # to prevent an infinite loop need to assume that there is no data matching
+                    # the search from this iteration on
                     return all_data
 
                 try:
-                    if data["page"] and (data["pageCount"] >= 0) and len(data["data"]) >= 0: # page count can be 0 (as of when this was written)
-                        all_data.extend(data["data"]) # Add data from the current page
+                    if (
+                        data["page"] and (data["pageCount"] >= 0) and len(data["data"]) >= 0
+                    ):  # page count can be 0 (as of when this was written)
+                        all_data.extend(data["data"])  # Add data from the current page
                     else:
-                        all_data.append(data) # data was not paginated, append and return
+                        all_data.append(data)  # data was not paginated, append and return
                         return all_data
-                except:
+                except Exception:
                     all_data.append(data)  # data was not paginated, append and return
                     return all_data
 
@@ -192,7 +195,7 @@ def combine_climate_units_and_metadata(self, search: Dict[str, Any]) -> List[Dic
                 warehouse_project_id = unit["issuance"]["warehouseProjectId"]
                 project = project_by_id[warehouse_project_id]
             except (KeyError, TypeError):
-                logger.warning(f"Can not get project by warehouse_project_id")
+                logger.warning("Can not get project by warehouse_project_id")
                 continue
 
             org_metadata = metadata_by_id.get(unit_org_uid)

app/crud/db.py

@@ -63,7 +63,7 @@ def select_first_db(self, model: Any, order_by: Any) -> Any:
             raise errorcode.internal_server_error(message="Select DB Failure")
 
     def select_activity_with_pagination(
-            self, model: Any, filters: Any, order_by: Any, limit: int = None, page: int = None
+        self, model: Any, filters: Any, order_by: Any, limit: Optional[int] = None, page: Optional[int] = None
     ) -> Tuple[Any, int]:
         try:
             query = self.db.query(model).filter(or_(*filters["or"]), and_(*filters["and"]))

app/schemas/__init__.py

@@ -4,9 +4,9 @@
     ActivitiesResponse,
     Activity,
     ActivityBase,
+    ActivityRecordResponse,
     ActivitySearchBy,
     ActivityWithCW,
-    ActivityRecordResponse
 )
 from app.schemas.key import Key  # noqa: F401
 from app.schemas.metadata import (  # noqa: F401

app/schemas/activity.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import enum
-from email.policy import default
 from typing import Any, Dict, List, Optional, Union
 
 from pydantic import Field, validator
@@ -60,5 +59,6 @@ class ActivitiesResponse(BaseModel):
     activities: List[ActivityWithCW] = Field(default_factory=list)
     total: int = 0
 
+
 class ActivityRecordResponse(BaseModel):
-    activity: ActivityWithCW = Field(default=None)
+    activity: Optional[ActivityWithCW] = Field(default=None)

tests/test_cat_lifecycle.py

@@ -59,15 +59,15 @@ async def test_cat_lifecycle(
         melt_secret_key: PrivateKey = AugSchemeMPL.key_gen(secrets.token_bytes(64))
         melt_public_key: G1Element = melt_secret_key.get_g1()
 
-        public_key_to_secret_key: Dict[bytes, PrivateKey] = {
+        public_key_to_secret_key: Dict[G1Element, PrivateKey] = {
             root_public_key: root_secret_key,
             mint_public_key: mint_secret_key,
             melt_public_key: melt_secret_key,
         }
 
         tail_program: Program = create_tail_program(
             public_key=root_public_key,
-            index=Program.to(["registry", "project", "vintage"]).get_tree_hash(),
+            index=Program.to(Program.to(["registry", "project", "vintage"]).get_tree_hash()),
         )
         tail_program_hash: bytes32 = tail_program.get_tree_hash()
 

tests/test_cat_workflow.py

@@ -57,6 +57,7 @@ async def test_cat_tokenization_workflow(
     wallet_2: Wallet = env_2.xch_wallet
 
     fingerprint = (await wallet_client_1.get_logged_in_fingerprint()).fingerprint
+    assert fingerprint is not None
     result = await wallet_client_1.get_private_key(GetPrivateKey(fingerprint=fingerprint))
     root_secret_key: PrivateKey = master_sk_to_root_sk(result.private_key.sk)
 
@@ -71,7 +72,7 @@ async def test_cat_tokenization_workflow(
         root_secret_key=root_secret_key,
         wallet_client=wallet_client_1,
     )
-    result = await climate_wallet_1.send_tokenization_transaction(
+    await climate_wallet_1.send_tokenization_transaction(
         to_puzzle_hash=await wallet_2.get_new_puzzlehash(),
         amount=amount,
         fee=fee,
@@ -151,6 +152,7 @@ async def test_cat_detokenization_workflow(
     wallet_2: Wallet = env_2.xch_wallet
 
     fingerprint = (await wallet_client_1.get_logged_in_fingerprint()).fingerprint
+    assert fingerprint is not None
     result = await wallet_client_1.get_private_key(GetPrivateKey(fingerprint=fingerprint))
     root_secret_key: PrivateKey = master_sk_to_root_sk(result.private_key.sk)
 
@@ -163,7 +165,8 @@ async def test_cat_detokenization_workflow(
         root_secret_key=root_secret_key,
         wallet_client=wallet_client_1,
     )
-    result = await climate_wallet_1.send_tokenization_transaction(
+
+    await climate_wallet_1.send_tokenization_transaction(
         to_puzzle_hash=await wallet_2.get_new_puzzlehash(),
         amount=amount,
         fee=fee,
@@ -220,22 +223,22 @@ async def test_cat_detokenization_workflow(
         wallet_client=wallet_client_2,
         constants=climate_wallet_1.constants,
     )
-    result = await climate_wallet_2.create_detokenization_request(
+    detok_result = await climate_wallet_2.create_detokenization_request(
         amount=amount,
         fee=fee,
         wallet_id=env_2.wallet_aliases["cat"],
     )
-    content: str = result["content"]
+    content: str = detok_result["content"]
 
-    result = await ClimateWallet.parse_detokenization_request(
+    detok_result = await ClimateWallet.parse_detokenization_request(
         content=content,
     )
-    assert result["mode"] == GatewayMode.DETOKENIZATION
-    assert result["amount"] == amount
-    assert result["fee"] == fee
-    assert result["asset_id"] == climate_wallet_1.tail_program_hash
+    assert detok_result["mode"] == GatewayMode.DETOKENIZATION
+    assert detok_result["amount"] == amount
+    assert detok_result["fee"] == fee
+    assert detok_result["asset_id"] == climate_wallet_1.tail_program_hash
 
-    result = await climate_wallet_1.sign_and_send_detokenization_request(
+    detok_result = await climate_wallet_1.sign_and_send_detokenization_request(
         content=content,
     )
 
@@ -327,6 +330,7 @@ async def test_cat_permissionless_retirement_workflow(
     wallet_2: Wallet = env_2.xch_wallet
 
     fingerprint = (await wallet_client_1.get_logged_in_fingerprint()).fingerprint
+    assert fingerprint is not None
     result = await wallet_client_1.get_private_key(GetPrivateKey(fingerprint=fingerprint))
     root_secret_key: PrivateKey = master_sk_to_root_sk(result.private_key.sk)
 
@@ -335,7 +339,7 @@ async def test_cat_permissionless_retirement_workflow(
         root_secret_key=root_secret_key,
         wallet_client=wallet_client_1,
     )
-    result = await climate_wallet_1.send_tokenization_transaction(
+    await climate_wallet_1.send_tokenization_transaction(
         to_puzzle_hash=await wallet_2.get_new_puzzlehash(),
         amount=amount,
         fee=fee,
@@ -389,7 +393,7 @@ async def test_cat_permissionless_retirement_workflow(
     )
 
     test_address = "This is a fake address".encode()
-    result = await climate_wallet_2.send_permissionless_retirement_transaction(
+    await climate_wallet_2.send_permissionless_retirement_transaction(
         amount=amount,
         fee=fee,
         beneficiary_name=beneficiary_name,

tests/test_disallow.py

@@ -11,11 +11,11 @@
 async def test_disallow() -> None:
     settings.MODE = ExecutionMode.DEV
 
-    @disallow([ExecutionMode.DEV])
+    @disallow([ExecutionMode.DEV])  # type: ignore[misc]
     async def disallow_dev() -> int:
         return 5
 
-    @disallow([ExecutionMode.REGISTRY])
+    @disallow([ExecutionMode.REGISTRY])  # type: ignore[misc]
     async def allow_dev() -> int:
         return 5