Refactor: Replace status codes with HTTP enums

operate/cli.py

@@ -71,14 +71,15 @@
 ).encode()
 DEFAULT_MAX_RETRIES = 3
 USER_NOT_LOGGED_IN_ERROR = JSONResponse(
-    content={"error": "User not logged in!"}, status_code=401
+    content={"error": "User not logged in!"}, status_code=HTTPStatus.UNAUTHORIZED
 )
 
 
 def service_not_found_error(service_config_id: str) -> JSONResponse:
     """Service not found error response"""
     return JSONResponse(
-        content={"error": f"Service {service_config_id} not found"}, status_code=404
+        content={"error": f"Service {service_config_id} not found"},
+        status_code=HTTPStatus.NOT_FOUND,
     )
 
 
@@ -351,14 +352,19 @@ async def _call(request: Request) -> JSONResponse:
                     else:
                         error["error"] = str(e)
                     errors.append(error)
-                    return JSONResponse(content={"errors": errors}, status_code=500)
+                    return JSONResponse(
+                        content={"errors": errors},
+                        status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
+                    )
                 except Exception as e:  # pylint: disable=broad-except
                     errors.append(
                         {"error": str(e.args[0]), "traceback": traceback.format_exc()}
                     )
                     logger.error(f"Error {str(e.args[0])}\n{traceback.format_exc()}")
                 retries += 1
-            return JSONResponse(content={"errors": errors}, status_code=500)
+            return JSONResponse(
+                content={"errors": errors}, status_code=HTTPStatus.INTERNAL_SERVER_ERROR
+            )
 
         return _call
 
@@ -396,7 +402,7 @@ async def _setup_account(request: Request) -> t.Dict:
         if operate.user_account is not None:
             return JSONResponse(
                 content={"error": "Account already exists"},
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         data = await request.json()
@@ -414,7 +420,7 @@ async def _update_password(  # pylint: disable=too-many-return-statements
         if operate.user_account is None:
             return JSONResponse(
                 content={"error": "Account does not exist."},
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         data = await request.json()
@@ -427,15 +433,15 @@ async def _update_password(  # pylint: disable=too-many-return-statements
                 content={
                     "error": "You must provide exactly one of 'old_password' or 'mnemonic' (seed phrase).",
                 },
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         if old_password and mnemonic:
             return JSONResponse(
                 content={
                     "error": "You must provide exactly one of 'old_password' or 'mnemonic' (seed phrase), but not both.",
                 },
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         try:
@@ -457,11 +463,13 @@ async def _update_password(  # pylint: disable=too-many-return-statements
                 content={"error": None, "message": "Password not updated."}
             )
         except ValueError as e:
-            return JSONResponse(content={"error": str(e)}, status_code=400)
+            return JSONResponse(
+                content={"error": str(e)}, status_code=HTTPStatus.BAD_REQUEST
+            )
         except Exception as e:  # pylint: disable=broad-except
             return JSONResponse(
                 content={"error": str(e), "traceback": traceback.format_exc()},
-                status_code=400,
+                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
             )
 
     @app.post("/api/account/login")
@@ -471,20 +479,20 @@ async def _validate_password(request: Request) -> t.Dict:
         if operate.user_account is None:
             return JSONResponse(
                 content={"error": "Account does not exist"},
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         data = await request.json()
         if not operate.user_account.is_valid(password=data["password"]):
             return JSONResponse(
                 content={"error": "Password is not valid"},
-                status_code=401,
+                status_code=HTTPStatus.UNAUTHORIZED,
             )
 
         operate.password = data["password"]
         return JSONResponse(
             content={"message": "Login successful"},
-            status_code=200,
+            status_code=HTTPStatus.OK,
         )
 
     @app.get("/api/wallet")
@@ -503,13 +511,13 @@ async def _create_wallet(request: Request) -> t.List[t.Dict]:
         if operate.user_account is None:
             return JSONResponse(
                 content={"error": "Cannot create wallet; User account does not exist!"},
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         if operate.password is None:
             return JSONResponse(
                 content={"error": "You need to login before creating a wallet"},
-                status_code=401,
+                status_code=HTTPStatus.UNAUTHORIZED,
             )
 
         data = await request.json()
@@ -556,7 +564,7 @@ async def _get_safe(request: Request) -> t.List[t.Dict]:
         if not manager.exists(ledger_type=ledger_type):
             return JSONResponse(
                 content={"error": "Wallet does not exist"},
-                status_code=404,
+                status_code=HTTPStatus.NOT_FOUND,
             )
         safes = manager.load(ledger_type=ledger_type).safes
         if safes is None or safes.get(chain) is None:
@@ -576,13 +584,13 @@ async def _create_safe(  # pylint: disable=too-many-return-statements
         if operate.user_account is None:
             return JSONResponse(
                 content={"error": "Cannot create safe; User account does not exist!"},
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         if operate.password is None:
             return JSONResponse(
                 content={"error": "You need to login before creating a safe"},
-                status_code=401,
+                status_code=HTTPStatus.UNAUTHORIZED,
             )
 
         data = await request.json()
@@ -592,7 +600,7 @@ async def _create_safe(  # pylint: disable=too-many-return-statements
                 content={
                     "error": "You can only specify 'initial_funds' or 'transfer_excess_assets', but not both."
                 },
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         logger.info(f"POST /api/wallet/safe {data=}")
@@ -668,7 +676,7 @@ async def _create_safe(  # pylint: disable=too-many-return-statements
         except Exception as e:  # pylint: disable=broad-except
             logger.error(traceback.format_exc())
             return JSONResponse(
-                status_code=500,
+                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
                 content={"error": str(e), "traceback": traceback.format_exc()},
             )
 
@@ -680,21 +688,21 @@ async def _update_safe(request: Request) -> t.List[t.Dict]:
         if operate.user_account is None:
             return JSONResponse(
                 content={"error": "Cannot update safe; User account does not exist!"},
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         if operate.password is None:
             return JSONResponse(
                 content={"error": "You need to login before updating a safe."},
-                status_code=401,
+                status_code=HTTPStatus.UNAUTHORIZED,
             )
 
         data = await request.json()
 
         if "chain" not in data:
             return JSONResponse(
                 content={"error": "You need to specify a chain to update a safe."},
-                status_code=401,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         chain = Chain(data["chain"])
@@ -703,7 +711,7 @@ async def _update_safe(request: Request) -> t.List[t.Dict]:
         if not manager.exists(ledger_type=ledger_type):
             return JSONResponse(
                 content={"error": "Wallet does not exist"},
-                status_code=401,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         wallet = manager.load(ledger_type=ledger_type)
@@ -906,7 +914,7 @@ async def _withdraw_onchain(request: Request) -> JSONResponse:
         if withdrawal_address is None:
             return JSONResponse(
                 content={"error": "withdrawal_address is required"},
-                status_code=400,
+                status_code=HTTPStatus.BAD_REQUEST,
             )
 
         try:
@@ -949,7 +957,7 @@ async def _withdraw_onchain(request: Request) -> JSONResponse:
         except Exception as e:  # pylint: disable=broad-except
             logger.error(traceback.format_exc())
             return JSONResponse(
-                status_code=500,
+                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
                 content={"error": str(e), "traceback": traceback.format_exc()},
             )
 

operate/data/contracts/foreign_omnibridge/contract.py

@@ -30,7 +30,7 @@
 
 PLACEHOLDER_NATIVE_TOKEN_ADDRESS = "0xEeeeeEeeeEeEeeEeEeEeeEEEeeeeEeeeeeeeEEeE"  # nosec
 
-#DEFAULT_GAS_BRIDGE_ETH_TO = 800_000
+# DEFAULT_GAS_BRIDGE_ETH_TO = 800_000
 DEFAULT_GAS_RELAY_TOKENS = 800_000
 
 # By simulations, nonzero-ERC20-bridge gas ~ 1.05 zero-ERC20-bridge gas

operate/data/contracts/foreign_omnibridge/contract.yaml

@@ -8,7 +8,7 @@ aea_version: '>=1.0.0, <2.0.0'
 fingerprint:
   __init__.py: bafybeibsmumov3s36vfo24xp2niilcp3ywju2d4yqfadllyjncqtgtndly
   build/ForeignOmnibridge.json: bafybeibmcflt7w5p5szgii7glbtrvjahweclowz2k7e6qjq7yfbvszy6em
-  contract.py: bafybeidjjdhjr2rj3q2q27fhm366kt4qpsvjqg3ytpa2titic5sari6r24
+  contract.py: bafybeihywi67spsgsmcbqgjrcqlwgiq2e36ymssznqubfxwedforxek4ri
 fingerprint_ignore_patterns: []
 contracts: []
 class_name: ForeignOmnibridge

operate/operate_http/__init__.py

@@ -23,6 +23,7 @@
 import traceback
 import typing as t
 from abc import ABC
+from http import HTTPStatus
 
 from starlette.requests import Request
 from starlette.responses import JSONResponse
@@ -142,7 +143,7 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> t.Any:
             tb = traceback.format_exc()
             response = JSONResponse(
                 content={"error": str(e), "traceback": tb},
-                status_code=500,
+                status_code=HTTPStatus.INTERNAL_SERVER_ERROR,
             )
             print(tb)
         await response(scope=scope, receive=receive, send=send)

operate/operate_http/exceptions.py

@@ -19,6 +19,8 @@
 
 """Exceptions."""
 
+from http import HTTPStatus
+
 
 class ResourceException(Exception):
     """Base resource exceptio."""
@@ -29,22 +31,22 @@ class ResourceException(Exception):
 class BadRequest(ResourceException):
     """Bad request error."""
 
-    code = 400
+    code = HTTPStatus.BAD_REQUEST
 
 
 class ResourceAlreadyExists(ResourceException):
     """Bad request error."""
 
-    code = 409
+    code = HTTPStatus.CONFLICT
 
 
 class NotFound(ResourceException):
     """Not found error."""
 
-    code = 404
+    code = HTTPStatus.NOT_FOUND
 
 
 class NotAllowed(ResourceException):
     """Not allowed error."""
 
-    code = 405
+    code = HTTPStatus.METHOD_NOT_ALLOWED

operate/quickstart/run_service.py

@@ -25,6 +25,7 @@
 import time
 import typing as t
 import warnings
+from http import HTTPStatus
 
 import requests
 from aea.crypto.registries import make_ledger_api
@@ -247,7 +248,7 @@ def configure_local_config(
                     metadata_hash = instance.functions.metadataHash().call().hex()
                     ipfs_address = IPFS_ADDRESS.format(hash=metadata_hash)
                     response = requests.get(ipfs_address)
-                    if response.status_code != 200:
+                    if response.status_code != HTTPStatus.OK:
                         raise requests.RequestException(
                             f"Failed to fetch data from {ipfs_address}: {response.status_code}"
                         )

operate/services/health_checker.py

@@ -22,6 +22,7 @@
 import json
 import typing as t
 from concurrent.futures import ThreadPoolExecutor
+from http import HTTPStatus
 from pathlib import Path
 from traceback import print_exc
 
@@ -32,7 +33,7 @@
 from operate.services.manage import ServiceManager  # type: ignore
 
 
-HTTP_OK = 200
+HTTP_OK = HTTPStatus.OK
 
 
 class HealthChecker:

operate/services/manage.py

@@ -30,6 +30,7 @@
 from collections import Counter, defaultdict
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import suppress
+from http import HTTPStatus
 from pathlib import Path
 
 import requests
@@ -295,7 +296,7 @@ def _get_on_chain_metadata(self, chain_config: ChainConfig) -> t.Dict:
         url = f"{IPFS_GATEWAY}f01701220{config_hash}"
         self.logger.info(f"Fetching {url=}...")
         res = requests.get(url, timeout=30)
-        if res.status_code == 200:
+        if res.status_code == HTTPStatus.OK:
             return res.json()
         raise ValueError(
             f"Something went wrong while trying to get the on-chain metadata from IPFS: {res}"