Code formatting

Author: cmoutafidis

.github/workflows/ci.yml

@@ -17,13 +17,20 @@ jobs:
       - name: Set up Python
         uses: actions/setup-python@v4
         with:
-          python-version: "3.10"
+          python-version: "3.9"
 
       - name: Install Poetry
         run: pip install poetry
 
       - name: Install Dependencies
         run: poetry install --with dev
 
+      - name: Set up Environment Variables
+        env:
+          PRIVATE_KEY: ${{ secrets.PRIVATE_KEY }}
+        run: echo "PRIVATE_KEY retrieved"
+
       - name: Run Tests
+        env:
+          PRIVATE_KEY: ${{ secrets.PRIVATE_KEY }}
         run: poetry run pytest --cov=jup_ag_sdk

check_code.sh

@@ -1,4 +1,4 @@
-black jup_ag_sdk tests
+black jup_ag_sdk tests --line-length 79
 isort jup_ag_sdk tests
 flake8 jup_ag_sdk tests
 mypy jup_ag_sdk tests

jup_ag_sdk/clients/jupiter_client.py

@@ -1,37 +1,47 @@
 import base64
 import os
+from typing import Dict, Optional
 
 import base58
 import httpx
 from solana.rpc.api import Client
-from solders.solders import Keypair, VersionedTransaction
+from solders.solders import Keypair, SendTransactionResp, VersionedTransaction
 
 
 class JupiterClient:
     """
-    Base client for interacting with Jupiter API. Also acts as a parent class for all sub-clients.
+    Base client for interacting with Jupiter API.
+    Also acts as a parent class for all sub-clients.
     """
 
-    def __init__(self, api_key, rpc_url, private_key_env_var, timeout):
+    def __init__(
+        self,
+        api_key: Optional[str],
+        rpc_url: Optional[str],
+        private_key_env_var: str,
+        timeout: int,
+    ):
         self.api_key = api_key
         self.rpc = Client(rpc_url) if rpc_url else None
-        self.base_url = "https://api.jup.ag" if api_key else "https://lite-api.jup.ag"
+        self.base_url = (
+            "https://api.jup.ag" if api_key else "https://lite-api.jup.ag"
+        )
         self.private_key_env_var = private_key_env_var
         self.timeout = timeout
         self.client = httpx.Client(timeout=self.timeout)
 
-    def close(self):
+    def close(self) -> None:
         self.client.close()
 
-    def _get_headers(self):
+    def _get_headers(self) -> Dict[str, str]:
         headers = {
             "Accept": "application/json",
         }
         if self.api_key:
             headers["x-api-key"] = self.api_key
         return headers
 
-    def _post_headers(self):
+    def _post_headers(self) -> Dict[str, str]:
         headers = {
             "Accept": "application/json",
             "Content-Type": "application/json",
@@ -40,31 +50,45 @@ def _post_headers(self):
             headers["x-api-key"] = self.api_key
         return headers
 
-    def _get_public_key(self):
+    def _get_public_key(self) -> str:
         wallet = Keypair.from_bytes(
-            base58.b58decode(os.getenv(self.private_key_env_var))
+            base58.b58decode(os.getenv(self.private_key_env_var, ""))
         )
         return str(wallet.pubkey())
 
-    def _send_transaction(self, transaction):
+    def _send_transaction(
+        self, transaction: VersionedTransaction
+    ) -> SendTransactionResp:
+        if not self.rpc:
+            raise ValueError("Client was initialized without RPC URL.")
         return self.rpc.send_transaction(transaction)
 
-    def _sign_base64_transaction(self, transaction_base64: str):
+    def _sign_base64_transaction(
+        self, transaction_base64: str
+    ) -> VersionedTransaction:
         transaction_bytes = base64.b64decode(transaction_base64)
-        versioned_transaction = VersionedTransaction.from_bytes(transaction_bytes)
+        versioned_transaction = VersionedTransaction.from_bytes(
+            transaction_bytes
+        )
         return self._sign_versioned_transaction(versioned_transaction)
 
-    def _sign_versioned_transaction(self, versioned_transaction: VersionedTransaction):
+    def _sign_versioned_transaction(
+        self, versioned_transaction: VersionedTransaction
+    ) -> VersionedTransaction:
         wallet = Keypair.from_bytes(
-            base58.b58decode(os.getenv(self.private_key_env_var))
+            base58.b58decode(os.getenv(self.private_key_env_var, ""))
         )
         account_keys = versioned_transaction.message.account_keys
         wallet_index = account_keys.index(wallet.pubkey())
 
         signers = list(versioned_transaction.signatures)
-        signers[wallet_index] = wallet
+        signers[wallet_index] = wallet  # type: ignore
 
-        return VersionedTransaction(versioned_transaction.message, signers)
+        return VersionedTransaction(
+            versioned_transaction.message, signers  # type: ignore
+        )
 
-    def _serialize_versioned_transaction(self, versioned_transaction: VersionedTransaction):
+    def _serialize_versioned_transaction(
+        self, versioned_transaction: VersionedTransaction
+    ) -> str:
         return base64.b64encode(bytes(versioned_transaction)).decode("utf-8")

jup_ag_sdk/clients/swap_api_client.py

@@ -1,15 +1,24 @@
+from typing import Any, Dict, Optional
+
+from solders.solders import SendTransactionResp
+
 from jup_ag_sdk.clients.jupiter_client import JupiterClient
 from jup_ag_sdk.models.swap_api.quote_request_model import QuoteRequest
 from jup_ag_sdk.models.swap_api.swap_request_model import SwapRequest
 
 
 class SwapApiClient(JupiterClient):
     """
-    A client for interacting with the Jupiter Swap API. Inherits from JupiterClient.
+    A client for interacting with the Jupiter Swap API.
+    Inherits from JupiterClient.
     """
 
     def __init__(
-        self, api_key=None, rpc_url="https://api.mainnet-beta.solana.com", private_key_env_var="PRIVATE_KEY", timeout=10
+        self,
+        api_key: Optional[str] = None,
+        rpc_url: str = "https://api.mainnet-beta.solana.com",
+        private_key_env_var: str = "PRIVATE_KEY",
+        timeout: int = 10,
     ):
         super().__init__(
             api_key=api_key,
@@ -18,7 +27,7 @@ def __init__(
             timeout=timeout,
         )
 
-    def quote(self, request: QuoteRequest) -> dict:
+    def quote(self, request: QuoteRequest) -> Dict[str, Any]:
         """
         Get a swap quote from the Jupiter API.
 
@@ -31,38 +40,44 @@ def quote(self, request: QuoteRequest) -> dict:
         params = request.to_dict()
 
         url = f"{self.base_url}/swap/v1/quote"
-        response = self.client.get(url, params=params, headers=self._get_headers())
+        response = self.client.get(
+            url, params=params, headers=self._get_headers()
+        )
         response.raise_for_status()
 
-        return response.json()
+        return response.json()  # type: ignore
 
-    def swap(self, request: SwapRequest) -> dict:
+    def swap(self, request: SwapRequest) -> Dict[str, Any]:
         """
         Submit a swap request to the Jupiter API.
 
         Args:
             request (SwapRequest): The swap request parameters.
 
         Returns:
-            dict: The parsed response containing transaction details and metadata.
+            dict: The parsed response containing transaction details
+            and metadata.
         """
         payload = request.to_dict()
 
         url = f"{self.base_url}/swap/v1/swap"
-        response = self.client.post(url, json=payload, headers=self._get_headers())
+        response = self.client.post(
+            url, json=payload, headers=self._get_headers()
+        )
         response.raise_for_status()
 
-        return response.json()
+        return response.json()  # type: ignore
 
-    def swap_and_execute(self, request: SwapRequest) -> dict:
+    def swap_and_execute(self, request: SwapRequest) -> SendTransactionResp:
         """
         Submit a swap request to the Jupiter API and execute it immediately.
 
         Args:
             request (SwapRequest): The swap request parameters.
 
         Returns:
-            dict: The raw RPC response containing the result of the transaction execution.
+            dict: The raw RPC response containing the result of the transaction
+            execution.
         """
         swap_response = self.swap(request)
         signed_transaction = self._sign_base64_transaction(

jup_ag_sdk/clients/ultra_api_client.py

@@ -1,15 +1,25 @@
+from typing import Any, Dict, Optional
+
 from jup_ag_sdk.clients.jupiter_client import JupiterClient
-from jup_ag_sdk.models.ultra_api.ultra_execute_request_model import UltraExecuteRequest
-from jup_ag_sdk.models.ultra_api.ultra_order_request_model import UltraOrderRequest
+from jup_ag_sdk.models.ultra_api.ultra_execute_request_model import (
+    UltraExecuteRequest,
+)
+from jup_ag_sdk.models.ultra_api.ultra_order_request_model import (
+    UltraOrderRequest,
+)
 
 
 class UltraApiClient(JupiterClient):
     """
-    A client for interacting with the Jupiter Swap API. Inherits from JupiterClient.
+    A client for interacting with the Jupiter Swap API.
+    Inherits from JupiterClient.
     """
 
     def __init__(
-        self, api_key=None, private_key_env_var="PRIVATE_KEY", timeout=10
+        self,
+        api_key: Optional[str] = None,
+        private_key_env_var: str = "PRIVATE_KEY",
+        timeout: int = 10,
     ):
         super().__init__(
             api_key=api_key,
@@ -18,7 +28,7 @@ def __init__(
             timeout=timeout,
         )
 
-    def order(self, request: UltraOrderRequest) -> dict:
+    def order(self, request: UltraOrderRequest) -> Dict[str, Any]:
         """
         Get an order from the Jupiter Ultra API.
 
@@ -31,12 +41,14 @@ def order(self, request: UltraOrderRequest) -> dict:
         params = request.to_dict()
 
         url = f"{self.base_url}/ultra/v1/order"
-        response = self.client.get(url, params=params, headers=self._get_headers())
+        response = self.client.get(
+            url, params=params, headers=self._get_headers()
+        )
         response.raise_for_status()
 
-        return response.json()
+        return response.json()  # type: ignore
 
-    def execute(self, request: UltraExecuteRequest) -> dict:
+    def execute(self, request: UltraExecuteRequest) -> Dict[str, Any]:
         """
         Execute the order with the Jupiter Ultra API.
 
@@ -49,12 +61,14 @@ def execute(self, request: UltraExecuteRequest) -> dict:
         payload = request.to_dict()
 
         url = f"{self.base_url}/ultra/v1/execute"
-        response = self.client.post(url, json=payload, headers=self._get_headers())
+        response = self.client.post(
+            url, json=payload, headers=self._get_headers()
+        )
         response.raise_for_status()
 
-        return response.json()
+        return response.json()  # type: ignore
 
-    def order_and_execute(self, request: UltraOrderRequest) -> dict:
+    def order_and_execute(self, request: UltraOrderRequest) -> Dict[str, Any]:
         """
         Get an order from the Jupiter Ultra API and execute it immediately.
 
@@ -73,7 +87,9 @@ def order_and_execute(self, request: UltraOrderRequest) -> dict:
 
         execute_request = UltraExecuteRequest(
             request_id=request_id,
-            signed_transaction=self._serialize_versioned_transaction(signed_transaction),
+            signed_transaction=self._serialize_versioned_transaction(
+                signed_transaction
+            ),
         )
 
         return self.execute(execute_request)

jup_ag_sdk/models/swap_api/quote_request_model.py

@@ -1,4 +1,4 @@
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
 
 from pydantic import BaseModel
 from pydantic.alias_generators import to_camel
@@ -21,10 +21,12 @@ class QuoteRequest(BaseModel):
     platform_fee_bps: Optional[int] = None
     dynamic_slippage: Optional[bool] = None
 
-    def to_dict(self):
+    def to_dict(self) -> Dict[str, Any]:
         params = self.model_dump(exclude_none=True)
 
-        camel_case_params = {to_camel(key): value for key, value in params.items()}
+        camel_case_params = {
+            to_camel(key): value for key, value in params.items()
+        }
 
         if "dexes" in camel_case_params:
             camel_case_params["dexes"] = ",".join(

jup_ag_sdk/models/swap_api/swap_request_model.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Any, Dict, Optional
 
 from pydantic import BaseModel
 from pydantic.alias_generators import to_camel
@@ -10,13 +10,15 @@ class PriorityLevelWithMaxLamports(BaseModel):
 
 
 class PrioritizationFeeLamports(BaseModel):
-    priority_level_with_max_lamports: Optional[PriorityLevelWithMaxLamports] = None
+    priority_level_with_max_lamports: Optional[
+        PriorityLevelWithMaxLamports
+    ] = None
     jito_tip_lamports: Optional[int] = None
 
 
 class SwapRequest(BaseModel):
     user_public_key: str
-    quote_response: dict
+    quote_response: Dict[str, Any]
 
     wrap_and_unwrap_sol: Optional[bool] = None
     use_shared_accounts: Optional[bool] = None
@@ -35,9 +37,11 @@ class Config:
         alias_generator = to_camel
         populate_by_name = True
 
-    def to_dict(self):
+    def to_dict(self) -> Dict[str, Any]:
         params = self.model_dump(exclude_none=True)
 
-        camel_case_params = {to_camel(key): value for key, value in params.items()}
+        camel_case_params = {
+            to_camel(key): value for key, value in params.items()
+        }
 
         return camel_case_params

jup_ag_sdk/models/ultra_api/ultra_execute_request_model.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Any, Dict
 
 from pydantic import BaseModel
 from pydantic.alias_generators import to_camel
@@ -8,9 +8,11 @@ class UltraExecuteRequest(BaseModel):
     signed_transaction: str
     request_id: str
 
-    def to_dict(self):
+    def to_dict(self) -> Dict[str, Any]:
         params = self.model_dump(exclude_none=True)
 
-        camel_case_params = {to_camel(key): value for key, value in params.items()}
+        camel_case_params = {
+            to_camel(key): value for key, value in params.items()
+        }
 
         return camel_case_params