Added implementation for Ultra and Swap API

Author: cmoutafidis

jup_ag_sdk/client.py

@@ -0,0 +1,70 @@
+import base64
+import os
+
+import base58
+import httpx
+from solana.rpc.api import Client
+from solders.solders import Keypair, VersionedTransaction
+
+
+class JupiterClient:
+    """
+    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):
+        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.private_key_env_var = private_key_env_var
+        self.timeout = timeout
+        self.client = httpx.Client(timeout=self.timeout)
+
+    def close(self):
+        self.client.close()
+
+    def _get_headers(self):
+        headers = {
+            "Accept": "application/json",
+        }
+        if self.api_key:
+            headers["x-api-key"] = self.api_key
+        return headers
+
+    def _post_headers(self):
+        headers = {
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        }
+        if self.api_key:
+            headers["x-api-key"] = self.api_key
+        return headers
+
+    def _get_public_key(self):
+        wallet = Keypair.from_bytes(
+            base58.b58decode(os.getenv(self.private_key_env_var))
+        )
+        return str(wallet.pubkey())
+
+    def _send_transaction(self, transaction):
+        return self.rpc.send_transaction(transaction)
+
+    def _sign_base64_transaction(self, transaction_base64: str):
+        transaction_bytes = base64.b64decode(transaction_base64)
+        versioned_transaction = VersionedTransaction.from_bytes(transaction_bytes)
+        return self._sign_versioned_transaction(versioned_transaction)
+
+    def _sign_versioned_transaction(self, versioned_transaction: VersionedTransaction):
+        wallet = Keypair.from_bytes(
+            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
+
+        return VersionedTransaction(versioned_transaction.message, signers)
+
+    def _serialize_versioned_transaction(self, versioned_transaction: VersionedTransaction):
+        return base64.b64encode(bytes(versioned_transaction)).decode("utf-8")

jup_ag_sdk/clients/__init__.py

@@ -0,0 +1,71 @@
+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.
+    """
+
+    def __init__(
+        self, api_key=None, rpc_url="https://api.mainnet-beta.solana.com", private_key_env_var="PRIVATE_KEY", timeout=10
+    ):
+        super().__init__(
+            api_key=api_key,
+            rpc_url=rpc_url,
+            private_key_env_var=private_key_env_var,
+            timeout=timeout,
+        )
+
+    def quote(self, request: QuoteRequest) -> dict:
+        """
+        Get a swap quote from the Jupiter API.
+
+        Args:
+            request (QuoteRequest): The request parameters for the quote.
+
+        Returns:
+            dict: The dict api response.
+        """
+        params = request.to_dict()
+
+        url = f"{self.base_url}/swap/v1/quote"
+        response = self.client.get(url, params=params, headers=self._get_headers())
+        response.raise_for_status()
+
+        return response.json()
+
+    def swap(self, request: SwapRequest) -> dict:
+        """
+        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.
+        """
+        payload = request.to_dict()
+
+        url = f"{self.base_url}/swap/v1/swap"
+        response = self.client.post(url, json=payload, headers=self._get_headers())
+        response.raise_for_status()
+
+        return response.json()
+
+    def swap_and_execute(self, request: SwapRequest) -> dict:
+        """
+        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.
+        """
+        swap_response = self.swap(request)
+        signed_transaction = self._sign_base64_transaction(
+            swap_response["swapTransaction"]
+        )
+        return self._send_transaction(signed_transaction)

jup_ag_sdk/clients/jupiter_client.py

@@ -0,0 +1,79 @@
+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
+
+
+class UltraApiClient(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
+    ):
+        super().__init__(
+            api_key=api_key,
+            rpc_url=None,
+            private_key_env_var=private_key_env_var,
+            timeout=timeout,
+        )
+
+    def order(self, request: UltraOrderRequest) -> dict:
+        """
+        Get an order from the Jupiter Ultra API.
+
+        Args:
+            request (UltraOrderRequest): The request parameters for the order.
+
+        Returns:
+            dict: The dict api response.
+        """
+        params = request.to_dict()
+
+        url = f"{self.base_url}/ultra/v1/order"
+        response = self.client.get(url, params=params, headers=self._get_headers())
+        response.raise_for_status()
+
+        return response.json()
+
+    def execute(self, request: UltraExecuteRequest) -> dict:
+        """
+        Execute the order with the Jupiter Ultra API.
+
+        Args:
+            request (UltraExecuteRequest): The execute request parameters.
+
+        Returns:
+            dict: The dict api response.
+        """
+        payload = request.to_dict()
+
+        url = f"{self.base_url}/ultra/v1/execute"
+        response = self.client.post(url, json=payload, headers=self._get_headers())
+        response.raise_for_status()
+
+        return response.json()
+
+    def order_and_execute(self, request: UltraOrderRequest) -> dict:
+        """
+        Get an order from the Jupiter Ultra API and execute it immediately.
+
+        Args:
+            request (UltraOrderRequest): The request parameters for the order.
+
+        Returns:
+            dict: The dict api response.
+        """
+        order_response = self.order(request)
+
+        request_id = order_response["requestId"]
+        signed_transaction = self._sign_base64_transaction(
+            order_response["transaction"]
+        )
+
+        execute_request = UltraExecuteRequest(
+            request_id=request_id,
+            signed_transaction=self._serialize_versioned_transaction(signed_transaction),
+        )
+
+        return self.execute(execute_request)

jup_ag_sdk/clients/swap_api_client.py

@@ -0,0 +1,57 @@
+from enum import Enum
+from urllib.parse import quote
+
+
+class DexEnum(str, Enum):
+    WOOFI = "Woofi"
+    PUMP_FUN = "Pump.fun"
+    WHIRLPOOL = "Whirlpool"
+    VIRTUALS = "Virtuals"
+    DAOS_FUN = "Daos.fun"
+    LIFINITY_V2 = "Lifinity V2"
+    STABBLE_STABLE_SWAP = "Stabble Stable Swap"
+    TOKEN_MILL = "Token Mill"
+    METEORA = "Meteora"
+    OASIS = "Oasis"
+    ALDRIN = "Aldrin"
+    GOOSEFX_GAMMA = "GooseFX GAMMA"
+    PERPS = "Perps"
+    SOLFI = "SolFi"
+    DEXLAB = "DexLab"
+    TOKEN_SWAP = "Token Swap"
+    ZEROFI = "ZeroFi"
+    CROPPER = "Cropper"
+    OBRIC_V2 = "Obric V2"
+    STABBLE_WEIGHTED_SWAP = "Stabble Weighted Swap"
+    SANCTUM_INFINITY = "Sanctum Infinity"
+    MOONIT = "Moonit"
+    SANCTUM = "Sanctum"
+    RAYDIUM_CP = "Raydium CP"
+    PHOENIX = "Phoenix"
+    PUMP_FUN_AMM = "Pump.fun Amm"
+    SABER = "Saber"
+    SABER_DECIMALS = "Saber (Decimals)"
+    RAYDIUM_CLMM = "Raydium CLMM"
+    DEX_1 = "1DEX"
+    PENGUIN = "Penguin"
+    ORCA_V2 = "Orca V2"
+    FLUXBEAM = "FluxBeam"
+    RAYDIUM = "Raydium"
+    METEORA_DLMM = "Meteora DLMM"
+    BONKSWAP = "Bonkswap"
+    SOLAYER = "Solayer"
+    STEPN = "StepN"
+    HELIUM_NETWORK = "Helium Network"
+    MERCURIAL = "Mercurial"
+    PERENA = "Perena"
+    ORCA_V1 = "Orca V1"
+    ALDRIN_V2 = "Aldrin V2"
+    SAROS = "Saros"
+    OPENBOOK_V2 = "OpenBook V2"
+    CREMA = "Crema"
+    OPENBOOK = "Openbook"
+    INVARIANT = "Invariant"
+    GUACSWAP = "Guacswap"
+
+    def __str__(self) -> str:
+        return quote(self.value)