testnet / mainnet compatibility

constants.py

@@ -0,0 +1,7 @@
+IS_TESTNET = os.getenv('TESTNET', 'false').lower() == 'true'
+REGEX_KASPA_ADDRESS = "^kaspa\:[a-z0-9]{61,63}$" if not IS_TESTNET else "^kaspatest\:[a-z0-9]{61,63}$"
+
+# address constants
+ADDRESS_PREFIX = "kaspatest" if IS_TESTNET else "kaspa"
+ADDRESS_EXAMPLE = "kaspatest:qpqz2vxj23kvh0m73ta2jjn2u4cv4tlufqns2eap8mxyyt0rvrxy6ejkful67" if IS_TESTNET \
+    else "kaspa:qqkqkzjvr7zwxxmjxjkmxxdwju9kjs6e9u82uh59z07vgaks6gg62v8707g73"

endpoints/__init__.py

@@ -1,8 +1,9 @@
 # encoding: utf-8
 import os
+from fastapi import HTTPException
 from functools import wraps
 
-from fastapi import HTTPException
+from constants import IS_TESTNET
 
 
 def filter_fields(response_dict, fields):
@@ -23,3 +24,15 @@ async def wrapper(*args, **kwargs):
         return await func(*args, **kwargs)
 
     return wrapper
+
+
+def mainnet_only(func):
+    @wraps(func)
+    async def wrapper(*args, **kwargs):
+        if not IS_TESTNET:
+            raise HTTPException(status_code=503, detail="Endpoint not available. "
+                                                        "This endpoint is only available in mainnet.")
+        return await func(*args, **kwargs)
+
+    return wrapper
+

endpoints/get_address_transactions.py

@@ -1,12 +1,13 @@
 # encoding: utf-8
+import os
 from enum import Enum
-from typing import List
-
 from fastapi import Path, Query
 from pydantic import BaseModel
 from sqlalchemy import text, func
 from sqlalchemy.future import select
+from typing import List
 
+from constants import ADDRESS_EXAMPLE, REGEX_KASPA_ADDRESS
 from dbsession import async_session
 from endpoints import sql_db_only
 from endpoints.get_transactions import search_for_transactions, TxSearch, TxModel
@@ -46,9 +47,8 @@ class PreviousOutpointLookupMode(str, Enum):
 @sql_db_only
 async def get_transactions_for_address(
         kaspaAddress: str = Path(
-            description="Kaspa address as string e.g. "
-                        "kaspa:pzhh76qc82wzduvsrd9xh4zde9qhp0xc8rl7qu2mvl2e42uvdqt75zrcgpm00",
-            regex="^kaspa\:[a-z0-9]{61,63}$")):
+            description=f"Kaspa address as string e.g. {ADDRESS_EXAMPLE}",
+            regex=REGEX_KASPA_ADDRESS)):
     """
     Get all transactions for a given address from database
     """
@@ -92,8 +92,8 @@ async def get_transactions_for_address(
 async def get_full_transactions_for_address(
         kaspaAddress: str = Path(
             description="Kaspa address as string e.g. "
-                        "kaspa:pzhh76qc82wzduvsrd9xh4zde9qhp0xc8rl7qu2mvl2e42uvdqt75zrcgpm00",
-            regex="^kaspa\:[a-z0-9]{61,63}$"),
+                        f"{ADDRESS_EXAMPLE}",
+            regex=REGEX_KASPA_ADDRESS),
         limit: int = Query(
             description="The number of records to get",
             ge=1,
@@ -136,8 +136,8 @@ async def get_full_transactions_for_address(
 async def get_transaction_count_for_address(
         kaspaAddress: str = Path(
             description="Kaspa address as string e.g. "
-                        "kaspa:pzhh76qc82wzduvsrd9xh4zde9qhp0xc8rl7qu2mvl2e42uvdqt75zrcgpm00",
-            regex="^kaspa\:[a-z0-9]{61,63}$")
+                        f"{ADDRESS_EXAMPLE}",
+            regex=REGEX_KASPA_ADDRESS)
 ):
     """
     Count the number of transactions associated with this address

endpoints/get_balance.py

@@ -1,21 +1,22 @@
 # encoding: utf-8
-
+import os
 from fastapi import Path, HTTPException
 from pydantic import BaseModel
 
+from constants import ADDRESS_EXAMPLE, REGEX_KASPA_ADDRESS, IS_TESTNET
 from server import app, kaspad_client
 
 
 class BalanceResponse(BaseModel):
-    address: str = "kaspa:pzhh76qc82wzduvsrd9xh4zde9qhp0xc8rl7qu2mvl2e42uvdqt75zrcgpm00"
+    address: str = ADDRESS_EXAMPLE
     balance: int = 38240000000
 
 
 @app.get("/addresses/{kaspaAddress}/balance", response_model=BalanceResponse, tags=["Kaspa addresses"])
 async def get_balance_from_kaspa_address(
         kaspaAddress: str = Path(
-            description="Kaspa address as string e.g. kaspa:pzhh76qc82wzduvsrd9xh4zde9qhp0xc8rl7qu2mvl2e42uvdqt75zrcgpm00",
-            regex="^kaspa\:[a-z0-9]{61,63}$")):
+            description=f"Kaspa address as string e.g. {ADDRESS_EXAMPLE}",
+            regex=REGEX_KASPA_ADDRESS)):
     """
     Get balance for a given kaspa address
     """

endpoints/get_price.py

@@ -12,6 +12,7 @@ class PriceResponse(BaseModel):
 
 
 @app.get("/info/price", response_model=PriceResponse | str, tags=["Kaspa network info"])
+@mainnet_only
 async def get_price(stringOnly: bool = False):
     """
     Returns the current price for Kaspa in USD.
@@ -25,6 +26,7 @@ async def get_price(stringOnly: bool = False):
 @app.get("/info/market-data",
          tags=["Kaspa network info"],
          include_in_schema=False)
+@mainnet_only
 async def get_market_data():
     """
     Returns market data for kaspa.

endpoints/get_utxos.py

@@ -1,10 +1,10 @@
 # encoding: utf-8
-
-from typing import List
-
+import os
 from fastapi import Path, HTTPException
 from pydantic import BaseModel
+from typing import List
 
+from constants import REGEX_KASPA_ADDRESS, ADDRESS_EXAMPLE
 from server import app, kaspad_client
 
 
@@ -24,15 +24,15 @@ class UtxoModel(BaseModel):
 
 
 class UtxoResponse(BaseModel):
-    address: str = "kaspa:qrzk988gtanp3nf76xkpexwud5cxfmfygqf42hz38pwea74s6qrj75jee85nj"
+    address: str = ADDRESS_EXAMPLE
     outpoint: OutpointModel
     utxoEntry: UtxoModel
 
 
 @app.get("/addresses/{kaspaAddress}/utxos", response_model=List[UtxoResponse], tags=["Kaspa addresses"])
 async def get_utxos_for_address(kaspaAddress: str = Path(
-    description="Kaspa address as string e.g. kaspa:qqkqkzjvr7zwxxmjxjkmxxdwju9kjs6e9u82uh59z07vgaks6gg62v8707g73",
-    regex="^kaspa\:[a-z0-9]{61,63}$")):
+    description=f"Kaspa address as string e.g. {ADDRESS_EXAMPLE}",
+    regex=REGEX_KASPA_ADDRESS)):
     """
     Lists all open utxo for a given kaspa address
     """